Upload

server/data/boxes.json

@@ -0,0 +1,15 @@
+{
+  "klangkiste-x1u8i4x5e4": {
+    "fingerprint": "ecfcaddfcdf73417c8b377cb807b006f7d0756ca717fda35f377e8d18b297c43",
+    "state": "UNPAIRED",
+    "first_seen": 1768244301,
+    "last_seen": 1768244591,
+    "firmware_version": "0.1.0",
+    "capabilities": {
+      "nfc": true,
+      "audio": true,
+      "spotify": true
+    },
+    "api_token": null
+  }
+}

server/main.py

@@ -0,0 +1,168 @@
+"""Minimal server API and GUI for pairing."""
+
+from __future__ import annotations
+
+import secrets
+import time
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI, HTTPException, Form
+from fastapi.responses import HTMLResponse, RedirectResponse
+from pydantic import BaseModel
+import uvicorn
+
+from storage import BoxRecord, BoxRegistry
+
+DATA_DIR = Path(__file__).resolve().parent / "data"
+DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+registry = BoxRegistry(DATA_DIR)
+app = FastAPI()
+
+
+class AnnounceRequest(BaseModel):
+    box_id: str
+    fingerprint: str
+    firmware_version: str
+    capabilities: dict[str, Any]
+
+
+class PollRequest(BaseModel):
+    box_id: str
+
+
+class PairRequest(BaseModel):
+    box_id: str
+
+
+def _now() -> int:
+    return int(time.time())
+
+
+def _validate_box_id(value: str) -> bool:
+    if not value.startswith("klangkiste-"):
+        return False
+    suffix = value.replace("klangkiste-", "", 1)
+    if len(suffix) != 10:
+        return False
+    return all(c.isdigit() or ("a" <= c <= "z") for c in suffix)
+
+
+@app.post("/api/boxes/announce")
+def announce(req: AnnounceRequest) -> dict[str, Any]:
+    if not _validate_box_id(req.box_id):
+        raise HTTPException(status_code=400, detail="invalid box_id")
+    if not req.fingerprint:
+        raise HTTPException(status_code=400, detail="fingerprint required")
+
+    boxes = registry.load()
+    now = _now()
+    if req.box_id in boxes:
+        record = boxes[req.box_id]
+        if record.fingerprint and record.fingerprint != req.fingerprint:
+            raise HTTPException(status_code=400, detail="fingerprint mismatch")
+        record.last_seen = now
+        record.firmware_version = req.firmware_version
+        record.capabilities = req.capabilities
+        if record.state != "PAIRED":
+            record.state = "UNPAIRED"
+    else:
+        record = BoxRecord(
+            box_id=req.box_id,
+            fingerprint=req.fingerprint,
+            state="UNPAIRED",
+            first_seen=now,
+            last_seen=now,
+            firmware_version=req.firmware_version,
+            capabilities=req.capabilities,
+        )
+    boxes[req.box_id] = record
+    registry.save(boxes)
+    return {"ok": True}
+
+
+@app.post("/api/boxes/poll-pairing")
+def poll_pairing(req: PollRequest) -> dict[str, Any]:
+    boxes = registry.load()
+    record = boxes.get(req.box_id)
+    if not record:
+        raise HTTPException(status_code=404, detail="unknown box")
+    if record.state == "PAIRED" and record.api_token:
+        return {"paired": True, "api_token": record.api_token}
+    return {"paired": False}
+
+
+@app.post("/api/boxes/pair")
+def pair_box(req: PairRequest) -> dict[str, Any]:
+    boxes = registry.load()
+    record = boxes.get(req.box_id)
+    if not record:
+        raise HTTPException(status_code=404, detail="unknown box")
+    if record.state != "PAIRED":
+        record.state = "PAIRED"
+        record.api_token = secrets.token_hex(32)
+        boxes[req.box_id] = record
+        registry.save(boxes)
+    return {"paired": True, "api_token": record.api_token}
+
+
+@app.get("/ui", response_class=HTMLResponse)
+def ui() -> str:
+    boxes = _apply_offline(registry.load())
+    unpaired = [b for b in boxes.values() if b.state != "PAIRED"]
+    paired = [b for b in boxes.values() if b.state == "PAIRED"]
+
+    def _row(box: BoxRecord) -> str:
+        return (
+            f"<tr><td>{box.box_id}</td><td>{box.firmware_version}</td>"
+            f"<td>{box.last_seen}</td>"
+            f"<td><form method='post' action='/ui/pair'>"
+            f"<input type='hidden' name='box_id' value='{box.box_id}'/>"
+            f"<button type='submit'>Box hinzufuegen</button>"
+            f"</form></td></tr>"
+        )
+
+    unpaired_rows = "".join(_row(b) for b in unpaired) or "<tr><td colspan='4'>Keine</td></tr>"
+    paired_rows = "".join(
+        f"<tr><td>{b.box_id}</td><td>{b.firmware_version}</td><td>{b.last_seen}</td></tr>"
+        for b in paired
+    ) or "<tr><td colspan='3'>Keine</td></tr>"
+
+    return f"""
+<!doctype html>
+<html>
+  <head><meta charset="utf-8"><title>Klangkiste Server</title></head>
+  <body>
+    <h1>Neue Boxen (UNPAIRED)</h1>
+    <table border="1">
+      <tr><th>box_id</th><th>firmware</th><th>last_seen</th><th>action</th></tr>
+      {unpaired_rows}
+    </table>
+    <h1>Gepaarte Boxen</h1>
+    <table border="1">
+      <tr><th>box_id</th><th>firmware</th><th>last_seen</th></tr>
+      {paired_rows}
+    </table>
+  </body>
+</html>
+"""
+
+
+def _apply_offline(boxes: dict[str, BoxRecord], threshold_seconds: int = 120) -> dict[str, BoxRecord]:
+    now = _now()
+    for record in boxes.values():
+        if now - record.last_seen > threshold_seconds:
+            record.state = "OFFLINE"
+    registry.save(boxes)
+    return boxes
+
+
+@app.post("/ui/pair")
+def ui_pair(box_id: str = Form(...)) -> RedirectResponse:
+    pair_box(PairRequest(box_id=box_id))
+    return RedirectResponse(url="/ui", status_code=303)
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7000, log_level="info")

