"""Grabette fleet — operator dashboard + command broker (Docker HF Space). A free-tier-friendly Docker Space. It is self-contained (no `grabette` import) so it deploys standalone to HF. Responsibilities: * Operator UI + login via HF Spaces native OAuth (`hf_oauth: true`). * Command broker: an in-memory, per-owner device registry + command queue. * Device auth: devices call with `Authorization: Bearer `; we resolve the owner via `whoami` (cached) and group devices by HF identity. Transport is short-polling (devices GET /api/devices/poll every couple seconds). That polling traffic doubles as the keep-alive that stops a free Space sleeping (sleep is timed from the last request). State is in-memory, so a restart drops it — devices simply re-register on their next poll; durable data lives in the device's own HF datasets, not here. Designed to be duplicated per user: one Space = one owner = one fleet. """ from __future__ import annotations import os import time import uuid from dataclasses import dataclass, field from typing import Any, Optional from fastapi import Depends, FastAPI, Header, HTTPException, Request from fastapi.responses import HTMLResponse from huggingface_hub import attach_huggingface_oauth, parse_huggingface_oauth, whoami from pydantic import BaseModel ONLINE_WINDOW = 15.0 # device is "online" if seen within this many seconds WHOAMI_TTL = 300.0 # --- device identity (Bearer token -> HF username, cached) ------------------- _whoami_cache: dict[str, tuple[str, float]] = {} def verify_user(token: str) -> str: now = time.time() hit = _whoami_cache.get(token) if hit and hit[1] > now: return hit[0] try: info = whoami(token=token) except Exception as e: # noqa: BLE001 raise ValueError(f"Invalid HF token: {e}") from e name = (info or {}).get("name", "") if not name: raise ValueError("Could not resolve HF username") _whoami_cache[token] = (name, now + WHOAMI_TTL) return name async def device_owner(authorization: Optional[str] = Header(None)) -> str: """FastAPI dep: resolve the device's owner from its Bearer HF token.""" if not authorization or not authorization.startswith("Bearer "): raise HTTPException(401, "Missing 'Authorization: Bearer '") try: return verify_user(authorization[len("Bearer ") :].strip()) except ValueError as e: raise HTTPException(401, str(e)) from e def operator_name(request: Request) -> str: """Resolve the logged-in operator's HF username, or 401.""" info = parse_huggingface_oauth(request) if info is None: raise HTTPException(401, "Not logged in") return info.user_info.preferred_username or info.user_info.name # --- in-memory fleet state --------------------------------------------------- @dataclass class Command: id: str type: str args: dict[str, Any] status: str = "pending" # pending | sent | done result: Optional[dict[str, Any]] = None created_at: float = field(default_factory=time.time) @dataclass class Device: device_id: str name: str capabilities: list[str] last_seen: float = field(default_factory=time.time) queue: list[Command] = field(default_factory=list) history: list[Command] = field(default_factory=list) @property def online(self) -> bool: return (time.time() - self.last_seen) < ONLINE_WINDOW FLEET: dict[str, dict[str, Device]] = {} # owner -> device_id -> Device def _fleet_of(owner: str) -> dict[str, Device]: return FLEET.setdefault(owner, {}) # --- app + OAuth ------------------------------------------------------------- app = FastAPI(title="Grabette fleet") attach_huggingface_oauth(app) # adds /oauth/huggingface/{login,logout,callback} # === device-facing API (Bearer auth) ======================================== class RegisterReq(BaseModel): device_id: str name: str = "" capabilities: list[str] = [] class ResultReq(BaseModel): device_id: str command_id: str result: dict[str, Any] = {} @app.post("/api/devices/register") async def register(req: RegisterReq, owner: str = Depends(device_owner)) -> dict[str, Any]: fleet = _fleet_of(owner) dev = fleet.get(req.device_id) if dev is None: dev = Device(req.device_id, req.name or req.device_id, req.capabilities) fleet[req.device_id] = dev else: dev.name = req.name or dev.name dev.capabilities = req.capabilities or dev.capabilities dev.last_seen = time.time() return {"status": "ok", "pending": len(dev.queue)} @app.get("/api/devices/poll") async def poll(device_id: str, owner: str = Depends(device_owner)) -> dict[str, Any]: dev = _fleet_of(owner).get(device_id) if dev is None: raise HTTPException(404, "Device not registered") dev.last_seen = time.time() pending = [c for c in dev.queue if c.status == "pending"] for c in pending: c.status = "sent" return {"commands": [{"id": c.id, "type": c.type, "args": c.args} for c in pending]} @app.post("/api/devices/result") async def result(req: ResultReq, owner: str = Depends(device_owner)) -> dict[str, str]: dev = _fleet_of(owner).get(req.device_id) if dev is None: raise HTTPException(404, "Device not registered") for c in dev.queue: if c.id == req.command_id: c.status = "done" c.result = req.result dev.queue.remove(c) dev.history.insert(0, c) del dev.history[20:] break return {"status": "ok"} # === operator-facing API (session OAuth) ===================================== class DispatchReq(BaseModel): device_id: str type: str args: dict[str, Any] = {} @app.get("/api/fleet/me") async def me(request: Request) -> dict[str, Any]: info = parse_huggingface_oauth(request) if info is None: return {"logged_in": False} return {"logged_in": True, "username": info.user_info.preferred_username or info.user_info.name} @app.get("/api/fleet/devices") async def list_devices(request: Request) -> dict[str, Any]: owner = operator_name(request) devices = _fleet_of(owner).values() return { "owner": owner, "devices": [ { "device_id": d.device_id, "name": d.name, "online": d.online, "capabilities": d.capabilities, "pending": len([c for c in d.queue if c.status != "done"]), "history": [ {"type": c.type, "args": c.args, "result": c.result} for c in d.history[:5] ], } for d in devices ], } @app.post("/api/fleet/dispatch") async def dispatch(req: DispatchReq, request: Request) -> dict[str, Any]: owner = operator_name(request) dev = _fleet_of(owner).get(req.device_id) if dev is None: raise HTTPException(404, "Device not found in your fleet") cmd = Command(id=uuid.uuid4().hex[:12], type=req.type, args=req.args) dev.queue.append(cmd) return {"status": "queued", "command_id": cmd.id} # === UI ====================================================================== @app.get("/") async def index() -> HTMLResponse: return HTMLResponse(_INDEX) _INDEX = """ Grabette fleet

🤖 Grabette fleet — operator dashboard

HuggingFace login

Checking…

Your devices

devicecommandslast result
No devices reporting yet. Start one with device_app.py.
"""