VIDEO NOT IMAGES + history4all

you immediately see what subject materials im reading and where im at

.env.example

@@ -0,0 +1,5 @@
+# what the helly
+DB_USER=smoke_admin
+DB_PASSWORD=123456
+DB_HOST=localhost
+DB_NAME=smoke_detection

.gitattributes

@@ -3,3 +3,4 @@
 *.jsonl filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
+*.db filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -9,4 +9,14 @@ wheels/
 # Virtual environments
 .venv
 
-Group_project_*
+# Local sqlite
+*.db*
+
+# artifacts/logs i forgor what are the paths uhhhhhhh
+alerts.jsonl
+output.jsonl
+output.mp4
+static/alerts/
+screenshots/
+events/
+logs/

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "SmokeDetectionApp"]
+	path = SmokeDetectionApp
+	url = https://github.com/ldlda/SmokeDihtectionApp.git

.vscode/settings.json

@@ -5,6 +5,6 @@
     "**/*.rpymc": true,
     "**/cache/": true
   },
-  "pylint.args": ["--generated-members=cv2.*", "--extension-pkg-whitelist=cv2"],
-//   "editor.fontFamily": "'JetbrainsMono Nerd Font', Consolas, 'Courier New', monospace"
+  "pylint.args": ["--generated-members=cv2.*", "--extension-pkg-whitelist=cv2"]
+  // "editor.fontFamily": "'JetbrainsMono Nerd Font', Consolas, 'Courier New', monospace"
 }

README.md

@@ -7,19 +7,16 @@ A robust, multi-stage computer vision pipeline designed to detect smoking events
 The pipeline operates in a "Fail-Fast" cascade with **early exit** optimization:
 
 1. **Broad Phase (Person Detection)**:
-
    - Detects humans in the frame.
    - _Optimization_: If no people are found, processing stops for that frame.
 
 2. **Narrow Phase (Cigarette Candidate Detection)**:
-
    - Crops the detected person (with configurable margin for context).
    - Runs a specialized YOLO detector to find small objects (cigarettes/lighters).
    - _Early Exit_: If confidence ≥ 0.8, skips verification phase entirely.
    - _Filter_: Rejects candidates smaller than 10x10 pixels.
 
 3. **Verification Phase (Classification)**:
-
    - Crops the candidate object with padding.
    - Runs a dedicated Binary Classifier to verify "Smoking" vs "Not Smoking".
    - _Output_: Only high-confidence events are flagged as alerts.
@@ -34,6 +31,8 @@ The pipeline operates in a "Fail-Fast" cascade with **early exit** optimization:
   - **Active Mode (Max FPS)**: Switches to full speed when activity is detected.
   - **Cooldown**: Stays active for 2 seconds after people leave frame.
 - **Non-Maximum Suppression (NMS)**: Cleans up overlapping boxes per-class.
+- **Weighted Boxes Fusion (WBF)**: Alternative to NMS that merges overlapping boxes for better localization.
+- **Inference Budget Caps**: Limits expensive model calls (e.g., max 3 person crops, max 5 verifications).
 - **Smart Event Recording**: Captures 2s before + 3s after each smoking event as separate clips.
 - **Live Display**: Real-time visualization window with 'q' to quit.
 - **Metrics Tracking**: FPS, latency, detection counts with summary on exit.
@@ -56,7 +55,52 @@ cd gp
 uv sync
 ```
 
-### Running the Pipeline
+### Running the Server
+
+The server runs the detection pipeline in the background while serving a REST API for the mobile app.
+
+```bash
+# Default: webcam 0, classic pipeline
+uv run server.py
+
+# Use stage-based pipeline with a specific camera
+uv run server.py --staged --source 2
+
+# RTSP camera
+uv run server.py --source rtsp://admin:[email protected]:554/stream
+
+# Custom host/port
+uv run server.py --host 0.0.0.0 --port 9000
+
+# Via uvicorn directly (uses env vars for config)
+uvicorn server:app --host 0.0.0.0 --port 8000
+```
+
+On startup the server prints the LAN address for the mobile app:
+
+```
+==================================================
+  Server ready
+  Local:    http://127.0.0.1:8000
+  Network:  http://192.168.1.42:8000
+  Hostname: http://lda.local:8000
+==================================================
+```
+
+Configuration can also be set via environment variables:
+
+| Variable        | Default                    | Description          |
+| :-------------- | :------------------------- | :------------------- |
+| `CAMERA_SOURCE` | `0`                        | Webcam index or RTSP |
+| `PERSON_MODEL`  | `models/human-det-best.pt` | Person model path    |
+| `CIG_MODEL`     | `models/cig-det-best.pt`   | Cigarette model path |
+| `VERIFY_MODEL`  | `models/cig-cls-best.pt`   | Classifier path      |
+| `STAGED`        | `false`                    | Use stage pipeline   |
+| `USE_VLM`       | `false`                    | Enable VLM stage     |
+| `VLM_API_KEY`   |                            | API key for VLM      |
+| `VLM_PROVIDER`  | `openai`                   | `openai`/`anthropic` |
+
+### Running the Pipeline CLI (standalone)
 
 ```bash
 # Run on a local video file
@@ -105,51 +149,50 @@ uv run -m tools.webcam_check
 The system generates multiple artifacts during execution:
 
 1. **Visual Debug (`output.mp4`)**:
-
    - A video file with bounding boxes drawn for all stages.
    - **Blue**: Detected Person.
    - **Yellow**: Cigarette Candidate.
    - **Red**: Confirmed Smoking Event.
 
 2. **Frame Metadata (`output.jsonl`)**:
-
    - A synchronized JSON Lines file containing raw detection data for every frame.
 
 3. **Alert Log (`alerts.jsonl`)**:
