From 25e635e2076324fe2321f22714536e983a02c1a8 Mon Sep 17 00:00:00 2001 From: Matthias Guillitte Date: Thu, 6 Aug 2026 16:55:24 +0200 Subject: [PATCH] MQTT hello is saved in database --- orchestrateur/main.py | 53 ++++++++++++++++++++++++++++++++++++------- shared/payloads.py | 4 +++- 2 files changed, 48 insertions(+), 9 deletions(-) diff --git a/orchestrateur/main.py b/orchestrateur/main.py index b376d85..fd15ea4 100644 --- a/orchestrateur/main.py +++ b/orchestrateur/main.py @@ -6,12 +6,40 @@ import asyncio import requests from orchestrateur.sensors import gps -from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads +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 +# --- DB SETUP --- +DB_PATH = "orchestrateur/db.sqlite" + +def init_db(): + """Ensures the connected_components table exists on startup.""" + sql = """ + CREATE TABLE IF NOT EXISTS connected_components ( + id TEXT PRIMARY KEY, + type TEXT, + timestamp INTEGER + ); + """ + db.execute(DB_PATH, sql) + print(f"[DB] Initialized database table at {DB_PATH}") + +def save_connected_component(component_id: str, component_type: str): + """Upserts component information into the database (blocking sync worker).""" + current_time = int(time.time()) + sql = """ + INSERT INTO connected_components (id, type, timestamp) + VALUES (?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + type = excluded.type, + timestamp = excluded.timestamp; + """ + 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}") + # --- Read Unique Device ID --- def get_device_id(): for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]: @@ -287,13 +315,19 @@ async def process_messages_task(): if topic == hello_topic: if data.get("id_orchestrator") != DEVICE_ID: - mw_id = data.get("id_microwave") - print(f"[MQTT] Hello from {mw_id}. Sending ACK.") - mqtt_client.publish( - config.MQTT_TOPIC_HELLO, - payloads.mqtt_hello_ack(DEVICE_ID, mw_id), - qos=config.MQTT_QOS - ) + component_id = data.get("id_microwave") + component_type = data.get("type", deviceTypes.DEVICE_TYPES["MICROWAVE"]) + + if component_id: + print(f"[MQTT] Hello received from '{component_id}' ({component_type}). Updating DB & sending ACK.") + # Offload DB insertion to async thread execution pool + await asyncio.to_thread(save_connected_component, component_id, component_type) + + mqtt_client.publish( + config.MQTT_TOPIC_HELLO, + payloads.mqtt_hello_ack(DEVICE_ID, component_id), + qos=config.MQTT_QOS + ) elif topic == sensor_topic: mw_id = str(data.get("id_microwave")) @@ -372,6 +406,9 @@ async def main(): global async_event_queue print("🚀 Orchestrateur Asyncio prêt. Lancement des tâches...") + # Initialize SQLite database table + init_db() + async_event_queue = asyncio.Queue() await asyncio.gather( diff --git a/shared/payloads.py b/shared/payloads.py index 659641d..6460673 100644 --- a/shared/payloads.py +++ b/shared/payloads.py @@ -1,4 +1,5 @@ from time import time +from shared.deviceTypes import DEVICE_TYPES try: @@ -16,7 +17,8 @@ def as_json(data): def mqtt_hello(id_microwave): return as_json({ - "id_microwave": id_microwave + "id_microwave": id_microwave, + "type": DEVICE_TYPES["MICROWAVE"] }) def mqtt_hello_ack(id_orchestrator, id_microwave):