MQTT esp_wifi

orchestrateur/main.py

@@ -1,69 +1,114 @@
 import threading
 import queue
 import time
-from shared import get_lora, deviceTypes
+from shared import get_lora, get_mqtt_client, deviceTypes, config
 
-# --- Lecture de l'ID unique du Raspberry Pi ---
+# --- Read Unique Device ID ---
 try:
     with open("device_id.txt", "r") as f:
         DEVICE_ID = f.read().strip()
 except Exception:
-    # Alternative si le script est lancé depuis un autre dossier
     try:
         with open("/home/pi/SmartWave/orchestrateur/device_id.txt", "r") as f:
             DEVICE_ID = f.read().strip()
     except Exception:
         DEVICE_ID = "RPI_Orchestrateur_Default"
 
-# Création de la file d'attente pour les messages (thread-safe)
+# Thread-safe queue for application messages
 data_queue = queue.Queue()
 
 lora = get_lora()
 lora.configure()
 
 def lora_listener():
-    """Thread de fond : écoute en permanence et répond aux Heartbeats."""
-    print("Thread Écouteur démarré.")
+    """Background Thread: Listens to LoRa traffic and responds to Heartbeats."""
+    print("Thread Écouteur LoRa démarré.")
     while True:
-        # On attend un paquet (timeout court pour rester réactif)
         paquet = lora.receive_packet(timeout_ms=1000)
-        
         if paquet:
             donnees = paquet["data"]
             expediteur_type = donnees.get("type")
             
-            # --- Cas 1 : Gestion automatique du Heartbeat ---
             if expediteur_type == deviceTypes.DEVICE_TYPES["MICROWAVE"]:
-                print(f"\n[Thread Fond] Heartbeat reçu de {donnees.get('id')}")
-                
+                print(f"\n[Thread LoRa] Heartbeat reçu de {donnees.get('id')}")
                 reponse = {
-                    "id": DEVICE_ID, # Ou lecture de ton fichier device_id.txt
+                    "id": DEVICE_ID,
                     "type": deviceTypes.DEVICE_TYPES["ORCHESTRATOR"]
                 }
                 lora.send(reponse)
-                
-            # --- Cas 2 : Donnée applicative, on l'envoie vers le thread principal ---
             else:
-                data_queue.put(paquet)
+                data_queue.put({"source": "LoRa", "data": paquet})
 
-# 2. Lancement du thread d'écoute
-listener_thread = threading.Thread(target=lora_listener, daemon=True)
-listener_thread.start()
+# --- Setup & Connect MQTT ---
+mqtt_client = get_mqtt_client(
+    host="192.168.50.1", # Using explicit gateway IP to dodge Docker loopback blocks
+    client_id="smartwave-orchestrateur-"+DEVICE_ID,
+    use_tls=config.USE_TLS,
+    cafile="/home/pi/SmartWave/orchestrateur/mqtt/certs/ca.crt",
+    keepalive=config.MQTT_KEEPALIVE,
+)
+mqtt_client.connect()
+
+# --- THE CRUCIAL PAHO FIX ---
+# Start Paho's internal background thread. This handles all network packets,
+# automatic keepalive pings, and delivery receipts cleanly.
+if hasattr(mqtt_client._client, "loop_start"):
+    mqtt_client._client.loop_start()
+    print("Paho MQTT asynchronous network loop started.")
+
+
+def mqtt_listener():
+    """Background Thread: Constantly inspects incoming MQTT message cache."""
+    print("Thread MQTT démarré.")
+    while True:
+        message = mqtt_client.get_message()
+        
+        if message:
+            print(f"\n[Thread MQTT] Message reçu : {message}")
+            data_queue.put({"source": "MQTT", "data": message})
+        
+        # --- THE CPU FIX ---
+        # Sleep for 100ms. Prevents the thread from turning into an infinite 100% CPU hog.
+        time.sleep(0.1)
+
+
+# Launch background monitoring workers
+threading.Thread(target=lora_listener, daemon=True).start()
+threading.Thread(target=mqtt_listener, daemon=True).start()
 
-# 3. Boucle principale (Main Thread) : tu es libre de faire autre chose !
 print("Orchestrateur prêt. Le main loop est libre.")
+
+# --- MAIN EXECUTION LOOP ---
 while True:
     try:
-        # On regarde si on a reçu des données applicatives (non-heartbeat)
-        # On utilise block=False pour ne pas bloquer si la queue est vide
+        # Check for non-heartbeat data safely
         try:
             msg = data_queue.get(block=False)
             print(f"\n[Main Loop] Données traitées : {msg['data']}")
         except queue.Empty:
-            pass # Rien à traiter, on fait autre chose...
+            pass
             
-        # Ici tu peux faire tes autres tâches
         time.sleep(1)
         
+        # Publish debug telemetry message
+        print("[Main Loop] Envoi d'un message de debug sur MQTT...")
+        response = mqtt_client.publish(
+            config.MQTT_TOPIC,
+            f"Orchestrateur actif, ID: {DEVICE_ID}",
+            qos=config.MQTT_QOS
+        )
+        
+        # This will now unblock instantly because loop_start() handles the delivery confirmation!
+        response.wait_for_publish()  
+        print("[Main Loop] Message de debug publié avec succès.")
+        
+        time.sleep(9)
+        
+        
     except KeyboardInterrupt:
-        break
+        break
+
+# Clean termination
+if hasattr(mqtt_client._client, "loop_stop"):
+    mqtt_client._client.loop_stop()
+mqtt_client.close()