Device registry (/devices + hello), static-IP fallback, WS keepalive, uncompressed fonts
Test, Build and Push / test-gateway (push) Successful in 11s
Test, Build and Push / release (push) Skipped
Test, Build and Push / build-gateway (push) Skipped

- gateway: GET /devices registry (per-IP connect count, timestamps,
  device/fw from new 'hello' protocol event); graceful utterance
  errors -> error event; uvicorn ws-ping-timeout 120s
- firmware: hello on WS connect; static 192.168.86.53 fallback after
  15s without DHCP; scan-on-boot removed (kept for diagnostics);
  ping_interval 8s; fonts regenerated --no-compress (compressed
  glyphs render blank with LV_USE_FONT_COMPRESSED off); 8MB factory
  partition; montserrat_28; LVGL lock timeout semantics fixed
- docs: sauron endpoint plan, power ladder, versioning policy

Known issue: 'Outside' AP association succeeds but no L2 traffic
flows (no DHCP, no ARP) — worked once at 22:05 then never again;
ToJ side verified clean. Investigation ongoing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 22:34:57 +02:00
co-authored by Claude Fable 5
parent ed0e8221f1
commit 3a402220b0
10 changed files with 4928 additions and 1568 deletions
+2 -1
View File
@@ -9,4 +9,5 @@ COPY src ./src
RUN pip install --no-cache-dir .
EXPOSE 8600
CMD ["uvicorn", "desklock_gateway.main:app", "--host", "0.0.0.0", "--port", "8600"]
# generous ws ping timeout: the device stays quiet during 10-25s Tatlock turns
CMD ["uvicorn", "desklock_gateway.main:app", "--host", "0.0.0.0", "--port", "8600", "--ws-ping-interval", "25", "--ws-ping-timeout", "120"]
+39 -3
View File
@@ -7,6 +7,7 @@ Protocol (see docs/architecture.md — keep in sync):
import asyncio
import logging
from datetime import datetime, timezone
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
@@ -20,21 +21,40 @@ app = FastAPI(title="DeskLock Gateway")
PCM_CHUNK_BYTES = 4096
# device registry: every endpoint that ever connected, keyed by client IP
DEVICES: dict[str, dict] = {}
def _now() -> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
@app.get("/healthz")
async def healthz() -> dict[str, str]:
return {"status": "ok"}
@app.get("/devices")
async def devices() -> dict[str, dict]:
return DEVICES
@app.websocket("/ws/voice")
async def voice(ws: WebSocket) -> None:
await ws.accept()
ip = ws.client.host if ws.client else "unknown"
device = DEVICES.setdefault(ip, {"connections": 0})
device["connections"] += 1
device.update(connected=True, connected_at=_now(), last_seen=_now())
logger.info("device connected from %s (connection #%d)", ip, device["connections"])
tatlock = TatlockClient()
pcm = bytearray()
recording = False
try:
while True:
message = await ws.receive()
device["last_seen"] = _now()
if "bytes" in message and message["bytes"] is not None:
if recording:
pcm.extend(message["bytes"])
@@ -45,15 +65,31 @@ async def voice(ws: WebSocket) -> None:
import json
event = json.loads(message["text"])
if event["type"] == "utterance_start":
if event["type"] == "hello":
device["device"] = event.get("device", "unknown")
device["fw"] = event.get("fw", "unknown")
logger.info("device identified: %s fw=%s", device["device"], device["fw"])
elif event["type"] == "utterance_start":
pcm.clear()
recording = True
elif event["type"] == "utterance_end":
recording = False
await _handle_utterance(ws, tatlock, bytes(pcm))
try:
await _handle_utterance(ws, tatlock, bytes(pcm))
except WebSocketDisconnect:
raise
except Exception:
logger.exception("utterance failed")
try:
await ws.send_json({"type": "error"})
await ws.send_json({"type": "state", "value": "idle"})
except Exception:
logger.info("device gone before error could be reported")
break
except WebSocketDisconnect:
logger.info("device disconnected")
logger.info("device disconnected: %s", ip)
finally:
device.update(connected=False, disconnected_at=_now())
await tatlock.aclose()
+7
View File
@@ -8,3 +8,10 @@ def test_healthz() -> None:
response = client.get("/healthz")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_devices_registry_empty() -> None:
client = TestClient(app)
response = client.get("/devices")
assert response.status_code == 200
assert isinstance(response.json(), dict)