tf

server/routes.py

@@ -1,11 +1,20 @@
 """API route definitions for the smoking detection server."""
 
+import asyncio
 import shutil
 import time
 from pathlib import Path
 
 import cv2
-from fastapi import APIRouter, File, Form, Response, UploadFile
+from fastapi import (
+    APIRouter,
+    File,
+    Form,
+    Response,
+    UploadFile,
+    WebSocket,
+    WebSocketDisconnect,
+)
 from fastapi.responses import StreamingResponse
 
 import server.runtime as rt
@@ -80,6 +89,43 @@ def live_status():
     return {"status": "success", "live": rt.manager.status()}
 
 
+@router.websocket("/ws/updates")
+async def ws_updates(websocket: WebSocket, history_limit: int = 10):
+    await websocket.accept()
+    clamped_limit = max(1, min(history_limit, 50))
+
+    try:
+        while True:
+            try:
+                records = rt.db.get_detections(None)[:clamped_limit]
+                history = [
+                    {
+                        "id": r.id,
+                        "user_id": r.user_id,
+                        "image_path": r.image_path,
+                        "confidence": r.confidence,
+                        "timestamp": r.timestamp,
+                        "camera_source": r.camera_source,
+                    }
+                    for r in records
+                ]
+            except Exception:
+                history = []
+
+            await websocket.send_json(
+                {
+                    "type": "snapshot",
+                    "live": rt.manager.status(),
+                    "history": history,
+                    "history_count": len(history),
+                    "server_time": int(time.time()),
+                }
+            )
+            await asyncio.sleep(1.0)
+    except WebSocketDisconnect:
+        return
+
+
 @router.post("/live/start")
 def live_start(source: str | None = Form(default=None)):
     """Start live pipeline (optionally with source override)."""
@@ -99,7 +145,26 @@ def live_start_phone():
         "status": "success" if ok else "fail",
         "message": message,
         "live": rt.manager.status(),
-        "hint": "Push JPEG frames to /live/push_frame",
+        "hint": "Push JPEG frames to /live/ws/push_frames",
+    }
+
+
+@router.get("/live/source_history")
+def live_source_history(limit: int = 10):
+    return {
+        "status": "success",
+        "current_source": rt.manager.status().get("source"),
+        "history": rt.manager.source_history(limit=limit),
+    }
+
+
+@router.post("/live/switch_previous")
+def live_switch_previous():
+    ok, message = rt.manager.switch_previous()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
     }
 
 
@@ -166,6 +231,44 @@ async def live_push_frame(frame: UploadFile = File(...)):
     }
 
 
+@router.websocket("/live/ws/push_frames")
+async def live_ws_push_frames(websocket: WebSocket):
+    await websocket.accept()
+    accepted = 0
+    dropped = 0
+
+    try:
+        while True:
+            message = await websocket.receive()
+            if message.get("type") == "websocket.disconnect":
+                break
+
+            payload = message.get("bytes")
+            if payload is None:
+                text_payload = message.get("text")
+                if text_payload == "ping":
+                    await websocket.send_json({"type": "pong"})
+                continue
+
+            ok, _ = rt.push_live_frame_bytes(payload)
+            if ok:
+                accepted += 1
+            else:
+                dropped += 1
+
+            if (accepted + dropped) % 60 == 0:
+                await websocket.send_json(
+                    {
+                        "type": "stats",
+                        "accepted": accepted,
+                        "dropped": dropped,
+                        "live": rt.manager.status(),
+                    }
+                )
+    except WebSocketDisconnect:
+        return
+
+
 @router.post("/clips/submit")
 async def clips_submit(
     file: UploadFile | None = File(default=None),

server/runtime.py

@@ -134,12 +134,28 @@ class PipelineManager:
         self._pipeline_factory: PipelineFactory | None = None
 
         self._source_spec: str = "0"
+        self._source_history: list[dict[str, Any]] = []
         self.running = False
         self.starting = False
         self.stopping = False
         self.paused = False
         self.last_error: str | None = None
 
+    def _record_source_locked(self, source_spec: str):
+        if (
+            self._source_history
+            and self._source_history[-1].get("source") == source_spec
+        ):
+            return
+        self._source_history.append(
+            {
+                "source": source_spec,
+                "timestamp": datetime.now().isoformat(timespec="seconds"),
+            }
+        )
+        if len(self._source_history) > 50:
+            self._source_history = self._source_history[-50:]
+
     def configure(
         self,
         source_factory: SourceFactory,
@@ -152,6 +168,7 @@ class PipelineManager:
             self._sink_factory = sink_factory
             self._pipeline_factory = pipeline_factory
             self._source_spec = initial_source
+            self._record_source_locked(initial_source)
 
     def _close_sinks(self, sinks: list[Sink] | None = None):
         sink_list = self._sinks if sinks is None else sinks
@@ -334,6 +351,7 @@ class PipelineManager:
     def switch_source(self, source_spec: str) -> tuple[bool, str]:
         with self._lock:
             self._source_spec = source_spec
+            self._record_source_locked(source_spec)
             running = self.running
             paused = self.paused
 
@@ -346,6 +364,27 @@ class PipelineManager:
 
         return self.start(source_spec)
 
+    def source_history(self, limit: int = 10) -> list[dict[str, Any]]:
+        with self._lock:
+            clamped = max(1, min(limit, 50))
+            items = self._source_history[-clamped:]
+            return list(reversed(items))
+
+    def switch_previous(self) -> tuple[bool, str]:
+        with self._lock:
+            current = self._source_spec
+            previous = None
+            for item in reversed(self._source_history):
+                source_spec = item.get("source")
+                if source_spec and source_spec != current:
+                    previous = source_spec
+                    break
+
+        if previous is None:
+            return False, "No previous source in history"
+
+        return self.switch_source(previous)
+
     def status(self) -> dict[str, Any]:
         with self._lock:
             return {
@@ -354,6 +393,7 @@ class PipelineManager:
                 "stopping": self.stopping,
                 "paused": self.paused,
                 "source": self._source_spec,
+                "source_history": self.source_history(limit=10),
                 "last_error": self.last_error,
                 "thread_alive": bool(self._thread and self._thread.is_alive()),
             }