server/requirements.txt

@@ -0,0 +1,3 @@
+fastapi==0.115.6
+uvicorn==0.30.6
+pydantic==2.9.2

server/storage.py

@@ -0,0 +1,60 @@
+"""Server-side storage for box registry."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class BoxRecord:
+    box_id: str
+    fingerprint: str
+    state: str
+    first_seen: int
+    last_seen: int
+    firmware_version: str
+    capabilities: dict[str, Any]
+    api_token: str | None = None
+
+
+class BoxRegistry:
+    def __init__(self, base_dir: Path) -> None:
+        self._path = base_dir / "boxes.json"
+
+    def load(self) -> dict[str, BoxRecord]:
+        if not self._path.exists():
+            return {}
+        raw = json.loads(self._path.read_text(encoding="utf-8"))
+        result: dict[str, BoxRecord] = {}
+        for box_id, data in raw.items():
+            if not isinstance(data, dict):
+                continue
+            result[box_id] = BoxRecord(
+                box_id=box_id,
+                fingerprint=str(data.get("fingerprint", "")),
+                state=str(data.get("state", "UNPAIRED")),
+                first_seen=int(data.get("first_seen", 0)),
+                last_seen=int(data.get("last_seen", 0)),
+                firmware_version=str(data.get("firmware_version", "")),
+                capabilities=dict(data.get("capabilities", {})),
+                api_token=data.get("api_token"),
+            )
+        return result
+
+    def save(self, records: dict[str, BoxRecord]) -> None:
+        data = {
+            box_id: {
+                "fingerprint": rec.fingerprint,
+                "state": rec.state,
+                "first_seen": rec.first_seen,
+                "last_seen": rec.last_seen,
+                "firmware_version": rec.firmware_version,
+                "capabilities": rec.capabilities,
+                "api_token": rec.api_token,
+            }
+            for box_id, rec in records.items()
+        }
+        self._path.write_text(json.dumps(data, indent=2), encoding="utf-8")