The Store Intelligence system is a real-time retail analytics platform built for Purplle offline stores. It processes video feeds from in-store cameras to generate behavioural events, which are ingested into a REST API that powers live dashboards and anomaly alerts for store operations teams.
The system has three logical layers:
┌─────────────────────────────────────────────────────────────────┐
│ CAMERAS (MP4 clips / RTSP streams) │
│ CAM3-entry CAM1-zone CAM2-zone CAM5-billing │
└───────────────────────┬─────────────────────────────────────────┘
│ frames
▼
┌─────────────────────────────────────────────────────────────────┐
│ DETECTION PIPELINE (pipeline/) │
│ │
│ detect.py │
│ ├── YOLOv8n ──► person bboxes per frame │
│ ├── IoUTracker ──► stable track_id per person │
│ ├── LineCrossDetector ──► ENTRY / EXIT events │
│ ├── ZoneMatcher (polygon) ──► ZONE_ENTERED/EXITED events │
│ └── QueueManager ──► QUEUE_COMPLETED/ABANDONED events │
│ │
│ emit.py ──► JSONL file + POST /events/ingest │
└───────────────────────┬─────────────────────────────────────────┘
│ JSON events
▼
┌─────────────────────────────────────────────────────────────────┐
│ API (app/) FastAPI + SQLAlchemy + SQLite │
│ │
│ POST /events/ingest ──► validate, dedup, persist │
│ GET /stores/{id}/metrics ──► live KPIs │
│ GET /stores/{id}/funnel ──► entry→purchase funnel │
│ GET /stores/{id}/anomalies ──► threshold-based alerts │
│ GET /health ──► liveness + feed lag │
└───────────────────────┬─────────────────────────────────────────┘
│ REST JSON
▼
┌─────────────────────────────────────────────────────────────────┐
│ DASHBOARD (dashboard/) │
│ index.html + app.js ──► auto-refresh every 5s via fetch() │
│ Chart.js for zone dwell bar chart │
└─────────────────────────────────────────────────────────────────┘
We use YOLOv8n (the nano variant) for person detection. It is the lightest model in the YOLOv8 family that still achieves competitive mAP on COCO for class 0 (person), making it suitable for near-real-time processing on CPU when running offline on video clips. The ultralytics library handles model downloading, inference, and result parsing.
Frames are sampled at configurable intervals (--skip-frames, default 2) to balance accuracy against processing time. A confidence threshold of 0.35 is used, calibrated to the output range observed from YOLOv8n: the model tends to produce noisy detections below 0.3, while genuine person detections typically exceed 0.4.
The lightweight IoU tracker (tracker.py) avoids the need to install external ByteTrack or DeepSORT packages. It uses:
- Greedy IoU matching between existing tracks and new detections in each frame
- A two-stage recovery: first match active tracks, then attempt to recover lost tracks with a relaxed IoU threshold
- A
max_ageparameter (default 30 frames) to keep tracks alive through occlusion - HSV histogram Re-ID computed over the torso region to detect re-entries after exit
This gives ByteTrack-like behaviour without C++ dependencies, making it runnable in any Python 3.11 environment.
Events are emitted to JSONL files and optionally POSTed to the API in batches (default 50 events per HTTP request) with retry logic (exponential back-off, 3 retries). The EventEmitter is a context manager ensuring the final batch is always flushed.
Event types are modelled to exactly match the sample events schema provided in sample_eventsbe42122.jsonl. Key design decisions:
- Entry/exit events use
id_token(string like "ID_00001") as the visitor identifier, matching the sample schema. This is derived from the tracker's integertrack_id. - Zone events use integer
track_id(consistent with the sample). - Queue events carry a
queue_event_idUUID for deduplication at ingest. store_code(e.g. "store_1008") is used in entry/exit;store_id(e.g. "ST1008") in zone/queue events — consistent with the sample data.- All timestamps are ISO 8601 with UTC timezone.
- Demographics (
gender_pred,age_pred,age_bucket) are optional to support cameras where face is not visible.
The FastAPI application is structured with one module per domain concern:
| Module | Responsibility |
|---|---|
ingestion.py |
Batch event ingest, validation, dedup |
metrics.py |
Real-time KPI computation |
funnel.py |
Stage-by-stage conversion funnel |
anomalies.py |
Threshold-based anomaly detection |
health.py |
Liveness and feed lag monitoring |
database.py |
SQLAlchemy models + session management |
models.py |
Pydantic schemas for all I/O |
Idempotent ingest: each event gets a deterministic SHA-256 dedup key derived from its fields. Posting the same event twice inserts once and returns duplicate_skipped: 1 on the second call.
Graceful degradation: all metrics endpoints return valid (empty) responses on an empty database — no endpoint ever returns null or crashes on missing data. Database errors return HTTP 503.
Three specific areas where AI (Kiro/Claude) shaped the design:
AI suggested implementing a lightweight IoU tracker rather than integrating ByteTrack directly. The reasoning: ByteTrack requires compiled C++ extensions that often fail on Windows paths with spaces. The IoU tracker replicates the two-stage matching (active → lost) that ByteTrack uses internally, without the build-time friction. This made the pipeline runnable on any machine with just pip install ultralytics opencv-python-headless numpy.
AI recommended HSV histogram comparison over the torso region for re-entry detection. Hue-Saturation histograms are robust to small lighting changes and can be computed in microseconds per frame. The Bhattacharyya coefficient was chosen as the similarity measure because it is bounded [0,1] and invariant to histogram scale, making it easy to set a meaningful threshold (0.55).
AI suggested the partial-success pattern for the /events/ingest endpoint: process valid events, accumulate errors for invalid ones, and return a structured {accepted, rejected, errors} response rather than failing the whole batch on a single bad event. This is critical in production where a pipeline may emit 500 events per batch and a single schema deviation should not block the remaining 499.
Camera Video (MP4)
│
▼ frame-by-frame read
┌──────────────┐
│ YOLOv8n │ ──► person bboxes [x1,y1,x2,y2, conf]
└──────────────┘
│
▼
┌──────────────┐
│ IoUTracker │ ──► track_id per person, state: active/lost/removed
└──────────────┘
│
├──[camera_type=entry]──► LineCrossDetector
│ │
│ direction=in ─────┼──► emit EntryEvent / ReentryEvent
│ direction=out ────┼──► emit ExitEvent
│
└──[camera_type=zone/billing]──► ZoneMatcher (polygon hit-test)
│
zone_id changed ───────┼──► emit ZoneEnteredEvent
│──► emit ZoneExitedEvent
│
zone_type=BILLING ─────┼──► QueueManager
│──► emit QueueCompletedEvent
│──► emit QueueAbandonedEvent
│
▼
EventEmitter ──► events.jsonl
│
└──[if api_url set]──► POST /events/ingest (batched, with retry)
│
▼
SQLite DB (events, sessions,
pos_transactions, anomalies)
│
┌─────────────┴──────────────┐
│ │
/metrics /anomalies
/funnel /health
│
▼
Dashboard (fetch every 5s)