add mqtt protocol setup and reorganize fastapi project

.github/workflows/ci.yml

@@ -44,6 +44,12 @@ jobs:
           poetry run python scripts/gen_proto.py
           git diff --exit-code || (echo "Error: Generated protobuf stubs are out of sync with pzem_004t.proto. Run scripts/gen_proto.py and commit." && exit 1)
 
+      - name: Run Ruff linter
+        run: poetry run ruff check .
+
+      - name: Run Ruff formatter check
+        run: poetry run ruff format --check .
+
       - name: Run strict type checking (Pyrefly)
         run: poetry run pyrefly check
 

.vscode/settings.json

@@ -1,4 +1,12 @@
 {
     "python-envs.defaultEnvManager": "ms-python.python:poetry",
-    "python-envs.defaultPackageManager": "ms-python.python:poetry"
+    "python-envs.defaultPackageManager": "ms-python.python:poetry",
+    "[python]": {
+        "editor.defaultFormatter": "charliermarsh.ruff",
+        "editor.formatOnSave": true,
+        "editor.codeActionsOnSave": {
+            "source.fixAll": "explicit",
+            "source.organizeImports": "explicit"
+        }
+    }
 }

README.md

@@ -11,42 +11,44 @@ interact with the gRPC microservice.
 
 ```mermaid
 flowchart LR
-    subgraph RESTConsumer["External REST Consumer"]
-        Web["Web / Mobile App / cURL"]
+    subgraph Microcontrollers["IoT Microcontrollers (e.g. ESP32)"]
+        ESP32["ESP32 + PZEM-004T\n(apps/mqtt_sensor_node)\n[paho-mqtt]"]
     end
 
-    subgraph Gateway["REST-to-gRPC Gateway (FastAPI)"]
-        GW["gateway/ (app.py)"]
+    subgraph MQTTBroker["MQTT Messaging Broker (Port 1883)"]
+        Broker["Eclipse Mosquitto\n(devices/+/telemetry)"]
     end
 
-    subgraph Device["PZEM-004t Device Gateway (client)"]
-        A1["client/ (telemetry.py)"]
-        A2["device/ (pzem_004t.py)"]
+    subgraph Bridges["Ingestion Bridges"]
+        Bridge["MQTT-to-gRPC Bridge\n(apps/mqtt_bridge)\n[paho-mqtt + grpc.aio]"]
     end
 
-    subgraph Collector["Telemetry Collector Service (server)"]
-        B1["server/ (servicer.py + app.py)"]
-        B2["Health Checking (grpc.health.v1)"]
-        B3["Interceptors (Tracing, Metrics, Recovery)"]
-        B4["in-memory store + logging"]
+    subgraph Collector["Telemetry Collector Service (Port 50051)"]
+        Coll["Collector Service\n(apps/collector)\n(gRPC Ingestion & Pub/Sub Hub)"]
     end
 
-    Web --> |"HTTP/JSON REST API (port 8000)"| GW
-    GW --> |"gRPC Unary / Client-Stream"| B1
-    A1 <==> |"gRPC (HTTP/2) - 4 RPC call types (port 50051)"| B1
-    A1 -- "simulated readings" --> A2
+    subgraph RESTConsumer["Web / Mobile / Dashboard"]
+        REST["REST-to-gRPC Gateway\n(apps/rest_gateway)\n(FastAPI - Port 8000)"]
+    end
+
+    ESP32 -->|"MQTT Publish (JSON)"| Broker
+    Broker -->|"MQTT Subscribe"| Bridge
+    Bridge ==>|"gRPC ReportReading (HTTP/2)"| Coll
+    REST ==>|"gRPC Unary & Batch (HTTP/2)"| Coll
 ```
 
 ### Industry-Grade Capabilities Implemented
 
-- **Dual-App Separation**: Telemetry Collector (Server) and IoT Gateway (Client) operate as decoupled microservice applications.
+- **Industrial IoT Protocol Hierarchy**:
+  - **MQTT**: Lightweight pub/sub for resource-constrained microcontrollers (ESP32) reading PZEM-004T sensors.
+  - **MQTT-to-gRPC Ingestion Bridge**: Seamlessly consumes MQTT telemetry topics and bridges them into gRPC.
+  - **FastAPI REST Gateway**: Modular HTTP backend for external web/mobile dashboards and REST API consumers.
 - **gRPC Interceptors**:
   - **Client-Side**: Injects distributed tracing headers (`x-request-id`) and `x-client-version`.
   - **Server-Side**: Performance metrics logging (RPC duration, peer IP, status code) and unhandled exception recovery translating errors safely into gRPC status codes.
 - **Official Health Checking (`grpc.health.v1`)**: Exposes standard gRPC health checks for Kubernetes liveness/readiness probes and load balancers.
 - **Connection Resilience**: Configured HTTP/2 keepalive pings (`grpc.keepalive_time_ms`), request timeouts (deadlines), and auto-reconnects.
