sillyhome-next-dev/app/ha/history.py

from __future__ import annotations

import math
from datetime import datetime

from pydantic import BaseModel

from app.ha.exceptions import HaUnexpectedPayloadError


class NumericHistoryPoint(BaseModel):
    timestamp: datetime
    value: float


class EntityHistorySeries(BaseModel):
    entity_id: str
    points: list[NumericHistoryPoint]


class StateHistoryPoint(BaseModel):
    timestamp: datetime
    state: str


class StateHistorySeries(BaseModel):
    entity_id: str
    points: list[StateHistoryPoint]


class LogbookEntry(BaseModel):
    entity_id: str
    timestamp: datetime
    message: str = ""
    context_user_id: str | None = None
    context_domain: str | None = None
    context_service: str | None = None


def normalize_history_payload(payload: object) -> list[EntityHistorySeries]:
    if not isinstance(payload, list):
        raise HaUnexpectedPayloadError("History-Payload muss eine Liste sein.")

    normalized: list[EntityHistorySeries] = []
    for raw_series in payload:
        if not isinstance(raw_series, list):
            raise HaUnexpectedPayloadError("History-Serie muss eine Liste sein.")
        series = _normalize_series(raw_series)
        if series is not None:
            normalized.append(series)

    return sorted(normalized, key=lambda item: item.entity_id)


def normalize_state_history_payload(payload: object) -> list[StateHistorySeries]:
    if not isinstance(payload, list):
        raise HaUnexpectedPayloadError("History-Payload muss eine Liste sein.")
    normalized: list[StateHistorySeries] = []
    for raw_series in payload:
        if not isinstance(raw_series, list):
            raise HaUnexpectedPayloadError("History-Serie muss eine Liste sein.")
        entity_id: str | None = None
        points: list[StateHistoryPoint] = []
        for raw_entry in raw_series:
            if not isinstance(raw_entry, dict):
                raise HaUnexpectedPayloadError("History-Eintrag muss ein Objekt sein.")
            raw_entity_id = raw_entry.get("entity_id")
            if raw_entity_id is not None:
                if not isinstance(raw_entity_id, str) or "." not in raw_entity_id:
                    raise HaUnexpectedPayloadError(
                        "History-Eintrag enthält ungültige entity_id."
                    )
                if entity_id is not None and entity_id != raw_entity_id:
                    raise HaUnexpectedPayloadError("History-Serie enthält mehrere Entities.")
                entity_id = raw_entity_id
            raw_state = raw_entry.get("state")
            if not isinstance(raw_state, str) or raw_state in {"unknown", "unavailable"}:
                continue
            if entity_id is None:
                raise HaUnexpectedPayloadError("History-Serie enthält keine entity_id.")
            timestamp = _parse_timestamp(
                raw_entry.get("last_changed") or raw_entry.get("last_updated")
            )
            if not points or points[-1].state != raw_state:
                points.append(StateHistoryPoint(timestamp=timestamp, state=raw_state))
        if entity_id is not None and points:
            points.sort(key=lambda point: point.timestamp)
            normalized.append(StateHistorySeries(entity_id=entity_id, points=points))
    return sorted(normalized, key=lambda item: item.entity_id)


def normalize_logbook_payload(payload: object, entity_id: str) -> list[LogbookEntry]:
    if not isinstance(payload, list):
        raise HaUnexpectedPayloadError("Logbook-Payload muss eine Liste sein.")
    entries: list[LogbookEntry] = []
    for raw_entry in payload:
        if not isinstance(raw_entry, dict):
            raise HaUnexpectedPayloadError("Logbook-Eintrag muss ein Objekt sein.")
        raw_entity_id = raw_entry.get("entity_id")
        if raw_entity_id != entity_id:
            continue
        entries.append(
            LogbookEntry(
                entity_id=entity_id,
                timestamp=_parse_timestamp(raw_entry.get("when")),
                message=str(raw_entry.get("message") or ""),
                context_user_id=_optional_string(raw_entry.get("context_user_id")),
                context_domain=_optional_string(
                    raw_entry.get("context_domain") or raw_entry.get("domain")
                ),
                context_service=_optional_string(raw_entry.get("context_service")),
            )
        )
    return sorted(entries, key=lambda item: item.timestamp)


def _normalize_series(raw_series: list[object]) -> EntityHistorySeries | None:
    entity_id: str | None = None
    points: list[NumericHistoryPoint] = []

    for raw_entry in raw_series:
        if not isinstance(raw_entry, dict):
            raise HaUnexpectedPayloadError("History-Eintrag muss ein Objekt sein.")

        raw_entity_id = raw_entry.get("entity_id")
        if raw_entity_id is not None:
            if not isinstance(raw_entity_id, str) or "." not in raw_entity_id:
                raise HaUnexpectedPayloadError("History-Eintrag enthält ungültige entity_id.")
            if entity_id is not None and entity_id != raw_entity_id:
                raise HaUnexpectedPayloadError("History-Serie enthält mehrere Entities.")
            entity_id = raw_entity_id

        raw_state = raw_entry.get("state")
        value = _finite_float(raw_state)
        if value is None:
            continue
        if entity_id is None:
            raise HaUnexpectedPayloadError("History-Serie enthält keine entity_id.")

        raw_timestamp = raw_entry.get("last_changed") or raw_entry.get("last_updated")
        timestamp = _parse_timestamp(raw_timestamp)
        points.append(NumericHistoryPoint(timestamp=timestamp, value=value))

    if entity_id is None or not points:
        return None

    points.sort(key=lambda point: point.timestamp)
    return EntityHistorySeries(entity_id=entity_id, points=points)


def _finite_float(value: object) -> float | None:
    if isinstance(value, bool) or value is None:
        return None
    if not isinstance(value, (str, int, float)):
        return None
    try:
        converted = float(value)
    except (TypeError, ValueError):
        return None
    return converted if math.isfinite(converted) else None


def _parse_timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise HaUnexpectedPayloadError("Numerischer History-Eintrag enthält keinen Zeitstempel.")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise HaUnexpectedPayloadError("History-Eintrag enthält ungültigen Zeitstempel.") from exc
    if parsed.tzinfo is None:
        raise HaUnexpectedPayloadError("History-Zeitstempel muss eine Zeitzone enthalten.")
    return parsed


def _optional_string(value: object) -> str | None:
    if value is None or value == "":
        return None
    return str(value)