A complete end-to-end system that transforms raw CCTV footage from physical retail stores into real-time business intelligence, specifically offline conversion rates.
# 1. Clone and navigate
cd store-intelligence
# 2. Install dependencies
pip install -r requirements.txt
# 3. Download YOLOv8n weights
python -c "from ultralytics import YOLO; YOLO('yolov8n.pt')"
# 4. Start API (with Docker Compose)
docker compose up
# 5. Send events
curl -X POST http://localhost:8000/events/ingest \
-H "Content-Type: application/json" \
-d '{"events": []}'store-intelligence/
├── data/ ← CCTV clips, POS data, store layout
├── pipeline/ ← Detection pipeline (YOLO + ByteTrack)
├── app/ ← FastAPI intelligence API
├── tests/ ← Pytest test suite
├── docs/ ← DESIGN.md, CHOICES.md
├── dashboard/ ← Live dashboard (bonus)
├── docker-compose.yml ← One-command system startup
├── Dockerfile ← Container definition
├── requirements.txt ← Python dependencies
└── README.md ← This file
- YOLOv8n person detection on each CCTV frame
- ByteTrack persistent person tracking (track_id per visitor)
- Re-ID logic to detect re-entries
- Emits structured JSON events
- 8 event types: ENTRY, EXIT, ZONE_ENTER, ZONE_EXIT, ZONE_DWELL, etc.
- Strict Pydantic validation with UUID, timestamps, session_seq ordering
/events/ingest— Batch event ingestion with deduplication/stores/{id}/metrics— Real-time conversion rate, visitor count, dwell time/stores/{id}/funnel— Entry → Zone → Billing → Purchase funnel/stores/{id}/heatmap— Zone frequency and dwell analysis/stores/{id}/anomalies— Queue spikes, conversion drops, dead zones/health— Service status and feed staleness monitoring
- Docker Compose with API + SQLite
- Structured logging on every request (trace_id, latency_ms, status_code)
- 70%+ pytest coverage with edge case handling
- Graceful error responses (503 on DB unavailable)
DESIGN.md— Architecture overview (2–3 AI-assisted decisions)CHOICES.md— 3 key architectural decisions
- Terminal UI (rich library) or Web UI showing real-time metrics
- QUICKSTART.md — Get started in 5 commands
- DEPLOYMENT.md — Complete deployment guide (local + Docker + testing)
- SUBMISSION_CHECKLIST.md — Pre-submission verification
- DESIGN.md — Architecture overview
- CHOICES.md — Key architectural decisions
# 1. Start API
python -m uvicorn app.main:app --port 8000
# 2. Run end-to-end demo
python scripts/demo.py
# 3. Start dashboard
streamlit run dashboard/streamlit_app.py
# → Open http://localhost:8501curl -X POST http://localhost:8000/events/ingest \
-H "Content-Type: application/json" \
-d '{
"events": [
{
"event_id": "550e8400-e29b-41d4-a716-446655440000",
"store_id": "STORE_BLR_002",
"camera_id": "CAM_ENTRY_01",
"visitor_id": "VIS_c8a2f1",
"event_type": "ENTRY",
"timestamp": "2026-03-03T14:22:10Z",
"zone_id": null,
"dwell_ms": 0,
"is_staff": false,
"confidence": 0.95,
"metadata": {"queue_depth": null, "sku_zone": null, "session_seq": 1}
}
]
}'curl http://localhost:8000/stores/STORE_BLR_002/metricsResponse:
{
"unique_visitors": 145,
"conversion_rate": 0.38,
"avg_dwell_ms": 480,
"queue_depth_current": 3,
"abandonment_rate": 0.12
}curl http://localhost:8000/stores/STORE_BLR_002/funnelResponse:
{
"entry": 145,
"zone_visit": 132,
"billing_queue": 85,
"purchased": 55
}curl http://localhost:8000/stores/STORE_BLR_002/heatmapResponse:
{
"SKINCARE": {"frequency": 87, "avg_dwell_ms": 420, "normalized": 85},
"MAKEUP": {"frequency": 62, "avg_dwell_ms": 310, "normalized": 58}
}curl http://localhost:8000/healthResponse:
{
"status": "healthy",
"stores": {
"STORE_BLR_002": {"last_event_timestamp": "2026-03-03T14:22:10Z", "lag_minutes": 2}
},
"warnings": []
}The North Star Metric this system tracks:
Conversion Rate = (Visitors who purchased) / (Total unique visitors)
Logic:
- Count unique
visitor_idwith ENTRY events - Correlate with POS transactions: if visitor was in billing zone within 5 min before transaction → purchased
- Calculate:
conversion_rate = purchased_count / total_visitors
- Group Entry: Multiple people detected separately (3 ENTRY events for 3 people)
- Staff: Uniform color heuristic → flag
is_staff: true(excluded from conversion calculation) - Re-entry: Person exits then returns → REENTRY event with same
visitor_id - Partial Occlusion: Emit event with low confidence, don't silently drop
- Billing Queue: Track
queue_depthin real-time as integer - Empty Store: Return zero counts gracefully, no crashes
- Camera Overlap: Acknowledged; cross-camera Re-ID optional
Every API request includes structured logging:
{
"event": "request_complete",
"timestamp": "2026-03-03T14:22:10Z",
"trace_id": "550e8400-e29b-41d4-a716-446655440000",
"endpoint": "/events/ingest",
"method": "POST",
"status_code": 200,
"latency_ms": 142,
"inserted_events": 495,
"duplicates": 3,
"errors": 2
}# Run all tests with coverage
pytest --cov=app --cov=pipeline tests/
# Run specific test file
pytest tests/test_metrics.py -v
# Run with print statements
pytest tests/ -s- ✅ System Execution:
docker compose upstarts without errors - ✅ API Responsiveness: POST /events/ingest and GET /metrics return valid JSON
- ✅ Event Generation: Detection pipeline outputs JSON with correct schema
- ✅ Documentation: DESIGN.md and CHOICES.md both >250 words
- ✅ Stability: No crashes during basic execution
- Detection: YOLOv8n + ByteTrack (person tracking)
- API: FastAPI + Uvicorn (async web server)
- Database: SQLite + SQLAlchemy ORM
- Testing: pytest + pytest-cov
- Logging: structlog (structured JSON logs)
- Dashboard: rich (terminal UI) or HTML+JS (web UI)
- Container: Docker + Docker Compose
All tools are 100% open source and free.
- Place CCTV clips in
data/clips/ - Place POS transactions CSV in
data/pos_transactions.csv - Convert Excel store layout to
data/store_layout.json - Run
docker compose upto start the API - Run detection pipeline to generate events
- Ingest events via POST
/events/ingest - Query metrics via GET endpoints