Traffic Violation Backend

A FastAPI service that ingests traffic-violation detections from a computer-vision pipeline, lets reviewers approve or reject each case, generates challan mocks, exposes analytics for the dashboard, and broadcasts live case events over WebSocket.

Status: Day 1-7 — All 8 endpoints implemented (Day 1-7 complete).

---

Highlights

• FastAPI + SQLAlchemy 2.0 + Pydantic v2 on Python 3.12.
• DB-portable: SQLite by default, Postgres via a single env var. No code changes.
• Zero infra: no auth, no microservices, no message queue. Suitable for a hackathon.
• Contract-first: docs/api-contract.md is the source of truth; every PR that changes an endpoint shape edits the contract in the same PR.
• Tested: 72 tests passing (smoke + evidence store + case service + analytics + cases endpoints + challan).

---

API Surface (v1)

#	Method	Path	Status	Planned (Day)
1	POST	/api/ingest	201 / 200	D1-2 (done)
2	GET	/api/cases	200	D3-5 (done)
3	GET	/api/cases/{case_id}	200	D3-5 (done)
4	PATCH	/api/cases/{case_id}/review	200	D3-5 (done)
5	POST	/api/challan	202	D5-7 (done)
6	GET	/api/analytics	200	D5-7 (done)
7	WS	/api/stream	101	D5-7 (done)
8	GET	/health	200	D1 (done)

Full request/response shapes, enums, and WS events: see docs/api-contract.md.

---

Quick Start

# from repo root
python -m venv .venv
& .venv\Scripts\Activate.ps1
pip install -r requirements.txt

Copy-Item .env.example .env             # default: SQLite at ./data/tv.db

uvicorn app.main:app --reload          # http://127.0.0.1:8000/docs
pytest -q                               # 72 passed

End-to-end smoke from PowerShell

$body = @'
{
  "client_request_id": "demo-001",
  "vehicle": { "plate": "dl01ab1234", "vehicle_type": "car", "color": "white" },
  "violation_type": "red_light_jump",
  "camera_id": "CAM-ND-042",
  "occurred_at": "2026-08-17T10:24:00Z",
  "evidence": [
    { "kind": "image", "ref": "s3://tv-bucket/demo-001/001.jpg" }
  ]
}
@
Invoke-WebRequest -Uri http://127.0.0.1:8000/api/ingest -Method POST `
  -ContentType application/json -Body $body
# -> 201 {"case_id":"...","status":"ingested","created_at":"...","replayed":false}

Invoke-WebRequest -Uri http://127.0.0.1:8000/health   # -> {"db":"ok", "status":"ok", ...}

# Analytics (implemented)
Invoke-WebRequest -Uri http://127.0.0.1:8000/api/analytics?group_by=type

# List cases (implemented)
Invoke-WebRequest -Uri http://127.0.0.1:8000/api/cases

# Review a case (implemented)
$review = @'{"decision":"approved","reviewer_id":"REV-001"}'@
Invoke-WebRequest -Uri http://127.0.0.1:8000/api/cases/{case_id}/review -Method PATCH -ContentType application/json -Body $review

# Issue challan (implemented)
$challan = @'{"case_id":"{case_id}","channel":"sms"}'@
Invoke-WebRequest -Uri http://127.0.0.1:8000/api/challan -Method POST -ContentType application/json -Body $challan

# WebSocket stream (implemented)
# Connect to ws://127.0.0.1:8000/api/stream

With Docker (Postgres)

# Start Postgres only
docker compose --profile db up -d

# Run API against Postgres
$env:DATABASE_URL="postgresql+psycopg://tv:tv@localhost:5432/tv"
uvicorn app.main:app --reload

# Full stack (API + Postgres in containers)
docker compose --profile db --profile api up -d

---

Project Structure

app/
+-- main.py                 FastAPI factory + lifespan (create_all on startup)
+-- config.py               pydantic-settings (.env)
+-- db.py                   SQLAlchemy engine / SessionLocal / Base / init_db / get_db
+-- api/                    One router per endpoint group
|   +-- health.py  ingest.py  cases.py  challan.py  analytics.py  stream.py  stub.py
+-- models/                 Case, EvidenceItem, Vehicle, Officer (SQLAlchemy ORM)
+-- schemas/                enums + Pydantic request/response models
+-- services/               Business logic (Indra owns routes, Alok owns services)
|   +-- ingest.py           Persist Case + cascades; idempotency on client_request_id
|   +-- evidence_store.py   Disk-backed media storage (frames/clips under media/{date}/)
|   +-- case_service.py     Status transitions, list filters, review workflow
|   +-- event_bus.py        In-process pub/sub for WebSocket broadcasting
|   +-- serialization.py    ORM -> Pydantic projectors
docs/
+-- api-contract.md         v1 contract — edit FIRST; code follows
tests/                      pytest; conftest pins a fresh SQLite DB
+-- test_smoke.py           Contract surface tests
+-- test_evidence_store.py  Media storage round-trip
+-- test_case_service.py    Status transitions, list filters, review workflow
+-- test_analytics.py       Group-by queries, date filters
+-- test_cases_endpoints.py Cases list/detail/review endpoints
+-- test_challan.py         Challan generation & eligibility
+-- test_stream.py          WebSocket event broadcasting (skipped: pytest-asyncio issue)
readme.md  plan.md  skill.md  requirements.txt  .env.example  docker-compose.yml  Dockerfile

