FabledSteward/plugins/homeassistant/service.py

# plugins/homeassistant/service.py
"""Reading what Home Assistant reports: which updates it knows of, which of them
Steward may install, and how its backups stand."""
from __future__ import annotations

from datetime import datetime, timezone

SOURCE = "homeassistant"
RESOURCE = "homeassistant"
# Backups run daily, so past a day and a bit one has been missed.
STALE_BACKUP_HOURS = 26

CORE = "update.home_assistant_core_update"
OS = "update.home_assistant_operating_system_update"
SUPERVISOR = "update.home_assistant_supervisor_update"
LAST_BACKUP = "sensor.backup_last_successful_automatic_backup"
NEXT_BACKUP = "sensor.backup_next_scheduled_automatic_backup"
BACKUP_MANAGER = "sensor.backup_backup_manager_state"
BACKUP_EVENT = "event.backup_automatic_backup"

_KIND_WORDS = {"core": "Core", "os": "Operating system", "addon": "Add-on"}
_KIND_ORDER = {"core": 0, "os": 1, "addon": 2, None: 3}

def classify(entity: dict) -> str | None:
    """"core", "os" or "addon" for an update Steward may install; None for the
    rest: Supervisor updates itself, and firmware and custom integrations are
    Home Assistant's own to offer."""
    entity_id = entity.get("entity_id")
    if entity_id == CORE:
        return "core"
    if entity_id == OS:
        return "os"
    if entity_id == SUPERVISOR:
        return None
    picture = (entity.get("attributes") or {}).get("entity_picture") or ""
    return "addon" if picture.startswith("/api/hassio/addons/") else None


def _update(entity: dict) -> dict:
    a = entity.get("attributes") or {}
    kind = classify(entity)
    url = a.get("release_url") or ""
    return {"entity_id": entity["entity_id"], "kind": kind, "kind_word": _KIND_WORDS.get(kind, "Other"),
            "title": a.get("title") or a.get("friendly_name") or entity["entity_id"],
            "installed": a.get("installed_version"), "latest": a.get("latest_version"),
            "waiting": entity.get("state") == "on", "in_progress": bool(a.get("in_progress")),
            "auto_update": bool(a.get("auto_update")),
            "release_url": url if url.startswith(("http://", "https://")) else None}


def _state(entity: dict | None) -> str | None:
    value = (entity or {}).get("state")
    return None if value in (None, "unknown", "unavailable") else value


def parse_moment(value: str | None) -> datetime | None:
    try:
        moment = datetime.fromisoformat(value) if value else None
    except ValueError:
        return None
    if moment is not None and moment.tzinfo is None:
        moment = moment.replace(tzinfo=timezone.utc)
    return moment


def is_newer(value: str | None, than: str | None) -> bool:
    """Whether reading `value` is a later moment than `than`, where `than` may be never."""
    moment, before = parse_moment(value), parse_moment(than)
    return moment is not None and (before is None or moment > before)


def read_snapshot(states: list[dict]) -> dict:
    """What a poll keeps of Home Assistant's entities."""
    by_id = {e.get("entity_id"): e for e in states}
    updates = [_update(e) for i, e in by_id.items() if (i or "").startswith("update.") and e.get("state") != "unavailable"]
    updates.sort(key=lambda u: (not u["waiting"], _KIND_ORDER[u["kind"]], u["title"].lower()))
    event = by_id.get(BACKUP_EVENT) or {}
    event_attrs = event.get("attributes") or {}
    return {
        "updates": updates,
        "backup": {"last_ok": _state(by_id.get(LAST_BACKUP)), "next": _state(by_id.get(NEXT_BACKUP)),
                   "manager": _state(by_id.get(BACKUP_MANAGER)), "last_result": event_attrs.get("event_type"),
                   "failed_reason": event_attrs.get("failed_reason"), "known": LAST_BACKUP in by_id},
    }


def backup_age_hours(backup: dict | None, now: datetime) -> float | None:
    last = parse_moment((backup or {}).get("last_ok"))
    return None if last is None else max(0.0, (now - last).total_seconds() / 3600)


def installable(connection) -> list[dict]:
    """The waiting updates Steward may install."""
    return [u for u in (connection.updates or []) if u["waiting"] and u["kind"]]


def backup_verdict(backup: dict | None, now: datetime) -> tuple[str, str]:
    """(tone, sentence) for how the backups stand."""
    backup = backup or {}
    if not backup.get("known"):
        return "warn", "Home Assistant reports no automatic backups. Turn them on under Settings → System → Backups."
    age = backup_age_hours(backup, now)
    if age is None:
        return "warn", "Home Assistant has no successful automatic backup yet."
    if backup.get("last_result") == "failed":
        reason = backup.get("failed_reason")
        return "warn", f"The newest backup attempt failed{f' ({reason})' if reason else ''}."
    if age > STALE_BACKUP_HOURS:
        return "warn", f"The last good backup is {age:.0f} hours old, past the {STALE_BACKUP_HOURS} hours expected."
    return "ok", "Backups are up to date."