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"
@@ -40,6 +40,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():
     for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]:
@@ -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"]
@@ -299,6 +350,10 @@ async def process_messages_task():
                 n_state = data["data"].get("new_cooking_state")
                 print(f"[LoRa] Microwave {mw_id} state changed to: {n_state}")
                 
+                if n_state in (CookingStates.DONE, CookingStates.STIRRING_REQUIRED, CookingStates.ALERT):
+                    print("buzzer alert triggered!")
+                    beats = 5 if n_state == CookingStates.ALERT else (1 if n_state == CookingStates.STIRRING_REQUIRED else 3)
+                    buzzer.buzzer_siren(beatsNb=beats)
                 if n_state == CookingStates.IDLE and microwave_states.get(mw_id) == MicrowaveState.COOKING:
                     microwave_states[mw_id] = MicrowaveState.DONE
                     print(f"[{mw_id}] Cooking finished. Waiting for user to remove dish.")
@@ -306,10 +361,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)
             
@@ -338,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 = []
@@ -414,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()
     )
@@ -426,4 +540,7 @@ if __name__ == "__main__":
     finally:
         if hasattr(mqtt_client._client, "loop_stop"):
             mqtt_client._client.loop_stop()
+        if hasattr(cloud_mqtt_client._client, "loop_stop"):
+            cloud_mqtt_client._client.loop_stop()
         mqtt_client.close()
+        cloud_mqtt_client.close()

orchestrateur/sensors/__init__.py

@@ -5,3 +5,4 @@ import sensors.temp_hum as temp_hum
 import sensors.button as button
 import sensors.gps as gps
 import sensors.camera as camera
+import sensors.buzzer as buzzer

orchestrateur/sensors/button.py

@@ -5,7 +5,6 @@ from sensors.lock import grove_lock
 from shared.logging import log
 
 button = 2
-button_switch_state = 0
 grovepi.pinMode(button, "INPUT")
 
 button_callback = None
@@ -15,7 +14,9 @@ def read_button_state():
     if not grove_lock.acquire(timeout=0.2):
         return None
     try:
-        return grovepi.digitalRead(button)
+        val = grovepi.digitalRead(button)
+        # Force strict binary output (0 or 1)
+        return 1 if val == 1 else 0
     except Exception as e:
         log(f"BTN Error: {e}")
         return None
@@ -23,21 +24,47 @@ def read_button_state():
         grove_lock.release()
 
 def monitor_button():
-    global button_switch_state
-    last_button_state = button_switch_state
+    last_stable_state = 0
+    candidate_state = 0
+    consecutive_count = 0
+    
+    # Require 3 consecutive identical reads (~60ms) to confirm a valid state change
+    REQUIRED_CONSECUTIVE_READS = 3
+    
+    # Minimum time gap (in seconds) between allowed button triggers (cooldown)
+    DEBOUNCE_COOLDOWN = 0.3
+    last_trigger_time = 0
 
     while True:
         current_state = read_button_state()
         
         if current_state is not None:
-            # Rising edge detection (0 -> 1 transition)
-            if current_state == 1 and last_button_state == 0:
-                if button_callback:
-                    button_callback()
-            last_button_state = current_state
-            time.sleep(0.02)  # Fast 20ms poll when lock is clear
+            if current_state == candidate_state:
+                consecutive_count += 1
+            else:
+                candidate_state = current_state
+                consecutive_count = 1
+
+            # State is confirmed stable across multiple reads
+            if consecutive_count >= REQUIRED_CONSECUTIVE_READS:
+                now = time.time()
+                
+                # Rising edge detection (0 -> 1 transition) with cooldown timer
+                if candidate_state == 1 and last_stable_state == 0:
+                    if (now - last_trigger_time) > DEBOUNCE_COOLDOWN:
+                        last_trigger_time = now
+                        if button_callback:
+                            try:
+                                button_callback()
+                            except Exception as e:
+                                log(f"[Button] Callback exception: {e}")
+                
+                last_stable_state = candidate_state
+
+            time.sleep(0.02)  # 20ms poll interval
         else:
-            # Lock was busy; retry quickly without updating last_button_state
+            # Lock was busy or read failed; reset candidate counter to reject noisy spikes
+            consecutive_count = 0
             time.sleep(0.01)
 
 def start_button_monitoring_thread():

orchestrateur/sensors/buzzer.py

@@ -0,0 +1,33 @@
+import time
+import grovepi
+from sensors.lock import grove_lock
+from shared import config
+
+# Pin definition
+BUZZER_PIN = 8
+
+grovepi.pinMode(BUZZER_PIN, "OUTPUT")
+buzzer_state = 0
+
+def set_buzzer(state: int):
+    global buzzer_state
+    buzzer_state = state
+    if not config.BUZZER_ACTIVATED:
+        return
+    try:
+        with grove_lock:
+            grovepi.digitalWrite(BUZZER_PIN, state)
+    except Exception as e:
+        print(f"[BUZZER] Error occurred while setting buzzer state: {e}")   
+
+def buzzer_toggle():
+    global buzzer_state
+    new_state = 1 - buzzer_state
+    set_buzzer(new_state)
+    
+def buzzer_siren(beatsNb: int = 3, delay: float = 0.3):
+    for _ in range(beatsNb):
+        set_buzzer(1)
+        time.sleep(delay)
+        set_buzzer(0)
+        time.sleep(delay)

shared/config.py

@@ -11,8 +11,11 @@ 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
 
 # Microwave Model
 COOKING_COMPARTMENT_HEIGHT = 30  # cm
+
+BUZZER_ACTIVATED = False

shared/payloads.py

@@ -52,3 +52,17 @@ def mqtt_cooking_config(id_microwave, cook_time, power_level, target_temp):
         "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
+    })