-
    - A structured log file containing **only** confirmed smoking events.
 
 4. **Event Clips (`events/event_<timestamp>.mp4`)**:
-
    - Smart clips containing only the smoking incidents.
    - Includes 2 seconds of pre-event context and 3 seconds post-event.
    - Automatically extends if smoking continues.
 
 5. **Screenshots (`screenshots/alert_<frame>_<conf>.jpg`)**:
-
    - Annotated JPG images saved when alerts fire.
    - Useful for evidence/audit trails without full video.
 
 6. **Application Logs (`logs/pipeline.log`)**:
-
    - Rotating log files with detailed debug information.
    - Includes thread names, function calls, and timing data.
 
 7. **Metrics Summary** (printed on exit):
-
    - Total frames, average FPS, latency, detection counts by label.
 
 ## 📂 Project Structure
 
 ```text
 gp/
-├── main.py                  # CLI Entry point
+├── server.py                # FastAPI server (runs pipeline + REST API)
+├── main.py                  # CLI entry point (standalone pipeline)
 ├── pyproject.toml           # Project config (uv/pip)
 ├── models/                  # Directory for .pt model weights
 ├── logs/                    # Rotating application logs
 ├── events/                  # Smart event clips (auto-created)
 ├── screenshots/             # Alert evidence images
 │
+├── db/                      # Database abstraction layer
+│   ├── repository.py        # Repository protocol + User/DetectionRecord
+│   ├── sqlite.py            # SQLiteRepository (zero-setup default)
+│   └── mysql.py             # MySQLRepository (optional)
+│
 ├── pipeline/
 │   ├── config.py            # PipelineConfig dataclass (all thresholds)
 │   ├── core.py              # SmokingDetectionPipeline (classic, fixed stages)
@@ -166,6 +209,7 @@ gp/
 │   │
 │   ├── sinks/               # Output handlers
 │   │   ├── alerts.py        # FileAlertStore (JSONL)
+│   │   ├── database.py      # DatabaseSink (saves alerts to DB)
 │   │   ├── display.py       # WindowDisplaySink (live preview)
 │   │   ├── events.py        # SmartEventSink (clip recorder)
 │   │   ├── metrics.py       # MetricsSink (FPS, latency, counts)
@@ -179,12 +223,24 @@ gp/
 │   │   └── api.py           # APIPersonDetector, VLMVerifier (cloud stages)
 │   │
 │   └── ops/                 # Algorithms & utilities
-│       ├── nms.py           # Non-Maximum Suppression
+│       ├── nms.py           # NMS + Weighted Boxes Fusion (WBF)
+│       ├── selection.py     # Budget-aware box selection (top-K, merge)
 │       ├── geometry.py      # Box expansion, zone filtering
 │       └── drawing.py       # Visualization helpers
 │
-└── tools/
-    └── webcam_check.py      # Utility to list available webcams
+├── tools/
+│   └── webcam_check.py      # Utility to list available webcams
+│
+└── SmokeDetectionApp/       # Android companion app
+    └── app/src/main/java/com/example/smokedetection/
+        ├── LoginActivity     # Auth screen + hidden server config
+        ├── MainActivity      # Dashboard with camera preview
+        ├── HistoryActivity   # Alert history list
+        ├── StreamActivity    # Live MJPEG viewer
+        ├── ViewEvidenceActivity # Full-screen alert image
+        ├── SmokeAlert        # Foreground polling service
+        ├── HistoryAdapter    # RecyclerView adapter
+        └── ApiClient         # Dynamic server URL + HTTP helpers
 ```
 
 ## ⚙️ Configuration
@@ -207,6 +263,12 @@ class PipelineConfig:
     idle_interval: float = 0.5   # 2 FPS when idle
     active_interval: float = 0.0 # Max FPS when active
     cooldown_frames: int = 60    # Stay active after detection
+
+    # Inference budget caps
+    max_person_crops: int = 3       # Max cig model runs (0 = unlimited)
+    person_priority: str = "area"   # "area", "confidence", "area_conf"
+    merge_people: bool = False      # Merge overlapping people into regions
+    max_cig_verifications: int = 5  # Max verifier runs (0 = unlimited)
 ```
 
 Pass a custom config to the pipeline:
@@ -222,6 +284,30 @@ config = PipelineConfig(
 pipeline = SmokingDetectionPipeline(..., config=config)
 ```
 
+## API Endpoints
+
+The server (`server.py`) exposes these endpoints:
+
+| Method | Endpoint         | Description                         |
+| :----- | :--------------- | :---------------------------------- |
+| GET    | `/ping`          | Health check (verify connectivity)  |
+| POST   | `/register`      | Create a new user account           |
+| POST   | `/login`         | Authenticate and get user ID        |
+| GET    | `/history`       | Get detection records for a user    |
+| POST   | `/clear_history` | Delete all records for a user       |
+| GET    | `/snapshot`      | Latest camera frame as JPEG         |
+| GET    | `/video_feed`    | MJPEG stream with detection overlay |
+
+Interactive docs at **<http://localhost:8000/docs>** (Swagger UI).
+
+## 📱 Android App
+
+The companion app (`SmokeDetectionApp/`) connects to the server for auth, alert history, and live camera feed.
+
+- **Dynamic server URL**: Tap the "Smoking Alert" title on the login screen **5 times** to configure the server address. The URL is persisted and verified via `/ping` before saving.
+- **Background polling**: `SmokeAlert` foreground service checks `/history` every 5 seconds and pushes a notification on new detections.
+- **Edge-to-edge**: All activities draw behind system bars with proper inset handling.
+
 ## 🔧 Performance Tuning
 
 For RTSP streams, the `ThreadedStreamSource` uses a background thread to prevent frame buildup and handles reconnection automatically.
