clawdmeter/daemon/server.py

"""Local control-panel HTTP API for the Clawdmeter app (Phase 7, M1).

A FastAPI app bound to 127.0.0.1 only. Serves the brand-styled web UI (``web/``)
and a small REST API over the unified config (``config.py``). Runs in a daemon
thread alongside the tray and the BLE daemon — one process, one exe.

Security: bound to loopback; the HA token is masked in GET responses and never
logged (mirrors ha_client / config).
"""
from __future__ import annotations

import copy
import os
import sys
import threading
from pathlib import Path

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

try:
    from daemon import config as cfgmod
except ImportError:  # running as a plain script (cwd on path)
    import config as cfgmod

MASK = "********"   # what the UI sees instead of the real token; echo it back to keep it


def _web_dir() -> Path:
    """The static UI directory: next to this file in source, under the PyInstaller
    bundle root when frozen (added via clawdmeter.spec datas)."""
    if getattr(sys, "frozen", False):
        return Path(sys._MEIPASS) / "daemon" / "web"  # type: ignore[attr-defined]
    return Path(__file__).parent / "web"


def _masked(cfg: dict) -> dict:
    c = copy.deepcopy(cfg)
    if c.get("ha", {}).get("token"):
        c["ha"]["token"] = MASK
    return c


app = FastAPI(title="Clawdmeter")


@app.get("/api/config")
def get_config() -> dict:
    return _masked(cfgmod.load_config())


class ConfigIn(BaseModel):
    ha: dict | None = None
    buttons: list | None = None
    settings: dict | None = None


@app.put("/api/config")
def put_config(incoming: ConfigIn) -> dict:
    cfg = cfgmod.load_config()
    data = incoming.model_dump(exclude_none=True)
    if "ha" in data:
        ha = dict(data["ha"])
        # Mask echoed back unchanged => keep the stored token (UI never holds it).
        if ha.get("token") == MASK:
            ha["token"] = cfg["ha"]["token"]
        cfg["ha"].update(ha)
    if "settings" in data:
        cfg["settings"].update(data["settings"])
    if "buttons" in data:
        cfg["buttons"] = data["buttons"]
    cfgmod.save_config(cfg)
    return _masked(cfg)


def _resolve_token(token: str) -> str:
    return cfgmod.load_config()["ha"]["token"] if token == MASK else token


@app.post("/api/ha/test")
async def ha_test(body: dict) -> dict:
    url = (body.get("url") or "").strip().rstrip("/")
    token = _resolve_token((body.get("token") or "").strip())
    if not url or not token:
        raise HTTPException(status_code=400, detail="url and token are required")
    try:
        async with httpx.AsyncClient(timeout=8.0) as client:
            resp = await client.get(f"{url}/api/",
                                    headers={"Authorization": f"Bearer {token}"})
    except httpx.HTTPError as e:
        return {"ok": False, "error": str(e)}
    if resp.status_code == 200:
        try:
            msg = resp.json().get("message", "API running")
        except ValueError:
            msg = "API running"
        return {"ok": True, "message": msg}
    return {"ok": False, "error": f"HTTP {resp.status_code}"}


@app.get("/api/ha/entities")
async def ha_entities() -> dict:
    ha = cfgmod.ha_settings()
    if not ha:
        return {"entities": [], "error": "Home Assistant not configured"}
    try:
        async with httpx.AsyncClient(timeout=8.0) as client:
            resp = await client.get(f"{ha['url']}/api/states",
                                    headers={"Authorization": f"Bearer {ha['token']}"})
    except httpx.HTTPError as e:
        return {"entities": [], "error": str(e)}
    if resp.status_code != 200:
        return {"entities": [], "error": f"HTTP {resp.status_code}"}
    try:
        states = resp.json()
    except ValueError:
        return {"entities": [], "error": "bad response"}
    lights = [s["entity_id"] for s in states
              if isinstance(s, dict) and str(s.get("entity_id", "")).startswith("light.")]
    return {"entities": sorted(lights)}


_status_provider = None  # set by the tray to expose live BLE/daemon state


def set_status_provider(fn) -> None:
    """The tray injects a callable returning the live status dict (connected,
    battery, state). Kept out of import-time so server.py runs standalone."""
    global _status_provider
    _status_provider = fn


@app.get("/api/status")
def get_status() -> dict:
    if _status_provider is not None:
        try:
            return _status_provider()
        except Exception:
            pass
    return {"connected": False, "battery": None, "state": "unknown"}


# Static UI mounted LAST so the /api/* routes above take precedence.
_wd = _web_dir()
if _wd.exists():
    app.mount("/", StaticFiles(directory=str(_wd), html=True), name="web")


def serve_in_thread(port: int | None = None) -> threading.Thread:
    """Start uvicorn on 127.0.0.1 in a daemon thread; return the thread."""
    import uvicorn

    p = port or cfgmod.DEFAULT_PORT
    # log_config=None disables uvicorn's default dictConfig. In the frozen,
    # windowed exe there is no console, so sys.stderr is None and uvicorn's
    # default logging setup dies with "Unable to configure formatter 'default'",
    # taking the whole control-panel server down. We don't need uvicorn's logs
    # (the daemon has its own file logger), so skip its logging config entirely.
    server = uvicorn.Server(uvicorn.Config(
        app, host="127.0.0.1", port=p, log_level="warning", log_config=None))
    t = threading.Thread(target=server.run, daemon=True, name="clawd-http")
    t.start()
    return t


if __name__ == "__main__":
    # Dev: run the server in the foreground with autoreload-free uvicorn.
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=cfgmod.DEFAULT_PORT, log_level="info")