From 0e89e42b488dc58bf826f787b3ec54bdffc6bb81 Mon Sep 17 00:00:00 2001 From: Matthias Guillitte Date: Fri, 17 Jul 2026 18:50:53 +0200 Subject: [PATCH] MQTT esp_wifi --- micro_ondes/esp_wifi/boot.py | 19 +++++ micro_ondes/esp_wifi/device_id.txt | 1 + micro_ondes/esp_wifi/main.py | 86 +++++++++++++++++++++++ orchestrateur/compose.yaml | 2 + orchestrateur/main.py | 93 ++++++++++++++++++------- orchestrateur/mqtt/mosquitto-plain.conf | 2 +- orchestrateur/mqtt/mosquitto-tls.conf | 2 +- shared/config.py | 9 +++ shared/mqtt.py | 22 +++++- 9 files changed, 207 insertions(+), 29 deletions(-) create mode 100644 micro_ondes/esp_wifi/boot.py create mode 100644 micro_ondes/esp_wifi/device_id.txt diff --git a/micro_ondes/esp_wifi/boot.py b/micro_ondes/esp_wifi/boot.py new file mode 100644 index 0000000..1711274 --- /dev/null +++ b/micro_ondes/esp_wifi/boot.py @@ -0,0 +1,19 @@ +# This file is executed on every boot (including wake-boot from deepsleep) +#import esp +#esp.osdebug(None) +#import webrepl +#webrepl.start() + +def do_connect(ssid, pwd): + import network + sta_if = network.WLAN(network.STA_IF) + if not sta_if.isconnected(): + print('connecting to network...') + sta_if.active(True) + sta_if.connect(ssid, pwd) + while not sta_if.isconnected(): + pass + print('network config:', sta_if.ifconfig()) + +# Attempt to connect to WiFi network +do_connect("Smartwave-1", 'Smartwave-prot-1') diff --git a/micro_ondes/esp_wifi/device_id.txt b/micro_ondes/esp_wifi/device_id.txt new file mode 100644 index 0000000..d8263ee --- /dev/null +++ b/micro_ondes/esp_wifi/device_id.txt @@ -0,0 +1 @@ +2 \ No newline at end of file diff --git a/micro_ondes/esp_wifi/main.py b/micro_ondes/esp_wifi/main.py index e69de29..184e187 100644 --- a/micro_ondes/esp_wifi/main.py +++ b/micro_ondes/esp_wifi/main.py @@ -0,0 +1,86 @@ +import _thread +import select # <--- Built-in module to handle TLS timeouts +from machine import Pin +from shared import get_mqtt_client +from shared import config +import time + +# --- Hardware Configuration --- +vext = Pin(19, Pin.OUT) +vext.value(0) +time.sleep_ms(100) + +# --- Read Unique Device ID --- +try: + with open("device_id.txt", "r") as f: + DEVICE_ID = f.read().strip() +except Exception: + DEVICE_ID = "ESP32_Inconnu" + +MQTT_CA_FILE = "/certs/ca.crt" + +# --- Setup MQTT Client --- +mqtt_client = get_mqtt_client( + host=config.MQTT_BROKER_HOST, + client_id="smartwave-esp32-" + DEVICE_ID, + use_tls=config.USE_TLS, + cafile=MQTT_CA_FILE, + keepalive=config.MQTT_KEEPALIVE, # Can safely be 30 now +) + +def on_mqtt_message(message): + print("[MQTT Thread] Received message:", message) + +mqtt_client.set_callback(on_mqtt_message) + + +def mqtt_background_thread(): + """Background MQTT worker using select.poll() for keepalive tracking.""" + print("[Thread] Background MQTT worker started.") + + while True: + try: + print("[Thread] Attempting connection to MQTT broker...") + mqtt_client.connect() + print("[Thread] Connected! Subscribing to topic...") + mqtt_client.subscribe(config.MQTT_TOPIC, qos=config.MQTT_QOS) + print("[Thread] Successfully subscribed. Setting up poller...") + + # --- THE SELECT POLLER SETUP --- + # Create a poller and register our active TLS socket to look for incoming data (POLLIN) + poller = select.poll() + poller.register(mqtt_client._client.sock, select.POLLIN) + + # Listening loop + while True: + # Wait for network events for a maximum of 15000 milliseconds (15 seconds) + events = poller.poll(15000) + + if not events: + # The 15 seconds expired with zero network traffic! + # Send a keepalive ping to Mosquitto. + print("[Thread] No data for 15s. Sending keepalive ping...") + mqtt_client._client.ping() + else: + # Data has physically arrived on the socket! + # Calling wait() now is completely safe and won't block indefinitely. + mqtt_client.wait() + + except Exception as e: + print("[Thread] Connection dropped or error encountered:", e) + print("[Thread] Cleaning up socket context. Retrying in 5 seconds...") + try: + mqtt_client.close() + except Exception: + pass + time.sleep(5) + +# --- Launch background worker --- +_thread.start_new_thread(mqtt_background_thread, ()) + + +# --- MAIN APPLICATION THREAD (Core 0) --- +print("[Main] Main execution path active.") +while True: + # Your main physical loop runs completely unhindered here + time.sleep(1) \ No newline at end of file diff --git a/orchestrateur/compose.yaml b/orchestrateur/compose.yaml index c064870..4b4ee2a 100644 --- a/orchestrateur/compose.yaml +++ b/orchestrateur/compose.yaml @@ -3,6 +3,7 @@ services: image: eclipse-mosquitto:2.0 environment: MQTT_TLS_ENABLED: ${MQTT_TLS_ENABLED:-true} + # restart: unless-stopped ports: - "192.168.50.1:8884:8884" volumes: @@ -13,6 +14,7 @@ services: - mqtt-data:/mosquitto/data - mqtt-log:/mosquitto/log command: ["/bin/sh", "/scripts/start-broker.sh"] + # command: ["tail", "-f", "/dev/null"] # Do nothing volumes: mqtt-data: diff --git a/orchestrateur/main.py b/orchestrateur/main.py index 78d1bbc..9610975 100644 --- a/orchestrateur/main.py +++ b/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 \ No newline at end of file + break + +# Clean termination +if hasattr(mqtt_client._client, "loop_stop"): + mqtt_client._client.loop_stop() +mqtt_client.close() \ No newline at end of file diff --git a/orchestrateur/mqtt/mosquitto-plain.conf b/orchestrateur/mqtt/mosquitto-plain.conf index faed2ca..f9855df 100644 --- a/orchestrateur/mqtt/mosquitto-plain.conf +++ b/orchestrateur/mqtt/mosquitto-plain.conf @@ -8,5 +8,5 @@ log_type notice log_type information allow_anonymous true -listener 8884 192.168.50.1 +listener 8884 protocol mqtt \ No newline at end of file diff --git a/orchestrateur/mqtt/mosquitto-tls.conf b/orchestrateur/mqtt/mosquitto-tls.conf index 8d58795..8b3cde3 100644 --- a/orchestrateur/mqtt/mosquitto-tls.conf +++ b/orchestrateur/mqtt/mosquitto-tls.conf @@ -8,7 +8,7 @@ log_type notice log_type information allow_anonymous true -listener 8884 192.168.50.1 +listener 8884 protocol mqtt cafile /mosquitto/certs/ca.crt certfile /mosquitto/certs/server.crt diff --git a/shared/config.py b/shared/config.py index 9afb033..8f8ce9d 100644 --- a/shared/config.py +++ b/shared/config.py @@ -1,2 +1,11 @@ + +# LoRa HEARTBEAT_INTERVAL = 10 + +# MQTT +MQTT_BROKER_HOST = "192.168.50.1" +MQTT_TOPIC = b"smartwave/demo" +MQTT_KEEPALIVE = 30 +USE_TLS = True +MQTT_QOS = 2 \ No newline at end of file diff --git a/shared/mqtt.py b/shared/mqtt.py index aceb8a3..7de4a9e 100644 --- a/shared/mqtt.py +++ b/shared/mqtt.py @@ -104,9 +104,17 @@ class BrokerClient: return self._client if IS_MICROPYTHON: + import ssl ssl_params = self.ssl_params - if self.use_tls and ssl_params is None and self.cafile is not None: - ssl_params = {"cadata": _read_file_bytes(self.cafile)} + + if self.use_tls and ssl_params is None: + # MicroPython uses context-less structures. + # If your CA is self-signed, validation can fail without a valid hostname match. + ssl_params = { + "cert_reqs": ssl.CERT_NONE, # Temporarily change to NONE to test if validation is the culprit + "cadata": _read_file_bytes(self.cafile) + } + client = _MQTTClient( self.client_id or "smartWave-client", self.host, @@ -114,7 +122,7 @@ class BrokerClient: user=self.username, password=self.password, keepalive=self.keepalive, - ssl=self.use_tls or ssl_params is not None, + ssl=self.use_tls, ssl_params=ssl_params, ) self._client = client @@ -155,6 +163,10 @@ class BrokerClient: payload_bytes = _ensure_bytes(payload) if IS_MICROPYTHON: return client.publish(topic, payload_bytes, retain=retain, qos=qos) + + if isinstance(topic, bytes): + topic = topic.decode('utf-8') + return client.publish(topic, payload_bytes, qos=qos, retain=retain) def subscribe(self, topic, qos=2): @@ -162,6 +174,10 @@ class BrokerClient: if IS_MICROPYTHON: client.set_callback(self._on_micropython_message) return client.subscribe(topic, qos=qos) + + if isinstance(topic, bytes): + topic = topic.decode('utf-8') + return client.subscribe(topic, qos=qos) def _on_micropython_message(self, topic, payload):