project_edge

Imported from Skrisps26/project_edge (AGENTS.md). Install upstream with npx skills add Skrisps26/project_edge. Copyright stays with the author.

Road Anomaly Detection and Geotagging System

Edge + Fog + Cloud | Raspberry Pi + Laptop + Phone

Project Overview

Real-time road anomaly detection system. A Raspberry Pi with a camera detects potholes and road damage using a pre-trained YOLO model. Detections are geotagged using GPS published from the phone via OwnTracks MQTT. Data flows through a fog aggregation layer on the laptop into LocalStack (local AWS emulator) and is visualised on a React + Leaflet web dashboard.

Hardware:

• Raspberry Pi 4 — camera, YOLO inference, MQTT publisher (Edge layer)
• Laptop — MQTT broker, Dockerized fog container with CPU/memory limits, LocalStack, FastAPI, React frontend (Fog + Cloud layer)
• Phone — mobile GPS source via OwnTracks MQTT

All three devices connect to the phone's mobile hotspot.

Architecture

Phone OwnTracks app
  └── publishes location to MQTT topic owntracks/phone/gps
        ↓
Laptop Mosquitto broker
  └── OwnTracks MQTT → updates in-memory GPS for GET /gps

Raspberry Pi
  └── Pi Camera or USB webcam → YOLO model (pre-trained .tflite file, no training needed)
  └── Detects anomaly → GET /gps from Laptop → attach GPS → MQTT publish

Laptop - Fog Container (Docker on fog-cloud-net)
  └── MQTT subscribe → deduplicate within 10m radius → forward to LocalStack
  └── Enforced limits: 0.5 CPU, 512MB RAM, throttled 10Mbps/15ms network link

Laptop - LocalStack (port 4566)
  └── S3 bucket: anomaly-images (stores cropped frame JPGs)
  └── DynamoDB table: anomalies (stores metadata)

Laptop - FastAPI (port 8000)
  └── POST /gps        ← phone pushes GPS here
  └── GET  /gps        ← Pi reads latest GPS
  └── GET  /anomalies  ← dashboard reads anomaly list
  └── GET  /gps-page   ← serves phone_gps.html to phone browser

Laptop - React + Leaflet (port 5173)
  └── Map with severity-coloured pins (red=high, orange=medium, yellow=low)
  └── Polls FastAPI /anomalies every 5 seconds

Layer Mapping

Project Structure

road-anomaly-detection/
├── AGENTS.md
├── README.md
├── .env
├── .gitignore                      # must include .env and edge/model/*.tflite
├── requirements-laptop.txt
├── requirements-pi.txt
├── tests/
│   ├── test_gps_api.py
│   ├── test_dedup.py
│   ├── test_localstack.py
│   ├── test_mqtt_pipeline.py
│   ├── test_detection.py
│   └── test_frontend_api.py
├── edge/
│   ├── detect.py                   # YOLO inference loop
│   ├── mqtt_publisher.py           # publishes anomaly JSON
│   └── model/
│       └── model.tflite            # pre-trained TFLite model (bring your own)
├── fog/
│   ├── fog_node.py                 # MQTT subscriber + dedup + cloud forward
│   └── dedup.py                    # Haversine distance dedup logic
├── cloud/
│   ├── localstack_client.py        # boto3 pointed at localhost:4566
│   └── setup_localstack.py         # creates S3 bucket + DynamoDB table
├── api/
│   └── main.py                     # FastAPI: /gps, /anomalies
├── gps/
│   └── owntracks.py                # MQTT listener for phone GPS
└── frontend/
    ├── src/
    │   ├── App.jsx
    │   ├── Map.jsx
    │   └── api.js
    └── package.json

Laptop stack (includes Mosquitto)

docker compose up -d

Apply fog traffic shaping

bash scripts/throttle_network.sh fog-node

LocalStack init

source .venv/bin/activate python cloud/setup_localstack.py

FastAPI (laptop)

pip install -r requirements-laptop.txt uvicorn api.main:app --host 0.0.0.0 --port 8000 --reload

React dashboard (laptop)

cd frontend && npm install && npm run dev

Edge detection (Pi)

python edge/detect.py --broker $LAPTOP_MDNS_HOST --confidence 0.45

Run all tests

pytest tests/ -v

Verify LocalStack manually

