usth/gp
3
0 Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: usth/gp:main
usth/gp:main usth/gp:what
...
compare: usth/gp:358f4650b5690afa0a7b2b543c4220162e8f3c5f
usth/gp:what usth/gp:main
3 Commits
main ... 358f4650b5
13 changed files with 1909 additions and 260 deletions
Expand all files Collapse all files

No files matched your search

.gitignore
+1 -1
View File
@@ -16,7 +16,7 @@ wheels/
alerts.jsonl
output.jsonl
output.mp4
static/alerts/
static/
screenshots/
events/
logs/
README.md
+67 -12
View File
@@ -63,6 +63,9 @@ The server runs the detection pipeline in the background while serving a REST AP
# Default: webcam 0, classic pipeline
uv run server.py
# Package entrypoint (same server)
uv run -m server
# Use stage-based pipeline with a specific camera
uv run server.py --staged --source 2
@@ -76,6 +79,14 @@ uv run server.py --host 0.0.0.0 --port 9000
uvicorn server:app --host 0.0.0.0 --port 8000
```
### Live + Clip Control (Server Runtime)
The server now runs a lifecycle manager for the live pipeline and a queued clip-job worker:
- **Live manager**: start/stop/pause/resume/switch source at runtime.
- **Clip jobs**: submit files to a queue, process one-at-a-time, track status.
- **Resource policy**: clip jobs can temporarily pause live processing and resume after job completion.
On startup the server prints the LAN address for the mobile app:
```
@@ -122,7 +133,7 @@ uv run main.py --source videos/test.mp4 --debug
### Check Available Webcams
```bash
uv run -m tools.webcam_check
uv run -m scripts.webcam_check
```
### CLI Arguments
@@ -181,6 +192,10 @@ The system generates multiple artifacts during execution:
```text
gp/
├── server.py # FastAPI server (runs pipeline + REST API)
├── server/ # Modular server runtime + routes
│ ├── runtime.py # PipelineManager + ClipJobManager
│ ├── routes.py # API routers (/live/*, /clips/*, auth/history)
│ └── __main__.py # `uv run -m server` entrypoint
├── main.py # CLI entry point (standalone pipeline)
├── pyproject.toml # Project config (uv/pip)
├── models/ # Directory for .pt model weights
@@ -228,8 +243,9 @@ gp/
│ ├── geometry.py # Box expansion, zone filtering
│ └── drawing.py # Visualization helpers
│
├── tools/
│ └── 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/
@@ -288,18 +304,57 @@ pipeline = SmokingDetectionPipeline(..., config=config)
The server (`server.py`) exposes these endpoints:
| Method | Endpoint | Description |
| :----- | :--------------- | :---------------------------------- |
| GET | `/ping` | Health check (verify connectivity) |
| POST | `/register` | Create a new user account |
| POST | `/login` | Authenticate and get user ID |
| GET | `/history` | Get detection records for a user |
| POST | `/clear_history` | Delete all records for a user |
| GET | `/snapshot` | Latest camera frame as JPEG |
| GET | `/video_feed` | MJPEG stream with detection overlay |
| 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.
SmokeDetectionApp
+1 -1
Submodule SmokeDetectionApp updated: 7ace087582...5a4d26289b.
pipeline/sinks/__init__.py
+2 -1
View File
@@ -1,7 +1,7 @@
"""Sinks for pipeline output handling."""
from pipeline.sinks.alerts import FileAlertStore
from pipeline.sinks.database import DatabaseSink
from pipeline.sinks.database import DatabaseSink, DatabaseSmartEventSink
from pipeline.sinks.display import WindowDisplaySink
from pipeline.sinks.events import SmartEventSink
from pipeline.sinks.metrics import MetricsSink
@@ -10,6 +10,7 @@ from pipeline.sinks.video import VideoFileStore
__all__ = [
"DatabaseSink",
"DatabaseSmartEventSink",
"FileAlertStore",
"MetricsSink",
"ScreenshotSink",
pipeline/sinks/database.py
+50
View File
@@ -8,6 +8,7 @@ import cv2
from db.repository import Repository
from pipeline.ops.drawing import draw_alert_banner, draw_detections
from pipeline.sinks.events import SmartEventSink
from pipeline.types import Frame
logger = logging.getLogger(__name__)
@@ -72,3 +73,52 @@ class DatabaseSink:
def close(self):
pass # DB connection is managed by the repository
class DatabaseSmartEventSink(SmartEventSink):
"""Records event clips and stores clip path in DB as alert evidence."""
def __init__(
self,
db_repo: Repository,
fps: float,
user_id: int = 1,
camera_source: str = "Pipeline",
output_dir: str = "static/events",
pre_event_seconds: float = 2,
post_event_seconds: float = 3,
):
super().__init__(
output_dir=output_dir,
fps=fps,
pre_event_seconds=pre_event_seconds,
post_event_seconds=post_event_seconds,
)
self.db_repo = db_repo
self.user_id = user_id
self.camera_source = camera_source
def _trigger_event(self, trigger_frame: Frame):
was_recording = self.recording
super()._trigger_event(trigger_frame)
# Save one DB row per newly started event clip
if was_recording or not self.recording or self.current_event_id is None:
return
ticket = trigger_frame.ticket
if not ticket:
return
filename = f"event_{self.current_event_id}.mp4"
web_path = f"/static/events/{filename}"
try:
self.db_repo.save_detection(
user_id=self.user_id,
image_path=web_path,
confidence=ticket.confidence,
camera_source=self.camera_source,
)
logger.info("Saved event alert to DB: %s", web_path)
except Exception as e:
logger.error("Failed to save event alert to DB: %s", e)
reports/live_stress_latest.json
+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
+117 -245
View File
@@ -15,122 +15,27 @@ Usage:
import argparse
import logging
import os
import socket
import threading
import time
from contextlib import asynccontextmanager
from pathlib import Path
import cv2
from fastapi import FastAPI, Form, Response
from fastapi.responses import StreamingResponse
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from db import SQLiteRepository
from db.repository import Repository
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
from pipeline.sinks import FileAlertStore, MetricsSink, SmartEventSink
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 pipeline.types import Frame
from server.routes import router as api_router
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Global state (set during lifespan)
# ---------------------------------------------------------------------------
db: Repository = SQLiteRepository() # default: data/smoke.db
pipeline = None # set at startup
source: ThreadedStreamSource | None = None # set at startup
live_frame: Frame | None = None # latest frame for MJPEG / snapshot
# ---------------------------------------------------------------------------
# Sink that exposes frames to the web layer
# ---------------------------------------------------------------------------
class LiveFrameSink(Sink):
"""Captures the latest frame so API endpoints can stream it."""
def handle(self, frame: Frame):
global live_frame
live_frame = frame
def close(self):
pass
class DatabaseSmartEventSink(SmartEventSink):
"""Records event clips and stores clip path in DB as alert evidence."""
def __init__(
self,
db_repo: Repository,
fps: float,
user_id: int = 1,
camera_source: str = "Pipeline",
output_dir: str = "static/events",
):
super().__init__(output_dir=output_dir, fps=fps)
self.db_repo = db_repo
self.user_id = user_id
self.camera_source = camera_source
def _trigger_event(self, trigger_frame: Frame):
was_recording = self.recording
super()._trigger_event(trigger_frame)
# Save one DB row per newly started event clip
if was_recording or not self.recording or self.current_event_id is None:
return
ticket = trigger_frame.ticket
if not ticket:
return
filename = f"event_{self.current_event_id}.mp4"
web_path = f"/static/events/{filename}"
try:
self.db_repo.save_detection(
user_id=self.user_id,
image_path=web_path,
confidence=ticket.confidence,
camera_source=self.camera_source,
)
logger.info("Saved event alert to DB: %s", web_path)
except Exception as e:
logger.error("Failed to save event alert to DB: %s", e)
def _get_lan_address() -> tuple[str, str]:
"""Return (ip, hostname) for the primary LAN interface."""
ip = "127.0.0.1"
try:
# Open a UDP socket to an external IP (doesn't actually send anything)
# to figure out which local NIC the OS would route through.
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
except Exception:
pass
hostname = socket.getfqdn()
return ip, hostname
# ---------------------------------------------------------------------------
# Pipeline runner (background thread)
# ---------------------------------------------------------------------------
def _run_pipeline():
"""Blocking call: runs in a daemon thread."""
assert pipeline is not None and source is not None
try:
pipeline.process_stream(source)
except Exception: # noqa: BLE001
logger.exception("Pipeline crashed")
# ---------------------------------------------------------------------------
# CLI args (parsed before uvicorn when run as __main__)
@@ -174,8 +79,6 @@ _args = argparse.Namespace(
# ---------------------------------------------------------------------------
@asynccontextmanager
async def app_lifespan(_app: FastAPI):
global pipeline, source # noqa: PLW0603
logging.basicConfig(
level=logging.DEBUG if _args.debug else logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
@@ -185,7 +88,7 @@ async def app_lifespan(_app: FastAPI):
# 1. Database
logger.info("Initializing database...")
db.initialize()
rt.db.initialize()
# 2. Models
logger.info("Loading models...")
@@ -194,44 +97,105 @@ async def app_lifespan(_app: FastAPI):
verifier = YoloClassifier(_args.verify_model)
logger.info("Models loaded.")
# 3. Source (webcam / RTSP)
cam = int(_args.source) if _args.source.isdigit() else _args.source
source = ThreadedStreamSource(cam)
logger.info("Camera source ready: %s (%.1f FPS)", _args.source, source.fps)
# 3. Configure live pipeline manager
def source_factory(source_spec: str) -> ThreadedStreamSource:
cam = int(source_spec) if source_spec.isdigit() else source_spec
return ThreadedStreamSource(cam)
# 4. Sinks
sinks: list[Sink] = [
LiveFrameSink(),
DatabaseSmartEventSink(db, fps=source.fps, output_dir="static/events"),
MetricsSink(log_interval=200),
FileAlertStore("alerts.jsonl"),
]
def sink_factory(src: ThreadedStreamSource) -> list[Sink]:
return [
rt.LiveFrameSink(),
DatabaseSmartEventSink(rt.db, fps=src.fps, output_dir="static/events"),
MetricsSink(log_interval=200),
FileAlertStore("alerts.jsonl"),
]
# 5. Pipeline config
config = PipelineConfig(
idle_interval=0.5,
active_interval=0.0,
cooldown_frames=int(source.fps * 2),
def pipeline_factory(src: ThreadedStreamSource, 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,
)
# 6. Build pipeline
if _args.staged or _args.vlm:
vlm_key = _args.vlm_key or os.environ.get(
"OPENAI_API_KEY" if _args.vlm_provider == "openai" else "ANTHROPIC_API_KEY"
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)),
)
pipeline = StageBasedPipeline(
person_model=person_model,
cigarette_model=cig_model,
verifier_model=verifier,
vlm_api_key=vlm_key if _args.vlm else None,
vlm_provider=_args.vlm_provider,
vlm_model=_args.vlm_model,
use_vlm=_args.vlm,
sinks=sinks,
config=config,
)
else:
pipeline = SmokingDetectionPipeline(
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,
@@ -239,13 +203,18 @@ async def app_lifespan(_app: FastAPI):
config=config,
)
# 7. Launch pipeline in background thread
t = threading.Thread(target=_run_pipeline, daemon=True, name="pipeline")
t.start()
logger.info("Pipeline running in background.")
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 = _get_lan_address()
ip, hostname = rt.get_lan_address()
port = _args.port if hasattr(_args, "port") else 8000
logger.info("=" * 50)
logger.info(" Server ready")
@@ -259,10 +228,9 @@ async def app_lifespan(_app: FastAPI):
# Shutdown
logger.info("Shutting down...")
source.stop()
for s in sinks:
s.close()
db.close()
rt.manager.stop()
rt.clip_jobs.shutdown()
rt.db.close()
logger.info("Goodbye.")
@@ -270,111 +238,15 @@ async def app_lifespan(_app: FastAPI):
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Smoking Detection", lifespan=app_lifespan) # type: ignore
app.include_router(api_router)
# Serve static assets (alert images, CSS/JS, etc.)
Path("static/alerts").mkdir(parents=True, exist_ok=True)
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")
# -- Auth ------------------------------------------------------------------
@app.get("/ping")
def ping():
"""Health check for the mobile app to verify connectivity."""
return {"status": "ok", "name": "SmokingDetection"}
@app.post("/register")
def register(username: str = Form(...), password: str = Form(...)):
try:
if db.get_user_by_username(username):
return {"status": "fail", "message": "Username already taken"}
new_id = db.create_user(username, password)
return {"status": "success", "user_id": new_id, "message": "Account created"}
except Exception as e:
return {"status": "error", "message": str(e)}
@app.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
try:
user = db.get_user_by_credentials(username, password)
if user:
return {"status": "success", "user_id": user.id, "username": user.username}
return {"status": "fail", "message": "Invalid username or password"}
except Exception as e:
return {"status": "error", "message": str(e)}
# -- Detection history -----------------------------------------------------
@app.get("/history")
def get_history(user_id: int | None = None):
try:
records = db.get_detections(user_id)
return [
{
"id": r.id,
"user_id": r.user_id,
"image_path": r.image_path,
"confidence": r.confidence,
"timestamp": r.timestamp,
"camera_source": r.camera_source,
}
for r in records
]
except Exception:
return []
@app.post("/clear_history")
def clear_history():
try:
deleted = db.clear_detections()
logger.info("Cleared %d records", deleted)
return {"status": "success", "message": "History cleared"}
except Exception as e:
return {"status": "error", "message": str(e)}
# -- Video / Snapshot ------------------------------------------------------
@app.get("/snapshot")
def snapshot():
"""Return the latest frame as a JPEG."""
if live_frame is not None:
_, jpg = cv2.imencode(".jpg", live_frame.data)
return Response(content=jpg.tobytes(), media_type="image/jpeg")
return Response(status_code=404)
@app.get("/video_feed")
def video_feed():
"""MJPEG stream - drop into an <img> tag."""
def generate():
while True:
if live_frame is not None:
from pipeline.ops.drawing import draw_detections
img = live_frame.data.copy()
draw_detections(img, live_frame.detections)
_, jpg = cv2.imencode(".jpg", img)
yield (
b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n"
)
time.sleep(0.033) # ~30 FPS cap
return StreamingResponse(
generate(), media_type="multipart/x-mixed-replace;boundary=frame"
)
# ---------------------------------------------------------------------------
# Entry-point
# ---------------------------------------------------------------------------
server/__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
+237
View File
@@ -0,0 +1,237 @@
"""API route definitions for the smoking detection server."""
import shutil
import time
from pathlib import Path
import cv2
from fastapi import APIRouter, File, Form, Response, UploadFile
from fastapi.responses import StreamingResponse
import server.runtime as rt
router = APIRouter()
@router.get("/ping")
def ping():
"""Health check for the mobile app to verify connectivity."""
return {"status": "ok", "name": "SmokingDetection"}
@router.post("/register")
def register(username: str = Form(...), password: str = Form(...)):
try:
if rt.db.get_user_by_username(username):
return {"status": "fail", "message": "Username already taken"}
new_id = rt.db.create_user(username, password)
return {"status": "success", "user_id": new_id, "message": "Account created"}
except Exception as e:
return {"status": "error", "message": str(e)}
@router.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
try:
user = rt.db.get_user_by_credentials(username, password)
if user:
return {"status": "success", "user_id": user.id, "username": user.username}
return {"status": "fail", "message": "Invalid username or password"}
except Exception as e:
return {"status": "error", "message": str(e)}
@router.get("/history")
def get_history(user_id: int | None = None):
try:
records = rt.db.get_detections(user_id)
return [
{
"id": r.id,
"user_id": r.user_id,
"image_path": r.image_path,
"confidence": r.confidence,
"timestamp": r.timestamp,
"camera_source": r.camera_source,
}
for r in records
]
except Exception:
return []
@router.post("/clear_history")
def clear_history(user_id: int | None = Form(default=None)):
try:
deleted = rt.db.clear_detections(user_id)
return {
"status": "success",
"message": "History cleared",
"deleted": deleted,
"user_id": user_id,
}
except Exception as e:
return {"status": "error", "message": str(e)}
@router.get("/live/status")
def live_status():
"""Get current live pipeline status."""
return {"status": "success", "live": rt.manager.status()}
@router.post("/live/start")
def live_start(source: str | None = Form(default=None)):
"""Start live pipeline (optionally with source override)."""
ok, message = rt.manager.start(source)
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.post("/live/stop")
def live_stop():
"""Stop live pipeline."""
ok, message = rt.manager.stop()
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.post("/live/pause")
def live_pause():
"""Pause live pipeline (stops source, resumable)."""
ok, message = rt.manager.pause()
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.post("/live/resume")
def live_resume():
"""Resume paused live pipeline."""
ok, message = rt.manager.resume()
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.post("/live/switch_source")
def live_switch_source(source: str = Form(...)):
"""Switch active live source (webcam index or RTSP URL)."""
ok, message = rt.manager.switch_source(source)
return {
"status": "success" if ok else "fail",
"message": message,
"live": rt.manager.status(),
}
@router.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
+538
View File
@@ -0,0 +1,538 @@
"""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
from db import SQLiteRepository
from db.repository import Repository
from pipeline.protocols import Sink
from pipeline.sources.file import FileStreamSource
from pipeline.sources.stream import ThreadedStreamSource
from pipeline.types import Frame
logger = logging.getLogger(__name__)
# Global runtime state
# (kept as module-level singletons for current single-instance server design)
db: Repository = SQLiteRepository()
pipeline = None
source: ThreadedStreamSource | None = None
live_frame: Frame | None = None
SourceFactory = Callable[[str], ThreadedStreamSource]
SinkFactory = Callable[[ThreadedStreamSource], list[Sink]]
PipelineFactory = Callable[[ThreadedStreamSource, list[Sink]], Any]
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 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.running = False
self.starting = False
self.stopping = False
self.paused = False
self.last_error: str | None = None
def configure(
self,
source_factory: SourceFactory,
sink_factory: SinkFactory,
pipeline_factory: PipelineFactory,
initial_source: str = "0",
):
with self._lock:
self._source_factory = source_factory
self._sink_factory = sink_factory
self._pipeline_factory = pipeline_factory
self._source_spec = initial_source
def _close_sinks(self, 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
running = self.running
paused = self.paused
if running:
self.stop()
return self.start(source_spec)
if paused:
return True, f"Source set to {source_spec} (will apply on resume)"
return self.start(source_spec)
def status(self) -> dict[str, Any]:
with self._lock:
return {
"running": self.running,
"starting": self.starting,
"stopping": self.stopping,
"paused": self.paused,
"source": self._source_spec,
"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")