"""The Home Assistant plugin's MCP tools and its get_overview findings. get_home_assistant reads what the poll stored. update_home_assistant starts an update as the page's Install does. The sign-in Steward holds is never returned. """ from __future__ import annotations from datetime import timezone from sqlalchemy import select from steward.core import weekly from steward.core.formatting import zone from steward.mcp._context import current_caller, mcp_app from steward.mcp.tools._audit import audit_mcp, mcp_username from steward.mcp.tools._shape import age_s, db_session, iso, now_utc, redact_url from steward.mcp.tools.overview import finding from steward.models.users import UserRole from . import schedule from .models import HomeAssistantConnection, HomeAssistantSchedule, HomeAssistantUpdate from .routes import connection_says from .service import STALE_BACKUP_HOURS, backup_age_hours, backup_verdict, installable, parse_moment, problems_open from .updates import CannotStart, start_update RECENT_UPDATES = 10 def _schedule_view(plan, now) -> dict | None: if plan is None: return None return {"enabled": plan.enabled, "when": weekly.describe(plan), "next_slot_at": iso(weekly.next_slot(plan, now)) if plan.enabled else None, "last_slot_at": iso(weekly.fired_slot(plan)), "stopped": plan.stopped} def _run_view(r: HomeAssistantUpdate) -> dict: return {"update_id": r.id, "entity_id": r.entity_id, "title": r.title, "kind": r.kind, "from_version": r.from_version, "to_version": r.to_version, "state": r.state, "message": r.message, "started_by": r.started_by, "started_at": iso(r.started_at), "backup_at": iso(r.backup_at), "finished_at": iso(r.finished_at)} async def get_home_assistant() -> dict: """Home Assistant as of the last poll: whether it answers, which updates are waiting, how old its last backup is, and the updates Steward was asked to install. `installable` is what update_home_assistant takes (Core, OS, add-ons); `others_waiting` is firmware and custom integrations, which Steward leaves to Home Assistant. `can_install` is whether the Home Assistant user who signed Steward in is an administrator, which installing needs. `system.problems` is what's wrong with the system itself: open repairs, Supervisor health, low disk, recovery mode, add-ons that should be running. `schedule` is the weekly slot for installing what's waiting, null until one is set. """ now = now_utc() async with db_session() as db: c = (await db.execute(select(HomeAssistantConnection))).scalars().first() runs = list((await db.execute(select(HomeAssistantUpdate).order_by( HomeAssistantUpdate.started_at.desc()).limit(RECENT_UPDATES))).scalars()) plan = (await db.execute(select(HomeAssistantSchedule))).scalars().first() if c is None: return {"connected": False, "says": "No connection yet: an admin signs in on the Home Assistant page."} reach, says = connection_says(c, now) backup = c.backup or {} age = backup_age_hours(backup, now) return {"connected": True, "url": redact_url(c.url), "reach": reach, "says": says, "polled_at": iso(c.last_polled_at), "ok_age_s": age_s(c.last_ok_at, now), "can_install": c.can_install, "installable": [{k: u[k] for k in ("entity_id", "title", "kind", "installed", "latest", "in_progress")} for u in installable(c)], "others_waiting": [u["title"] for u in c.updates or [] if u["waiting"] and not u["kind"]], "backup": {"verdict": backup_verdict(backup, now)[1], "last_ok": iso(parse_moment(backup.get("last_ok"))), "age_hours": round(age, 1) if age is not None else None, "stale_after_hours": STALE_BACKUP_HOURS, "next": iso(parse_moment(backup.get("next"))), "last_result": backup.get("last_result")}, "system": c.health or {}, "schedule": _schedule_view(plan, now), "updates": [_run_view(r) for r in runs]} async def update_home_assistant(entity_id: str) -> dict: """Install one Core, OS or add-on update, as the page's Install does: Steward takes a backup, confirms it landed, installs, and follows it to its end. entity_id: from get_home_assistant's `installable`. An update restarts Home Assistant or the add-on. An OS update reads as done once it's written, and takes effect after a reboot, which Steward leaves to a person. Refused while another is under way, or if nothing is waiting there. Admin only, as on the page. Recorded in the audit log. """ caller = current_caller() try: run = await start_update(mcp_app(), entity_id, mcp_username(caller.username)) except CannotStart as exc: raise ValueError(f"Nothing was started: {exc}.") from None await audit_mcp("homeassistant.update_started", "homeassistant_update", run.id, {"entity_id": entity_id, "title": run.title, "to": run.to_version}) return {**_run_view(run), "next": "get_home_assistant follows it: `updates` shows its state."} async def set_home_assistant_schedule(enabled: bool, weekday: int = 5, at_time: str = "05:00", time_zone: str = "UTC") -> dict: """Set the weekly slot for installing what's waiting, or turn it off, as the page does. weekday: 0 for Monday to 6 for Sunday. at_time: "HH:MM" in time_zone. At each slot Steward installs add-ons, then Core, then the OS, one at a time, each after a confirmed backup, stopping at the first failure. A Core update restarts Home Assistant and an OS update reboots the machine, which Steward does itself here, with nobody watching. The first slot is the next one from now. Admin only, and audited. """ try: weekly.parse_at(at_time) except ValueError: raise ValueError(f"at_time is HH:MM, not {at_time!r}.") from None if not 0 <= weekday <= 6: raise ValueError("weekday is 0 for Monday to 6 for Sunday.") if zone(time_zone) is timezone.utc and time_zone.upper() != "UTC": raise ValueError(f"Unknown time zone {time_zone!r}.") row = await schedule.save(mcp_app(), enabled=enabled, weekday=weekday, at_time=at_time, time_zone=time_zone) await audit_mcp("homeassistant.schedule_saved", "homeassistant_schedule", row.id, {"enabled": row.enabled, "when": weekly.describe(row)}) return _schedule_view(row, now_utc()) def overview_findings(c, latest_run, now, plan=None) -> list[dict]: """A connection that isn't answering, a problem with the system, stale backups, a newest update that failed, and a scheduled run that stopped. Updates waiting aren't a problem to raise.""" if c is None: return [] found = [] reach, says = connection_says(c, now) if reach == "down": found.append(finding("warning", "home_assistant_down", "Home Assistant", f"Home Assistant isn't answering: {says}", since=c.last_ok_at, next="get_home_assistant()")) problems = problems_open(c.health) if problems: more = f" (and {len(problems) - 1} more)" if len(problems) > 1 else "" found.append(finding("warning", "home_assistant_problem", "Home Assistant", f"{problems[0]['title']}{more}", next="get_home_assistant()")) tone, backup_says = backup_verdict(c.backup, now) if c.last_ok_at is not None and tone == "warn": found.append(finding("warning", "home_assistant_backup", "Home Assistant", backup_says, since=parse_moment((c.backup or {}).get("last_ok")), next="get_home_assistant()")) if latest_run is not None and latest_run.state == "failed": found.append(finding("warning", "home_assistant_update_failed", latest_run.title, f"The update of {latest_run.title} failed: {latest_run.message}", since=latest_run.finished_at, next="get_home_assistant()")) if plan is not None and plan.stopped: found.append(finding("warning", "home_assistant_schedule_stopped", "Home Assistant", f"The scheduled update run stopped: {plan.stopped}", since=weekly.fired_slot(plan), next="get_home_assistant()")) return found async def mcp_overview(db) -> list[dict]: c = (await db.execute(select(HomeAssistantConnection))).scalars().first() latest = (await db.execute(select(HomeAssistantUpdate).order_by( HomeAssistantUpdate.started_at.desc()).limit(1))).scalars().first() plan = (await db.execute(select(HomeAssistantSchedule))).scalars().first() return overview_findings(c, latest, now_utc(), plan) def register_tools(register) -> None: register(get_home_assistant, role=UserRole.viewer, read_only=True) register(update_home_assistant, role=UserRole.admin, read_only=False) register(set_home_assistant_schedule, role=UserRole.admin, read_only=False)