@@ -232,19 +318,39 @@ Key tuning parameters:
 - `active_interval`: Set to 0.0 for max detection rate
 - `cig_high_confidence`: Lower for faster alerts, higher for fewer false positives
 
+### Inference Budget
+
+When using lightweight (nano) models that produce many false positives, cap expensive downstream calls:
+
+```python
+config = PipelineConfig(
+    max_person_crops=3,        # Only process 3 people max per frame
+    person_priority="area",    # Prioritize largest (closest to camera)
+    merge_people=True,         # Merge overlapping people into regions
+    merge_people_iou=0.3,      # IoU threshold for merging
+    max_cig_verifications=5,   # Only verify top 5 cigarette candidates
+)
+```
+
+| Priority     | Strategy                                            |
+| ------------ | --------------------------------------------------- |
+| `area`       | Largest boxes first (closer to camera, more detail) |
+| `confidence` | Highest scoring boxes first                         |
+| `area_conf`  | Balanced: area × confidence                         |
+
 ## 🔌 Pluggable Stage System
 
 The `--staged` flag enables the modular stage pipeline, allowing you to mix local models with cloud APIs:
 
 ### Stage Types
 
-| Stage | Description | Use Case |
-|:------|:------------|:---------|
-| `LocalPersonDetector` | YOLO person detection | Default, GPU/CPU |
-| `LocalCigaretteDetector` | YOLO cigarette detection | Default, GPU/CPU |
-| `LocalVerifier` | YOLO classifier | Default, GPU/CPU |
-| `APIPersonDetector` | Cloud person detection | No local GPU, scaling |
-| `VLMVerifier` | GPT-4V/Claude Vision | High-stakes verification |
+| Stage                    | Description              | Use Case                 |
+| :----------------------- | :----------------------- | :----------------------- |
+| `LocalPersonDetector`    | YOLO person detection    | Default, GPU/CPU         |
+| `LocalCigaretteDetector` | YOLO cigarette detection | Default, GPU/CPU         |
+| `LocalVerifier`          | YOLO classifier          | Default, GPU/CPU         |
+| `APIPersonDetector`      | Cloud person detection   | No local GPU, scaling    |
+| `VLMVerifier`            | GPT-4V/Claude Vision     | High-stakes verification |
 
 ### VLM Verification Example
 

db/__init__.py

@@ -0,0 +1,18 @@
+"""Database abstraction layer.
+
+Provides a Repository protocol and concrete implementations for SQLite and MySQL.
+The server doesn't care which one it talks to.
+"""
+
+from db.repository import Repository
+from db.sqlite import SQLiteRepository
+
+__all__ = ["Repository", "SQLiteRepository"]
+
+# Optional MySQL import (requires mysql-connector-python)
+try:
+    from db.mysql import MySQLRepository
+
+    __all__.append("MySQLRepository")
+except ImportError:
+    pass

db/mysql.py

@@ -0,0 +1,138 @@
+"""MySQL implementation of the Repository."""
+
+import os
+from typing import Optional
+
+import mysql.connector  # type: ignore[import-untyped]
+
+from db.repository import DetectionRecord, User
+
+
+class MySQLRepository:
+    """MySQL-backed repository. Needs a running MySQL server + .env config."""
+
+    def __init__(
+        self,
+        user: Optional[str] = None,
+        password: Optional[str] = None,
+        host: Optional[str] = None,
+        database: Optional[str] = None,
+    ):
+        self.config = {
+            "user": user or os.getenv("DB_USER", "smoke_admin"),
+            "password": password or os.getenv("DB_PASSWORD", "123456"),
+            "host": host or os.getenv("DB_HOST", "localhost"),
+            "database": database or os.getenv("DB_NAME", "smoke_detection"),
+        }
+
+    def _connect(self):
+        return mysql.connector.connect(**self.config)
+
+    def initialize(self) -> None:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            """CREATE TABLE IF NOT EXISTS users (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                username VARCHAR(50) UNIQUE NOT NULL,
+                password VARCHAR(255) NOT NULL
+            )"""
+        )
+        cursor.execute(
+            """CREATE TABLE IF NOT EXISTS detections (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                user_id INT,
+                image_path VARCHAR(255) NOT NULL,
+                confidence FLOAT,
+                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                camera_source VARCHAR(50) DEFAULT 'Laptop Webcam',
+                FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+            )"""
+        )
+        cursor.execute(
+            "INSERT IGNORE INTO users (username, password) VALUES ('admin', '123456')"
+        )
+        conn.commit()
+        conn.close()
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT * FROM users WHERE username = %s", (username,))
+        row = cursor.fetchone()
+        conn.close()
+        return User(**row) if row else None
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute(
+            "SELECT * FROM users WHERE username = %s AND password = %s",
+            (username, password),
+        )
+        row = cursor.fetchone()
+        conn.close()
+        return User(**row) if row else None
+
+    def create_user(self, username: str, password: str) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO users (username, password) VALUES (%s, %s)",
+            (username, password),
+        )
+        conn.commit()
+        new_id = cursor.lastrowid
+        conn.close()
+        return new_id  # type: ignore[return-value]
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (%s, %s, %s, %s)",
+            (user_id, image_path, confidence, camera_source),
+        )
+        conn.commit()
+        new_id = cursor.lastrowid
+        conn.close()
+        return new_id  # type: ignore[return-value]
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        if user_id is not None:
+            cursor.execute(
+                "SELECT * FROM detections WHERE user_id = %s ORDER BY timestamp DESC",
+                (user_id,),
+            )
+        else:
+            cursor.execute("SELECT * FROM detections ORDER BY timestamp DESC")
+        rows = cursor.fetchall()
+        conn.close()
+        return [DetectionRecord(**r) for r in rows]
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        if user_id is None:
+            cursor.execute("DELETE FROM detections")
+        else:
+            cursor.execute("DELETE FROM detections WHERE user_id = %s", (user_id,))
+        conn.commit()
+        count = cursor.rowcount
+        conn.close()
+        return count
+
+    def close(self) -> None:
+        pass  # MySQL uses per-call connections

