why

.gitignore

@@ -16,7 +16,7 @@ wheels/
 alerts.jsonl
 output.jsonl
 output.mp4
-static/alerts/
+static/
 screenshots/
 events/
 logs/

README.md

@@ -63,6 +63,9 @@ The server runs the detection pipeline in the background while serving a REST AP
 # Default: webcam 0, classic pipeline
 uv run server.py
 
+# Package entrypoint (same server)
+uv run -m server
+
 # Use stage-based pipeline with a specific camera
 uv run server.py --staged --source 2
 
@@ -76,6 +79,14 @@ uv run server.py --host 0.0.0.0 --port 9000
 uvicorn server:app --host 0.0.0.0 --port 8000
 ```
 
+### Live + Clip Control (Server Runtime)
+
+The server now runs a lifecycle manager for the live pipeline and a queued clip-job worker:
+
+- **Live manager**: start/stop/pause/resume/switch source at runtime.
+- **Clip jobs**: submit files to a queue, process one-at-a-time, track status.
+- **Resource policy**: clip jobs can temporarily pause live processing and resume after job completion.
+
 On startup the server prints the LAN address for the mobile app:
 
 ```
@@ -122,7 +133,7 @@ uv run main.py --source videos/test.mp4 --debug
 ### Check Available Webcams
 
 ```bash
-uv run -m tools.webcam_check
+uv run -m scripts.webcam_check
 ```
 
 ### CLI Arguments
