Telemetry

cloud/APIs/mqtt.py

@@ -0,0 +1,36 @@
+import json
+import os
+import paho.mqtt.publish as publish
+import sys
+
+# Retrieve broker parameters from shared config or environment variables
+MQTT_HOST = os.getenv("MQTT_HOST", "localhost")
+MQTT_PORT = int(os.getenv("MQTT_PORT", 8884))
+MQTT_USER = os.getenv("MQTT_USER", None)
+MQTT_PASS = os.getenv("MQTT_PASS", None)
+
+
+def send_command(topic: str = "cmd/all", payload: dict | str = None):
+    """
+    Publishes an MQTT command payload to a given topic.
+    """
+    if payload is None:
+        payload = {}
+
+    if isinstance(payload, dict):
+        payload_str = json.dumps(payload)
+    else:
+        payload_str = str(payload)
+
+    auth = None
+    if MQTT_USER:
+        auth = {"username": MQTT_USER, "password": MQTT_PASS or ""}
+
+    publish.single(
+        topic=topic,
+        payload=payload_str,
+        hostname=MQTT_HOST,
+        port=MQTT_PORT,
+        auth=auth
+    )
+    print(f"[MQTT] Published command to '{topic}': {payload_str}")

cloud/app.py

@@ -1,11 +1,14 @@
 import os
 import base64
 import uuid
+import datetime
+import sys
 from flask import Flask, request, jsonify
 from pymongo import MongoClient
 from APIs import generate, EdamamAPI
-import sys
+from APIs.mqtt import send_command
 from microwaveCookPlanner import MicrowaveCookPlanner
+
 sys.path.insert(0, '..')
 try:
     from shared import config
@@ -18,12 +21,13 @@ app = Flask(__name__)
 # Configuration & Setup
 # ---------------------------------------------------------
 
-# Configure MongoDB connection (adjust the URI as needed for your environment)
+# Configure MongoDB connection
 MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
 client = MongoClient(MONGO_URI)
 db = client["microwave_network_db"]
+
 cooking_collection = db["cooking_parameters"]
-device_network_collection = db["device_network"]
+telemetry_collection = db["telemetry_data"]
 
 # Ensure the camera image storage directory exists when the app starts
 CAMERA_IMAGE_DIR = "storage/dishCameraImages"
@@ -103,29 +107,45 @@ def cooking_params():
     # 5. Return complete output
     return jsonify(cook_plan), 201
 
-@app.route("/device-network", methods=["POST"])
-def device_network():
+
+@app.route("/telemetry", methods=["POST"])
+def telemetry():
     data = request.get_json()
     
     if not data:
         return jsonify({"error": "Invalid or missing JSON payload"}), 400
 