awslocal s3 ls s3://anomaly-images awslocal dynamodb scan --table-name anomalies

---

## Tech Stack — Use These Exactly

- **Python 3.11** on both Pi and laptop
- **tflite-runtime** for inference on Pi — `import tflite_runtime.interpreter as tflite`
- **The model is pre-trained** — load `edge/model/model.tflite` and run inference only. No training code.
- On laptop (for tests): use `tensorflow` full package if tflite-runtime unavailable
- **paho-mqtt** for all MQTT communication
- **FastAPI + uvicorn** for the API layer
- **boto3** with `endpoint_url="http://localhost:4566"` for LocalStack
- **LocalStack free tier** — only S3 and DynamoDB
- **React 18 + Vite** for frontend
- **react-leaflet** for map — not Google Maps, not Mapbox
- **pytest** for all tests

---

## MQTT Message Schema

```json
{
  "anomaly_id": "uuid-v4-string",
  "timestamp": "2026-04-11T10:30:00Z",
  "lat": 12.9234,
  "lng": 80.1276,
  "severity": "high",
  "confidence": 0.87,
  "image_key": "1712830200_12.9234_80.1276.jpg"
}

Severity mapping: confidence >= 0.75 → "high" | 0.55–0.74 → "medium" | < 0.55 → "low"

DynamoDB Schema

Table: anomalies, partition key: anomaly_id (String) Attributes: timestamp, lat, lng, severity, confidence, image_key, reported_at

Environment Variables (.env)

LAPTOP_IP=192.168.43.x
MQTT_BROKER=192.168.43.x
MQTT_PORT=1883
MQTT_TOPIC=road/anomaly
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_DEFAULT_REGION=us-east-1
LOCALSTACK_ENDPOINT=http://localhost:4566
S3_BUCKET=anomaly-images
DYNAMO_TABLE=anomalies
GPS_SERVER_URL=http://192.168.43.x:8000/gps
DEDUP_RADIUS_METRES=10
CONFIDENCE_THRESHOLD=0.45

Do

• Load the TFLite model once at startup using tflite.Interpreter('edge/model/model.tflite') and call allocate_tensors() — do not reload per frame
• Use interpreter.set_tensor() / interpreter.invoke() / interpreter.get_tensor() for inference
• Warm up model with a dummy numpy zeros frame at startup before the detection loop
• Use Haversine formula for dedup — 10 metre threshold
• Store images in S3 as {unix_timestamp}_{lat}_{lng}.jpg
• OwnTracks publishes GPS over MQTT to the owntracks/# topic tree
• Bind all services to 0.0.0.0 so Pi and phone can reach them on the hotspot
• Read all config from .env using python-dotenv
• Write pytest tests for every module — see step-by-step section
• GPS stored in FastAPI is an in-memory dict {"lat": float, "lng": float, "updated_at": str}
• If GPS not received yet, Pi retries for up to 5 seconds then skips that detection
• Use docker compose up -d for the laptop stack, then bash scripts/throttle_network.sh fog-node to apply fog emulation

Don't

• Do NOT write any model training code — the model is already trained, load and infer only
• Do NOT use real AWS cloud — LocalStack only at localhost:4566
• Do NOT hardcode any IP addresses — always read from .env
• Do NOT use Google Maps or any API-key-gated map tile service
• Do NOT use ultralytics or full PyTorch — use tflite-runtime only on the Pi
• Do NOT commit .env or edge/model/model.tflite to git
• Do NOT use sys.exit() on errors — raise exceptions and log them

Gotchas and Known Issues

• Phone GPS in browser: Browser geolocation is no longer the main path. Use OwnTracks on the phone and MQTT to the laptop broker.
• GPS lag: Coordinates may be 1-2 seconds old at 30 km/h, causing ~8-10 metre drift. Document as a known limitation. Do not attempt to compensate.
• Mosquitto on Mac: Binds to localhost by default. Add listener 1883 0.0.0.0 and allow_anonymous true to /usr/local/etc/mosquitto/mosquitto.conf.
• LocalStack persistence: Data lost on restart. Set LOCALSTACK_VOLUME_DIR to persist.
• TFLite first inference: Takes 1-2 seconds. Always warm up with a dummy frame on startup.
• Pi camera: Use picamera2 library on Pi OS Bookworm, not the legacy picamera.
• USB webcam fallback: The Pi detector may use cv2.VideoCapture(0) when picamera2 is unavailable.

Step-by-Step Build Instructions

Follow this order strictly. Do not proceed to the next step until all tests for the current step pass.

STEP 1 — Fog Deduplication Logic

File: fog/dedup.py

Implement the Haversine distance formula. Expose one function: is_duplicate(new_lat, new_lng, existing_anomalies, radius_metres=10) -> bool Returns True if any anomaly in existing_anomalies (list of dicts with lat/lng) is within radius_metres of the new point.

Tests to write (tests/test_dedup.py):

• Two identical coordinates → returns True
• Two points 5m apart → returns True (within 10m threshold)
• Two points 15m apart → returns False (outside threshold)
• Two points 9.9m apart → returns True (edge case, just inside)
• Two points 10.1m apart → returns False (edge case, just outside)
• Empty existing list → returns False
• Real Chennai coords (12.9236, 80.1275) vs (12.9236, 80.1276) — ~9m → returns True
• Custom radius_metres=5 with 7m separation → returns False (outside custom threshold)

Gate: pytest tests/test_dedup.py -v — all 8 tests must pass before Step 2.

STEP 2 — LocalStack Setup and Cloud Client

Files: cloud/setup_localstack.py, cloud/localstack_client.py

setup_localstack.py: Creates S3 bucket anomaly-images and DynamoDB table anomalies. Must be idempotent — safe to run multiple times without error.

localstack_client.py: Exposes:

• save_anomaly(anomaly: dict, image_bytes: bytes) -> bool
• get_all_anomalies() -> list[dict]

Tests to write (tests/test_localstack.py):

• Setup script creates the S3 bucket without error
• Setup script creates the DynamoDB table without error
• Setup script runs twice without error (idempotent)
• save_anomaly with valid data returns True
• save_anomaly → item retrievable from DynamoDB with correct fields
• save_anomaly → image retrievable from S3 with correct key
• get_all_anomalies returns empty list when no anomalies exist
• get_all_anomalies returns correct count after 3 saves
• save_anomaly with missing required field raises ValueError, does not write partial data

Gate: pytest tests/test_localstack.py -v — all 9 tests must pass before Step 3. LocalStack must be running (localstack start -d) for these tests.

STEP 3 — FastAPI GPS and Anomaly Endpoints

File: api/main.py

Endpoints:

• POST /gps body {"lat": float, "lng": float} — stores in memory, returns 200
• GET /gps — returns latest GPS dict or 404 if none received yet
• GET /anomalies — returns list from DynamoDB via get_all_anomalies()
• GET /gps — returns the latest GPS state held in memory
• api/owntracks.py — OwnTracks listener that updates the same in-memory GPS state

Tests to write (tests/test_gps_api.py):

• POST /gps with valid data → 200 response
• GET /gps before any POST → 404 response
• POST /gps then GET /gps → returns same lat/lng values
• POST /gps twice → GET /gps returns second (most recent) values
• POST /gps with missing lat field → 422 validation error
• POST /gps with string instead of float for lat → 422 validation error
• GET /anomalies → 200 and returns a list (even if empty)
• POST /gps and OwnTracks MQTT messages update the in-memory GPS state
• CORS headers present on responses (Pi and phone access from different origin)

Gate: pytest tests/test_gps_api.py -v — all 9 tests must pass before Step 4.

STEP 4 — Fog Node: MQTT Subscribe + Dedup + Forward

File: fog/fog_node.py

Subscribes to MQTT_TOPIC. On each message:

1. Parse JSON payload
2. Load existing anomalies from DynamoDB
3. Run is_duplicate() — if duplicate, log and discard
4. If not duplicate, call save_anomaly() and log success

Tests to write (tests/test_mqtt_pipeline.py):

• Valid MQTT JSON message is parsed correctly into a dict
• Invalid JSON message is caught and logged — node does not crash
• Message missing lat field is caught and logged
• Message with confidence 0.8 → severity is "high"
• Message with confidence 0.6 → severity is "medium"
• Message with confidence 0.4 → severity is "low"
• Duplicate message (same coords within 10m) → save_anomaly NOT called (mock it)
• Non-duplicate message → save_anomaly IS called exactly once (mock it)
• 10 rapid valid messages processed without crash (stress test with mock broker)

Gate: pytest tests/test_mqtt_pipeline.py -v — all 9 tests must pass before Step 5.

STEP 5 — Edge Detection (Raspberry Pi)

Files: edge/detect.py, edge/mqtt_publisher.py

detect.py:

• Load model: interpreter = tflite.Interpreter('edge/model/model.tflite'); interpreter.allocate_tensors()
• Get input/output details with get_input_details() / get_output_details()
• Warm up with one dummy numpy zeros frame (resized to model's expected input shape) before the loop
• Open Pi camera using picamera2
• Loop: capture frame → resize to input shape → set_tensor → invoke() → get_tensor for output → if detection above threshold, call publish()
• Handle camera open failure gracefully with a logged error

mqtt_publisher.py:

• publish(frame, confidence, lat, lng) -> dict
• Assigns UUID, ISO timestamp, severity from confidence
• Encodes frame as JPEG bytes
• Builds JSON payload per schema
• Publishes to MQTT_TOPIC
• Returns the anomaly dict

Tests to write (tests/test_detection.py):

• Interpreter loads from edge/model/model.tflite without error (skip with pytest.mark.skipif if file missing)
• get_input_details() returns a list with at least one entry
• get_output_details() returns a list with at least one entry
• Running inference on a correctly-shaped dummy numpy zeros frame does not crash
• invoke() completes without error on dummy input
• confidence 0.8 → severity "high"; 0.6 → "medium"; 0.4 → "low"
• publish() returns a dict with all required fields: anomaly_id, timestamp, lat, lng, severity, confidence, image_key
• publish() generates a valid UUID4 for anomaly_id
• MQTT client publish called exactly once per publish() call (mock paho client)
• image_key format: {unix_timestamp}_{lat}_{lng}.jpg

Gate: pytest tests/test_detection.py -v — all 9 tests must pass before Step 6.

STEP 6 — React + Leaflet Dashboard

Files: frontend/src/App.jsx, frontend/src/Map.jsx, frontend/src/api.js

api.js: fetchAnomalies() — calls GET /anomalies, returns the list.

Map.jsx: Leaflet map centred on Chennai (12.9236, 80.1275), zoom 13. Circle marker per anomaly — Red ("high"), Orange ("medium"), Yellow ("low"). Popup shows: severity, confidence %, timestamp, lat/lng.

App.jsx: Polls fetchAnomalies() every 5 seconds. Shows total anomaly count. Shows last updated timestamp.

Tests to write (tests/test_frontend_api.py): Integration tests against FastAPI /anomalies:

• GET /anomalies with 0 stored anomalies → returns []
• GET /anomalies with 1 stored anomaly → returns list with 1 item
• GET /anomalies with 3 stored anomalies → returns list with 3 items
• Each anomaly has all required fields (anomaly_id, lat, lng, severity, confidence, timestamp, image_key)
• severity values are only "low", "medium", or "high" — no other values accepted
• lat and lng are floats, not strings
• confidence is a float between 0.0 and 1.0
• timestamp is a string in ISO 8601 format

React build smoke test:

• npm run build in frontend/ exits with code 0

Gate: pytest tests/test_frontend_api.py -v && cd frontend && npm run build All tests pass and build succeeds.

FINAL GATE — Full System Check

pytest tests/ -v

Every single test across all 6 modules must pass.

Then perform a manual end-to-end verification:

1. Start LocalStack, Mosquitto, FastAPI, Fog Node, React Dev Server on laptop
2. Configure OwnTracks on the phone and confirm the Pi can read GET /gps
3. On Pi, run edge/detect.py — point camera at a road damage image
4. Confirm anomaly appears on the React map within 10 seconds

Manual Test Commands (Quick Reference)

# Inject a fake GPS reading
curl -X POST http://localhost:8000/gps \
  -H "Content-Type: application/json" \
  -d '{"lat": 12.9236, "lng": 80.1275}'

# Inject a fake anomaly via MQTT (bypasses Pi entirely)
mosquitto_pub -h localhost -t road/anomaly -m \
  '{"anomaly_id":"test-001","timestamp":"2026-04-11T10:00:00Z","lat":12.9236,"lng":80.1275,"severity":"high","confidence":0.89,"image_key":"test.jpg"}'

# Check DynamoDB
awslocal dynamodb scan --table-name anomalies

# Check S3
awslocal s3 ls s3://anomaly-images