-- **REST-to-gRPC Gateway**: FastAPI application translating external HTTP/JSON REST requests into strongly-typed gRPC calls.
-- **Container Orchestration**: Production `Dockerfile` and `docker-compose.yml` for multi-container deployment.
+- **Container Orchestration**: Multi-container Docker Compose topology orchestrating Mosquitto MQTT, Collector, MQTT Bridge, and REST Gateway across segmented bridge networks.
 
 ### The four gRPC call types
 
@@ -72,15 +74,22 @@ src/python_grpc/
     device/
       pzem_004t.py             # PZEM004TDevice hardware physics simulator
   apps/                        # autonomous deployable applications
-    collector/                 # Cloud-tier: Telemetry Collector Server
+    collector/                 # Cloud-tier: Telemetry Collector Server (gRPC only)
       app.py                   # server lifecycle, health check, graceful shutdown
       servicer.py              # in-memory pub/sub telemetry broadcast servicer
       __main__.py              # CLI entry point (python -m python_grpc.apps.collector)
-    device_agent/              # Edge-tier: IoT Hardware Agent
-      agent.py                 # resilient reporting, health checking, streaming
-      __main__.py              # CLI entry point (python -m python_grpc.apps.device_agent)
-    rest_gateway/              # Consumer-tier: FastAPI REST-to-gRPC Gateway
-      app.py                   # FastAPI proxy endpoints (unary & batch)
+    mqtt_sensor_node/          # Edge-tier: Microcontroller (ESP32) MQTT Sensor Node
+      app.py                   # sensor reading & MQTT JSON publishing loop
+      __main__.py              # CLI entry point (python -m python_grpc.apps.mqtt_sensor_node)
+    mqtt_bridge/               # Bridge-tier: MQTT-to-gRPC Telemetry Ingestion Bridge
+      app.py                   # MQTT subscriber forwarding to collector over gRPC
+      __main__.py              # CLI entry point (python -m python_grpc.apps.mqtt_bridge)
+    rest_gateway/              # Consumer-tier: Modular FastAPI REST-to-gRPC Gateway
+      app.py                   # FastAPI app factory & lifespan
+      config.py                # Gateway configuration settings
+      dependencies.py          # Dependency injection & stub resolver
+      schemas.py               # Pydantic models for validation
+      routers/                 # Modular APIRouters (telemetry, health)
       __main__.py              # CLI entry point (python -m python_grpc.apps.rest_gateway)
 scripts/
   gen_proto.py                 # regenerate stubs from the proto
@@ -91,8 +100,9 @@ tests/
   test_cross_host.py           # multi-client pub/sub broadcasting & disconnect resilience
   test_health_and_interceptors.py # gRPC health & interceptor integration tests
   test_gateway.py              # FastAPI REST-to-gRPC gateway integration tests
+  test_mqtt_pipeline.py        # MQTT sensor node + bridge + gRPC collector tests
 Dockerfile                     # multi-app container build
-docker-compose.yml             # multi-network orchestration (cloud-tier & edge-tier)
+docker-compose.yml             # multi-network orchestration with Mosquitto MQTT broker
 ```
 
 ## Requirements
@@ -116,24 +126,29 @@ poetry install
 poetry run python -m python_grpc.apps.collector --host 0.0.0.0 --port 50051
 ```
 
-#### 2. Run the IoT Device Gateway Client (Terminal 2)
-```bash
-poetry run python -m python_grpc.apps.device_agent --target localhost:50051 --device-id PZEM-004T-0001 --count 5
-```
-
-#### 3. Start the REST-to-gRPC Gateway (Terminal 3, optional)
+#### 2. Start the REST-to-gRPC Gateway (Terminal 2)
 ```bash
 poetry run python -m python_grpc.apps.rest_gateway --host 0.0.0.0 --port 8000 --grpc-target localhost:50051
 ```
 Open your browser at `http://localhost:8000/docs` to test Swagger UI or send a cURL request:
 ```bash
-curl -X POST "http://localhost:8000/api/v1/telemetry" \
+curl -X POST "http://localhost:8000/api/telemetry" \
   -H "Content-Type: application/json" \
   -d '{"device_id": "REST-01", "voltage": 230.2, "current": 2.1, "active_power": 483.4, "energy": 1.2, "frequency": 50.0, "power_factor": 0.99}'
 ```
 
+#### 3. Run the MQTT-to-gRPC Bridge & Simulated ESP32 Sensor Node (Terminal 3 & 4)
+If you have an MQTT broker running (such as Mosquitto on port 1883):
+```bash
+# Start Bridge to forward MQTT messages into gRPC Collector
+poetry run python -m python_grpc.apps.mqtt_bridge --mqtt-host localhost --mqtt-port 1883 --grpc-target localhost:50051
+
+# Start Simulated ESP32 reading PZEM-004T and publishing over MQTT
+poetry run python -m python_grpc.apps.mqtt_sensor_node --broker-host localhost --broker-port 1883 --device-id ESP32-PZEM-01 --count 5
+```
+
 ### Option B: Running with Docker Compose
