feat: add FastAPI backend for city CCTV surveillance

- Camera CRUD with go2rtc RTSP stream registration
- Incident model with type/severity classification (traffic_accident, fight, etc.)
- VLM analysis endpoint: snapshot capture → Celery job → VLM → incident record
- LLM agent query endpoint with DB context injection
- Shift/daily report generator via LLM agent
- WebSocket real-time alert stream via Redis pub/sub
- MinIO snapshot storage service
- PostgreSQL + asyncpg async ORM (SQLAlchemy 2.0)
- Celery worker for async VLM jobs
- Docker Compose stack: backend, worker, db, redis, minio, go2rtc

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DXvYppU6bAHzS9R8HbTHdN

Author: claude

docker/docker-compose-backend.yml

@@ -0,0 +1,89 @@
+version: '3.8'
+
+services:
+  db:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_DB: namucam
+      POSTGRES_USER: namucam
+      POSTGRES_PASSWORD: namucam
+    volumes:
+      - pg_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U namucam"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  redis:
+    image: redis:7-alpine
+    volumes:
+      - redis_data:/data
+
+  minio:
+    image: minio/minio
+    command: server /data --console-address ":9001"
+    environment:
+      MINIO_ROOT_USER: minioadmin
+      MINIO_ROOT_PASSWORD: minioadmin
+    volumes:
+      - minio_data:/data
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+
+  go2rtc:
+    image: alexxit/go2rtc
+    ports:
+      - "1984:1984"
+    volumes:
+      - ./go2rtc.yaml:/config/go2rtc.yaml
+
+  backend:
+    build:
+      context: ../src/backend
+      dockerfile: Dockerfile
+    ports:
+      - "8000:8000"
+    environment:
+      DATABASE_URL: postgresql+asyncpg://namucam:namucam@db:5432/namucam
+      REDIS_URL: redis://redis:6379/0
+      MINIO_ENDPOINT: minio:9000
+      MINIO_ACCESS_KEY: minioadmin
+      MINIO_SECRET_KEY: minioadmin
+      GO2RTC_URL: http://go2rtc:1984
+      VLM_BASE_URL: ${VLM_BASE_URL:-http://host.docker.internal:5405}
+      LLM_BASE_URL: ${LLM_BASE_URL:-http://host.docker.internal:5407}
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_started
+    volumes:
+      - snapshots:/tmp/snapshots
+
+  worker:
+    build:
+      context: ../src/backend
+      dockerfile: Dockerfile
+    command: celery -A backend.workers.celery_app worker --loglevel=info --concurrency=4
+    environment:
+      DATABASE_URL: postgresql+asyncpg://namucam:namucam@db:5432/namucam
+      REDIS_URL: redis://redis:6379/0
+      MINIO_ENDPOINT: minio:9000
+      MINIO_ACCESS_KEY: minioadmin
+      MINIO_SECRET_KEY: minioadmin
+      GO2RTC_URL: http://go2rtc:1984
+      VLM_BASE_URL: ${VLM_BASE_URL:-http://host.docker.internal:5405}
+      LLM_BASE_URL: ${LLM_BASE_URL:-http://host.docker.internal:5407}
+    depends_on:
+      - db
+      - redis
+    volumes:
+      - snapshots:/tmp/snapshots
+
+volumes:
+  pg_data:
+  redis_data:
+  minio_data:
+  snapshots:

docker/go2rtc.yaml

@@ -0,0 +1,12 @@
+api:
+  listen: ":1984"
+
+# Streams are added dynamically via the go2rtc REST API
+# Example:
+# streams:
+#   cam_001: rtsp://192.168.1.100/stream1
+#   cam_002: rtsp://192.168.1.101/stream1
+streams: {}
+
+webrtc:
+  listen: ":8555"

src/backend/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libpq-dev gcc && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . /app/backend
+
+ENV PYTHONPATH=/app
+
+# API server
+CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

src/backend/__init__.py

@@ -0,0 +1,131 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime, timedelta, timezone
+
+from backend.db.database import get_db
+from backend.models.incident import Incident
+from backend.models.camera import Camera
+from backend.services.agent_service import query_agent
+
+router = APIRouter(prefix="/agent", tags=["agent"])
+
+
+class QueryRequest(BaseModel):
+    question: str
+    # Optional time window — defaults to last 24 hours
+    from_time: Optional[datetime] = None
+    to_time: Optional[datetime] = None
+
+
+class QueryResponse(BaseModel):
+    question: str
+    answer: str
+    incidents_in_context: int
+
+
+@router.post("/query", response_model=QueryResponse)
+async def query(payload: QueryRequest, db: AsyncSession = Depends(get_db)):
+    """Natural language query over incident data."""
+    now = datetime.now(timezone.utc)
+    from_time = payload.from_time or (now - timedelta(hours=24))
+    to_time = payload.to_time or now
+
+    # Build context from DB
+    incidents_result = await db.execute(
+        select(Incident)
+        .where(Incident.occurred_at >= from_time, Incident.occurred_at <= to_time)
+        .order_by(Incident.occurred_at.desc())
+        .limit(100)
+    )
+    incidents = incidents_result.scalars().all()
+
+    cameras_result = await db.execute(select(Camera))
+    cameras = {str(c.id): c.name for c in cameras_result.scalars().all()}
+
+    context = {
+        "query_window": {"from": from_time.isoformat(), "to": to_time.isoformat()},
+        "total_incidents": len(incidents),
+        "incidents": [
+            {
+                "id": str(inc.id),
+                "camera": cameras.get(str(inc.camera_id), str(inc.camera_id)),
+                "type": inc.incident_type,
+                "severity": inc.severity,
+                "description": inc.description,
+                "occurred_at": inc.occurred_at.isoformat(),
+                "resolved": inc.resolved,
+                "confidence": inc.confidence,
+            }
+            for inc in incidents
+        ],
+    }
+
+    answer = await query_agent(payload.question, context)
+
+    return QueryResponse(
+        question=payload.question,
+        answer=answer,
+        incidents_in_context=len(incidents),
+    )
+
+
+@router.post("/report")
+async def generate_report(
+    from_time: Optional[datetime] = None,
+    to_time: Optional[datetime] = None,
+    district: Optional[str] = None,
+    db: AsyncSession = Depends(get_db),
+):
+    """Generate a structured shift/daily report using the LLM agent."""
+    now = datetime.now(timezone.utc)
+    from_time = from_time or (now - timedelta(hours=8))
+    to_time = to_time or now
+
+    q = select(Incident).where(
+        Incident.occurred_at >= from_time,
+        Incident.occurred_at <= to_time,
+    )
+    result = await db.execute(q)
+    incidents = result.scalars().all()
+
+    cam_q = select(Camera)
+    if district:
+        cam_q = cam_q.where(Camera.district == district)
+    cam_result = await db.execute(cam_q)
+    cameras = {str(c.id): {"name": c.name, "location": c.location} for c in cam_result.scalars().all()}
+
+    context = {
+        "report_period": {"from": from_time.isoformat(), "to": to_time.isoformat()},
+        "district": district or "all",
+        "cameras": cameras,
+        "incidents": [
+            {
+                "camera": cameras.get(str(inc.camera_id), {}).get("name", "unknown"),
+                "type": inc.incident_type,
+                "severity": inc.severity,
+                "description": inc.description,
+                "occurred_at": inc.occurred_at.isoformat(),
+                "resolved": inc.resolved,
+            }
+            for inc in incidents
+            if str(inc.camera_id) in cameras
+        ],
+    }
+
+    prompt = (
+        "Generate a formal shift surveillance report based on the provided incident data. "
+        "Include: executive summary, incident breakdown by type and severity, "
+        "notable events, unresolved incidents requiring attention, and recommendations."
+    )
+
+    report_text = await query_agent(prompt, context)
+
+    return {
+        "period": {"from": from_time.isoformat(), "to": to_time.isoformat()},
+        "district": district or "all",
+        "total_incidents": len(context["incidents"]),
+        "report": report_text,
+    }

src/backend/api/__init__.py

@@ -0,0 +1,100 @@
+from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
+from sqlalchemy.ext.asyncio import AsyncSession
+from pydantic import BaseModel
+from typing import Optional
+from uuid import UUID
+import httpx
+import tempfile
+import os
+
+from backend.db.database import get_db
+from backend.models.camera import Camera
+from backend.models.incident import AnalysisJob
+from backend.workers.tasks import analyze_frame_task
+from backend.config import settings
+
+router = APIRouter(prefix="/analyze", tags=["analyze"])
+
+
+class AnalyzeRequest(BaseModel):
+    camera_id: UUID
+    custom_prompt: Optional[str] = None
+
+
+class AnalyzeResponse(BaseModel):
+    job_id: str
+    status: str
+    message: str
+
+
+class JobStatusResponse(BaseModel):
+    job_id: str
+    status: str
+    result: Optional[dict] = None
+    error: Optional[str] = None
+
+    class Config:
+        from_attributes = True
+
+
+@router.post("/trigger", response_model=AnalyzeResponse)
+async def trigger_analysis(payload: AnalyzeRequest, db: AsyncSession = Depends(get_db)):
+    """Grab a snapshot from the camera and queue a VLM analysis job."""
+    cam = await db.get(Camera, payload.camera_id)
+    if not cam:
+        raise HTTPException(status_code=404, detail="Camera not found")
+    if not cam.is_active:
+        raise HTTPException(status_code=400, detail="Camera is inactive")
+
+    # Capture snapshot from go2rtc
+    snap_path = await _capture_snapshot(cam.name)
+
+    # Create job record
+    job = AnalysisJob(camera_id=payload.camera_id, frame_path=snap_path, status="pending")
+    db.add(job)
+    await db.flush()
+
+    # Queue Celery task
+    task = analyze_frame_task.delay(
+        str(payload.camera_id),
+        snap_path,
+        str(job.id),
+    )
+    job.celery_task_id = task.id
+
+    return AnalyzeResponse(
+        job_id=str(job.id),
+        status="queued",
+        message=f"Analysis queued for camera '{cam.name}'",
+    )
+
+
+@router.get("/jobs/{job_id}", response_model=JobStatusResponse)
+async def get_job_status(job_id: UUID, db: AsyncSession = Depends(get_db)):
+    job = await db.get(AnalysisJob, job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return JobStatusResponse(
+        job_id=str(job.id),
+        status=job.status,
+        result=job.result,
+        error=job.error,
+    )
+
+
+async def _capture_snapshot(camera_name: str) -> str:
+    """Download a JPEG snapshot from go2rtc and save to a temp file."""
+    name = camera_name.replace(" ", "_")
+    url = f"{settings.GO2RTC_URL}/snapshot?src={name}"
+    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.get(url)
+            resp.raise_for_status()
+            tmp.write(resp.content)
+            tmp.flush()
+        return tmp.name
+    except Exception as exc:
+        tmp.close()
+        os.unlink(tmp.name)
+        raise HTTPException(status_code=502, detail=f"Failed to capture snapshot: {exc}")