jexck/gp
1
0 Fork 0
forked from usth/gp
Code Pull Requests Activity

Compare commits

merge into: jexck/gp:main
jexck/gp:main usth/gp:what usth/gp:main
..
pull from: usth/gp:what
usth/gp:what usth/gp:main jexck/gp:main
13 Commits
main .. what
33 changed files with 4015 additions and 368 deletions
Expand all files Collapse all files

No files matched your search

.env.example
+5
View File
@@ -0,0 +1,5 @@
# what the helly
DB_USER=smoke_admin
DB_PASSWORD=123456
DB_HOST=localhost
DB_NAME=smoke_detection
.gitattributes
+1
View File
@@ -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
+12 -1
View File
@@ -9,4 +9,15 @@ 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/
screenshots/
events/
logs/
media/
.gitmodules
+3
View File
@@ -0,0 +1,3 @@
[submodule "SmokeDetectionApp"]
path = SmokeDetectionApp
url = https://github.com/ldlda/SmokeDihtectionApp.git
.vscode/settings.json
Vendored
+2 -1
View File
@@ -6,5 +6,6 @@
"**/cache/": true
},
"pylint.args": ["--generated-members=cv2.*", "--extension-pkg-whitelist=cv2"],
// "editor.fontFamily": "'JetbrainsMono Nerd Font', Consolas, 'Courier New', monospace"
"java.configuration.updateBuildConfiguration": "interactive"
// "editor.fontFamily": "'JetbrainsMono Nerd Font', Consolas, 'Courier New', monospace"
}
README.md
+178 -17
View File
@@ -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,63 @@ 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
# Package entrypoint (same server)
uv run -m server
# 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
```
### 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:
```
==================================================
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
@@ -78,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
@@ -105,51 +160,54 @@ 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)
├── 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
├── 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 +224,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 +238,25 @@ 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
├── scripts/
│ ├── 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/
├── 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 +279,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 +300,69 @@ 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 (optionally by `user_id`) |
| POST | `/clear_history` | Delete detection records (optional `user_id`) |
| GET | `/snapshot` | Latest camera frame as JPEG |
| 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.
- **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,6 +373,26 @@ 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:
@@ -239,7 +400,7 @@ The `--staged` flag enables the modular stage pipeline, allowing you to mix loca
### 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 |
SmokeDetectionApp
Submodule
+1
Submodule SmokeDetectionApp added at e296efdac5.
db/__init__.py
+18
View File
@@ -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
+138
View File
@@ -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
+68
View File
@@ -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
+126
View File
@@ -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
models/cig-det-best.pt
LFS
BIN
View File
Binary file not shown.
pipeline/config.py
+18 -2
View File
@@ -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
+13 -2
View File
@@ -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
+117 -1
View File
@@ -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
+174
View File
@@ -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
+3
View File
@@ -1,6 +1,7 @@
"""Sinks for pipeline output handling."""
from pipeline.sinks.alerts import FileAlertStore
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
@@ -8,6 +9,8 @@ from pipeline.sinks.screenshots import ScreenshotSink
from pipeline.sinks.video import VideoFileStore
__all__ = [
"DatabaseSink",
"DatabaseSmartEventSink",
"FileAlertStore",
"MetricsSink",
"ScreenshotSink",
pipeline/sinks/database.py
+124
View File
@@ -0,0 +1,124 @@
"""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.sinks.events import SmartEventSink
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
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)
pipeline/sources/file.py
+35 -1
View File
@@ -3,6 +3,7 @@
import glob
import logging
import time
from pathlib import Path
from typing import Iterator
import cv2
@@ -19,9 +20,10 @@ class FileStreamSource(StreamSource):
def __init__(self, path: str):
self.path = path
self._fps = 30.0
self._image_exts = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
# Probe for FPS if it's a single file
if "*" not in self.path:
if "*" not in self.path and not self._is_image_path(self.path):
try:
cap = cv2.VideoCapture(self.path)
if cap.isOpened():
@@ -30,6 +32,16 @@ class FileStreamSource(StreamSource):
except Exception as e:
logger.warning("Could not probe FPS from %s: %s", path, e)
def _is_image_path(self, value: str) -> bool:
return Path(value).suffix.lower() in self._image_exts
def _iter_image_files(self, directory: Path) -> list[str]:
files: list[str] = []
for ext in self._image_exts:
files.extend(glob.glob(str(directory / f"*{ext}")))
files.extend(glob.glob(str(directory / f"*{ext.upper()}")))
return sorted(set(files))
@property
def fps(self) -> float:
return self._fps
@@ -45,6 +57,28 @@ class FileStreamSource(StreamSource):
yield Frame(timestamp=time.time(), data=img, frame_id=frame_id)
return
file_path = Path(self.path)
# Handle single image file
if self._is_image_path(self.path):
img = cv2.imread(self.path)
if img is None:
logger.warning("Could not read image: %s", self.path)
return
logger.info("Loaded single image: %s", self.path)
yield Frame(timestamp=time.time(), data=img, frame_id=0)
return
# Handle directory of images
if file_path.exists() and file_path.is_dir():
files = self._iter_image_files(file_path)
logger.info("Loading %d images from directory: %s", len(files), self.path)
for frame_id, file in enumerate(files):
img = cv2.imread(file)
if img is not None:
yield Frame(timestamp=time.time(), data=img, frame_id=frame_id)
return
logger.info("Opening video file: %s", self.path)
cap = cv2.VideoCapture(self.path)
frame_id = 0
pipeline/stages/api.py
+1
View File
@@ -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
+56 -6
View File
@@ -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
+1 -1
View File
@@ -60,7 +60,7 @@ class StagePipeline:
stage.name,
ctx.stop_reason,
)
continue
continue # break?
try:
result = stage.process(ctx)
pipeline/types.py
+1 -1
View File
@@ -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
+6
View File
@@ -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[standard]>=0.41.0",
]
[[tool.uv.index]]
reports/live_stress_latest.json
+622
View File
@@ -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
+265
View File
@@ -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())
tools/webcam_check.py → scripts/webcam_check.py
View File
File renamed without changes.
server.py
+260
View File
@@ -0,0 +1,260 @@
"""
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
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import server.runtime as rt
from pipeline.config import PipelineConfig
from pipeline.core import SmokingDetectionPipeline
from pipeline.models import YoloCigaretteDetector, YoloClassifier, YoloPersonDetector
from pipeline.protocols import Sink, StreamSource
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
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# 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):
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...")
rt.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. Configure live pipeline manager
def source_factory(source_spec: str) -> StreamSource:
if source_spec.lower() in {"phone", "android", "mobile"}:
rt.phone_source.reset()
return rt.phone_source
cam = int(source_spec) if source_spec.isdigit() else source_spec
return ThreadedStreamSource(cam)
def sink_factory(src: StreamSource) -> list[Sink]:
return [
rt.LiveFrameSink(),
DatabaseSmartEventSink(rt.db, fps=src.fps, output_dir="static/events"),
MetricsSink(log_interval=200),
FileAlertStore("alerts.jsonl"),
]
def pipeline_factory(src: StreamSource, sinks: list[Sink]):
config = PipelineConfig(
idle_interval=0.5,
active_interval=0.0,
cooldown_frames=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.manager.configure(
source_factory=source_factory,
sink_factory=sink_factory,
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)
logger.info("%s (source=%s)", msg, _args.source)
# Show real addresses
ip, hostname = rt.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...")
rt.manager.stop()
rt.clip_jobs.shutdown()
rt.db.close()
logger.info("Goodbye.")
# ---------------------------------------------------------------------------
# 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)
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")
# ---------------------------------------------------------------------------
# Entry-point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
_args = parse_args()
uvicorn.run(app, host=_args.host, port=_args.port)
server/__init__.py
View File
Whitespace-only changes.
server/__main__.py
+9
View File
@@ -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
+371
View File
@@ -0,0 +1,371 @@
"""API route definitions for the smoking detection server."""
import asyncio
import shutil
import time
from pathlib import Path
import cv2
from fastapi import (
APIRouter,
File,
Form,
Response,
UploadFile,
WebSocket,
WebSocketDisconnect,
)
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.websocket("/ws/updates")
async def ws_updates(websocket: WebSocket, history_limit: int = 10):
await websocket.accept()
clamped_limit = max(1, min(history_limit, 50))
try:
while True:
try:
records = rt.db.get_detections(None)[:clamped_limit]
history = [
{
"id": r.id,
"user_id": r.user_id,
"image_path": r.image_path,
"confidence": r.confidence,
"timestamp": r.timestamp,
"camera_source": r.camera_source,
}
for r in records
]
except Exception:
history = []
await websocket.send_json(
{
"type": "snapshot",
"live": rt.manager.status(),
"history": history,
"history_count": len(history),
"server_time": int(time.time()),
}
)
await asyncio.sleep(1.0)
except WebSocketDisconnect:
return
@router.post("/live/start")
def live_start(source: str | None = Form(default=None)):
"""Start live pipeline (optionally with source override)."""
ok, message = rt.manager.start(source)
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.post("/live/start_phone")
def live_start_phone():
"""Start live pipeline using phone frame push source."""
ok, message = rt.manager.switch_source("phone")
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
"hint": "Push JPEG frames to /live/ws/push_frames",
}
@router.get("/live/source_history")
def live_source_history(limit: int = 10):
return {
"status": "success",
"current_source": rt.manager.status().get("source"),
"history": rt.manager.source_history(limit=limit),
}
@router.post("/live/switch_previous")
def live_switch_previous():
ok, message = rt.manager.switch_previous()
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@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.post("/live/push_frame")
async def live_push_frame(frame: UploadFile = File(...)):
"""Push one JPEG frame from phone/app into live source queue."""
try:
data = await frame.read()
finally:
await frame.close()
if not data:
return {"status": "fail", "message": "Empty frame payload"}
ok, message = rt.push_live_frame_bytes(data)
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.websocket("/live/ws/push_frames")
async def live_ws_push_frames(websocket: WebSocket):
await websocket.accept()
accepted = 0
dropped = 0
try:
while True:
message = await websocket.receive()
if message.get("type") == "websocket.disconnect":
break
payload = message.get("bytes")
if payload is None:
text_payload = message.get("text")
if text_payload == "ping":
await websocket.send_json({"type": "pong"})
continue
ok, _ = rt.push_live_frame_bytes(payload)
if ok:
accepted += 1
else:
dropped += 1
if (accepted + dropped) % 60 == 0:
await websocket.send_json(
{
"type": "stats",
"accepted": accepted,
"dropped": dropped,
"live": rt.manager.status(),
}
)
except WebSocketDisconnect:
return
@router.post("/clips/submit")
async def clips_submit(
file: UploadFile | None = File(default=None),
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."""
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
+650
View File
@@ -0,0 +1,650 @@
"""Runtime state and utility helpers for the FastAPI server."""
import logging
import socket
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from queue import Empty, Queue
from typing import Any, Callable
import cv2
import numpy as np
from db import SQLiteRepository
from db.repository import Repository
from pipeline.protocols import Sink, StreamSource
from pipeline.sources.file import FileStreamSource
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: StreamSource | None = None
live_frame: Frame | None = None
SourceFactory = Callable[[str], StreamSource]
SinkFactory = Callable[[StreamSource], list[Sink]]
PipelineFactory = Callable[[StreamSource, 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."""
def handle(self, frame: Frame):
global live_frame
live_frame = frame
def close(self):
pass
class PushFrameSource(StreamSource):
"""Frame source that receives JPEG frames pushed via API."""
def __init__(self, fps: float = 15.0, max_queue: int = 120):
self._fps = fps
self._queue: Queue[Frame] = Queue(maxsize=max_queue)
self._running = True
self._frame_id = 0
self._lock = threading.Lock()
@property
def fps(self) -> float:
return self._fps
def reset(self):
with self._lock:
self._running = True
self._frame_id = 0
while not self._queue.empty():
try:
self._queue.get_nowait()
self._queue.task_done()
except Empty:
break
def push_jpeg(self, data: bytes) -> tuple[bool, str]:
if not self._running:
return False, "phone source not running"
arr = np.frombuffer(data, dtype=np.uint8)
image = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if image is None:
return False, "invalid image payload"
with self._lock:
frame = Frame(timestamp=time.time(), data=image, frame_id=self._frame_id)
self._frame_id += 1
if self._queue.full():
try:
self._queue.get_nowait()
self._queue.task_done()
except Empty:
pass
self._queue.put(frame)
return True, "frame pushed"
def stream(self):
while self._running:
try:
frame = self._queue.get(timeout=0.5)
except Empty:
continue
try:
yield frame
finally:
self._queue.task_done()
def stop(self):
self._running = False
phone_source = PushFrameSource()
def push_live_frame_bytes(data: bytes) -> tuple[bool, str]:
return phone_source.push_jpeg(data)
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._session_id = 0
self._source_factory: SourceFactory | None = None
self._sink_factory: SinkFactory | None = None
self._pipeline_factory: PipelineFactory | None = None
self._source_spec: str = "0"
self._source_history: list[dict[str, Any]] = []
self.running = False
self.starting = False
self.stopping = False
self.paused = False
self.last_error: str | None = None
def _record_source_locked(self, source_spec: str):
if (
self._source_history
and self._source_history[-1].get("source") == source_spec
):
return
self._source_history.append(
{
"source": source_spec,
"timestamp": datetime.now().isoformat(timespec="seconds"),
}
)
if len(self._source_history) > 50:
self._source_history = self._source_history[-50:]
def configure(
self,
source_factory: SourceFactory,
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
self._record_source_locked(initial_source)
def _close_sinks(self, sinks: list[Sink] | None = None):
sink_list = self._sinks if sinks is None else sinks
for sink in sink_list:
try:
sink.close()
except Exception:
logger.exception("Error closing sink: %s", sink.__class__.__name__)
if sinks is None:
self._sinks = []
def _run_wrapper(self, session_id: int):
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:
if self._session_id == session_id:
self.running = False
self.stopping = False
self.starting = False
self.paused = False
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.stopping:
return False, "Pipeline is stopping"
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._session_id += 1
session_id = self._session_id
self._thread = threading.Thread(
target=self._run_wrapper,
args=(session_id,),
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:
if self.stopping:
return False, "Pipeline is stopping"
self.stopping = True
thread = self._thread
source_obj = source
stop_session_id = self._session_id
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:
if self._session_id != stop_session_id:
self.stopping = False
return True, "Pipeline stop request completed"
self.starting = False
self.running = False
self.stopping = 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
self._record_source_locked(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 source_history(self, limit: int = 10) -> list[dict[str, Any]]:
with self._lock:
clamped = max(1, min(limit, 50))
items = self._source_history[-clamped:]
return list(reversed(items))
def switch_previous(self) -> tuple[bool, str]:
with self._lock:
current = self._source_spec
previous = None
for item in reversed(self._source_history):
source_spec = item.get("source")
if source_spec and source_spec != current:
previous = source_spec
break
if previous is None:
return False, "No previous source in history"
return self.switch_source(previous)
def status(self) -> dict[str, Any]:
with self._lock:
return {
"running": self.running,
"starting": self.starting,
"stopping": self.stopping,
"paused": self.paused,
"source": self._source_spec,
"source_history": self.source_history(limit=10),
"last_error": self.last_error,
"thread_alive": bool(self._thread and self._thread.is_alive()),
}
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,
"outputs": clip_job_outputs(self.job_id),
}
def clip_job_outputs(job_id: str) -> dict[str, Any]:
base_local = Path("static") / "clip_jobs" / job_id
events_local = base_local / "events"
event_clips: list[str] = []
if events_local.exists():
for event_file in sorted(events_local.glob("event_*.mp4")):
event_clips.append(f"/static/clip_jobs/{job_id}/events/{event_file.name}")
return {
"processed_video": f"/static/clip_jobs/{job_id}/processed.mp4",
"alerts_log": f"/static/clip_jobs/{job_id}/alerts.jsonl",
"events_dir": f"/static/clip_jobs/{job_id}/events/",
"event_clips": event_clips,
}
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"
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")
uv.lock
Generated
+735 -333
View File
File diff suppressed because it is too large. Load diff