For an in-depth tour including layering and conventions, read skill.md.

---

Configuration

All settings come from .env (or real environment variables):

Variable	Default	Description
DATABASE_URL	sqlite:///./data/tv.db	SQLAlchemy URL. Postgres: see below.
APP_NAME	tv-backend	Surfaced by /health.
APP_VERSION	v1	Surfaced by /health.
ENV	dev	dev / test / prod.
HOST / PORT	127.0.0.1 / 8000	For local uvicorn only.
LOG_LEVEL	info	root logger level.
CORS_ORIGINS	*	CSV or *.
MEDIA_DIR	./media	Evidence frames/clips storage root.

Postgres:

DATABASE_URL=postgresql+psycopg://tv:tv@localhost:5432/tv
pip install "psycopg[binary]"
uvicorn app.main:app --reload

---

Data Model (summary)

Case 1 --- 1 Vehicle
Case * --- 1 Officer (optional)
Case 1 --- * EvidenceItem

Key enumerations (see app/schemas/enums.py):

• CaseStatus: ingested | in_review | approved | rejected | challan_issued
• ViolationType: red_light_jump, speeding, no_helmet, wrong_side, seatbelt_violation, pollution_certificate, distracted_driving, illegal_parking, oversized_vehicle, other
• ReviewDecision: approved | rejected

---

Services (Alok's scope)

Module	Purpose
evidence_store.py	save_frame(bytes) -> evidence_id, save_clip(path) -> evidence_id, get_media(evidence_id) -> FileResponse. Files stored under media/{YYYY-MM-DD}/{uuid}.ext.
case_service.py	list_cases(filters, pagination), review_case(case_id, decision), transition_to_in_review(), is_challan_eligible(), mark_challan_issued().
analytics.py (endpoint)	Real SQLAlchemy GROUP BY on violation_type, camera_id, hour(occurred_at) with from/to date filters.
event_bus.py	In-process pub/sub: subscribe(), unsubscribe(), broadcast_sync() for sync endpoints, broadcast() for async.

---

Testing

pytest                 # all 72 tests
pytest -q              # quiet
pytest tests/test_smoke.py::test_health   # single test
pytest tests/test_analytics.py            # analytics tests
pytest tests/test_case_service.py         # case service tests
pytest tests/test_evidence_store.py       # evidence store tests
pytest tests/test_cases_endpoints.py      # cases endpoints
pytest tests/test_challan.py              # challan endpoints

Tests use FastAPI's TestClient with the lifespan enabled, so create_all() runs on the test DB once per session. The conftest.py sets DATABASE_URL to a fresh ./data/tv-test.db before any from app.main import .... Tables are cleared after each test for isolation.

Test file	Coverage
test_smoke.py	7 tests — health, ingest, idempotency, validation
test_evidence_store.py	8 tests — save/load/delete frames & clips, daily subdirs
test_case_service.py	24 tests — list filters, status transitions, review workflow, challan eligibility
test_analytics.py	10 tests — group_by type/camera/hour, date filters, multiple group_by, invalid params
test_cases_endpoints.py	15 tests — list/detail/review endpoints, filters, pagination, 404/409 errors
test_challan.py	6 tests — challan creation, eligibility checks, channels
test_stream.py	5 tests (skipped — pytest-asyncio event loop cleanup issue; WS verified manually)

---

Roadmap

The day-by-day delivery plan, owners, decisions log, risks, and progress log live in plan.md. Headline:

• Day 1-2 (done) contract + bootstrap + /api/ingest + /api/analytics + services + tests + Docker
• Day 3-5 (done) reviewer flow — list, detail, review endpoints (using case_service.py)
• Day 5-7 (done) challan, WS stream, hook WS broadcast into /ingest, /review, /challan
• Day 9-11 merge CV branch and run end-to-end

---

Contributing

Please read skill.md first — it covers layering, conventions, and where things live. Short version:

1. Contract first — edit docs/api-contract.md in the same PR if you change a route shape.
2. Branch feat/<area>-<short> off main.
3. Add a test under tests/ for any new public surface.
4. Reviewers are listed in plan.md → Owners table.
5. After merge, append a one-liner to plan.md → Progress Log and skill.md → Update Log.

---

License

Hackathon project — internal only. Not licensed for distribution.