diff --git a/cloud/APIs/mqtt.py b/cloud/APIs/mqtt.py new file mode 100644 index 0000000..7f9d764 --- /dev/null +++ b/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}") \ No newline at end of file diff --git a/cloud/app.py b/cloud/app.py index c45672b..1250527 100644 --- a/cloud/app.py +++ b/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) \ No newline at end of file + app.run(debug=getattr(config, "DEBUG", True)) \ No newline at end of file diff --git a/orchestrateur/lib/systemd_logs.py b/orchestrateur/lib/systemd_logs.py new file mode 100644 index 0000000..b77cf55 --- /dev/null +++ b/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 [] \ No newline at end of file diff --git a/orchestrateur/main.py b/orchestrateur/main.py index a1abe02..440f943 100644 --- a/orchestrateur/main.py +++ b/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() \ No newline at end of file + if hasattr(cloud_mqtt_client._client, "loop_stop"): + cloud_mqtt_client._client.loop_stop() + mqtt_client.close() + cloud_mqtt_client.close() \ No newline at end of file diff --git a/shared/config.py b/shared/config.py index 66ab8a6..4540c5e 100644 --- a/shared/config.py +++ b/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 diff --git a/shared/payloads.py b/shared/payloads.py index 923a893..8d98b02 100644 --- a/shared/payloads.py +++ b/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 }) \ No newline at end of file