what

pipeline/sinks/__init__.py

@@ -1,7 +1,7 @@
 """Sinks for pipeline output handling."""
 
 from pipeline.sinks.alerts import FileAlertStore
-from pipeline.sinks.database import DatabaseSink
+from pipeline.sinks.database import DatabaseSink, DatabaseSmartEventSink
 from pipeline.sinks.display import WindowDisplaySink
 from pipeline.sinks.events import SmartEventSink
 from pipeline.sinks.metrics import MetricsSink
@@ -10,6 +10,7 @@ from pipeline.sinks.video import VideoFileStore
 
 __all__ = [
     "DatabaseSink",
+    "DatabaseSmartEventSink",
     "FileAlertStore",
     "MetricsSink",
     "ScreenshotSink",

pipeline/sinks/database.py

@@ -8,6 +8,7 @@ import cv2
 
 from db.repository import Repository
 from pipeline.ops.drawing import draw_alert_banner, draw_detections
+from pipeline.sinks.events import SmartEventSink
 from pipeline.types import Frame
 
 logger = logging.getLogger(__name__)
@@ -72,3 +73,52 @@ class DatabaseSink:
 
     def close(self):
         pass  # DB connection is managed by the repository
+
+
+class DatabaseSmartEventSink(SmartEventSink):
+    """Records event clips and stores clip path in DB as alert evidence."""
+
+    def __init__(
+        self,
+        db_repo: Repository,
+        fps: float,
+        user_id: int = 1,
+        camera_source: str = "Pipeline",
+        output_dir: str = "static/events",
+        pre_event_seconds: float = 2,
+        post_event_seconds: float = 3,
+    ):
+        super().__init__(
+            output_dir=output_dir,
+            fps=fps,
+            pre_event_seconds=pre_event_seconds,
+            post_event_seconds=post_event_seconds,
+        )
+        self.db_repo = db_repo
+        self.user_id = user_id
+        self.camera_source = camera_source
+
+    def _trigger_event(self, trigger_frame: Frame):
+        was_recording = self.recording
+        super()._trigger_event(trigger_frame)
+
+        # Save one DB row per newly started event clip
+        if was_recording or not self.recording or self.current_event_id is None:
+            return
+
+        ticket = trigger_frame.ticket
+        if not ticket:
+            return
+
+        filename = f"event_{self.current_event_id}.mp4"
+        web_path = f"/static/events/{filename}"
+        try:
+            self.db_repo.save_detection(
+                user_id=self.user_id,
+                image_path=web_path,
+                confidence=ticket.confidence,
+                camera_source=self.camera_source,
+            )
+            logger.info("Saved event alert to DB: %s", web_path)
+        except Exception as e:
+            logger.error("Failed to save event alert to DB: %s", e)

reports/live_stress_latest.json

@@ -0,0 +1,622 @@
+{
+  "started_at": "2026-03-01T19:28:11",
+  "base_url": "http://127.0.0.1:8000",
+  "config": {
+    "cycles": 20,
+    "request_timeout": 8.0,
+    "sleep_ms": 120,
+    "pause_every": 3,
+    "pause_sleep_ms": 120,
+    "source_a": "0",
+    "source_b": "2"
+  },
+  "summary": {
+    "total": 75,
+    "ok": 75,
+    "err": 0,
+    "avg_ms": 286.4,
+    "max_ms": 1051,
+    "over_500ms": 16,
+    "over_1000ms": 12
+  },
+  "errors": [],
+  "tail": [
+    {
+      "name": "cycle18-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle18-resume",
+      "ok": true,
+      "duration_ms": 357,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-switch-0",
+      "ok": true,
+      "duration_ms": 1001,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-after",
+      "ok": true,
+      "duration_ms": 18,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-switch-2",
+      "ok": true,
+      "duration_ms": 401,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "final-status",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    }
+  ],
+  "all_results": [
+    {
+      "name": "warmup-status",
+      "ok": true,
+      "duration_ms": 23,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "warmup-start",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"fail\",\"message\":\"Pipeline already running\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-switch-0",
+      "ok": true,
+      "duration_ms": 275,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-switch-2",
+      "ok": true,
+      "duration_ms": 520,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-status-after",
+      "ok": true,
+      "duration_ms": 24,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-switch-0",
+      "ok": true,
+      "duration_ms": 1051,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle3-resume",
+      "ok": true,
+      "duration_ms": 151,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-switch-2",
+      "ok": true,
+      "duration_ms": 494,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-switch-0",
+      "ok": true,
+      "duration_ms": 1024,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-switch-2",
+      "ok": true,
+      "duration_ms": 451,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-pause",
+      "ok": true,
+      "duration_ms": 1006,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle6-resume",
+      "ok": true,
+      "duration_ms": 341,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-status-before",
+      "ok": true,
+      "duration_ms": 25,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-switch-0",
+      "ok": true,
+      "duration_ms": 998,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-status-after",
+      "ok": true,
+      "duration_ms": 21,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-switch-2",
+      "ok": true,
+      "duration_ms": 424,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-status-before",
+      "ok": true,
+      "duration_ms": 17,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-switch-0",
+      "ok": true,
+      "duration_ms": 1034,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-pause",
+      "ok": true,
+      "duration_ms": 1013,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle9-resume",
+      "ok": true,
+      "duration_ms": 150,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-status-after",
+      "ok": true,
+      "duration_ms": 17,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-status-before",
+      "ok": true,
+      "duration_ms": 16,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-switch-2",
+      "ok": true,
+      "duration_ms": 389,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-switch-0",
+      "ok": true,
+      "duration_ms": 1013,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-status-after",
+      "ok": true,
+      "duration_ms": 21,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-switch-2",
+      "ok": true,
+      "duration_ms": 461,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-pause",
+      "ok": true,
+      "duration_ms": 1004,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle12-resume",
+      "ok": true,
+      "duration_ms": 340,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-switch-0",
+      "ok": true,
+      "duration_ms": 999,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-status-after",
+      "ok": true,
+      "duration_ms": 14,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-switch-2",
+      "ok": true,
+      "duration_ms": 400,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-switch-0",
+      "ok": true,
+      "duration_ms": 1017,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-pause",
+      "ok": true,
+      "duration_ms": 1004,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle15-resume",
+      "ok": true,
+      "duration_ms": 130,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-status-after",
+      "ok": true,
+      "duration_ms": 22,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-switch-2",
+      "ok": true,
+      "duration_ms": 382,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-status-after",
+      "ok": true,
+      "duration_ms": 26,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-switch-0",
+      "ok": true,
+      "duration_ms": 977,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-status-after",
+      "ok": true,
+      "duration_ms": 14,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-before",
+      "ok": true,
+      "duration_ms": 15,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-switch-2",
+      "ok": true,
+      "duration_ms": 353,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle18-resume",
+      "ok": true,
+      "duration_ms": 357,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-switch-0",
+      "ok": true,
+      "duration_ms": 1001,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-after",
+      "ok": true,
+      "duration_ms": 18,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-switch-2",
+      "ok": true,
+      "duration_ms": 401,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "final-status",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    }
+  ]
+}

scripts/live_stress_test.py

@@ -0,0 +1,265 @@
+"""Stress-test live control endpoints for the smoking detection server.
+
+Usage example:
+    python scripts/live_stress_test.py --cycles 20 --request-timeout 8 --sleep-ms 120
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+
+@dataclass
+class StepResult:
+    name: str
+    ok: bool
+    duration_ms: int
+    status_code: int | None
+    body: str
+
+
+def _call(
+    base_url: str,
+    name: str,
+    method: str,
+    path: str,
+    timeout_seconds: float,
+    form: dict[str, Any] | None = None,
+) -> StepResult:
+    url = f"{base_url}{path}"
+    data = None
+    headers: dict[str, str] = {}
+
+    if form is not None:
+        data = urlencode(form).encode("utf-8")
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+
+    request = Request(url=url, data=data, method=method, headers=headers)
+
+    start = time.perf_counter()
+    try:
+        with urlopen(request, timeout=timeout_seconds) as response:
+            body = response.read().decode("utf-8", errors="replace")
+            status = getattr(response, "status", None)
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name, ok=True, duration_ms=duration_ms, status_code=status, body=body
+        )
+    except HTTPError as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        try:
+            body = err.read().decode("utf-8", errors="replace")
+        except Exception:
+            body = str(err)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=err.code,
+            body=body,
+        )
+    except URLError as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=None,
+            body=str(err.reason),
+        )
+    except Exception as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=None,
+            body=str(err),
+        )
+
+
+def _summarize(results: list[StepResult]) -> dict[str, Any]:
+    ok_results = [r for r in results if r.ok]
+    err_results = [r for r in results if not r.ok]
+    durations = [r.duration_ms for r in ok_results]
+
+    return {
+        "total": len(results),
+        "ok": len(ok_results),
+        "err": len(err_results),
+        "avg_ms": round(sum(durations) / len(durations), 1) if durations else 0,
+        "max_ms": max(durations) if durations else 0,
+        "over_500ms": sum(1 for d in durations if d > 500),
+        "over_1000ms": sum(1 for d in durations if d > 1000),
+    }
+
+
+def run_stress(args: argparse.Namespace) -> tuple[list[StepResult], dict[str, Any]]:
+    results: list[StepResult] = []
+
+    def add(result: StepResult) -> None:
+        results.append(result)
+        if args.verbose or not result.ok:
+            status = "OK" if result.ok else "ERR"
+            print(
+                f"{status}|{result.name}|{result.duration_ms}ms|"
+                f"status={result.status_code}|{result.body[:180]}"
+            )
+
+    add(
+        _call(
+            args.base_url, "warmup-status", "GET", "/live/status", args.request_timeout
+        )
+    )
+    add(
+        _call(
+            args.base_url,
+            "warmup-start",
+            "POST",
+            "/live/start",
+            args.request_timeout,
+            {"source": args.source_b},
+        )
+    )
+
+    for cycle in range(1, args.cycles + 1):
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-status-before",
+                "GET",
+                "/live/status",
+                args.request_timeout,
+            )
+        )
+
+        src = args.source_b if cycle % 2 == 0 else args.source_a
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-switch-{src}",
+                "POST",
+                "/live/switch_source",
+                args.request_timeout,
+                {"source": src},
+            )
+        )
+
+        if args.pause_every > 0 and cycle % args.pause_every == 0:
+            add(
+                _call(
+                    args.base_url,
+                    f"cycle{cycle}-pause",
+                    "POST",
+                    "/live/pause",
+                    args.request_timeout,
+                )
+            )
+            time.sleep(args.pause_sleep_ms / 1000.0)
+            add(
+                _call(
+                    args.base_url,
+                    f"cycle{cycle}-resume",
+                    "POST",
+                    "/live/resume",
+                    args.request_timeout,
+                )
+            )
+
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-status-after",
+                "GET",
+                "/live/status",
+                args.request_timeout,
+            )
+        )
+
+        time.sleep(args.sleep_ms / 1000.0)
+
+    add(
+        _call(
+            args.base_url, "final-status", "GET", "/live/status", args.request_timeout
+        )
+    )
+
+    summary = _summarize(results)
+    return results, summary
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Stress test /live/* endpoints")
+    parser.add_argument("--base-url", default="http://127.0.0.1:8000")
+    parser.add_argument("--cycles", type=int, default=20)
+    parser.add_argument("--request-timeout", type=float, default=8.0)
+    parser.add_argument("--sleep-ms", type=int, default=120)
+    parser.add_argument("--pause-every", type=int, default=3)
+    parser.add_argument("--pause-sleep-ms", type=int, default=120)
+    parser.add_argument("--source-a", default="0")
+    parser.add_argument("--source-b", default="2")
+    parser.add_argument("--out", default="reports/live_stress_latest.json")
+    parser.add_argument("--verbose", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    started_at = datetime.now().isoformat(timespec="seconds")
+
+    results, summary = run_stress(args)
+
+    report = {
+        "started_at": started_at,
+        "base_url": args.base_url,
+        "config": {
+            "cycles": args.cycles,
+            "request_timeout": args.request_timeout,
+            "sleep_ms": args.sleep_ms,
+            "pause_every": args.pause_every,
+            "pause_sleep_ms": args.pause_sleep_ms,
+            "source_a": args.source_a,
+            "source_b": args.source_b,
+        },
+        "summary": summary,
+        "errors": [asdict(r) for r in results if not r.ok],
+        "tail": [asdict(r) for r in results[-10:]],
+        "all_results": [asdict(r) for r in results],
+    }
+
+    out_path = Path(args.out)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    out_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
+
+    print(
+        "SUMMARY|"
+        f"total={summary['total']}|ok={summary['ok']}|err={summary['err']}|"
+        f"avg_ms={summary['avg_ms']}|max_ms={summary['max_ms']}|"
+        f"over_500ms={summary['over_500ms']}|over_1000ms={summary['over_1000ms']}"
+    )
+    print(f"REPORT|{out_path.as_posix()}")
+
+    if summary["err"] > 0:
+        print("ERRORS_BEGIN")
+        for err in report["errors"][:8]:
+            print(
+                f"ERR|{err['name']}|{err['duration_ms']}ms|"
+                f"status={err['status_code']}|{err['body'][:180]}"
+            )
+        print("ERRORS_END")
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server.py

@@ -15,122 +15,25 @@ Usage:
 import argparse
 import logging
 import os
-import socket
-import threading
-import time
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-import cv2
+from fastapi import FastAPI
-from fastapi import FastAPI, Form, Response
-from fastapi.responses import StreamingResponse
 from fastapi.staticfiles import StaticFiles
 
-from db import SQLiteRepository
+import server.runtime as rt
-from db.repository import Repository
 from pipeline.config import PipelineConfig
 from pipeline.core import SmokingDetectionPipeline
 from pipeline.models import YoloCigaretteDetector, YoloClassifier, YoloPersonDetector
 from pipeline.protocols import Sink
-from pipeline.sinks import FileAlertStore, MetricsSink, SmartEventSink
+from pipeline.sinks import FileAlertStore, MetricsSink
+from pipeline.sinks.database import DatabaseSmartEventSink
 from pipeline.sources.stream import ThreadedStreamSource
 from pipeline.staged_core import StageBasedPipeline
-from pipeline.types import Frame
+from server.routes import router as api_router
 
 logger = logging.getLogger(__name__)
 
-# ---------------------------------------------------------------------------
-# Global state (set during lifespan)
-# ---------------------------------------------------------------------------
-db: Repository = SQLiteRepository()  # default: data/smoke.db
-pipeline = None  # set at startup
-source: ThreadedStreamSource | None = None  # set at startup
-live_frame: Frame | None = None  # latest frame for MJPEG / snapshot
-
-
-# ---------------------------------------------------------------------------
-# Sink that exposes frames to the web layer
-# ---------------------------------------------------------------------------
-class LiveFrameSink(Sink):
-    """Captures the latest frame so API endpoints can stream it."""
-
-    def handle(self, frame: Frame):
-        global live_frame
-        live_frame = frame
-
-    def close(self):
-        pass
-
-
-class DatabaseSmartEventSink(SmartEventSink):
-    """Records event clips and stores clip path in DB as alert evidence."""
-
-    def __init__(
-        self,
-        db_repo: Repository,
-        fps: float,
-        user_id: int = 1,
-        camera_source: str = "Pipeline",
-        output_dir: str = "static/events",
-    ):
-        super().__init__(output_dir=output_dir, fps=fps)
-        self.db_repo = db_repo
-        self.user_id = user_id
-        self.camera_source = camera_source
-
-    def _trigger_event(self, trigger_frame: Frame):
-        was_recording = self.recording
-        super()._trigger_event(trigger_frame)
-
-        # Save one DB row per newly started event clip
-        if was_recording or not self.recording or self.current_event_id is None:
-            return
-
-        ticket = trigger_frame.ticket
-        if not ticket:
-            return
-
-        filename = f"event_{self.current_event_id}.mp4"
-        web_path = f"/static/events/{filename}"
-        try:
-            self.db_repo.save_detection(
-                user_id=self.user_id,
-                image_path=web_path,
-                confidence=ticket.confidence,
-                camera_source=self.camera_source,
-            )
-            logger.info("Saved event alert to DB: %s", web_path)
-        except Exception as e:
-            logger.error("Failed to save event alert to DB: %s", e)
-
-
-def _get_lan_address() -> tuple[str, str]:
-    """Return (ip, hostname) for the primary LAN interface."""
-    ip = "127.0.0.1"
-    try:
-        # Open a UDP socket to an external IP (doesn't actually send anything)
-        # to figure out which local NIC the OS would route through.
-        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        s.connect(("8.8.8.8", 80))
-        ip = s.getsockname()[0]
-        s.close()
-    except Exception:
-        pass
-    hostname = socket.getfqdn()
-    return ip, hostname
-
-
-# ---------------------------------------------------------------------------
-# Pipeline runner (background thread)
-# ---------------------------------------------------------------------------
-def _run_pipeline():
-    """Blocking call: runs in a daemon thread."""
-    assert pipeline is not None and source is not None
-    try:
-        pipeline.process_stream(source)
-    except Exception:  # noqa: BLE001
-        logger.exception("Pipeline crashed")
-
 
 # ---------------------------------------------------------------------------
 # CLI args  (parsed before uvicorn when run as __main__)
@@ -174,8 +77,6 @@ _args = argparse.Namespace(
 # ---------------------------------------------------------------------------
 @asynccontextmanager
 async def app_lifespan(_app: FastAPI):
-    global pipeline, source  # noqa: PLW0603
-
     logging.basicConfig(
         level=logging.DEBUG if _args.debug else logging.INFO,
         format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
@@ -185,7 +86,7 @@ async def app_lifespan(_app: FastAPI):
 
     # 1. Database
     logger.info("Initializing database...")
-    db.initialize()
+    rt.db.initialize()
 
     # 2. Models
     logger.info("Loading models...")
@@ -194,32 +95,33 @@ async def app_lifespan(_app: FastAPI):
     verifier = YoloClassifier(_args.verify_model)
     logger.info("Models loaded.")
 
-    # 3. Source (webcam / RTSP)
+    # 3. Configure live pipeline manager
-    cam = int(_args.source) if _args.source.isdigit() else _args.source
+    def source_factory(source_spec: str) -> ThreadedStreamSource:
-    source = ThreadedStreamSource(cam)
+        cam = int(source_spec) if source_spec.isdigit() else source_spec
-    logger.info("Camera source ready: %s (%.1f FPS)", _args.source, source.fps)
+        return ThreadedStreamSource(cam)
 
-    # 4. Sinks
+    def sink_factory(src: ThreadedStreamSource) -> list[Sink]:
-    sinks: list[Sink] = [
+        return [
-        LiveFrameSink(),
+            rt.LiveFrameSink(),
-        DatabaseSmartEventSink(db, fps=source.fps, output_dir="static/events"),
+            DatabaseSmartEventSink(rt.db, fps=src.fps, output_dir="static/events"),
             MetricsSink(log_interval=200),
             FileAlertStore("alerts.jsonl"),
         ]
 
-    # 5. Pipeline config
+    def pipeline_factory(src: ThreadedStreamSource, sinks: list[Sink]):
         config = PipelineConfig(
             idle_interval=0.5,
             active_interval=0.0,
-        cooldown_frames=int(source.fps * 2),
+            cooldown_frames=int(src.fps * 2),
         )
 
-    # 6. Build pipeline
         if _args.staged or _args.vlm:
             vlm_key = _args.vlm_key or os.environ.get(
-            "OPENAI_API_KEY" if _args.vlm_provider == "openai" else "ANTHROPIC_API_KEY"
+                "OPENAI_API_KEY"
+                if _args.vlm_provider == "openai"
+                else "ANTHROPIC_API_KEY"
             )
-        pipeline = StageBasedPipeline(
+            return StageBasedPipeline(
                 person_model=person_model,
                 cigarette_model=cig_model,
                 verifier_model=verifier,
@@ -230,8 +132,8 @@ async def app_lifespan(_app: FastAPI):
                 sinks=sinks,
                 config=config,
             )
-    else:
+
-        pipeline = SmokingDetectionPipeline(
+        return SmokingDetectionPipeline(
             person_model=person_model,
             cigarette_model=cig_model,
             verifier_model=verifier,
@@ -239,13 +141,19 @@ async def app_lifespan(_app: FastAPI):
             config=config,
         )
 
-    # 7. Launch pipeline in background thread
+    rt.manager.configure(
-    t = threading.Thread(target=_run_pipeline, daemon=True, name="pipeline")
+        source_factory=source_factory,
-    t.start()
+        sink_factory=sink_factory,
-    logger.info("Pipeline running in background.")
+        pipeline_factory=pipeline_factory,
+        initial_source=_args.source,
+    )
+    ok, msg = rt.manager.start_async()
+    if not ok:
+        raise RuntimeError(msg)
+    logger.info("%s (source=%s)", msg, _args.source)
 
     # Show real addresses
-    ip, hostname = _get_lan_address()
+    ip, hostname = rt.get_lan_address()
     port = _args.port if hasattr(_args, "port") else 8000
     logger.info("=" * 50)
     logger.info("  Server ready")
@@ -259,10 +167,8 @@ async def app_lifespan(_app: FastAPI):
 
     # Shutdown
     logger.info("Shutting down...")
-    source.stop()
+    rt.manager.stop()
-    for s in sinks:
+    rt.db.close()
-        s.close()
-    db.close()
     logger.info("Goodbye.")
 
 
@@ -270,6 +176,7 @@ async def app_lifespan(_app: FastAPI):
 # FastAPI app
 # ---------------------------------------------------------------------------
 app = FastAPI(title="Smoking Detection", lifespan=app_lifespan)  # type: ignore
+app.include_router(api_router)
 
 # Serve static assets (alert images, CSS/JS, etc.)
 Path("static/alerts").mkdir(parents=True, exist_ok=True)
@@ -277,104 +184,6 @@ Path("static/events").mkdir(parents=True, exist_ok=True)
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
 
-# -- Auth ------------------------------------------------------------------
-
-
-@app.get("/ping")
-def ping():
-    """Health check for the mobile app to verify connectivity."""
-    return {"status": "ok", "name": "SmokingDetection"}
-
-
-@app.post("/register")
-def register(username: str = Form(...), password: str = Form(...)):
-    try:
-        if db.get_user_by_username(username):
-            return {"status": "fail", "message": "Username already taken"}
-        new_id = db.create_user(username, password)
-        return {"status": "success", "user_id": new_id, "message": "Account created"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-@app.post("/login")
-def login(username: str = Form(...), password: str = Form(...)):
-    try:
-        user = db.get_user_by_credentials(username, password)
-        if user:
-            return {"status": "success", "user_id": user.id, "username": user.username}
-        return {"status": "fail", "message": "Invalid username or password"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-# -- Detection history -----------------------------------------------------
-
-
-@app.get("/history")
-def get_history(user_id: int | None = None):
-    try:
-        records = db.get_detections(user_id)
-        return [
-            {
-                "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:
-        return []
-
-
-@app.post("/clear_history")
-def clear_history():
-    try:
-        deleted = db.clear_detections()
-        logger.info("Cleared %d records", deleted)
-        return {"status": "success", "message": "History cleared"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-# -- Video / Snapshot ------------------------------------------------------
-
-
-@app.get("/snapshot")
-def snapshot():
-    """Return the latest frame as a JPEG."""
-    if live_frame is not None:
-        _, jpg = cv2.imencode(".jpg", live_frame.data)
-        return Response(content=jpg.tobytes(), media_type="image/jpeg")
-    return Response(status_code=404)
-
-
-@app.get("/video_feed")
-def video_feed():
-    """MJPEG stream - drop into an <img> tag."""
-
-    def generate():
-        while True:
-            if live_frame is not None:
-                from pipeline.ops.drawing import draw_detections
-
-                img = live_frame.data.copy()
-                draw_detections(img, live_frame.detections)
-                _, jpg = cv2.imencode(".jpg", img)
-                yield (
-                    b"--frame\r\n"
-                    b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n"
-                )
-            time.sleep(0.033)  # ~30 FPS cap
-
-    return StreamingResponse(
-        generate(), media_type="multipart/x-mixed-replace;boundary=frame"
-    )
-
-
 # ---------------------------------------------------------------------------
 # Entry-point
 # ---------------------------------------------------------------------------

server/__main__.py

@@ -0,0 +1,9 @@
+"""Package entrypoint so `python -m server` / `uv run -m server` works."""
+
+import runpy
+from pathlib import Path
+
+# Execute top-level server.py as __main__
+runpy.run_path(
+    str(Path(__file__).resolve().parent.parent / "server.py"), run_name="__main__"
+)

server/routes.py

@@ -0,0 +1,165 @@
+"""API route definitions for the smoking detection server."""
+
+import time
+
+import cv2
+from fastapi import APIRouter, Form, Response
+from fastapi.responses import StreamingResponse
+
+import server.runtime as rt
+
+router = APIRouter()
+
+
+@router.get("/ping")
+def ping():
+    """Health check for the mobile app to verify connectivity."""
+    return {"status": "ok", "name": "SmokingDetection"}
+
+
+@router.post("/register")
+def register(username: str = Form(...), password: str = Form(...)):
+    try:
+        if rt.db.get_user_by_username(username):
+            return {"status": "fail", "message": "Username already taken"}
+        new_id = rt.db.create_user(username, password)
+        return {"status": "success", "user_id": new_id, "message": "Account created"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.post("/login")
+def login(username: str = Form(...), password: str = Form(...)):
+    try:
+        user = rt.db.get_user_by_credentials(username, password)
+        if user:
+            return {"status": "success", "user_id": user.id, "username": user.username}
+        return {"status": "fail", "message": "Invalid username or password"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.get("/history")
+def get_history(user_id: int | None = None):
+    try:
+        records = rt.db.get_detections(user_id)
+        return [
+            {
+                "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:
+        return []
+
+
+@router.post("/clear_history")
+def clear_history(user_id: int | None = Form(default=None)):
+    try:
+        deleted = rt.db.clear_detections(user_id)
+        return {
+            "status": "success",
+            "message": "History cleared",
+            "deleted": deleted,
+            "user_id": user_id,
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.get("/live/status")
+def live_status():
+    """Get current live pipeline status."""
+    return {"status": "success", "live": rt.manager.status()}
+
+
+@router.post("/live/start")
+def live_start(source: str | None = Form(default=None)):
+    """Start live pipeline (optionally with source override)."""
+    ok, message = rt.manager.start(source)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/stop")
+def live_stop():
+    """Stop live pipeline."""
+    ok, message = rt.manager.stop()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/pause")
+def live_pause():
+    """Pause live pipeline (stops source, resumable)."""
+    ok, message = rt.manager.pause()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/resume")
+def live_resume():
+    """Resume paused live pipeline."""
+    ok, message = rt.manager.resume()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/switch_source")
+def live_switch_source(source: str = Form(...)):
+    """Switch active live source (webcam index or RTSP URL)."""
+    ok, message = rt.manager.switch_source(source)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.get("/snapshot")
+def snapshot():
+    """Return the latest frame as a JPEG."""
+    if rt.live_frame is not None:
+        _, jpg = cv2.imencode(".jpg", rt.live_frame.data)
+        return Response(content=jpg.tobytes(), media_type="image/jpeg")
+    return Response(status_code=404)
+
+
+@router.get("/video_feed")
+def video_feed():
+    """MJPEG stream - drop into an <img> tag."""
+
+    def generate():
+        while True:
+            if rt.live_frame is not None:
+                from pipeline.ops.drawing import draw_detections
+
+                img = rt.live_frame.data.copy()
+                draw_detections(img, rt.live_frame.detections)
+                _, jpg = cv2.imencode(".jpg", img)
+                yield (
+                    b"--frame\r\n"
+                    b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n"
+                )
+            time.sleep(0.033)
+
+    return StreamingResponse(
+        generate(), media_type="multipart/x-mixed-replace;boundary=frame"
+    )

server/runtime.py

@@ -0,0 +1,280 @@
+"""Runtime state and utility helpers for the FastAPI server."""
+
+import logging
+import socket
+import threading
+from typing import Any, Callable
+
+from db import SQLiteRepository
+from db.repository import Repository
+from pipeline.protocols import Sink
+from pipeline.sources.stream import ThreadedStreamSource
+from pipeline.types import Frame
+
+logger = logging.getLogger(__name__)
+
+# Global runtime state
+# (kept as module-level singletons for current single-instance server design)
+db: Repository = SQLiteRepository()
+pipeline = None
+source: ThreadedStreamSource | None = None
+live_frame: Frame | None = None
+
+
+SourceFactory = Callable[[str], ThreadedStreamSource]
+SinkFactory = Callable[[ThreadedStreamSource], list[Sink]]
+PipelineFactory = Callable[[ThreadedStreamSource, list[Sink]], Any]
+
+
+class LiveFrameSink(Sink):
+    """Captures the latest frame so API endpoints can stream it."""
+
+    def handle(self, frame: Frame):
+        global live_frame
+        live_frame = frame
+
+    def close(self):
+        pass
+
+
+class PipelineManager:
+    """Controls the live pipeline lifecycle (start/stop/pause/resume/switch)."""
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._thread: threading.Thread | None = None
+        self._start_thread: threading.Thread | None = None
+        self._sinks: list[Sink] = []
+
+        self._source_factory: SourceFactory | None = None
+        self._sink_factory: SinkFactory | None = None
+        self._pipeline_factory: PipelineFactory | None = None
+
+        self._source_spec: str = "0"
+        self.running = False
+        self.starting = False
+        self.paused = False
+        self.last_error: str | None = None
+
+    def configure(
+        self,
+        source_factory: SourceFactory,
+        sink_factory: SinkFactory,
+        pipeline_factory: PipelineFactory,
+        initial_source: str = "0",
+    ):
+        with self._lock:
+            self._source_factory = source_factory
+            self._sink_factory = sink_factory
+            self._pipeline_factory = pipeline_factory
+            self._source_spec = initial_source
+
+    def _close_sinks(self):
+        for sink in self._sinks:
+            try:
+                sink.close()
+            except Exception:
+                logger.exception("Error closing sink: %s", sink.__class__.__name__)
+        self._sinks = []
+
+    def _run_wrapper(self):
+        global pipeline, source
+        assert pipeline is not None and source is not None
+        try:
+            pipeline.process_stream(source)
+        except Exception as e:
+            with self._lock:
+                self.last_error = str(e)
+            logger.exception("Pipeline crashed")
+        finally:
+            with self._lock:
+                self.running = False
+                # Only auto-close sinks when pipeline exits by itself
+                self._close_sinks()
+                pipeline = None
+                source = None
+
+    def start(self, source_spec: str | None = None) -> tuple[bool, str]:
+        global pipeline, source, live_frame
+        with self._lock:
+            if self.starting:
+                return False, "Pipeline is starting"
+
+            if self.running:
+                return False, "Pipeline already running"
+
+            if source_spec is not None:
+                self._source_spec = source_spec
+
+            if (
+                not self._source_factory
+                or not self._sink_factory
+                or not self._pipeline_factory
+            ):
+                return False, "PipelineManager not configured"
+
+            source_spec_local = self._source_spec
+            source_factory = self._source_factory
+            sink_factory = self._sink_factory
+            pipeline_factory = self._pipeline_factory
+            self.starting = True
+
+        try:
+            source_obj = source_factory(source_spec_local)
+            sinks_obj = sink_factory(source_obj)
+            pipeline_obj = pipeline_factory(source_obj, sinks_obj)
+        except Exception as e:
+            with self._lock:
+                self.starting = False
+                self.last_error = str(e)
+            return False, f"Failed to build pipeline: {e}"
+
+        with self._lock:
+            if self.running:
+                self.starting = False
+                try:
+                    source_obj.stop()
+                except Exception:
+                    logger.exception("Error stopping source after race")
+                return False, "Pipeline already running"
+
+            source = source_obj
+            self._sinks = sinks_obj
+            pipeline = pipeline_obj
+            live_frame = None
+
+            self._thread = threading.Thread(
+                target=self._run_wrapper,
+                daemon=True,
+                name="pipeline",
+            )
+            self._thread.start()
+            self.starting = False
+            self.running = True
+            self.paused = False
+            self.last_error = None
+            return True, f"Pipeline started (source={self._source_spec})"
+
+    def start_async(self, source_spec: str | None = None) -> tuple[bool, str]:
+        with self._lock:
+            if self.starting:
+                return False, "Pipeline is starting"
+            if self.running:
+                return False, "Pipeline already running"
+
+            def _runner():
+                ok, msg = self.start(source_spec)
+                if ok:
+                    logger.info(msg)
+                else:
+                    logger.warning(msg)
+
+            self._start_thread = threading.Thread(
+                target=_runner,
+                daemon=True,
+                name="pipeline-start",
+            )
+            self._start_thread.start()
+            return True, "Pipeline start requested"
+
+    def stop(self) -> tuple[bool, str]:
+        global pipeline, source
+        with self._lock:
+            thread = self._thread
+            source_obj = source
+
+        if source_obj is not None:
+            try:
+                source_obj.stop()
+            except Exception:
+                logger.exception("Error stopping source")
+
+        if thread and thread.is_alive():
+            thread.join(timeout=3.0)
+
+        with self._lock:
+            self.starting = False
+            self.running = False
+            self.paused = False
+            self._close_sinks()
+            pipeline = None
+            source = None
+            return True, "Pipeline stopped"
+
+    def pause(self) -> tuple[bool, str]:
+        with self._lock:
+            if not self.running:
+                return False, "Pipeline is not running"
+
+        ok, msg = self.stop()
+        if ok:
+            with self._lock:
+                self.paused = True
+            return True, "Pipeline paused"
+        return False, msg
+
+    def resume(self) -> tuple[bool, str]:
+        with self._lock:
+            if self.running:
+                return False, "Pipeline already running"
+            if not self.paused:
+                return False, "Pipeline is not paused"
+
+        ok, msg = self.start()
+        if ok:
+            with self._lock:
+                self.paused = False
+            return True, "Pipeline resumed"
+        return False, msg
+
+    def switch_source(self, source_spec: str) -> tuple[bool, str]:
+        with self._lock:
+            self._source_spec = source_spec
+            running = self.running
+            paused = self.paused
+
+        if running:
+            self.stop()
+            return self.start(source_spec)
+
+        if paused:
+            return True, f"Source set to {source_spec} (will apply on resume)"
+
+        return self.start(source_spec)
+
+    def status(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "running": self.running,
+                "starting": self.starting,
+                "paused": self.paused,
+                "source": self._source_spec,
+                "last_error": self.last_error,
+                "thread_alive": bool(self._thread and self._thread.is_alive()),
+            }
+
+
+manager = PipelineManager()
+
+
+def get_lan_address() -> tuple[str, str]:
+    """Return (ip, hostname) for the primary LAN interface."""
+    ip = "127.0.0.1"
+    try:
+        probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        probe.connect(("8.8.8.8", 80))
+        ip = probe.getsockname()[0]
+        probe.close()
+    except Exception:
+        pass
+    hostname = socket.getfqdn()
+    return ip, hostname
+
+
+def run_pipeline():
+    """Blocking call used by the background pipeline thread."""
+    assert pipeline is not None and source is not None
+    try:
+        pipeline.process_stream(source)
+    except Exception:
+        logger.exception("Pipeline crashed")