db/repository.py

@@ -0,0 +1,68 @@
+"""Repository protocol - the interface all DB backends must implement."""
+
+from dataclasses import dataclass
+from typing import Optional, Protocol, runtime_checkable
+
+
+@dataclass
+class User:
+    id: int
+    username: str
+    password: str  # plaintext for now (legacy compat) - hash this later
+
+
+@dataclass
+class DetectionRecord:
+    id: int
+    user_id: int
+    image_path: str
+    confidence: float
+    timestamp: str
+    camera_source: str
+
+
+@runtime_checkable
+class Repository(Protocol):
+    """Database operations for the smoking detection server."""
+
+    def initialize(self) -> None:
+        """Create tables if they don't exist."""
+        ...
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        """Find a user by username. Returns None if not found."""
+        ...
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        """Find a user by username + password. Returns None if invalid."""
+        ...
+
+    def create_user(self, username: str, password: str) -> int:
+        """Create a new user. Returns the new user ID."""
+        ...
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        """Save a detection record. Returns the new record ID."""
+        ...
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        """Get detection history, optionally filtered by user. Newest first."""
+        ...
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        """Delete detections, optionally filtered by user. Returns rows deleted."""
+        ...
+
+    def close(self) -> None:
+        """Close the database connection."""
+        ...

db/sqlite.py

@@ -0,0 +1,126 @@
+"""SQLite implementation of the Repository."""
+
+import sqlite3
+from pathlib import Path
+from typing import Optional
+
+from db.repository import DetectionRecord, User
+
+# Schema that works with SQLite (no MySQL-specific syntax)
+SCHEMA = """\
+CREATE TABLE IF NOT EXISTS users (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    username TEXT UNIQUE NOT NULL,
+    password TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS detections (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id INTEGER,
+    image_path TEXT NOT NULL,
+    confidence REAL,
+    timestamp TEXT DEFAULT (datetime('now', 'localtime')),
+    camera_source TEXT DEFAULT 'Laptop Webcam',
+    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+);
+
+INSERT OR IGNORE INTO users (username, password) VALUES ('admin', '123456');
+"""
+
+
+class SQLiteRepository:
+    """SQLite-backed repository. Zero setup, file-based."""
+
+    def __init__(self, db_path: str | Path = "data/smoke.db"):
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn: Optional[sqlite3.Connection] = None
+
+    def _get_conn(self) -> sqlite3.Connection:
+        if self._conn is None:
+            self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
+            self._conn.row_factory = sqlite3.Row
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA foreign_keys=ON")
+        return self._conn
+
+    def initialize(self) -> None:
+        conn = self._get_conn()
+        conn.executescript(SCHEMA)
+        conn.commit()
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        row = (
+            self._get_conn()
+            .execute("SELECT * FROM users WHERE username = ?", (username,))
+            .fetchone()
+        )
+        return User(**dict(row)) if row else None
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        row = (
+            self._get_conn()
+            .execute(
+                "SELECT * FROM users WHERE username = ? AND password = ?",
+                (username, password),
+            )
+            .fetchone()
+        )
+        return User(**dict(row)) if row else None
+
+    def create_user(self, username: str, password: str) -> int:
+        conn = self._get_conn()
+        cursor = conn.execute(
+            "INSERT INTO users (username, password) VALUES (?, ?)",
+            (username, password),
+        )
+        conn.commit()
+        return cursor.lastrowid  # type: ignore[return-value]
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        conn = self._get_conn()
+        cursor = conn.execute(
+            "INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (?, ?, ?, ?)",
+            (user_id, image_path, confidence, camera_source),
+        )
+        conn.commit()
+        return cursor.lastrowid  # type: ignore[return-value]
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        conn = self._get_conn()
+        if user_id is not None:
+            rows = conn.execute(
+                "SELECT * FROM detections WHERE user_id = ? ORDER BY timestamp DESC",
+                (user_id,),
+            ).fetchall()
+        else:
+            rows = conn.execute(
+                "SELECT * FROM detections ORDER BY timestamp DESC"
+            ).fetchall()
+        return [DetectionRecord(**dict(r)) for r in rows]
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        conn = self._get_conn()
+        if user_id is None:
+            cursor = conn.execute("DELETE FROM detections")
+        else:
+            cursor = conn.execute(
+                "DELETE FROM detections WHERE user_id = ?", (user_id,)
+            )
+        conn.commit()
+        return cursor.rowcount
+
+    def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None

pipeline/config.py

@@ -2,6 +2,8 @@
 
 from dataclasses import dataclass
 
+from pipeline.ops.selection import SelectPriority
+
 
 @dataclass
 class PipelineConfig:
@@ -22,16 +24,30 @@ class PipelineConfig:
     min_cig_size: int = 10  # Minimum cig box dimension
 
     # Person crop expansion
-    person_margin: float = 0.2  # 20% expansion around person box
+    person_margin: float = 1.0  # 100% expansion around person box
 
     # Verification crop padding
-    cig_crop_padding: float = 0.5  # 50% padding around cig for context
+    cig_crop_padding: float = 1.2  # 120% padding around cig for context
 
     # Adaptive sampling
     idle_interval: float = 0.5  # Seconds between inference when idle
     active_interval: float = 0.0  # Seconds between inference when active
     cooldown_frames: int = 60  # Frames to stay active after last detection
 
