Compressor/models/common/stats.py

"""Robust statistics and trend helpers shared by both algorithms."""
from __future__ import annotations

import math
from statistics import median

from .constants import MIN_HISTORY_HOURS


def number(value):
    if value is None:
        return None
    try:
        value = float(value)
    except (TypeError, ValueError):
        return None
    return value if math.isfinite(value) else None


def finite(value):
    return number(value)


def median_value(values):
    values = [value for value in values if value is not None]
    return median(values) if values else None


def max_consecutive(values):
    longest = 0
    current = 0
    for value in values:
        if value:
            current += 1
            longest = max(longest, current)
        else:
            current = 0
    return longest


def robust_baseline(values, minimum: int = MIN_HISTORY_HOURS):
    values = [value for value in values if value is not None]
    if len(values) < minimum:
        return None, None
    center = median(values)
    mad = median(abs(value - center) for value in values)
    # Keep a meaningful relative scale when a sensor is very stable.
    scale = max(1.4826 * mad, abs(center) * 0.05, 1e-6)
    return center, scale


def linear_slope(values):
    pairs = [(index, value) for index, value in enumerate(values) if value is not None]
    if len(pairs) < 3:
        return None
    xbar = sum(x for x, _ in pairs) / len(pairs)
    ybar = sum(y for _, y in pairs) / len(pairs)
    denominator = sum((x - xbar) ** 2 for x, _ in pairs)
    if denominator == 0:
        return None
    return sum((x - xbar) * (y - ybar) for x, y in pairs) / denominator


def consecutive_decline_hours(history: list[dict], current: dict) -> int:
    """Count contiguous valid hours that have not risen from the prior hour."""
    from datetime import timedelta

    count = 1
    newer = current
    for older in reversed(history):
        if newer["_dt"] - older["_dt"] != timedelta(hours=1):
            break
        if newer["oil_avg"] is None or older["oil_avg"] is None or newer["oil_avg"] > older["oil_avg"]:
            break
        count += 1
        newer = older
    return count


def consecutive_flag_hours(history: list[dict], current: dict, field: str) -> int:
    from datetime import timedelta

    count = 1 if current.get(field) else 0
    if not count:
        return 0
    newer = current
    for older in reversed(history):
        if newer["_dt"] - older["_dt"] != timedelta(hours=1) or not older.get(field):
            break
        count += 1
        newer = older
    return count


__all__ = [
    "number", "finite", "median_value", "max_consecutive", "robust_baseline",
    "linear_slope", "consecutive_decline_hours", "consecutive_flag_hours",
]