-Spin up the entire microservice topology:
+Spin up the entire microservice topology (Mosquitto MQTT broker, Collector, MQTT Bridge, and REST Gateway):
 ```bash
 docker compose up --build
 ```
@@ -144,12 +159,30 @@ docker compose up --build
 poetry run pytest -v --cov=python_grpc
 ```
 
-Tests run 17 automated integration and unit tests covering:
+Tests run 18 automated integration and unit tests covering:
 - Sensor physics and energy accumulation ([`tests/test_device.py`](tests/test_device.py))
 - All 4 gRPC streaming patterns ([`tests/test_telemetry.py`](tests/test_telemetry.py))
 - Cross-host multi-client pub/sub broadcasting & disconnect resilience ([`tests/test_cross_host.py`](tests/test_cross_host.py))
 - Standard gRPC Health Checking (`grpc.health.v1`) & Interceptors ([`tests/test_health_and_interceptors.py`](tests/test_health_and_interceptors.py))
 - FastAPI REST-to-gRPC unary and batch forwarding ([`tests/test_gateway.py`](tests/test_gateway.py))
+- Simulated ESP32 MQTT pub/sub ingestion into gRPC Collector ([`tests/test_mqtt_pipeline.py`](tests/test_mqtt_pipeline.py))
+
+## Code Formatting & Linting
+
+Ruff is used for ultra-fast linting and code formatting:
+```bash
+# Check for lint violations
+poetry run ruff check .
+
+# Automatically apply safe lint fixes
+poetry run ruff check --fix .
+
+# Check formatting without modifying files
+poetry run ruff format --check .
+
+# Automatically format the entire codebase
+poetry run ruff format .
+```
 
 ## Type Checking
 

docker-compose.yml

@@ -18,40 +18,74 @@ services:
           - collector.internal
     restart: unless-stopped
 
+
   # -------------------------------------------------------------
-  # Edge Device Tier (Simulates Remote IoT Edge Unit on Host 2)
+  # External Consumer Tier (Simulates REST Consumer on Host 3)
   # -------------------------------------------------------------
-  device-agent:
+  rest-gateway:
     image: ${IMAGE_NAME:-python-grpc}:${IMAGE_TAG:-latest}
     build:
       context: .
       dockerfile: Dockerfile
-    command: python -m python_grpc.apps.device_agent --target collector.internal:50051 --device-id PZEM-EDGE-001 --count 5
+    command: python -m python_grpc.apps.rest_gateway --host 0.0.0.0 --port 8000 --grpc-target collector.internal:50051
+    ports:
+      - "8000:8000"
     depends_on:
       - collector-service
     environment:
-      - LOG_LEVEL=${LOG_LEVEL:-INFO}
+      - GRPC_TARGET=collector.internal:50051
+    networks:
+      cloud-tier:
+    restart: unless-stopped
+
+  # -------------------------------------------------------------
+  # MQTT Broker (Industrial Standard IoT Messaging Tier)
+  # -------------------------------------------------------------
+  mqtt-broker:
+    image: eclipse-mosquitto:2.0
+    command: mosquitto -c /mosquitto-no-auth.conf
+    ports:
+      - "1883:1883"
     networks:
       edge-tier:
+        aliases:
+          - broker.internal
       cloud-tier:
     restart: unless-stopped
 
   # -------------------------------------------------------------
-  # External Consumer Tier (Simulates REST Consumer on Host 3)
+  # IoT Microcontroller Sensor Node (Simulates ESP32 + PZEM-004T)
   # -------------------------------------------------------------
-  rest-gateway:
+  mqtt-sensor-node:
     image: ${IMAGE_NAME:-python-grpc}:${IMAGE_TAG:-latest}
     build:
       context: .
       dockerfile: Dockerfile
-    command: python -m python_grpc.apps.rest_gateway --host 0.0.0.0 --port 8000 --grpc-target collector.internal:50051
-    ports:
-      - "8000:8000"
+    command: python -m python_grpc.apps.mqtt_sensor_node --broker-host broker.internal --broker-port 1883 --device-id ESP32-PZEM-01 --count 5
+    depends_on:
+      - mqtt-broker
+    environment:
+      - LOG_LEVEL=${LOG_LEVEL:-INFO}
+    networks:
+      edge-tier:
+    restart: unless-stopped
+
+  # -------------------------------------------------------------
+  # MQTT-to-gRPC Bridge Service (Translates MQTT pub/sub to gRPC)
+  # -------------------------------------------------------------
+  mqtt-bridge:
+    image: ${IMAGE_NAME:-python-grpc}:${IMAGE_TAG:-latest}
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: python -m python_grpc.apps.mqtt_bridge --mqtt-host broker.internal --mqtt-port 1883 --grpc-target collector.internal:50051
     depends_on:
+      - mqtt-broker
       - collector-service
     environment:
-      - GRPC_TARGET=collector.internal:50051
+      - LOG_LEVEL=${LOG_LEVEL:-INFO}
     networks:
+      edge-tier:
       cloud-tier:
     restart: unless-stopped