UART communication

micro_ondes/esp_lora/main.py

@@ -1,8 +1,6 @@
 import _thread
 from machine import Pin
-from shared import get_lora
+from shared import get_lora, get_uart, deviceTypes, config
-from shared import deviceTypes
-from shared import config
 import time
 
 # --- Configuration Matérielle ---
@@ -48,10 +46,24 @@ def heartbeat_loop():
             print("ESP32 : Pas de réponse de l'orchestrateur (Le RPI est-il éteint ?)")
         
         time.sleep(config.HEARTBEAT_INTERVAL)
+        
+# UART
+uart_device = get_uart(uart_id=1, tx_pin=46, rx_pin=45)
 
 # Lancer la boucle de heartbeat dans un thread séparé
 _thread.start_new_thread(heartbeat_loop, ())
 
+# --- MAIN APPLICATION THREAD ---
+print("[Main] Main execution path active.")
 while True:
-    # Fait rien pour l'instant
+    # 1. Listen for incoming UART serial packets from the WROOM board
-    time.sleep(1)
+    while uart_device.any():
+        command = uart_device.read()
+        print(f"[Main] Received command from WiFi Board: {command}")
+        
+    uart_device.send(f"Hello from esp-32 lora ID {DEVICE_ID}") 
+
+    # 2. Send local metrics over the wire to the WiFi board every few seconds
+    # uart_device.send("Data Pack: LoRa Link RSSI -72dBm")
+
+    time.sleep_ms(200)

micro_ondes/esp_wifi/main.py

@@ -1,16 +1,23 @@
 import _thread
-import select # <--- Built-in module to handle TLS timeouts
+import select
 from machine import Pin
-from shared import get_mqtt_client
+from shared import get_mqtt_client, get_uart, config
-from shared import config
 import time
 
-# --- Hardware Configuration ---
+# Simple thread-safe queue list
+msg_queue = []
+queue_lock = _thread.allocate_lock()
+
+def queue_publish(topic, payload):
+    """Safely queues a message from the main thread."""
+    with queue_lock:
+        msg_queue.append((topic, payload))
+
+# --- Hardware & Client Setup ---
 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()
@@ -19,13 +26,12 @@ except Exception:
     
 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
+    keepalive=config.MQTT_KEEPALIVE,
 )
 
 def on_mqtt_message(message):
@@ -35,7 +41,7 @@ mqtt_client.set_callback(on_mqtt_message)
 
 
 def mqtt_background_thread():
-    """Background MQTT worker using select.poll() for keepalive tracking."""
+    """Background MQTT worker handling ALL socket operations safely."""
     print("[Thread] Background MQTT worker started.")
     
     while True:
@@ -43,37 +49,59 @@ def mqtt_background_thread():
             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)
+            mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, 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
+            last_check = time.time()
+            
             while True:
-                # Wait for network events for a maximum of 15000 milliseconds (15 seconds)
+                # 1. Process outbound messages queued by the main thread
-                events = poller.poll(15000)
+                while len(msg_queue) > 0:
+                    with queue_lock:
+                        topic, payload = msg_queue.pop(0)
+                    print(f"[Thread] Safely publishing queued message to {topic}...")
+                    mqtt_client.publish(topic, payload, qos=config.MQTT_QOS)
                 
-                if not events:
+                # 2. Check for incoming messages (non-blocking poll)
-                    # The 15 seconds expired with zero network traffic! 
+                # Shortened timeout to keep the queue responsive
-                    # Send a keepalive ping to Mosquitto.
+                events = poller.poll(200) 
-                    print("[Thread] No data for 15s. Sending keepalive ping...")
+                if events:
-                    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()
+                
+                # 3. Handle Keepalive tracking manually
+                if time.time() - last_check >= 15:
+                    print("[Thread] Sending keepalive ping...")
+                    mqtt_client._client.ping()
+                    last_check = time.time()
+                    
+                # Small breathe room for the CPU core
+                time.sleep_ms(50)
                         
         except Exception as e:
             print("[Thread] Connection dropped or error encountered:", e)
             print("[Thread] Cleaning up socket context. Retrying in 5 seconds...")
+            
+            # --- FIX FOR ERROR 23 (SOCKET LEAK) ---
+            # Manually force-kill the underlying socket file descriptor if it exists
+            try:
+                if mqtt_client._client and hasattr(mqtt_client._client, "sock"):
+                    if mqtt_client._client.sock is not None:
+                        mqtt_client._client.sock.close()
+            except Exception:
+                pass # Already dead or closed
+
+            # Now we let the wrapper do its normal cleanup safely
             try:
                 mqtt_client.close()
             except Exception:
                 pass
             time.sleep(5)
+            
+# UART
+uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16)
 
 # --- Launch background worker ---
 _thread.start_new_thread(mqtt_background_thread, ())
@@ -81,6 +109,19 @@ _thread.start_new_thread(mqtt_background_thread, ())
 
 # --- MAIN APPLICATION THREAD (Core 0) ---
 print("[Main] Main execution path active.")
+time.sleep(2) # Give the thread a moment to initial connect
+
 while True:
-    # Your main physical loop runs completely unhindered here
+    print("[Main] Queueing a test message for MQTT...")
-    time.sleep(1)
+    # Instead of direct publishing, push it to the queue safely
+    queue_publish(config.MQTT_TOPIC_SENSOR, "Hello from ESP32!")
+    
+    # 1. Check if the Heltec V3 sent us something over the wire
+    while uart_device.any():
+        incoming_msg = uart_device.read()
+        print(f"[Main] Received from esp-lora over UART: {incoming_msg}")
+        
+    # 2. Example: Send data to the Heltec board every 5 seconds
+    # uart_device.send("Status Check: WiFi Active")
+    
+    time.sleep_ms(200) # Fast responsive polling loop for local UART

