diff --git a/micro_ondes/esp_lora/main.py b/micro_ondes/esp_lora/main.py index 9a1c695..f6fb6f3 100644 --- a/micro_ondes/esp_lora/main.py +++ b/micro_ondes/esp_lora/main.py @@ -1,8 +1,6 @@ import _thread from machine import Pin -from shared import get_lora -from shared import deviceTypes -from shared import config +from shared import get_lora, get_uart, deviceTypes, 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 - time.sleep(1) \ No newline at end of file + # 1. Listen for incoming UART serial packets from the WROOM board + 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) \ No newline at end of file diff --git a/micro_ondes/esp_wifi/main.py b/micro_ondes/esp_wifi/main.py index 184e187..979fba2 100644 --- a/micro_ondes/esp_wifi/main.py +++ b/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 config +from shared import get_mqtt_client, get_uart, 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) - events = poller.poll(15000) + # 1. Process outbound messages queued by the main thread + 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: - # 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. + # 2. Check for incoming messages (non-blocking poll) + # Shortened timeout to keep the queue responsive + events = poller.poll(200) + if events: 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 - time.sleep(1) \ No newline at end of file + print("[Main] Queueing a test message for MQTT...") + # 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 \ No newline at end of file diff --git a/orchestrateur/main.py b/orchestrateur/main.py index 9610975..54ee21a 100644 --- a/orchestrateur/main.py +++ b/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 ) diff --git a/shared/__init__.py b/shared/__init__.py index 0f66a7e..d782b38 100644 --- a/shared/__init__.py +++ b/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) \ No newline at end of file + return BrokerClient(*args, **kwargs) + +def get_uart(*args, **kwargs): + from .uart_comm import SafeUART + return SafeUART(*args, **kwargs) \ No newline at end of file diff --git a/shared/config.py b/shared/config.py index 8f8ce9d..1752375 100644 --- a/shared/config.py +++ b/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 \ No newline at end of file +MQTT_QOS = 1 \ No newline at end of file diff --git a/shared/lora_device.py b/shared/lora_device.py index 6cead42..b3bca03 100644 --- a/shared/lora_device.py +++ b/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: diff --git a/shared/uart_comm.py b/shared/uart_comm.py new file mode 100644 index 0000000..2b701c2 --- /dev/null +++ b/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 \ No newline at end of file