Project Watershed

Edge-resilient IoT telemetry pipeline — buffers sensor data locally during connectivity loss, syncs to AWS IoT Core on reconnect, uses Claude Sonnet for anomaly detection.

Built by Skipper as a portfolio project.

Stack: Python · MQTT (Mosquitto) · AWS IoT Core · Claude Sonnet · Terraform · SQLite · Grafana

---

Architecture

Edge Device
  ├── Mosquitto (local MQTT broker)
  ├── collector.py   — normalizes sensor payloads from MQTT topics
  ├── buffer.py      — SQLite-backed offline buffer (survives power/connectivity loss)
  ├── sync.py        — probes connectivity; drains buffer → AWS IoT Core
  └── anomaly.py     — batches readings → Claude Sonnet for anomaly detection

Cloud (AWS)
  ├── AWS IoT Core   — ingests telemetry over MQTT/TLS
  ├── Timestream     — time-series storage
  └── Grafana        — dashboards

See docs/case-study.md for a full technical write-up.

---

Quick Start

1. Install dependencies

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

2. Configure

cp .env.example .env
# Edit .env — fill in AWS_IOT_ENDPOINT, ANTHROPIC_API_KEY, cert paths

3. Start the local MQTT broker

mosquitto -c mosquitto/mosquitto.conf

4. Run the agent

python -m agent.main

---

Testing

pytest tests/ -v

Tests are fully offline — no AWS or Anthropic credentials needed.

---

Infrastructure

IMPORTANT: Never run terraform apply without explicit approval. Always review terraform plan output first.

cd infra
terraform init
terraform plan
# terraform apply   ← only after reviewing the plan

Required AWS resources: IoT Core Thing + cert, Timestream DB/table, IAM role, IoT topic rule.

See infra/variables.tf for configurable inputs and create a terraform.tfvars to override defaults (gitignored).

---

Grafana Dashboard

Import dashboard/grafana-dashboard.json via Dashboards → Import in the Grafana UI. The dashboard expects an Amazon Timestream data source configured with read access to the watershed_telemetry database.

---

Project Structure

watershed/
├── agent/
│   ├── main.py        # entry point and event loop
│   ├── collector.py   # MQTT subscriber + payload normalizer
│   ├── buffer.py      # SQLite offline buffer
│   ├── sync.py        # connectivity probe + AWS IoT sync
│   ├── anomaly.py     # Claude Sonnet anomaly detection
│   └── config.py      # configuration from env vars
├── infra/             # Terraform (AWS IoT Core, Timestream)
├── mosquitto/         # local Mosquitto broker config
├── dashboard/         # Grafana dashboard JSON
├── docs/              # case study and portfolio post
├── tests/             # pytest unit tests
├── .env.example       # required environment variables
├── requirements.txt
└── CLAUDE.md          # project context for Claude Code

---

License

MIT