+    # === Inference Budget Caps ===
+    # Limits expensive model calls when there are too many detections
+
+    # Max person crops to run cig detection on (0 = unlimited)
+    max_person_crops: int = 3
+    # How to prioritize which people to process: "area", "confidence", "area_conf"
+    person_priority: SelectPriority = "area"
+    # Merge overlapping people into regions instead of processing separately
+    merge_people: bool = False
+    merge_people_iou: float = 0.3
+
+    # Max cigarette candidates to run verifier on (0 = unlimited)
+    max_cig_verifications: int = 5
+
 
 # Default config instance
 DEFAULT_CONFIG = PipelineConfig()

pipeline/ops/__init__.py

@@ -17,12 +17,23 @@ from pipeline.ops.geometry import (
     point_in_box,
     point_in_polygon,
 )
-from pipeline.ops.nms import nms, nms_by_class
+from pipeline.ops.nms import nms, nms_by_class, wbf, wbf_by_class
+from pipeline.ops.selection import (
+    merge_overlapping,
+    select_for_verification,
+    select_top_k,
+)
 
 __all__ = [
-    # NMS
+    # NMS / Box Fusion
     "nms",
     "nms_by_class",
+    "wbf",
+    "wbf_by_class",
+    # Selection / Budget
+    "select_top_k",
+    "select_for_verification",
+    "merge_overlapping",
     # Geometry
     "expand_box",
     "point_in_box",

pipeline/ops/nms.py

@@ -1,4 +1,4 @@
-"""Non-Maximum Suppression algorithms for detection filtering."""
+"""Box filtering algorithms: NMS and Weighted Boxes Fusion."""
 
 from typing import List
 
@@ -59,3 +59,119 @@ def nms_by_class(
         result.extend(nms(dets, iou_threshold))
 
     return result
+
+
+def wbf(
+    detections: List[Detection],
+    iou_threshold: float = 0.5,
+    skip_threshold: float = 0.0,
+) -> List[Detection]:
+    """
+    Weighted Boxes Fusion: Merge overlapping boxes instead of suppressing.
+
+    Unlike NMS which picks the best box and discards overlapping ones,
+    WBF fuses overlapping boxes by averaging their coordinates weighted
+    by confidence. This often gives better localization.
+
+    Args:
+        detections: List of Detection objects
+        iou_threshold: Boxes with IoU > threshold are fused together
+        skip_threshold: Ignore boxes with confidence below this
+
+    Returns:
+        Fused list of detections
+
+    Reference:
+        https://arxiv.org/abs/1910.13302
+    """
+    if not detections:
+        return []
+
+    # Filter low confidence
+    dets = [d for d in detections if d.score >= skip_threshold]
+    if not dets:
+        return []
+
+    # Sort by confidence (highest first)
+    dets = sorted(dets, key=lambda d: d.score, reverse=True)
+
+    # Clusters of fused boxes
+    clusters: List[List[Detection]] = []
+
+    for det in dets:
+        matched_cluster = None
+
+        # Find a cluster this box belongs to
+        for cluster in clusters:
+            # Check IoU with the fused box (first element is always the fused result)
+            if cluster[0].box.iou(det.box) >= iou_threshold:
+                matched_cluster = cluster
+                break
+
+        if matched_cluster is not None:
+            # Add to existing cluster and recompute fused box
+            matched_cluster.append(det)
+            _update_fused_box(matched_cluster)
+        else:
+            # Start new cluster
+            clusters.append([det])
+
+    # Return the fused box from each cluster (first element)
+    return [cluster[0] for cluster in clusters]
+
+
+def _update_fused_box(cluster: List[Detection]):
+    """
+    Recompute the fused box for a cluster.
+    The fused box is stored as cluster[0].
+    """
+    if len(cluster) == 1:
+        return
+
+    # Weighted average of coordinates
+    total_weight = sum(d.score for d in cluster)
+
+    x1 = sum(d.box.x1 * d.score for d in cluster) / total_weight
+    y1 = sum(d.box.y1 * d.score for d in cluster) / total_weight
+    x2 = sum(d.box.x2 * d.score for d in cluster) / total_weight
+    y2 = sum(d.box.y2 * d.score for d in cluster) / total_weight
+
+    # Fused confidence: average boosted by number of boxes
+    # More boxes agreeing = higher confidence (capped at 1.0)
+    avg_conf = total_weight / len(cluster)
+    fused_conf = min(1.0, avg_conf * (1 + 0.1 * (len(cluster) - 1)))
+
+    # Update the first element with fused values
+    fused = cluster[0]
+    fused.box.x1 = int(x1)
+    fused.box.y1 = int(y1)
+    fused.box.x2 = int(x2)
+    fused.box.y2 = int(y2)
+    fused.box.conf = fused_conf
+    fused.score = fused_conf
+    fused.metadata["wbf_count"] = len(cluster)
+
+
+def wbf_by_class(
+    detections: List[Detection],
+    iou_threshold: float = 0.5,
+    skip_threshold: float = 0.0,
+) -> List[Detection]:
+    """
+    Run WBF separately for each class label.
+    This prevents boxes of different classes from being fused together.
+    """
+    if not detections:
+        return []
+
+    # Group by label
+    by_label: dict[str, List[Detection]] = {}
+    for det in detections:
+        by_label.setdefault(det.label, []).append(det)
+
+    # Run WBF per class
+    result = []
+    for dets in by_label.values():
+        result.extend(wbf(dets, iou_threshold, skip_threshold))
+
+    return result

pipeline/ops/selection.py

@@ -0,0 +1,174 @@
+"""Budget-aware box selection for capping expensive inference calls."""
+
+from typing import List, Literal
+
+from pipeline.types import BoundingBox, Detection
+
+type SelectPriority = Literal["area", "confidence", "area_conf"]
+
+
+def select_top_k(
+    detections: List[Detection],
+    k: int,
+    diversity_iou: float = 0.5,
+    priority: SelectPriority = "area",
+) -> List[Detection]:
+    """
+    Select top K detections with spatial diversity.
+
+    Picks the K "best" boxes while ensuring they don't overlap too much.
+    This caps expensive downstream inference (e.g., max 3 person crops → cig model).
+
+    Args:
+        detections: Candidate detections
+        k: Maximum number to select
+        diversity_iou: Reject candidates overlapping selected boxes above this IoU
+        priority: How to rank candidates:
+            - "area": Larger boxes first (closer to camera, more detail)
+            - "confidence": Higher confidence first
+            - "area_conf": area * confidence (balanced)
+
+    Returns:
+        Up to K diverse, high-priority detections
+    """
+    if not detections or k <= 0:
+        return []
+
+    # Score each detection
+    def score(det: Detection) -> float:
+        box = det.box
+        # area = (box.x2 - box.x1) * (box.y2 - box.y1)
+        area = box.area
+        if priority == "area":
+            return area
+        elif priority == "confidence":
+            return det.score
+        else:  # area_conf
+            return area * det.score
+
+    # Sort by priority (highest first)
+    ranked = sorted(detections, key=score, reverse=True)
+
+    selected = []
+    for det in ranked:
+        if len(selected) >= k:
+            break
+
+        # Check diversity: don't select if overlaps too much with already selected
+        dominated = False
+        for sel in selected:
+            if det.box.iou(sel.box) > diversity_iou:
+                dominated = True
+                break
+
+        if not dominated:
+            selected.append(det)
+
+    return selected
+
+
+def merge_overlapping(
+    detections: List[Detection],
+    iou_threshold: float = 0.3,
+) -> List[Detection]:
+    """
+    Merge overlapping boxes into larger regions.
+
+    Instead of processing each overlapping person separately, merge them
+    into one bigger crop. Good when people are clustered together.
+
+    Args:
+        detections: List of detections (e.g., people)
+        iou_threshold: Merge boxes with IoU above this
+
+    Returns:
+        Merged detections (fewer, larger boxes)
+    """
+    if not detections:
+        return []
+
+    # Sort by confidence
+    dets = sorted(detections, key=lambda d: d.score, reverse=True)
+
+    clusters: List[List[Detection]] = []
+
+    for det in dets:
+        merged = False
+        for cluster in clusters:
+            # Check if this det overlaps with any box in cluster
+            for member in cluster:
+                if det.box.iou(member.box) >= iou_threshold:
+                    cluster.append(det)
+                    merged = True
+                    break
+            if merged:
+                break
+
+        if not merged:
+            clusters.append([det])
+
+    # Create merged boxes from clusters
+    result = []
+    for cluster in clusters:
+        merged_det = _merge_cluster(cluster)
+        result.append(merged_det)
+
+    return result
+
+
+def _merge_cluster(cluster: List[Detection]) -> Detection:
+    """Merge a cluster of detections into one encompassing box."""
+    if len(cluster) == 1:
+        return cluster[0]
+
+    # Bounding box that contains all
+    x1 = min(d.box.x1 for d in cluster)
+    y1 = min(d.box.y1 for d in cluster)
+    x2 = max(d.box.x2 for d in cluster)
+    y2 = max(d.box.y2 for d in cluster)
+
+    # Confidence: max of cluster (could also use average)
+    conf = max(d.score for d in cluster)
+
+    # Use label from highest confidence detection
+    best = max(cluster, key=lambda d: d.score)
+
+    return Detection(
+        box=BoundingBox(x1=x1, y1=y1, x2=x2, y2=y2, conf=conf),
+        score=conf,
+        label=best.label,
+        metadata={
+            "merged_count": len(cluster),
+            "original_boxes": [
+                (d.box.x1, d.box.y1, d.box.x2, d.box.y2) for d in cluster
+            ],
+        },
+    )
+
+
+def select_for_verification(
+    detections: List[Detection],
+    k: int,
+    min_confidence: float = 0.0,
+) -> List[Detection]:
+    """
+    Select top K cigarette candidates for expensive verification.
+
+    Simple confidence-based selection for the verifier stage.
+    Higher confidence cigs are more likely real → verify those first.
+
+    Args:
+        detections: Cigarette candidates
+        k: Max verifications to run
+        min_confidence: Skip candidates below this
+
+    Returns:
+        Top K candidates by confidence
+    """
+    if not detections or k <= 0:
+        return []
+
+    filtered = [d for d in detections if d.score >= min_confidence]
+    ranked = sorted(filtered, key=lambda d: d.score, reverse=True)
+
+    return ranked[:k]

pipeline/sinks/__init__.py

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

pipeline/sinks/database.py

@@ -0,0 +1,74 @@
+"""Database sink - saves detection alerts to the database."""
+
+import logging
+import time
+from pathlib import Path
+
+import cv2
+
+from db.repository import Repository
+from pipeline.ops.drawing import draw_alert_banner, draw_detections
+from pipeline.types import Frame
+
+logger = logging.getLogger(__name__)
+
+
+class DatabaseSink:
+    """
+    Pipeline sink that saves smoking alerts to a database.
+
+    On ticket (alert), saves an annotated screenshot and inserts a detection record.
+    """
+
+    def __init__(
+        self,
+        db: Repository,
+        user_id: int = 1,
+        image_dir: str = "static/alerts",
+        camera_source: str = "Pipeline",
+    ):
+        self.db = db
+        self.user_id = user_id
+        self.camera_source = camera_source
+        self.image_dir = Path(image_dir)
+        self.image_dir.mkdir(parents=True, exist_ok=True)
+        self.count = 0
+
+        logger.info("DatabaseSink initialized (user_id=%d, dir=%s)", user_id, image_dir)
+
+    def handle(self, frame: Frame):
+        if not frame.ticket:
+            return
+
+        self.count += 1
+        ticket = frame.ticket
+
+        # Save annotated image
+        filename = f"alert_{int(time.time())}_{self.count}.jpg"
+        filepath = self.image_dir / filename
+
+        img = frame.data.copy()
+        draw_detections(img, frame.detections)
+        draw_alert_banner(img, f"SMOKING {ticket.confidence:.0%}")
+        cv2.imwrite(str(filepath), img)
+
+        # Insert into database
+        web_path = f"/static/alerts/{filename}"
+        try:
+            self.db.save_detection(
+                user_id=self.user_id,
+                image_path=web_path,
+                confidence=ticket.confidence,
+                camera_source=self.camera_source,
+            )
+            logger.info(
+                "Alert #%d saved to DB: %s (%.2f)",
+                self.count,
+                filename,
+                ticket.confidence,
+            )
+        except Exception as e:
+            logger.error("Failed to save alert to DB: %s", e)
+
+    def close(self):
+        pass  # DB connection is managed by the repository

pipeline/stages/api.py

@@ -149,6 +149,7 @@ class APIPersonDetector(APIStageBase):
         return resp.json()
 
 
+# haiku 4.5 is multimodal but i fear its going to ping 65 million images per second (we need cache control? who are we? why cant we upload video?)
 class VLMVerifier(APIStageBase):
     """
     Vision-Language Model verification stage.

pipeline/stages/local.py

@@ -5,7 +5,12 @@ import time
 from typing import List, Optional
 
 from pipeline.config import DEFAULT_CONFIG, PipelineConfig
-from pipeline.ops import nms_by_class
+from pipeline.ops import (
+    merge_overlapping,
+    nms_by_class,
+    select_for_verification,
+    select_top_k,
+)
 from pipeline.protocols import Model, Verifier
 from pipeline.stages.protocol import StageContext, StageResult
 from pipeline.types import BoundingBox, Detection, Ticket
@@ -66,10 +71,39 @@ class LocalCigaretteDetector:
         cfg = self.cfg
         h_img, w_img = ctx.frame_data.shape[:2]
 
+        # === Budget: Limit how many people we process ===
+        people_to_process = ctx.people
+
+        if cfg.merge_people:
+            # Merge overlapping people into regions
+            people_to_process = merge_overlapping(
+                people_to_process, cfg.merge_people_iou
+            )
+            if len(people_to_process) < len(ctx.people):
+                logger.debug(
+                    "Merged %d people into %d regions",
+                    len(ctx.people),
+                    len(people_to_process),
+                )
+
+        if cfg.max_person_crops > 0 and len(people_to_process) > cfg.max_person_crops:
+            # Select top K diverse people by priority
+            people_to_process = select_top_k(
+                people_to_process,
+                k=cfg.max_person_crops,
+                priority=cfg.person_priority,
+            )
+            logger.debug(
+                "Budget: Processing %d/%d people (priority=%s)",
+                len(people_to_process),
+                len(ctx.people),
+                cfg.person_priority,
+            )
+
         all_cigs: List[Detection] = []
         early_hit = None
 
-        for person in ctx.people:
+        for person in people_to_process:
             # Expand person box for crop
             crop_box = self._expand_box(person.box, cfg.person_margin, w_img, h_img)
             ex1, ey1, ex2, ey2 = crop_box
@@ -195,12 +229,28 @@ class LocalVerifier:
         cfg = self.cfg
         h_img, w_img = ctx.frame_data.shape[:2]
 
+        # === Budget: Limit how many cigs we verify ===
+        cigs_to_verify = ctx.cigarettes
+
+        if (
+            cfg.max_cig_verifications > 0
+            and len(cigs_to_verify) > cfg.max_cig_verifications
+        ):
+            # Select top K by confidence (most likely real → verify first)
+            cigs_to_verify = select_for_verification(
+                cigs_to_verify,
+                k=cfg.max_cig_verifications,
+                min_confidence=cfg.cig_min_confidence,
+            )
+            logger.debug(
+                "Budget: Verifying %d/%d cigarettes",
+                len(cigs_to_verify),
+                len(ctx.cigarettes),
+            )
+
         confirmed: list[tuple[Detection, float]] = []
 
-        for cig_det in ctx.cigarettes:
-            offset = cig_det.metadata.get("crop_offset", (0, 0))
-            ex1, ey1 = offset
-
+        for cig_det in cigs_to_verify:
             # Extract padded crop for classifier
             crop = self._extract_cig_crop(ctx.frame_data, cig_det, w_img, h_img)
             if crop.size == 0:

pipeline/stages/runner.py

@@ -60,7 +60,7 @@ class StagePipeline:
                     stage.name,
                     ctx.stop_reason,
                 )
-                continue
+                continue  # break?
 
             try:
                 result = stage.process(ctx)

pipeline/types.py

@@ -10,7 +10,7 @@ class BoundingBox:
     y1: int
     x2: int
     y2: int
-    conf: float = 0.0
+    conf: float = 0.0  # legacy
 
     @property
     def area(self):

pyproject.toml

@@ -5,11 +5,17 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "dotenv>=0.9.9",
+    "fastapi>=0.133.0",
+    "mysql-connector-python>=9.6.0",
     "numpy>=2.4.0",
     "opencv-python>=4.11.0.86",
+    "python-dotenv>=1.2.1",
+    "python-multipart>=0.0.22",
     "torch>=2.9.1",
     "torchvision>=0.24.1",
     "ultralytics>=8.3.248",
+    "uvicorn>=0.41.0",
 ]
 
 [[tool.uv.index]]

server.py

@@ -0,0 +1,385 @@
+"""
+Smoking Detection Server
+========================
+
+FastAPI server that runs the full detection pipeline on a live webcam feed
+and exposes REST endpoints for the frontend (auth, history, video stream).
+
+Usage:
+    uvicorn server:app --host 0.0.0.0 --port 8000
+    python server.py                          # dev mode
+    python server.py --staged                 # use stage-based pipeline
+    python server.py --source rtsp://...      # RTSP camera
+"""
+
+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, Form, Response
+from fastapi.responses import StreamingResponse
+from fastapi.staticfiles import StaticFiles
+
+from db import SQLiteRepository
+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.sources.stream import ThreadedStreamSource
+from pipeline.staged_core import StageBasedPipeline
+from pipeline.types import Frame
+
+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__)
+# ---------------------------------------------------------------------------
+def parse_args():
+    p = argparse.ArgumentParser(description="Smoking Detection Server")
+    p.add_argument(
+        "--source", type=str, default="0", help="Webcam index or RTSP URL (default: 0)"
+    )
+    p.add_argument("--person-model", default="models/human-det-best.pt")
+    p.add_argument("--cig-model", default="models/cig-det-best.pt")
+    p.add_argument("--verify-model", default="models/cig-cls-best.pt")
+    p.add_argument("--staged", action="store_true", help="Use stage-based pipeline")
+    p.add_argument("--vlm", action="store_true")
+    p.add_argument("--vlm-key", default=None)
+    p.add_argument("--vlm-provider", default="openai", choices=["openai", "anthropic"])
+    p.add_argument("--vlm-model", default="gpt-4-vision-preview")
+    p.add_argument("--host", default="0.0.0.0")
+    p.add_argument("--port", type=int, default=8000)
+    p.add_argument("--debug", action="store_true")
+    return p.parse_args()
+
+
+# Module-level defaults (overridden by __main__ block or env)
+_args = argparse.Namespace(
+    source=os.environ.get("CAMERA_SOURCE", "0"),
+    person_model=os.environ.get("PERSON_MODEL", "models/human-det-best.pt"),
+    cig_model=os.environ.get("CIG_MODEL", "models/cig-det-best.pt"),
+    verify_model=os.environ.get("VERIFY_MODEL", "models/cig-cls-best.pt"),
+    staged=os.environ.get("STAGED", "").lower() in ("1", "true", "yes"),
+    vlm=os.environ.get("USE_VLM", "").lower() in ("1", "true", "yes"),
+    vlm_key=os.environ.get("VLM_API_KEY"),
+    vlm_provider=os.environ.get("VLM_PROVIDER", "openai"),
+    vlm_model=os.environ.get("VLM_MODEL", "gpt-4-vision-preview"),
+    debug=False,
+)
+
+
+# ---------------------------------------------------------------------------
+# Lifespan - startup & shutdown
+# ---------------------------------------------------------------------------
+@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",
+        datefmt="%H:%M:%S",
+    )
+    logging.getLogger("ultralytics").setLevel(logging.WARNING)
+
+    # 1. Database
+    logger.info("Initializing database...")
+    db.initialize()
+
+    # 2. Models
+    logger.info("Loading models...")
+    person_model = YoloPersonDetector(_args.person_model)
+    cig_model = YoloCigaretteDetector(_args.cig_model)
+    verifier = YoloClassifier(_args.verify_model)
+    logger.info("Models loaded.")
+
+    # 3. Source (webcam / RTSP)
+    cam = int(_args.source) if _args.source.isdigit() else _args.source
+    source = ThreadedStreamSource(cam)
+    logger.info("Camera source ready: %s (%.1f FPS)", _args.source, source.fps)
+
+    # 4. Sinks
+    sinks: list[Sink] = [
+        LiveFrameSink(),
+        DatabaseSmartEventSink(db, fps=source.fps, output_dir="static/events"),
+        MetricsSink(log_interval=200),
+        FileAlertStore("alerts.jsonl"),
+    ]
+
+    # 5. Pipeline config
+    config = PipelineConfig(
+        idle_interval=0.5,
+        active_interval=0.0,
+        cooldown_frames=int(source.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"
+        )
+        pipeline = 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,
+        )
+    else:
+        pipeline = SmokingDetectionPipeline(
+            person_model=person_model,
+            cigarette_model=cig_model,
+            verifier_model=verifier,
+            sinks=sinks,
+            config=config,
+        )
+
+    # 7. Launch pipeline in background thread
+    t = threading.Thread(target=_run_pipeline, daemon=True, name="pipeline")
+    t.start()
+    logger.info("Pipeline running in background.")
+
+    # Show real addresses
+    ip, hostname = _get_lan_address()
+    port = _args.port if hasattr(_args, "port") else 8000
+    logger.info("=" * 50)
+    logger.info("  Server ready")
+    logger.info("  Local:    http://127.0.0.1:%d", port)
+    logger.info("  Network:  http://%s:%d", ip, port)
+    if hostname and hostname != ip:
+        logger.info("  Hostname: http://%s:%d", hostname, port)
+    logger.info("=" * 50)
+
+    yield  # --- app serves requests ---
+
+    # Shutdown
+    logger.info("Shutting down...")
+    source.stop()
+    for s in sinks:
+        s.close()
+    db.close()
+    logger.info("Goodbye.")
+
+
+# ---------------------------------------------------------------------------
+# FastAPI app
+# ---------------------------------------------------------------------------
+app = FastAPI(title="Smoking Detection", lifespan=app_lifespan)  # type: ignore
+
+# 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)
+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
+# ---------------------------------------------------------------------------
+if __name__ == "__main__":
+    import uvicorn
+
+    _args = parse_args()
+    uvicorn.run(app, host=_args.host, port=_args.port)