orchestrateur/main.py

@@ -48,6 +48,8 @@ mqtt_client = get_mqtt_client(
     keepalive=config.MQTT_KEEPALIVE,
 )
 mqtt_client.connect()
+mqtt_client.subscribe(config.MQTT_TOPIC_SENSOR, qos=config.MQTT_QOS)
+print(f"Subscribed to topic: {config.MQTT_TOPIC_SENSOR}")
 
 # --- THE CRUCIAL PAHO FIX ---
 # Start Paho's internal background thread. This handles all network packets,
@@ -93,7 +95,7 @@ while True:
         # Publish debug telemetry message
         print("[Main Loop] Envoi d'un message de debug sur MQTT...")
         response = mqtt_client.publish(
-            config.MQTT_TOPIC,
+            config.MQTT_TOPIC_COOKING,
             f"Orchestrateur actif, ID: {DEVICE_ID}",
             qos=config.MQTT_QOS
         )

shared/__init__.py

@@ -14,4 +14,8 @@ def get_database(*args, **kwargs):
 
 def get_mqtt_client(*args, **kwargs):
     from .mqtt import BrokerClient
-    return BrokerClient(*args, **kwargs)
+    return BrokerClient(*args, **kwargs)
+
+def get_uart(*args, **kwargs):
+    from .uart_comm import SafeUART
+    return SafeUART(*args, **kwargs)

shared/config.py

@@ -5,7 +5,8 @@ HEARTBEAT_INTERVAL = 10
 
 # MQTT
 MQTT_BROKER_HOST = "192.168.50.1"
-MQTT_TOPIC = b"smartwave/demo"
+MQTT_TOPIC_SENSOR = b"smartwave/sensor"
+MQTT_TOPIC_COOKING = b"smartwave/cooking"
 MQTT_KEEPALIVE = 30
 USE_TLS = True
-MQTT_QOS = 2
+MQTT_QOS = 1

shared/lora_device.py

@@ -112,7 +112,7 @@ else:
                 
                 # La clé ajoute d'elle-même l'octet de groupe configuré dans ses registres
                 cmd = f"AT+SEND=1,{hex_payload},1,3\r\n"
-                print(f"RPI : Envoi de la commande HEX -> AT+SEND=1,[HEX_DATA],1,3")
+                # print(f"RPI : Envoi de la commande HEX -> AT+SEND=1,[HEX_DATA],1,3")
                 self.ser.write(cmd.encode('utf-8'))
                 
                 time.sleep(0.2)
@@ -123,7 +123,7 @@ else:
                         response += self.ser.readline().decode('utf-8', errors='ignore')
                     time.sleep(0.05)
                 
-                print(f"[RPI LA66 TX STATUS] :\n{response.strip()}")
+                # print(f"[RPI LA66 TX STATUS] :\n{response.strip()}")
 
         def receive_packet(self, timeout_ms=5000):
             with self.lock:

shared/uart_comm.py

@@ -0,0 +1,66 @@
+# shared/uart_comm.py
+import _thread
+from machine import UART
+import time
+
+class SafeUART:
+    def __init__(self, uart_id, tx_pin, rx_pin, baudrate=115200):
+        # Initialize the hardware UART channel
+        self.uart = UART(uart_id, baudrate=baudrate, tx=tx_pin, rx=rx_pin, timeout=10)
+        
+        # Core thread-safety assets
+        self.lock = _thread.allocate_lock()
+        self.rx_queue = []
+        self.buffer = b""
+        
+        # Start the background data worker thread
+        _thread.stack_size(4096) # Cap the stack size for the UART listener
+        _thread.start_new_thread(self._listener_worker, ())
+        _thread.stack_size(0) 
+        
+        print(f"[UART] Thread initialized on UART{uart_id} (TX:{tx_pin}, RX:{rx_pin})")
+
+    def _listener_worker(self):
+        """Asynchronous internal loop parsing incoming stream lines into the queue."""
+        while True:
+            try:
+                if self.uart.any():
+                    with self.lock:
+                        # Pull all raw bytes waiting in the hardware ring buffer
+                        chunk = self.uart.read(self.uart.any())
+                        if chunk:
+                            self.buffer += chunk
+                        
+                        # Process complete lines terminated by a newline character
+                        while b'\n' in self.buffer:
+                            line, self.buffer = self.buffer.split(b'\n', 1)
+                            try:
+                                decoded_line = line.decode('utf-8').strip()
+                                if decoded_line:
+                                    self.rx_queue.append(decoded_line)
+                            except Exception:
+                                pass # Discard corrupt data frames safely
+            except Exception as e:
+                print("[UART Thread Error]:", e)
+                
+            time.sleep_ms(20) # Give other background threads breathing room
+
+    def send(self, message):
+        """Safely pushes strings across the serial wire from any thread context."""
+        if not message.endswith('\n'):
+            message += '\n'
+        
+        with self.lock:
+            self.uart.write(message.encode('utf-8'))
+
+    def any(self):
+        """Checks if any complete messages are waiting to be read."""
+        with self.lock:
+            return len(self.rx_queue) > 0
+
+    def read(self):
+        """Pulls the oldest unread string from the queue. Returns None if empty."""
+        with self.lock:
+            if self.rx_queue:
+                return self.rx_queue.pop(0)
+        return None