Smart Home Sensor Monitoring System

Real-time IoT monitoring API designed for smart home scenarios. The project ingests sensor data, processes it asynchronously, detects anomalies, and exposes monitoring and alerting endpoints.

This project was built to demonstrate advanced backend engineering skills with FastAPI, asynchronous concurrency, object-oriented design, and MLOps-ready architecture.

How To Run In 1 Minute

Local (Windows PowerShell)

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload

Then open:

http://127.0.0.1:8000/docs

Docker

docker compose up --build -d
docker compose ps

Check API:

Invoke-RestMethod -Uri "http://127.0.0.1:8000/health" -Method Get | ConvertTo-Json -Depth 5

What This Project Does

• Collects sensor readings for temperature, humidity, motion, and energy
• Processes incoming events using an asynchronous queue and concurrent workers
• Normalizes sensor values by sensor type for consistent downstream analytics
• Detects anomalies with a rolling z-score approach
• Exposes operational endpoints for health, metrics, and alerts
• Provides an integrated IoT simulator to generate realistic sensor traffic
• Runs locally or inside Docker for reproducible deployment

Why It Is Useful

• Provides a production-style architecture for IoT monitoring systems
• Separates responsibilities clearly across domain models, service layer, processing, and storage
• Demonstrates how to handle high-frequency telemetry with async workflows
• Makes it easy to evolve from rule-based anomaly detection to ML-based inference

Architecture Overview

1. Client or simulator sends data to POST /ingest
2. Payload is queued in an asyncio queue
3. Concurrent workers process queued readings
4. Readings are normalized and aggregated
5. Detector evaluates anomaly score on rolling windows
6. Runtime metrics, sensor history, and alerts are exposed via API endpoints

Core Features

• Real-time ingestion endpoint: POST /ingest
• Sensor history endpoint: GET /sensors/{sensor_id}/readings
• Runtime monitoring endpoint: GET /monitoring/stats
• Alert feed endpoint: GET /alerts
• Health endpoint: GET /health
• Simulator control endpoints:
  • POST /simulator/start
  • POST /simulator/stop
  • POST /simulator/tick

Technology Mapping

• Advanced OOP
  • Service orchestration and dedicated processing/store classes
• FastAPI
  • Async API endpoints and schema validation with Pydantic
• Async and Concurrency
  • asyncio.Queue, worker tasks, and coordinated async execution
• IoT Data Processing
  • Sensor-type normalization and rolling analytics
• Decorators
  • Async timing and bounded-value decorators
• Functional Programming
  • Stateless normalization and aggregation helpers
• MLOps Readiness
  • Detector abstraction can be replaced by model inference pipeline

Project Structure

• app/main.py: FastAPI app lifecycle and endpoints
• app/models.py: Pydantic models, enums, and runtime dataclasses
• app/service.py: Ingestion pipeline, worker loops, anomaly-triggered alerts
• app/processing.py: Normalization, aggregation, and detector logic
• app/store.py: In-memory persistence for readings, metrics, and alerts
• app/simulator.py: Async synthetic IoT data generator
• app/decorators.py: Reusable decorators for timing and value bounding
• app/config.py: Centralized runtime settings
• tests/test_api.py: API smoke and behavior tests
• check-all.ps1: One-command environment and runtime verification script

Quick Start (Local)

Windows PowerShell

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload

Open documentation at:

http://127.0.0.1:8000/docs

Quick Start (Docker)

docker compose up --build -d
docker compose ps

Check health:

Invoke-RestMethod -Uri "http://127.0.0.1:8000/health" -Method Get | ConvertTo-Json -Depth 5

Validation and Test Commands

• Run automated tests:

.\.venv\Scripts\python.exe -m pytest -q

• Run full environment checks:

powershell -ExecutionPolicy Bypass -File .\check-all.ps1

Example API Calls

Ingest a sensor reading

$payload = @{ sensor_id = "temp-77"; sensor_type = "temperature"; value = 28.3 } | ConvertTo-Json
Invoke-RestMethod -Uri "http://127.0.0.1:8000/ingest" -Method Post -ContentType "application/json" -Body $payload

Get monitoring stats

Invoke-RestMethod -Uri "http://127.0.0.1:8000/monitoring/stats" -Method Get | ConvertTo-Json -Depth 8

Get latest alerts

Invoke-RestMethod -Uri "http://127.0.0.1:8000/alerts?limit=20" -Method Get | ConvertTo-Json -Depth 8

Integrating Real IoT Sources

Recommended data sources:

• MQTT streams
• Arduino and ESP32 telemetry logs
• Kaggle IoT datasets replay

Integration strategy:

1. Build an adapter that maps external payloads to SensorReadingIn
2. Send normalized event objects to the service ingest pipeline
3. Persist historical telemetry in a production database
4. Add model retraining and drift monitoring for anomaly quality

Optional Extensions

• Open WebUI integration for conversational monitoring and alert explanation
• Authentication and RBAC for secure API exposure
• Persistent backend with PostgreSQL, TimescaleDB, or Redis
• Prometheus and Grafana for advanced observability

Suggested GCP Deployment Path

1. Build and push image to Artifact Registry
2. Deploy API to Cloud Run
3. Move state to managed data store
4. Store long-term data in BigQuery
5. Use Vertex AI pipelines for anomaly model lifecycle

Current Limitations

• In-memory persistence only
• No authentication by default
• Rule-based anomaly detection baseline

Author

Maintained for portfolio and applied backend engineering practice in IoT monitoring and cloud-native deployment workflows.