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, how its backups stand, and whether the system has a problem."""
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
LOW_DISK_PCT = 10

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."


def _issue_title(issue: dict, titles: dict) -> str:
    key = issue.get("translation_key") or issue.get("issue_id") or ""
    title = titles.get(f"component.{issue.get('domain')}.issues.{key}.title")
    if not title:
        return key.removeprefix("issue_").replace("_", " ").capitalize() or "A repair is open"
    for name, value in (issue.get("translation_placeholders") or {}).items():
        title = title.replace("{" + name + "}", str(value))
    return title


def read_health(found: dict) -> dict:
    """What a poll keeps of client.read_system: the problems to show, and the disk.
    A part Home Assistant wouldn't answer is left out rather than read as fine:
    `repairs_read` says whether the repairs were seen, and `supervised` whether a
    Supervisor answered, as it doesn't on every install or for every user."""
    problems = []
    if (found.get("config") or {}).get("recovery_mode"):
        problems.append({"kind": "recovery", "severity": "critical",
                         "title": "Home Assistant started in recovery mode: its configuration didn't load."})
    issues = found.get("issues")
    for issue in issues or []:
        if not issue.get("ignored") and not issue.get("dismissed_version"):
            problems.append({"kind": "repair", "severity": issue.get("severity") or "warning",
                             "title": _issue_title(issue, found.get("titles") or {})})
    resolution = found.get("resolution") or {}
    for word in ("unhealthy", "unsupported"):
        for reason in resolution.get(word) or []:
            problems.append({"kind": "supervisor", "severity": "error" if word == "unhealthy" else "warning",
                             "title": f"The Supervisor calls this system {word}: {str(reason).replace('_', ' ')}."})
    host = found.get("host") or {}
    free, total = host.get("disk_free"), host.get("disk_total")
    free_pct = round(100 * free / total, 1) if free is not None and total else None
    if free_pct is not None and free_pct < LOW_DISK_PCT:
        problems.append({"kind": "disk", "severity": "error", "title": f"Only {free:g} GB of {total:g} GB free."})
    boots = found.get("boots") or {}
    for addon in found.get("addons") or []:
        if boots.get(addon.get("slug")) == "auto":
            problems.append({"kind": "addon", "severity": "error",
                             "title": f"{addon.get('name') or addon.get('slug')} should start on its own and isn't running."})
    return {"repairs_read": issues is not None, "supervised": found.get("resolution") is not None, "problems": problems,
            "disk_free_gb": free, "disk_total_gb": total, "disk_free_pct": free_pct}


def problems_open(health: dict | None) -> list[dict]:
    return (health or {}).get("problems") or []