From 01eccf47c6d3b4ba9c8c18c9aaaad17b2c8c804d Mon Sep 17 00:00:00 2001 From: Matthias Guillitte Date: Sun, 16 Aug 2026 16:28:58 +0200 Subject: [PATCH] Better LoRa --- micro_ondes/esp_lora/main.py | 62 ++++++----- orchestrateur/main.py | 4 +- shared/lora_device.py | 198 +++++++++++++++++++---------------- 3 files changed, 141 insertions(+), 123 deletions(-) diff --git a/micro_ondes/esp_lora/main.py b/micro_ondes/esp_lora/main.py index 6ee9cfe..3348ae7 100644 --- a/micro_ondes/esp_lora/main.py +++ b/micro_ondes/esp_lora/main.py @@ -1,8 +1,6 @@ import gc import sys import time -import _thread -from lib.microwaveScreen import MicrowaveScreen import uasyncio as asyncio from machine import Pin, SoftI2C import ssd1306 @@ -19,6 +17,7 @@ from shared.logging import log from shared.lora_device import LoraCommands from shared.alerts import Alert, AlertType, AlertManager from shared.microwave_state import MicrowaveState +from lib.microwaveScreen import MicrowaveScreen # --- READ DEVICE ID --- try: @@ -112,10 +111,9 @@ def init_hardware(): print(f"[Main] ESP32 initialized with ID: '{DEVICE_ID}' (Type: {deviceTypes.DEVICE_TYPES['MICROWAVE']})") - -# --- DEDICATED LORA HARDWARE THREAD --- -def lora_hardware_thread(): - """Runs in a separate OS thread to keep the LoRa radio in continuous RX mode.""" +# --- ASYNC LORA TASK (REPLACES _THREAD) --- +async def lora_rx_and_heartbeat_task(): + """Replaces the hardware OS thread with a non-blocking async task to avoid SPI collisions.""" last_heartbeat_time = 0 while True: @@ -124,19 +122,22 @@ def lora_hardware_thread(): # 1. Send periodic heartbeat if now - last_heartbeat_time >= config.LORA_HEARTBEAT_INTERVAL: last_heartbeat_time = now - log("\n[LoRa Thread] Sending Heartbeat...") + log("\n[LoRa Task] Sending Heartbeat...") if lora: lora.send(PING_PAYLOAD) + # CRITICAL: Force the radio chip out of Standby mode and back into RX mode + if hasattr(lora, 'receive'): + lora.receive() - # 2. Blocking 300ms RX listen window (keeps radio actively listening) + # 2. Listen for incoming packets (Must be longer than LoRa Time-on-Air) if lora: - paquet = lora.receive_reliable(timeout_ms=300) + paquet = lora.receive_reliable(timeout_ms=350) if paquet is not None: - log(f"[LoRa Thread] New Packet Received: {paquet}") + log(f"[LoRa Task] New Packet Received: {paquet}") data_queue.put(paquet) - time.sleep_ms(10) - + # Give control back to event loop + await asyncio.sleep_ms(10) # --- COOKING STATE CALLBACKS --- def cooking_state_temperature_provider(): @@ -228,7 +229,7 @@ def update_screen(): async def uart_polling_task(): """Polls UART for incoming messages from the WiFi board.""" - global cooking_state, cooking_start_time, temperature_asked + global cooking_state, cooking_start_time, temperature_asked, microwave_state while True: if uart_device and uart_device.any(): @@ -256,6 +257,7 @@ async def uart_polling_task(): await asyncio.sleep_ms(200) cooking_state_on_state_change(cooking_state) + microwave_state = MicrowaveState.COOKING elif command.command_type == UARTCommandType.TEMPERATURE_RESPONSE: try: # Check if payload is already a dict or needs JSON decoding @@ -282,40 +284,43 @@ async def lora_process_task(): if paquet and not paquet.get("raw"): data = paquet.get("data", {}) + # RECIPIENT FILTERING: Drop packets not addressed to this ESP32 or ALL + target = data.get("target_id") or data.get("recipient") + if target and target not in [DEVICE_ID, "ALL", "BROADCAST"]: + log(f"[LoRa Process] Ignoring packet intended for {target}") + continue + if "action" in data: if data["action"] == LoraCommands.TOGGLE_PAUSE: if cooking_state is not None: if cooking_state.state == cookingState.CookingStates.DONE: - print("[LoRa Process] Cooking is done. Resetting microwave for the next session.") + print("[LoRa Process] Cooking is done. Resetting microwave for next session.") cooking_state.set_state(cookingState.CookingStates.IDLE) await asyncio.sleep_ms(20) cooking_state = None else: cooking_state.toggle_pause() - if cooking_state.paused: - print("[LoRa Process] Cooking paused via orchestrator command.") - else: - print("[LoRa Process] Cooking resumed via orchestrator command.") + print(f"[LoRa Process] Cooking paused state toggled to {cooking_state.paused}") else: log("[LoRa Process] No active cooking state to toggle pause/resume.") + elif data["action"] == LoraCommands.TOGGLE_DEFROST: print("[LoRa Process] Toggling defrost mode via orchestrator command.") - defrost_mode = data["defrost_state"] + defrost_mode = data.get("defrost_state", False) update_screen() + elif data["action"] == LoraCommands.NEW_ALERT: alert = Alert(data.get("alert_type"), data.get("message", "Unsafe area")) alert.timestamp = data.get("timestamp", time.time()) print(f"[LoRa Process] New alert received via orchestrator: {alert.to_dict()}") alert_manager.add_alert(alert) + elif data["action"] == LoraCommands.MICROVAVE_STATE_UPDATE: new_state = data.get("new_microwave_state") if new_state: print(f"[LoRa Process] Microwave state update received: {new_state}") microwave_state = new_state update_screen() - else: - print("[LoRa Process] Received microwave state update with no state specified.") - await asyncio.sleep_ms(100) @@ -351,23 +356,16 @@ async def memory_cleanup_task(): # --- BOOTSTRAP --- async def main(): - global cooking_state, defrost_mode, microwave_screen, alert_manager + global alert_manager print("[Main] Starting application...") init_hardware() alert_manager = AlertManager() alert_manager.set_on_alert_callback(on_new_alert) - - # Launch dedicated hardware thread for LoRa RX - try: - _thread.stack_size(16 * 1024) - except Exception: - pass - _thread.start_new_thread(lora_hardware_thread, ()) - print("[Main] LoRa hardware background thread started.") - # Launch background async tasks + # Launch all tasks within the same single-threaded uasyncio event loop + asyncio.create_task(lora_rx_and_heartbeat_task()) asyncio.create_task(uart_polling_task()) asyncio.create_task(lora_process_task()) asyncio.create_task(cooking_loop_task()) diff --git a/orchestrateur/main.py b/orchestrateur/main.py index 6b1702e..33ef0a5 100644 --- a/orchestrateur/main.py +++ b/orchestrateur/main.py @@ -268,11 +268,11 @@ def button_callback(): global button_state if microwave_states.get("2").cooking_state == CookingStates.COOKING or microwave_states.get("2").cooking_state == CookingStates.DONE or microwave_states.get("2").cooking_state == CookingStates.STIRRING_REQUIRED: print("[Button] Toggling pause/resume for microwave '2'.") - lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_PAUSE}) + lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_PAUSE}, max_retries=6) else: button_state = not button_state print(f"[Button] Defrost state toggled to: {button_state}") - lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_DEFROST, "defrost_state": button_state}) + lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_DEFROST, "defrost_state": button_state}, max_retries=5) button.set_callback(button_callback) button.start_button_monitoring_thread() diff --git a/shared/lora_device.py b/shared/lora_device.py index 1002783..a8440e8 100644 --- a/shared/lora_device.py +++ b/shared/lora_device.py @@ -27,6 +27,12 @@ class BaseLoraDevice: self.received_acks = set() self.pending_rx_queue = [] self.default_group = 2 + + # NEW: Global TX lock to prevent concurrent overlapping transmissions + if IS_MICROPYTHON: + self.tx_lock = _thread.allocate_lock() + else: + self.tx_lock = threading.Lock() def _generate_msg_id(self): return random.getrandbits(16) @@ -77,51 +83,49 @@ class BaseLoraDevice: def send_reliable(self, payload, max_retries=4, ack_timeout=3.0): """Sends a payload and listens in a single continuous RX window for the ACK.""" - lock = getattr(self, 'lock', None) - if isinstance(payload, dict): - payload = dict(payload) - else: - payload = {"data": payload} + with self.tx_lock: + lock = getattr(self, 'lock', None) + + if isinstance(payload, dict): + payload = dict(payload) + else: + payload = {"data": payload} - msg_id = self._generate_msg_id() - payload["_msg_id"] = msg_id + msg_id = self._generate_msg_id() + payload["_msg_id"] = msg_id - log(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===") + log(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===") - for attempt in range(max_retries): - log(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}") - self.send(payload) + for attempt in range(max_retries): + log(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}") + self.send(payload) - # 1. Open a single continuous RX window for the full timeout duration - # (Defaulted to 3.0s to account for LA66 UART + transmission time) - timeout_ms = int(ack_timeout * 1000) - packet = self.receive_packet(timeout_ms=timeout_ms) + timeout_ms = int(ack_timeout * 1000) + packet = self.receive_packet(timeout_ms=timeout_ms) + + if packet: + if lock: lock.acquire() + try: + filtered_packet = self._process_incoming_packet(packet) + if filtered_packet: + self.pending_rx_queue.append(filtered_packet) + finally: + if lock: lock.release() - # 2. Process incoming packet if received - if packet: if lock: lock.acquire() try: - filtered_packet = self._process_incoming_packet(packet) - if filtered_packet: - self.pending_rx_queue.append(filtered_packet) + if msg_id in self.received_acks: + self.received_acks.remove(msg_id) + log(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===") + return True finally: if lock: lock.release() - # 3. Check if matching ACK was received - if lock: lock.acquire() - try: - if msg_id in self.received_acks: - self.received_acks.remove(msg_id) - log(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===") - return True - finally: - if lock: lock.release() + log(f"[ReliableLoRa] Attempt {attempt + 1} timed out waiting for ACK for msg_id {msg_id}") - log(f"[ReliableLoRa] Attempt {attempt + 1} timed out waiting for ACK for msg_id {msg_id}") - - print(f"[ReliableLoRa] ERROR: Failed to receive ACK for msg_id {msg_id} after {max_retries} attempts.") - return False + print(f"[ReliableLoRa] ERROR: Failed to receive ACK for msg_id {msg_id} after {max_retries} attempts.") + return False def receive_reliable(self, timeout_ms=1000): """Receives a packet, automatically sending ACKs and filtering duplicate retries.""" @@ -296,19 +300,28 @@ else: self.ser.reset_output_buffer() self.lock = threading.Lock() - # Initial configuration + self._emergency_rx_buffer = [] + self.configure(freq=868.1, sf=7, bw=125) def _send_at_cmd(self, cmd, timeout=1.5): - """Sends AT command and reads response until line received or timeout.""" + """Sends AT command and safely handles async incoming packets.""" self.ser.write(f"{cmd}\r\n".encode('utf-8')) start = time.time() resp = "" while (time.time() - start) < timeout: if self.ser.in_waiting > 0: - line = self.ser.readline().decode('utf-8', errors='ignore') - resp += line + line = self.ser.readline().decode('utf-8', errors='ignore').strip() + + if not line: + continue + + if any(marker in line for marker in ["+RECV:", "+RCV=", "+DRX:", "(HEX:)", "Data:"]): + self._emergency_rx_buffer.append(line) + continue + + resp += line + "\n" if "OK" in line or "ERROR" in line or "AT_BUSY" in line: break time.sleep(0.02) @@ -364,64 +377,70 @@ else: timeout_s = timeout_ms / 1000.0 while (time.time() - start_time) < timeout_s: - if self.ser.in_waiting > 0: + line = "" + + # NEW: Process emergency buffer first before checking serial + if self._emergency_rx_buffer: + line = self._emergency_rx_buffer.pop(0) + elif self.ser.in_waiting > 0: line = self.ser.readline().decode('utf-8', errors='ignore').strip() - if line: - payload_bytes = None - - # Robust parsing for LA66 response variants (+RECV:, +RCV=, +DRX:, HEX:, Data:) - if "+RECV:" in line: - parts = line.split("+RECV:")[1].strip().split(",") - hex_str = parts[2].strip() if len(parts) >= 3 else parts[0].strip() - try: payload_bytes = bytes.fromhex(hex_str) + + if line: + payload_bytes = None + + # Robust parsing for LA66 response variants (+RECV:, +RCV=, +DRX:, HEX:, Data:) + if "+RECV:" in line: + parts = line.split("+RECV:")[1].strip().split(",") + hex_str = parts[2].strip() if len(parts) >= 3 else parts[0].strip() + try: payload_bytes = bytes.fromhex(hex_str) + except ValueError: pass + elif "+RCV=" in line: + parts = line.split("+RCV=")[1].strip().split(",") + if len(parts) >= 4: + try: payload_bytes = bytes.fromhex(parts[3].strip()) except ValueError: pass - elif "+RCV=" in line: - parts = line.split("+RCV=")[1].strip().split(",") - if len(parts) >= 4: - try: payload_bytes = bytes.fromhex(parts[3].strip()) - except ValueError: pass - elif "+DRX:" in line: - parts = line.split("+DRX:")[1].strip().split(",") - if len(parts) >= 2: - try: payload_bytes = bytes.fromhex(parts[1].strip()) - except ValueError: pass - elif "(HEX:)" in line: - hex_part = line.split("(HEX:)")[1].strip().replace(" ", "") - try: payload_bytes = bytes.fromhex(hex_part) + elif "+DRX:" in line: + parts = line.split("+DRX:")[1].strip().split(",") + if len(parts) >= 2: + try: payload_bytes = bytes.fromhex(parts[1].strip()) except ValueError: pass - elif "Data:" in line: - payload_bytes = line.split("Data:")[1].strip().encode('utf-8') + elif "(HEX:)" in line: + hex_part = line.split("(HEX:)")[1].strip().replace(" ", "") + try: payload_bytes = bytes.fromhex(hex_part) + except ValueError: pass + elif "Data:" in line: + payload_bytes = line.split("Data:")[1].strip().encode('utf-8') + + if payload_bytes and len(payload_bytes) > 0: + if payload_bytes[0] in (0x7B, 0x5B): + group = self.default_group + payload_clean = payload_bytes.strip(b'\x00 \r\n\t') + elif len(payload_bytes) > 1: + group = payload_bytes[0] + payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t') + else: + continue - if payload_bytes and len(payload_bytes) > 0: - if payload_bytes[0] in (0x7B, 0x5B): - group = self.default_group - payload_clean = payload_bytes.strip(b'\x00 \r\n\t') - elif len(payload_bytes) > 1: - group = payload_bytes[0] - payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t') - else: - continue - + try: + text = payload_clean.decode('utf-8').strip('\x00 \r\n\t') + except UnicodeError: + continue + + if text.startswith('{') or text.startswith('['): + decoded_text = text + elif text.lower().startswith('7b') or text.lower().startswith('5b'): try: - text = payload_clean.decode('utf-8').strip('\x00 \r\n\t') - except UnicodeError: - continue - - if text.startswith('{') or text.startswith('['): + decoded_text = bytes.fromhex(text).decode('utf-8').strip('\x00 \r\n\t') + except Exception: decoded_text = text - elif text.lower().startswith('7b') or text.lower().startswith('5b'): - try: - decoded_text = bytes.fromhex(text).decode('utf-8').strip('\x00 \r\n\t') - except Exception: - decoded_text = text - else: - decoded_text = text - - try: - parsed_json = json.loads(decoded_text) - return {"group": group, "data": parsed_json, "raw": False} - except json.JSONDecodeError: - return {"group": group, "data": decoded_text, "raw": True} + else: + decoded_text = text + + try: + parsed_json = json.loads(decoded_text) + return {"group": group, "data": parsed_json, "raw": False} + except json.JSONDecodeError: + return {"group": group, "data": decoded_text, "raw": True} time.sleep(0.01) return None @@ -438,6 +457,7 @@ def get_lora_device(port_or_pins=None): class LoraCommands: PING = "ping" + MICROVAVE_STATE_UPDATE = "microwave_state_update" COOKING_STATE_UPDATE = "cooking_state_update" COOKING_UPDATE = "cooking_update" TOGGLE_PAUSE = "toggle_pause"