app/scripts/tinyhouse/api.py

"""Talking to the installation, and the constants this house is wired with.

Addresses, MAC addresses and the DMX map live in ``house.json`` at the
workspace root rather than in this package: they are this installation's
private data, so they are git-ignored the way the Node-RED inventory is.
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any

import httpx

from . import dashboards

API = os.environ.get("API_URL", "http://api.localhost")
EMAIL = os.environ.get("FIRST_SUPERUSER", "")
PASSWORD = os.environ.get("FIRST_SUPERUSER_PASSWORD", "")

HOUSE_FILE = Path(
    os.environ.get("HOUSE_FILE", Path(__file__).resolve().parents[3] / "house.json")
)


def house() -> dict[str, Any]:
    """The wiring of this particular house."""
    if not HOUSE_FILE.exists():
        raise SystemExit(
            f"No {HOUSE_FILE}. It holds this installation's addresses and DMX "
            "map; copy the one from the workspace root or write it from "
            "docs/private/node-red-transition.md."
        )
    data: dict[str, Any] = json.loads(HOUSE_FILE.read_text())
    return data


class Api:
    """The REST API, logged in, raising on anything that is not a 2xx."""

    def __init__(self) -> None:
        if not EMAIL or not PASSWORD:
            raise SystemExit("FIRST_SUPERUSER / FIRST_SUPERUSER_PASSWORD are unset.")
        self.http = httpx.Client(base_url=f"{API}/api/v1", timeout=60)
        answer = self.http.post(
            "/login/access-token",
            data={"username": EMAIL, "password": PASSWORD},
        )
        if answer.status_code != 200:
            raise SystemExit(f"Could not log in to {API}: {answer.text}")
        self.http.headers["Authorization"] = f"Bearer {answer.json()['access_token']}"

    def __call__(self, method: str, path: str, body: Any = None) -> Any:
        response = self.http.request(method, path, json=body)
        response.raise_for_status()
        return response.json() if response.content else None

    def drop(self, path: str) -> None:
        """Delete something that may not be there."""
        try:
            self("DELETE", path)
        except httpx.HTTPStatusError:
            pass

    # ── flows ────────────────────────────────────────────────────────────

    def put_flow(self, flow: Flow) -> list[str]:
        """Create the flow, write its node code, publish it, leave it stopped.

        Stopped is the only safe default here: publishing a flow that drives
        the house would command every fixture it owns from whatever its inputs
        happen to hold. Starting one is a decision someone makes while
        watching.
        """
        self.drop(f"/flows/{flow.name}")
        self(
            "PUT",
            f"/flows/{flow.name}",
            {
                "name": flow.name,
                "title": flow.title,
                "nodes": flow.nodes,
                "inputs": flow.inputs,
            },
        )
        for node_id, code in flow.sources.items():
            self("PUT", f"/flows/{flow.name}/nodes/{node_id}/source", {"code": code})
        version = self("GET", f"/flows/{flow.name}")["definition"]["version"]
        self("POST", f"/flows/{flow.name}/publish", {"version": version})
        self("POST", f"/flows/{flow.name}/stop")
        issues = self("POST", f"/flows/{flow.name}/validate")["issues"]
        return [_issue_line(i) for i in issues]

    def share(self, flow: str, node_id: str, lib_name: str) -> None:
        """Move a node's code into the library, if it is not there already."""
        library = {node["name"] for node in self("GET", "/flows/library")}
        if lib_name in library:
            return
        self("POST", f"/flows/{flow}/nodes/{node_id}/share", {"lib_name": lib_name})

    # ── dashboards ───────────────────────────────────────────────────────

    def put_dashboard(self, name: str, title: str, icon: str, widgets: list) -> None:
        self.drop(f"/dashboards/{name}")
        self("POST", f"/dashboards/{name}")
        current = self("GET", f"/dashboards/{name}?draft=true")
        self(
            "PUT",
            f"/dashboards/{name}",
            {
                **current,
                "title": title,
                "icon": icon,
                "columns": dashboards.COLUMNS,
                "canvas_width": dashboards.CANVAS[0],
                "canvas_height": dashboards.CANVAS[1],
                "pages": [
                    {
                        "id": "main",
                        "title": title,
                        "sections": [{"id": "main", "widgets": widgets}],
                    }
                ],
            },
        )
        version = self("GET", f"/dashboards/{name}?draft=true")["version"]
        self("POST", f"/dashboards/{name}/publish", {"version": version})


def _issue_line(issue: Any) -> str:
    if isinstance(issue, dict):
        return f"{issue.get('code')}: {issue.get('message')}"
    return str(issue)


class Flow:
    """One flow's definition, assembled before it is sent."""

    def __init__(self, name: str, title: str) -> None:
        self.name = name
        self.title = title
        self.nodes: list[dict[str, Any]] = []
        self.inputs: list[dict[str, Any]] = []
        self.sources: dict[str, str] = {}

    def add(self, node: dict[str, Any], source: str | None = None) -> dict[str, Any]:
        self.nodes.append(node)
        if source is not None:
            self.sources[node["id"]] = source
        return node

    def input(self, name: str, dtype: str, initial: Any, **spec: Any) -> None:
        self.inputs.append(
            {"spec": {"name": name, "dtype": dtype, **spec}, "initial": initial}
        )

    def msg(self, name: str) -> str:
        """This flow's name for a message, as a dashboard widget spells it."""
        return f"{self.name}.{name}"


def report(name: str, nodes: int, issues: list[str]) -> None:
    print(f"  {name}: {nodes} nodes, published, stopped")
    for issue in issues:
        print(f"    ! {issue}", file=sys.stderr)