operating-room-monitor-server/scripts/demo_client/server.py

"""Tiny stdlib HTTP server for the demo client page.

- Serves `index.html` (and other sibling files) from this directory.
- Exposes `GET /labels.json`, which parses the repo's
  `app/resources/consumable_classifier_labels.yaml` and returns its label
  list so the page can prefill the candidate-consumables input.
- ``--live`` 使用 dev 依赖 ``livereload``：修改本目录下 HTML/JS/CSS
  或更新 ``labels.json``/YAML 后，浏览器自动刷新（需 ``uv run --group dev``）。

Run:
    python scripts/demo_client/server.py            # 127.0.0.1:38081
    python scripts/demo_client/server.py -p 9000    # custom port
    uv run --group dev python scripts/demo_client/server.py --live
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import threading
import time
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

SCRIPT_DIR = Path(__file__).resolve().parent
REPO_ROOT = SCRIPT_DIR.parents[1]
LABELS_YAML = REPO_ROOT / "app" / "resources" / "consumable_classifier_labels.yaml"
# ``--live`` 模式写入静态 `labels.json`，与 livereload 兼容（勿提交，见 .gitignore）
LABELS_JSON_ARTIFACT = SCRIPT_DIR / "labels.json"


def _load_labels_with_pyyaml(path: Path) -> list[str] | None:
    try:
        import yaml  # type: ignore[import-untyped]
    except ImportError:
        return None
    try:
        data: Any = yaml.safe_load(path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError):
        return None
    if not isinstance(data, dict):
        return None
    names = data.get("names")
    if isinstance(names, dict):
        try:
            items = sorted(names.items(), key=lambda kv: int(kv[0]))
        except (TypeError, ValueError):
            items = list(names.items())
        return [str(v) for _, v in items]
    if isinstance(names, list):
        return [str(v) for v in names]
    return None


def _load_labels_fallback(path: Path) -> list[str]:
    """Minimal parser for the known labels.yaml shape: `<int>: <text>` under `names:`."""
    labels: list[tuple[int, str]] = []
    in_names = False
    pattern = re.compile(r"^\s+(\d+)\s*:\s*(.+?)\s*$")
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if stripped.startswith("names:"):
            in_names = True
            continue
        if in_names and not line.startswith((" ", "\t")):
            in_names = False
        if not in_names:
            continue
        match = pattern.match(line)
        if match:
            labels.append((int(match.group(1)), match.group(2)))
    labels.sort(key=lambda kv: kv[0])
    return [name for _, name in labels]


def load_labels() -> list[str]:
    if not LABELS_YAML.is_file():
        return []
    labels = _load_labels_with_pyyaml(LABELS_YAML)
    if labels is None:
        labels = _load_labels_fallback(LABELS_YAML)
    return labels


def write_labels_json_artifact() -> None:
    body = json.dumps({"labels": load_labels()}, ensure_ascii=False) + "\n"
    LABELS_JSON_ARTIFACT.write_text(body, encoding="utf-8")


def _spawn_labels_yaml_poll() -> None:
    p = LABELS_YAML
    state: list[float] = [0.0]
    if p.is_file():
        state[0] = p.stat().st_mtime

    def loop() -> None:
        while True:
            time.sleep(1.0)
            if not p.is_file():
                continue
            m = p.stat().st_mtime
            if m > state[0]:
                state[0] = m
                try:
                    write_labels_json_artifact()
                except OSError as e:
                    print(
                        f"[demo-client] labels.json 写入失败: {e}",
                        file=sys.stderr,
                    )

    threading.Thread(target=loop, daemon=True, name="DemoClientLabelsYamlWatch").start()


def run_livereload(host: str, port: int) -> None:
    try:
        from livereload import Server
    except ImportError as exc:  # pragma: no cover
        print(
            "``--live`` 需要 dev 依赖。请执行: uv sync --group dev",
            file=sys.stderr,
        )
        raise SystemExit(1) from exc

    write_labels_json_artifact()
    _spawn_labels_yaml_poll()
    server = Server()
    server.watch(str(SCRIPT_DIR))
    if LABELS_YAML.is_file():
        server.watch(str(LABELS_YAML))
    url = f"http://{host}:{port}/"
    print(f"Demo client (livereload) at {url}")
    print(f"  static dir : {SCRIPT_DIR}")
    print(f"  labels yaml: {LABELS_YAML}  ->  {LABELS_JSON_ARTIFACT.name}")
    print("Press Ctrl+C to stop.")
    server.serve(
        root=str(SCRIPT_DIR),
        host=host,
        port=port,
        open_url=False,
        debug=False,
    )


class DemoHandler(SimpleHTTPRequestHandler):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, directory=str(SCRIPT_DIR), **kwargs)

    def do_GET(self) -> None:  # noqa: N802 (stdlib override)
        if self.path.split("?", 1)[0] == "/labels.json":
            self._send_labels()
            return
        super().do_GET()

    def _send_labels(self) -> None:
        body = json.dumps({"labels": load_labels()}, ensure_ascii=False).encode("utf-8")
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: Any) -> None:  # noqa: A002
        sys.stderr.write(
            "[demo-client] %s - %s\n" % (self.address_string(), format % args)
        )


def run_plain(host: str, port: int) -> None:
    server = ThreadingHTTPServer((host, port), DemoHandler)
    url = f"http://{host}:{port}/"
    print(f"Demo client serving at {url}")
    print(f"  static dir : {SCRIPT_DIR}")
    print(f"  labels yaml: {LABELS_YAML}")
    print("Press Ctrl+C to stop.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")
    finally:
        server.server_close()


def main() -> None:
    parser = argparse.ArgumentParser(description="Operation room demo client server")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=38081)
    parser.add_argument(
        "--live",
        action="store_true",
        help="浏览器热重载（需 uv sync --group dev；会生成同目录 labels.json）",
    )
    args = parser.parse_args()
    if args.live:
        run_livereload(args.host, args.port)
    else:
        run_plain(args.host, args.port)


if __name__ == "__main__":
    main()