+    # Stamp UTC timestamp for Node-RED queries
+    data["received_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
 
-    # 2. Save to MongoDB
     try:
-        # Insert the dictionary directly into Mongo (it will retain your exact JSON keys)
-        device_network_collection.insert_one(data)
-        
-        return "", 200
+        # Mongo creates '_id' automatically upon insertion
+        telemetry_collection.insert_one(data)
+        return jsonify({"status": "success", "message": "Telemetry saved"}), 200
         
     except Exception as e:
         return jsonify({"error": f"Database error: {str(e)}"}), 500
-    
+
+
 @app.route("/debug", methods=["GET"])
 def debug():
-    image_path = "microwaveDish.jpg"
-    edamam = EdamamAPI()
-    return edamam.analyze_dish_image(image_path)
+    # Construct external HTTP endpoint dynamically based on incoming request host
+    telemetry_url = f"{request.host_url.rstrip('/')}/telemetry"
+    
+    cmd_payload = {
+        "action": "request_telemetry",
+        "endpoint": telemetry_url
+    }
+    
+    try:
+        send_command(topic="cmd/all", payload=cmd_payload)
+        return jsonify({
+            "status": "Telemetry command sent to cmd/all",
+            "published_payload": cmd_payload
+        }), 200
+    except Exception as e:
+        return jsonify({"error": f"Failed to publish MQTT command: {str(e)}"}), 500
+
 
 if __name__ == "__main__":
-    app.run(debug=config.DEBUG)
+    app.run(debug=getattr(config, "DEBUG", True))

orchestrateur/lib/systemd_logs.py

@@ -0,0 +1,61 @@
+from datetime import datetime
+import asyncio
+import json
+
+async def get_systemd_logs(lines: int = 200, service_name: str = "smartwave") -> list[dict]:
+    """
+    Asynchronously fetches the last N log entries from a systemd service 
+    using native asyncio subprocess execution.
+    """
+    cmd = (
+        "journalctl",
+        "-u", service_name,
+        "-n", str(lines),
+        "-o", "json",
+        "--no-pager"
+    )
+    
+    try:
+        # Create non-blocking child process
+        proc = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+        
+        # Await completion and read output non-blockingly
+        stdout_bytes, stderr_bytes = await proc.communicate()
+        
+        if proc.returncode != 0:
+            print(f"[Logs] journalctl error (code {proc.returncode}): {stderr_bytes.decode('utf-8', errors='replace')}")
+            return []
+            
+        structured_logs = []
+        stdout_text = stdout_bytes.decode("utf-8", errors="replace")
+        
+        for line in stdout_text.strip().split("\n"):
+            if not line:
+                continue
+            try:
+                entry = json.loads(line)
+                
+                raw_ts = entry.get("__REALTIME_TIMESTAMP")
+                timestamp_iso = None
+                if raw_ts:
+                    timestamp_iso = datetime.fromtimestamp(int(raw_ts) / 1_000_000).isoformat()
+                
+                structured_logs.append({
+                    "timestamp": timestamp_iso,
+                    "timestamp_us": int(raw_ts) if raw_ts else None,
+                    "message": entry.get("MESSAGE", ""),
+                    "priority": entry.get("PRIORITY"),
+                    "pid": entry.get("_PID"),
+                })
+            except (json.JSONDecodeError, ValueError):
+                continue
+                
+        return structured_logs
+
+    except Exception as e:
+        print(f"[Logs] Failed to read systemd logs for '{service_name}': {e}")
+        return []

orchestrateur/main.py

@@ -1,7 +1,6 @@
 import base64
 import json
 import time
-import traceback
 import asyncio
 import requests
 
@@ -10,7 +9,8 @@ from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads, db
 from shared.logging import log
 from shared.cookingState import CookingStates
 from shared.lora_device import LoraCommands
-from sensors import ultrasonicRanger, temp_hum, button, camera
+from sensors import ultrasonicRanger, temp_hum, button, camera, buzzer
+from lib.systemd_logs import get_systemd_logs
 
 # --- DB SETUP ---
 DB_PATH = "orchestrateur/db.sqlite"
@@ -39,6 +39,12 @@ def save_connected_component(component_id: str, component_type: str):
     """
     db.execute(DB_PATH, sql, (str(component_id), str(component_type), current_time))
     print(f"[DB] Component saved/updated -> ID: {component_id}, Type: {component_type}, Timestamp: {current_time}")
+    
+def get_connected_components():
+    """Returns a list of all connected components from the database."""
+    sql = "SELECT id, type, timestamp FROM connected_components;"
+    rows = db.fetchall(DB_PATH, sql)
+    return [{"id": row[0], "type": row[1], "timestamp": row[2]} for row in rows]
 
 # --- Read Unique Device ID ---
 def get_device_id():
@@ -71,10 +77,11 @@ ir_data_cache = {}   # mw_id -> dict of IR readings
 ir_data_events = {}  # mw_id -> asyncio.Event()
 
 
-# --- HARDWARE SETUP ---
+# --- HARDWARE & MQTT SETUP ---
 lora = get_lora()
 lora.configure()
 
+# Local MQTT Broker (ESP32 IR & Sensors)
 mqtt_client = get_mqtt_client(
     host="192.168.50.1",
     client_id="smartwave-orchestrateur-" + DEVICE_ID,
@@ -88,7 +95,25 @@ mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS)
 
 if hasattr(mqtt_client._client, "loop_start"):
     mqtt_client._client.loop_start()
-    print("[MQTT] Paho background loop started.")
+    print("[MQTT Local] Paho background loop started.")
+
+# Cloud MQTT Broker (Remote Commands)
+cloud_mqtt_client = get_mqtt_client(
+    host="mqtt.matthiasg.dev",
+    client_id=DEVICE_ID,
+    use_tls=False,
+    username="microwave_device",
+    password="myMicrowaveVerySecret",
+    keepalive=config.MQTT_KEEPALIVE,
+)
+cloud_mqtt_client.connect()
+cloud_mqtt_client.subscribe("cmd/all", qos=config.CLOUD_MQTT_QOS)
+cloud_mqtt_client.subscribe(f"cmd/{DEVICE_ID}", qos=config.CLOUD_MQTT_QOS)
+
+if hasattr(cloud_mqtt_client._client, "loop_start"):
+    cloud_mqtt_client._client.loop_start()
+    print("[MQTT Cloud] Paho background loop started.")
+
 
 # --- BACKGROUND TASKS (PRODUCERS) ---
 async def lora_listener_task():
@@ -102,8 +127,8 @@ async def lora_listener_task():
         await asyncio.sleep(0.05)
 
 async def mqtt_listener_task():
-    """Polls MQTT cache and pushes to the async queue."""
-    print("[MQTT] Async listener started.")
+    """Polls local MQTT cache and pushes to the async queue."""
+    print("[MQTT Local] Async listener started.")
     while True:
         message = mqtt_client.get_message()
         if message:
@@ -124,6 +149,28 @@ async def mqtt_listener_task():
             })
         await asyncio.sleep(0.1)
 
+async def cloud_mqtt_listener_task():
+    """Polls Cloud MQTT cache and pushes commands to the async queue."""
+    print("[MQTT Cloud] Async listener started.")
+    while True:
+        message = cloud_mqtt_client.get_message()
+        if message:
+            try:
+                payload = json.loads(message['payload'])
+            except Exception:
+                payload = message['payload']
+            
+            topic = message['topic']
+            if isinstance(topic, bytes):
+                topic = topic.decode('utf-8')
+
+            await async_event_queue.put({
+                "source": "CloudMQTT",
+                "topic": topic,
+                "data": payload
+            })
+        await asyncio.sleep(0.1)
+
 def button_callback():
     """Button physical interrupt callback."""
     global button_state
@@ -288,6 +335,10 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
 # --- MAIN LOGIC TASKS ---
 async def process_messages_task():
     """Consumes the unified event queue."""
+    # Helper to normalize config topics to str
+    def to_str(val):
+        return val.decode('utf-8') if isinstance(val, bytes) else val
+    
     while True:
         msg = await async_event_queue.get()
         source = msg["source"]
@@ -309,10 +360,6 @@ async def process_messages_task():
                     
         elif source == "MQTT":
             topic = msg["topic"]
-            
-            # Helper to normalize config topics to str
-            def to_str(val):
-                return val.decode('utf-8') if isinstance(val, bytes) else val
 
             hello_topic = to_str(config.MQTT_TOPIC_HELLO)
             sensor_topic = to_str(config.MQTT_TOPIC_SENSOR)
@@ -342,6 +389,68 @@ async def process_messages_task():
                 if mw_id in ir_data_events:
                     ir_data_events[mw_id].set()
 
+        elif source == "CloudMQTT":
+            topic = msg["topic"]
+            print(f"[CloudMQTT] Command received on '{topic}': {data}")
+            
+            action = data.get("action")
+            
+            if action == "request_telemetry":
+                endpoint = data.get("endpoint")
+                if endpoint:
+                    # Schedule as a background task so processing the queue isn't stalled
+                    asyncio.create_task(handle_telemetry_request(endpoint))
+                else:
+                    print("[CloudMQTT] Received 'request_telemetry' but missing 'endpoint' field.")
+            else:
+                print(f"[CloudMQTT] Unknown action: {action}")
+
+async def handle_telemetry_request(endpoint: str):
+    """
+    Task to gather telemetry data (including structured systemd logs) 
+    and post it to the target HTTP endpoint.
+    """
+    print(f"[Telemetry] Starting telemetry collection for endpoint: {endpoint}")
+    try:
+        # 1. Fetch systemd logs asynchronously
+        logs_list = await get_systemd_logs(lines=200, service_name="smartwave")
+        
+        # 2. Unpack temperature and humidity from sensors.temp_hum
+        ambient_temp, ambient_humidity = temp_hum.get_temperature_and_humidity_with_retry()
+        
+        # 3. Build the telemetry payload
+        telemetry_payload = payloads.telemetry_payload(
+            device_id=DEVICE_ID,
+            microwave_states=microwave_states,
+            button_state=button_state,
+            cloud_alert=cloud_alert,
+            gps_data=gps.get_gps_data(),
+            ambient_temp=ambient_temp,
+            ambient_humidity=ambient_humidity,
+            connected_components=get_connected_components(),
+            logs=logs_list
+        )
+        
+        print(telemetry_payload)
+        print(f"[Telemetry] Sending payload with {len(logs_list)} log entries...")
+        
+        # 4. Offload blocking HTTP POST to thread pool
+        response = await asyncio.to_thread(
+            requests.post, 
+            endpoint, 
+            json=telemetry_payload, 
+            headers={"Content-Type": "application/json"},
+            timeout=15
+        )
+        
+        response.raise_for_status()
+        print(f"[Telemetry] Successfully dispatched telemetry (HTTP {response.status_code}).")
+
+    except requests.exceptions.RequestException as req_err:
+        print(f"[Telemetry] HTTP Request failed to {endpoint}: {req_err}")
+    except Exception as e:
+        print(f"[Telemetry] Error handling telemetry request: {e}")
+
 async def get_filtered_dish_height(samples=3, delay=0.04):
     """Reads ultrasonic sensor multiple times and returns the median, discarding invalid zeros."""
     valid_samples = []
@@ -418,6 +527,7 @@ async def main():
     await asyncio.gather(
         lora_listener_task(),
         mqtt_listener_task(),
+        cloud_mqtt_listener_task(),
         process_messages_task(),
         monitor_dish_height_task()
     )
@@ -430,4 +540,7 @@ if __name__ == "__main__":
     finally:
         if hasattr(mqtt_client._client, "loop_stop"):
             mqtt_client._client.loop_stop()
-        mqtt_client.close()
+        if hasattr(cloud_mqtt_client._client, "loop_stop"):
+            cloud_mqtt_client._client.loop_stop()
+        mqtt_client.close()
+        cloud_mqtt_client.close()

shared/config.py

@@ -11,6 +11,7 @@ MQTT_TOPIC_COOKING = b"smartwave/cooking"
 MQTT_KEEPALIVE = 30
 USE_TLS = True
 MQTT_QOS = 1
+CLOUD_MQTT_QOS = 2
 # Long because messages are stored into the broker and will be sent when the orchestrator is back online.
 MQTT_HELLO_INTERVAL = 30
 

shared/payloads.py

@@ -51,4 +51,18 @@ def mqtt_cooking_config(id_microwave, cook_time, power_level, target_temp):
         "cook_time": cook_time,
         "power_level": power_level,
         "target_temp": target_temp
+    })
+    
+def telemetry_payload(device_id, microwave_states, button_state, cloud_alert, gps_data, ambient_temp, ambient_humidity, connected_components, logs):
+    return as_json({
+        "device_id": device_id,
+        "timestamp": int(time()),
+        "states": microwave_states,
+        "button_state": button_state,
+        "cloud_alert": cloud_alert,
+        "gps": gps_data,
+        "ambient_temp": ambient_temp,
+        "ambient_humidity": ambient_humidity,
+        "connected_components": connected_components,
+        "logs": logs
     })