feat(workspace): ingest hardware test results and show on dashboard

- Add TEST_RESULTS_PATH storage alongside machine registry
- Add POST /api/test-results endpoint to ingest per-device results
- Add GET /api/test-results endpoint to list recent results
- Render recent hardware test results on investor dashboard
This commit is contained in:
Fábio Cunha 2026-07-05 17:38:02 +01:00
parent 2bf4c2183c
commit 22fa151d6c

View file

@ -79,6 +79,7 @@ SAVEARTH_WORKSPACE_DATA_DIR = Path(
MACHINE_REGISTRY_PATH = SAVEARTH_WORKSPACE_DATA_DIR / "machines.json"
SNAPSHOTS_DIR = SAVEARTH_WORKSPACE_DATA_DIR / "snapshots"
TOKENS_DIR = SAVEARTH_WORKSPACE_DATA_DIR / "tokens"
TEST_RESULTS_PATH = SAVEARTH_WORKSPACE_DATA_DIR / "test_results.json"
def _ensure_data_dirs() -> None:
@ -226,6 +227,32 @@ def _validate_token(token: str) -> Optional[str]:
return info.get("machine_name")
def _load_test_results(limit: int = 100) -> List[Dict[str, Any]]:
"""Load recent hardware test results."""
_ensure_data_dirs()
if not TEST_RESULTS_PATH.exists():
return []
try:
data = json.loads(TEST_RESULTS_PATH.read_text(encoding="utf-8"))
results = data.get("results", [])
return results[-limit:]
except Exception:
return []
def _append_test_result(result: Dict[str, Any]) -> None:
"""Append a single test result to the workspace store."""
_ensure_data_dirs()
data: Dict[str, Any] = {"results": []}
if TEST_RESULTS_PATH.exists():
try:
data = json.loads(TEST_RESULTS_PATH.read_text(encoding="utf-8"))
except Exception:
pass
data.setdefault("results", []).append(result)
TEST_RESULTS_PATH.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
def _machine_id(machine_name: str) -> str:
return hashlib.sha256(machine_name.encode()).hexdigest()[:12]
@ -698,8 +725,38 @@ bash bootstrap.sh</pre>
</div>
</div>
<div class="section">
<h2>🧪 Hardware Test Results</h2>
<div class="fleet-box">
{% if test_results %}
<p>Recent results: <span class="fleet-metric">{{ test_results|length }}</span></p>
<table>
<thead>
<tr><th>MAC</th><th>PIR</th><th>MIC</th><th>Battery</th><th>Display</th><th>Serial</th><th>Time</th></tr>
</thead>
<tbody>
{% for r in test_results %}
<tr>
<td>{{ r.mac }}</td>
<td><span class="badge badge-{{ 'green' if r.pir_ok else 'red' }}">{{ 'OK' if r.pir_ok else 'Fail' }}</span></td>
<td><span class="badge badge-{{ 'green' if r.mic_ok else 'red' }}">{{ 'OK' if r.mic_ok else 'Fail' }}</span></td>
<td><span class="badge badge-{{ 'green' if r.battery_ok else 'red' }}">{{ 'OK' if r.battery_ok else 'Fail' }}</span></td>
<td>{{ r.display_state }}</td>
<td>{{ r.serial_number }}</td>
<td>{{ r.timestamp }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p>No hardware test results ingested yet.</p>
<p>Test stations push results to <code>/api/test-results</code>.</p>
{% endif %}
</div>
</div>
<div class="footer">
<p>Data sources: savearth project workspaces, replica-omnisciente central brain realms, savearth-mcp (live fleet data).</p>
<p>Data sources: savearth project workspaces, replica-omnisciente central brain realms, savearth-mcp (live fleet data), hardware test stations.</p>
<p>Dashboard served by <strong>savearth-workspace</strong> MCP server.</p>
</div>
</div>
@ -725,11 +782,13 @@ def build_dashboard_html(fleet: Optional[Dict[str, Any]] = None) -> str:
{"name": name, **m}
for name, m in registry.get("machines", {}).items()
]
test_results = _load_test_results(limit=20)
template = Template(DASHBOARD_TEMPLATE)
return template.render(
generated_at=_now(),
executive_summary=executive_summary,
test_results=test_results,
projects=projects,
risks=risks,
fleet=fleet or {"available": False, "errors": []},
@ -1146,6 +1205,40 @@ async def register_api_handler(request: Request) -> JSONResponse:
return JSONResponse({"status": "registered", "machine_name": machine_name})
async def ingest_test_result_handler(request: Request) -> JSONResponse:
"""Ingest a single-device hardware test result from a test station."""
try:
body = await request.json()
except Exception:
return JSONResponse({"error": "Invalid JSON"}, status_code=400)
required = {"mac", "pir_ok", "mic_ok", "battery_ok", "display_state", "timestamp"}
missing = required - set(body.keys())
if missing:
return JSONResponse({"error": f"Missing fields: {sorted(missing)}"}, status_code=400)
result = {
"mac": body.get("mac"),
"pir_ok": bool(body.get("pir_ok")),
"mic_ok": bool(body.get("mic_ok")),
"battery_ok": bool(body.get("battery_ok")),
"display_state": body.get("display_state", "SKIPPED"),
"serial_number": body.get("serial_number", ""),
"notes": body.get("notes", ""),
"timestamp": body.get("timestamp"),
"station": body.get("station", "unknown"),
"received_at": _now(),
}
_append_test_result(result)
return JSONResponse({"status": "accepted", "mac": result["mac"]})
async def list_test_results_handler(request: Request) -> JSONResponse:
"""Return recent hardware test results."""
limit = int(request.query_params.get("limit", "100"))
return JSONResponse({"results": _load_test_results(limit=limit)})
async def latest_snapshot_handler(request: Request) -> JSONResponse:
"""Return metadata for the latest snapshot of a machine."""
machine_id = request.query_params.get("machine_id", "")
@ -1198,6 +1291,8 @@ def build_starlette_app() -> Starlette:
Route("/health", health_handler),
Route("/bootstrap/{os}", bootstrap_script_handler),
Route("/api/register", register_api_handler, methods=["POST"]),
Route("/api/test-results", ingest_test_result_handler, methods=["POST"]),
Route("/api/test-results", list_test_results_handler),
Route("/api/snapshots/latest", latest_snapshot_handler),
Route("/api/snapshots/{machine_id}/{snapshot_id}.tar.gz", download_snapshot_handler),
Mount("/", app=mcp_starlette),