@@ -181,6 +192,10 @@ The system generates multiple artifacts during execution:
 ```text
 gp/
 ├── server.py                # FastAPI server (runs pipeline + REST API)
+├── server/                  # Modular server runtime + routes
+│   ├── runtime.py           # PipelineManager + ClipJobManager
+│   ├── routes.py            # API routers (/live/*, /clips/*, auth/history)
+│   └── __main__.py          # `uv run -m server` entrypoint
 ├── main.py                  # CLI entry point (standalone pipeline)
 ├── pyproject.toml           # Project config (uv/pip)
 ├── models/                  # Directory for .pt model weights
@@ -228,8 +243,9 @@ gp/
 │       ├── geometry.py      # Box expansion, zone filtering
 │       └── drawing.py       # Visualization helpers
 │
-├── tools/
+├── scripts/
-│   └── webcam_check.py      # Utility to list available webcams
+│   ├── webcam_check.py      # Utility to list available webcams
+│   └── live_stress_test.py  # Stress test for /live/* and queue responsiveness
 │
 └── SmokeDetectionApp/       # Android companion app
     └── app/src/main/java/com/example/smokedetection/
@@ -288,18 +304,57 @@ pipeline = SmokingDetectionPipeline(..., config=config)
 
 The server (`server.py`) exposes these endpoints:
 
-| Method | Endpoint         | Description                         |
+| Method | Endpoint              | Description                                     |
-| :----- | :--------------- | :---------------------------------- |
+| :----- | :-------------------- | :---------------------------------------------- |
-| GET    | `/ping`          | Health check (verify connectivity)  |
+| GET    | `/ping`               | Health check (verify connectivity)              |
-| POST   | `/register`      | Create a new user account           |
+| POST   | `/register`           | Create a new user account                       |
-| POST   | `/login`         | Authenticate and get user ID        |
+| POST   | `/login`              | Authenticate and get user ID                    |
-| GET    | `/history`       | Get detection records for a user    |
+| GET    | `/history`            | Get detection records (optionally by `user_id`) |
-| POST   | `/clear_history` | Delete all records for a user       |
+| POST   | `/clear_history`      | Delete detection records (optional `user_id`)   |
-| GET    | `/snapshot`      | Latest camera frame as JPEG         |
+| GET    | `/snapshot`           | Latest camera frame as JPEG                     |
-| GET    | `/video_feed`    | MJPEG stream with detection overlay |
+| GET    | `/video_feed`         | MJPEG stream with detection overlay             |
+| GET    | `/live/status`        | Live pipeline state (`running/paused/source`)   |
+| POST   | `/live/start`         | Start live pipeline (optional `source`)         |
+| POST   | `/live/stop`          | Stop live pipeline                              |
+| POST   | `/live/pause`         | Pause live pipeline                             |
+| POST   | `/live/resume`        | Resume paused live pipeline                     |
+| POST   | `/live/switch_source` | Switch active webcam/RTSP source                |
+| POST   | `/clips/submit`       | Queue a clip job (upload file or local path)    |
+| GET    | `/clips`              | List recent clip jobs + queue stats             |
+| GET    | `/clips/{job_id}`     | Get status/details for a clip job               |
 
 Interactive docs at **<http://localhost:8000/docs>** (Swagger UI).
 
+### Quick Test Calls
+
+PowerShell examples:
+
+```powershell
+$base='http://127.0.0.1:8000'
+
+# Live controls
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/live/status')
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/switch_source') -Body @{source='2'}
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/pause')
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/resume')
+
+# Queue clip from local path
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/clips/submit') -Body @{path='test_media/139474-772542450_medium.mp4'}
+
+# List jobs and inspect one job
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/clips')
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/clips/<job_id>')
+```
+
+curl examples:
+
+```bash
+curl http://127.0.0.1:8000/live/status
+curl -X POST -F "source=2" http://127.0.0.1:8000/live/switch_source
+curl -X POST -F "path=test_media/139474-772542450_medium.mp4" http://127.0.0.1:8000/clips/submit
+curl http://127.0.0.1:8000/clips
+```
+
 ## 📱 Android App
 
 The companion app (`SmokeDetectionApp/`) connects to the server for auth, alert history, and live camera feed.

server.py

@@ -28,6 +28,8 @@ from pipeline.models import YoloCigaretteDetector, YoloClassifier, YoloPersonDet
 from pipeline.protocols import Sink
 from pipeline.sinks import FileAlertStore, MetricsSink
 from pipeline.sinks.database import DatabaseSmartEventSink
+from pipeline.sinks.video import VideoFileStore
+from pipeline.sources.file import FileStreamSource
 from pipeline.sources.stream import ThreadedStreamSource
 from pipeline.staged_core import StageBasedPipeline
 from server.routes import router as api_router
@@ -147,6 +149,65 @@ async def app_lifespan(_app: FastAPI):
         pipeline_factory=pipeline_factory,
         initial_source=_args.source,
     )
+
+    def clip_source_factory(path: str) -> FileStreamSource:
+        return FileStreamSource(path)
+
+    def clip_sink_factory(src: FileStreamSource, job_id: str) -> list[Sink]:
+        job_dir = Path("static/clip_jobs") / job_id
+        events_dir = job_dir / "events"
+        job_dir.mkdir(parents=True, exist_ok=True)
+        events_dir.mkdir(parents=True, exist_ok=True)
+        return [
+            VideoFileStore(str(job_dir / "processed.mp4"), fps=src.fps),
+            DatabaseSmartEventSink(
+                rt.db,
+                fps=src.fps,
+                camera_source=f"ClipJob:{job_id}",
+                output_dir=str(events_dir),
+            ),
+            MetricsSink(log_interval=200),
+            FileAlertStore(str(job_dir / "alerts.jsonl")),
+        ]
+
+    def clip_pipeline_factory(src: FileStreamSource, sinks: list[Sink]):
+        config = PipelineConfig(
+            idle_interval=0.5,
+            active_interval=0.0,
+            cooldown_frames=max(1, int(src.fps * 2)),
+        )
+
+        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"
+            )
+            return StageBasedPipeline(
+                person_model=person_model,
+                cigarette_model=cig_model,
+                verifier_model=verifier,
+                vlm_api_key=vlm_key if _args.vlm else None,
+                vlm_provider=_args.vlm_provider,
+                vlm_model=_args.vlm_model,
+                use_vlm=_args.vlm,
+                sinks=sinks,
+                config=config,
+            )
+
+        return SmokingDetectionPipeline(
+            person_model=person_model,
+            cigarette_model=cig_model,
+            verifier_model=verifier,
+            sinks=sinks,
+            config=config,
+        )
+
+    rt.clip_jobs.configure(
+        source_factory=clip_source_factory,
+        sink_factory=clip_sink_factory,
+        pipeline_factory=clip_pipeline_factory,
+    )
     ok, msg = rt.manager.start_async()
     if not ok:
         raise RuntimeError(msg)
@@ -168,6 +229,7 @@ async def app_lifespan(_app: FastAPI):
     # Shutdown
     logger.info("Shutting down...")
     rt.manager.stop()
+    rt.clip_jobs.shutdown()
     rt.db.close()
     logger.info("Goodbye.")
 
@@ -181,6 +243,7 @@ app.include_router(api_router)
 # Serve static assets (alert images, CSS/JS, etc.)
 Path("static/alerts").mkdir(parents=True, exist_ok=True)
 Path("static/events").mkdir(parents=True, exist_ok=True)
+Path("static/clip_jobs").mkdir(parents=True, exist_ok=True)
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
 

server/routes.py

@@ -1,9 +1,11 @@
 """API route definitions for the smoking detection server."""
 
+import shutil
 import time
+from pathlib import Path
 
 import cv2
-from fastapi import APIRouter, Form, Response
+from fastapi import APIRouter, File, Form, Response, UploadFile
 from fastapi.responses import StreamingResponse
 
 import server.runtime as rt
@@ -133,6 +135,76 @@ def live_switch_source(source: str = Form(...)):
     }
 
 
+@router.post("/clips/submit")
+async def clips_submit(
+    file: UploadFile | None = File(default=None),
+    path: str | None = Form(default=None),
+):
+    """Queue a clip processing job from uploaded file or existing local path."""
+    if file is None and not path:
+        return {
+            "status": "fail",
+            "message": "Provide either an uploaded file or a local file path",
+        }
+
+    clip_input_path = ""
+    source_name = None
+
+    try:
+        if file is not None:
+            uploads_dir = Path("media/clip_jobs/uploads")
+            uploads_dir.mkdir(parents=True, exist_ok=True)
+
+            source_name = file.filename or "upload.mp4"
+            suffix = Path(source_name).suffix or ".mp4"
+            safe_name = f"clip_{int(time.time() * 1000)}{suffix}"
+            saved_path = uploads_dir / safe_name
+
+            with saved_path.open("wb") as out:
+                shutil.copyfileobj(file.file, out)
+
+            clip_input_path = str(saved_path)
+        else:
+            local_path = Path(path or "")
+            if not local_path.exists() or not local_path.is_file():
+                return {
+                    "status": "fail",
+                    "message": f"File not found: {local_path}",
+                }
+            clip_input_path = str(local_path)
+            source_name = local_path.name
+    finally:
+        if file is not None:
+            await file.close()
+
+    job = rt.clip_jobs.submit(clip_input_path, source_name)
+    return {
+        "status": "success",
+        "message": "Clip job queued",
+        "job": job,
+        "queue": rt.clip_jobs.status(),
+    }
+
+
+@router.get("/clips")
+def clips_list(limit: int = 20):
+    """List recent clip jobs."""
+    return {
+        "status": "success",
+        "queue": rt.clip_jobs.status(),
+        "jobs": rt.clip_jobs.list(limit=max(1, min(limit, 200))),
+    }
+
+
+@router.get("/clips/{job_id}")
+def clips_get(job_id: str):
+    """Get one clip job state."""
+    job = rt.clip_jobs.get(job_id)
+    if not job:
+        return {"status": "fail", "message": "Job not found", "job_id": job_id}
+    return {"status": "success", "job": job}
+
+
 @router.get("/snapshot")
 def snapshot():
     """Return the latest frame as a JPEG."""

server/runtime.py

@@ -3,11 +3,16 @@
 import logging
 import socket
 import threading
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from queue import Empty, Queue
 from typing import Any, Callable
 
 from db import SQLiteRepository
 from db.repository import Repository
 from pipeline.protocols import Sink
+from pipeline.sources.file import FileStreamSource
 from pipeline.sources.stream import ThreadedStreamSource
 from pipeline.types import Frame
 
@@ -25,6 +30,10 @@ SourceFactory = Callable[[str], ThreadedStreamSource]
 SinkFactory = Callable[[ThreadedStreamSource], list[Sink]]
 PipelineFactory = Callable[[ThreadedStreamSource, list[Sink]], Any]
 
+ClipSourceFactory = Callable[[str], FileStreamSource]
+ClipSinkFactory = Callable[[FileStreamSource, str], list[Sink]]
+ClipPipelineFactory = Callable[[FileStreamSource, list[Sink]], Any]
+
 
 class LiveFrameSink(Sink):
     """Captures the latest frame so API endpoints can stream it."""
@@ -257,6 +266,213 @@ class PipelineManager:
 manager = PipelineManager()
 
 
+@dataclass
+class ClipJob:
+    job_id: str
+    input_path: str
+    submitted_at: str
+    source_name: str | None = None
+    status: str = "queued"  # queued | running | done | failed
+    started_at: str | None = None
+    finished_at: str | None = None
+    error: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "job_id": self.job_id,
+            "input_path": self.input_path,
+            "source_name": self.source_name,
+            "status": self.status,
+            "submitted_at": self.submitted_at,
+            "started_at": self.started_at,
+            "finished_at": self.finished_at,
+            "error": self.error,
+        }
+
+
+class ClipJobManager:
+    """Simple queued clip processor with a single worker."""
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._queue: Queue[str] = Queue()
+        self._jobs: dict[str, ClipJob] = {}
+        self._ordered_ids: list[str] = []
+
+        self._source_factory: ClipSourceFactory | None = None
+        self._sink_factory: ClipSinkFactory | None = None
+        self._pipeline_factory: ClipPipelineFactory | None = None
+
+        self._shutdown = False
+        self._worker = threading.Thread(
+            target=self._worker_loop,
+            daemon=True,
+            name="clip-worker",
+        )
+        self._worker.start()
+
+    def configure(
+        self,
+        source_factory: ClipSourceFactory,
+        sink_factory: ClipSinkFactory,
+        pipeline_factory: ClipPipelineFactory,
+    ):
+        with self._lock:
+            self._source_factory = source_factory
+            self._sink_factory = sink_factory
+            self._pipeline_factory = pipeline_factory
+
+    def submit(self, input_path: str, source_name: str | None = None) -> dict[str, Any]:
+        now = datetime.now().isoformat(timespec="seconds")
+        job_id = f"job_{int(time.time() * 1000)}"
+        job = ClipJob(
+            job_id=job_id,
+            input_path=input_path,
+            source_name=source_name,
+            submitted_at=now,
+        )
+
+        with self._lock:
+            self._jobs[job_id] = job
+            self._ordered_ids.append(job_id)
+            self._queue.put(job_id)
+            return job.to_dict()
+
+    def get(self, job_id: str) -> dict[str, Any] | None:
+        with self._lock:
+            job = self._jobs.get(job_id)
+            return job.to_dict() if job else None
+
+    def list(self, limit: int = 20) -> list[dict[str, Any]]:
+        with self._lock:
+            ids = self._ordered_ids[-limit:]
+            ids.reverse()
+            return [self._jobs[job_id].to_dict() for job_id in ids]
+
+    def status(self) -> dict[str, Any]:
+        with self._lock:
+            queued = sum(1 for job in self._jobs.values() if job.status == "queued")
+            running = sum(1 for job in self._jobs.values() if job.status == "running")
+            done = sum(1 for job in self._jobs.values() if job.status == "done")
+            failed = sum(1 for job in self._jobs.values() if job.status == "failed")
+            return {
+                "queued": queued,
+                "running": running,
+                "done": done,
+                "failed": failed,
+                "worker_alive": self._worker.is_alive(),
+            }
+
+    def shutdown(self):
+        with self._lock:
+            self._shutdown = True
+
+        if self._worker.is_alive():
+            self._worker.join(timeout=2.0)
+
+    def _set_job_state(
+        self,
+        job_id: str,
+        status: str,
+        *,
+        started_at: str | None = None,
+        finished_at: str | None = None,
+        error: str | None = None,
+    ):
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if not job:
+                return
+            job.status = status
+            if started_at is not None:
+                job.started_at = started_at
+            if finished_at is not None:
+                job.finished_at = finished_at
+            if error is not None:
+                job.error = error
+
+    def _worker_loop(self):
+        while True:
+            with self._lock:
+                if self._shutdown:
+                    return
+
+            try:
+                job_id = self._queue.get(timeout=0.5)
+            except Empty:
+                continue
+
+            self._process_job(job_id)
+            self._queue.task_done()
+
+    def _process_job(self, job_id: str):
+        with self._lock:
+            job = self._jobs.get(job_id)
+            source_factory = self._source_factory
+            sink_factory = self._sink_factory
+            pipeline_factory = self._pipeline_factory
+
+        if not job:
+            return
+
+        if not source_factory or not sink_factory or not pipeline_factory:
+            self._set_job_state(job_id, "failed", error="ClipJobManager not configured")
+            return
+
+        self._set_job_state(
+            job_id,
+            "running",
+            started_at=datetime.now().isoformat(timespec="seconds"),
+            error=None,
+        )
+
+        paused_live = False
+        try:
+            live_status = manager.status()
+            if live_status.get("running"):
+                ok, _ = manager.pause()
+                paused_live = ok
+
+            src = source_factory(job.input_path)
+            sinks = sink_factory(src, job_id)
+            pipe = pipeline_factory(src, sinks)
+
+            try:
+                pipe.process_stream(src)
+            finally:
+                try:
+                    src.stop()
+                except Exception:
+                    logger.exception("Failed to stop clip source for %s", job_id)
+
+                for sink in sinks:
+                    try:
+                        sink.close()
+                    except Exception:
+                        logger.exception("Failed to close clip sink for %s", job_id)
+
+            self._set_job_state(
+                job_id,
+                "done",
+                finished_at=datetime.now().isoformat(timespec="seconds"),
+            )
+            logger.info("Clip job completed: %s", job_id)
+        except Exception as e:
+            logger.exception("Clip job failed: %s", job_id)
+            self._set_job_state(
+                job_id,
+                "failed",
+                finished_at=datetime.now().isoformat(timespec="seconds"),
+                error=str(e),
+            )
+        finally:
+            if paused_live:
+                manager.resume()
+
+
+clip_jobs = ClipJobManager()
+
+
 def get_lan_address() -> tuple[str, str]:
     """Return (ip, hostname) for the primary LAN interface."""
     ip = "127.0.0.1"