diff --git a/micro_ondes/esp_wifi/boot.py b/micro_ondes/esp_wifi/boot.py index b587b00..40fd0d6 100644 --- a/micro_ondes/esp_wifi/boot.py +++ b/micro_ondes/esp_wifi/boot.py @@ -5,16 +5,50 @@ esp.osdebug(True) #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()) +# def do_connect(ssid, pwd): +# import network +# sta_if = network.WLAN(network.STA_IF) +# sta_if.config(pm=sta_if.PM_NONE) +# 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()) + +import network +import time + +def do_connect(ssid, password): + wlan = network.WLAN(network.STA_IF) + + # 1. ALWAYS activate the interface FIRST + if not wlan.active(): + wlan.active(True) + + # 2. Configure Wi-Fi options AFTER activation + try: + # Disable Wi-Fi modem sleep (0 = PM_NONE) + wlan.config(pm=0) + except Exception as e: + print("[Wi-Fi] Warning: Failed to set power management:", e) + + # 3. Connect to the access point + if not wlan.isconnected(): + print(f"[Wi-Fi] Connecting to {ssid}...") + wlan.connect(ssid, password) + + timeout = 15 + start_time = time.time() + while not wlan.isconnected(): + if time.time() - start_time > timeout: + print("[Wi-Fi] Connection timed out!") + return False + time.sleep(0.5) + + print("[Wi-Fi] Connected! Network config:", wlan.ifconfig()) + return True # Attempt to connect to WiFi network do_connect("Smartwave-1", 'Smartwave-prot-1') diff --git a/micro_ondes/esp_wifi/main.py b/micro_ondes/esp_wifi/main.py index 5cc066d..21d046f 100644 --- a/micro_ondes/esp_wifi/main.py +++ b/micro_ondes/esp_wifi/main.py @@ -1,23 +1,12 @@ -import _thread -import select -from machine import Pin, I2C -from sensors import temperature_sensor -from shared import get_mqtt_client, get_uart, config, payloads, cookingState -from shared.uart_comm import UARTCommand, UARTCommandType -from shared.sensors import RGBLED -from shared.logging import log +import gc +import sys import time import ujson as json -import sys +import uasyncio as asyncio +from machine import Pin, I2C -# 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)) +# 1. Clean memory immediately +gc.collect() # --- READ DEVICE ID --- try: @@ -26,214 +15,290 @@ try: except Exception: DEVICE_ID = "ESP32_Inconnu" -# --- Cooking State --- -cooking_state = None # This will hold the current cooking state if any +# --- GLOBAL APP STATE --- +orchestrator_id = None +cooking_state = None +mqtt_connected = False + +# --- MQTT SETUP (Initialized First!) --- +from shared import get_mqtt_client, config, payloads -# --- MQTT SETUP --- MQTT_CA_FILE = "/certs/ca.crt" mqtt_client = get_mqtt_client( - host=config.MQTT_BROKER_HOST, - client_id="smartwave-esp32-" + DEVICE_ID, - use_tls=config.USE_TLS, + host="192.168.50.1", + client_id="smartwave-esp32-demo", + use_tls=True, cafile=MQTT_CA_FILE, - keepalive=config.MQTT_KEEPALIVE, + keepalive=30, ) -global orchestrator_id -orchestrator_id = None -def on_mqtt_message(message): - print("[MQTT Thread] Received message:", message) +# --- HARDWARE & MODULE DEFERRED IMPORTS --- +# We declare variables here, but initialize them AFTER MQTT connects +status_led = None +uart_device = None +mlx_temperature_sensor = None +cookingState = None +log = None +UARTCommand = None +UARTCommandType = None + + +def init_hardware(): + """Initializes hardware peripherals AFTER MQTT TLS has reserved its memory.""" + global status_led, uart_device, mlx_temperature_sensor + global cookingState, log, UARTCommand, UARTCommandType - # Try and parse the payload as json, but if it fails, just print the raw payload - payload_data=None - try: - payload_data = json.loads(message['payload']) - except Exception as e: - print("[MQTT Thread] Error parsing JSON:", e) - sys.print_exception(e) - pass # Maybe it's not JSON + print("[Main] Initializing hardware peripherals...") - if message['topic'] == config.MQTT_TOPIC_HELLO and payload_data and "id_orchestrator" in payload_data and payload_data["id_microwave"] == DEVICE_ID: - print("[MQTT Thread] Hello response received from orchestrator:", payload_data["id_orchestrator"]) - global orchestrator_id - orchestrator_id = payload_data["id_orchestrator"] - # Unsubscribe from the hello topic since we got a response - mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO) - print("[MQTT Thread] Unsubscribed from topic:", config.MQTT_TOPIC_HELLO) - - # Handle cooking messages - elif message['topic'] == config.MQTT_TOPIC_COOKING and payload_data and payload_data["id_microwave"] == DEVICE_ID: - # Cooking sensors init request - if not "cook_time" in payload_data: - print("[MQTT Thread] Cooking sensors init received from the orchestrator") - obj_temp = mlx_temperature_sensor.read_object_temp() - amb_temp = mlx_temperature_sensor.read_ambient_temp() - queue_publish(config.MQTT_TOPIC_SENSOR, payloads.mqtt_sensor_data(DEVICE_ID, obj_temp, amb_temp)) - # Received cooking parameters from the orchestrator - else: - print("[MQTT Thread] Cooking parameters received from the orchestrator:", payload_data) - global cooking_state - cooking_state = cookingState.CookingState( - cook_time=payload_data["cook_time"], - power_level=payload_data["power_level"], - target_temp=payload_data["target_temp"] - ) - cooking_state.set_state_change_callback(on_cooking_state_change) - cooking_state.set_state(cookingState.CookingStates.IDLE) # Set initial state to IDLE - # Send to the LoRa board the cooking parameters - uart_device.send_as_command(UARTCommand(UARTCommandType.COOKING_PARAMS, payload_data)) - print("[MQTT Thread] Cooking parameters sent to LoRa board.") - - print("[MQTT Thread] Message processing complete.") + # Deferred module imports + from shared import get_uart, cookingState as cs, logging + from shared.uart_comm import UARTCommand as UC, UARTCommandType as UCT + from shared.sensors import RGBLED + from sensors import temperature_sensor -mqtt_client.set_callback(on_mqtt_message) + cookingState = cs + log = logging.log + UARTCommand = UC + UARTCommandType = UCT + + # Status LED + status_led = RGBLED(red_pin=21, green_pin=19, blue_pin=18) + + # Hardware UART 2 + uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16) + + # I2C Temperature Sensor + temperature_sensor_i2c = I2C( + 0, + scl=Pin(25, Pin.IN, Pin.PULL_UP), + sda=Pin(26, Pin.IN, Pin.PULL_UP), + freq=100000, + ) + devices = temperature_sensor_i2c.scan() + if 0x5A in devices: + print("[Main] MLX90614 found at address 0x5A!") + else: + print("[Main] MLX90614 not found on I2C bus.") + mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c) -def mqtt_background_thread(): - """Background MQTT worker handling ALL socket operations safely.""" - print("[Thread] Background MQTT worker started.") +# --- CALLBACKS --- +def on_cooking_state_change(state): + if status_led is None: + return + print(f"[Main] Cooking state changed to: {state.state}") + BLINK_INTERVAL_MS = 500 - 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_COOKING, qos=config.MQTT_QOS) - print("[Thread] Successfully subscribed. Setting up poller...") - - poller = select.poll() - poller.register(mqtt_client._client.sock, select.POLLIN) - - last_check = time.time() - - while True: - # 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) - - # 2. Check for incoming messages (non-blocking poll) - 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) - sys.print_exception(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 + from shared.sensors import RGBLED + if state.state == cookingState.CookingStates.IDLE: + status_led.color = RGBLED.OFF + status_led.blink_off() + elif state.state == cookingState.CookingStates.COOKING: + status_led.color = RGBLED.YELLOW + status_led.blink_off() + elif state.state == cookingState.CookingStates.STIRRING_REQUIRED: + status_led.color = RGBLED.ORANGE + status_led.blink_on(BLINK_INTERVAL_MS) + elif state.state == cookingState.CookingStates.DONE: + status_led.color = RGBLED.GREEN + status_led.blink_off() + elif state.state == cookingState.CookingStates.ALERT: + status_led.color = RGBLED.RED + status_led.blink_on(BLINK_INTERVAL_MS) - # 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) - -# Cooking Cycle -cooking_state = None def on_received_cooking_state_update(new_state): global cooking_state if cooking_state is None: print("[Main] No active cooking state to update.") return - - log(f"[Main] Updating cooking state to: {new_state}") + if log: + log(f"[Main] Updating cooking state to: {new_state}") cooking_state.set_state(new_state) - -# Status LED -status_led = RGBLED(red_pin=21, green_pin=19, blue_pin=18) -def on_cooking_state_change(state): - print(f"[Main] Cooking state changed to: {state.state}") - - # === STATUS LED UPDATE === - BLINK_INTERVAL_MS = 500 # Blink every 500ms - - if state.state == cookingState.CookingStates.IDLE: - status_led.color = RGBLED.OFF - status_led.blink_off() - if state.state == cookingState.CookingStates.COOKING: - status_led.color = RGBLED.YELLOW - status_led.blink_off() - if state.state == cookingState.CookingStates.STIRRING_REQUIRED: - status_led.color = RGBLED.ORANGE - status_led.blink_on(BLINK_INTERVAL_MS) - if state.state == cookingState.CookingStates.DONE: - status_led.color = RGBLED.GREEN - status_led.blink_off() - if state.state == cookingState.CookingStates.ALERT: - status_led.color = RGBLED.RED - status_led.blink_on(BLINK_INTERVAL_MS) - -# --- MAIN APPLICATION THREAD (Core 0) --- -print("[Main] Main execution path active.") -# Temperature sensor setup -temperature_sensor_i2c = I2C(0, scl=Pin(25, Pin.IN, Pin.PULL_UP), sda=Pin(26, Pin.IN, Pin.PULL_UP), freq=100000) -# Scan to verify the sensor is connected and detected -print("Scanning I2C bus...") -devices = temperature_sensor_i2c.scan() -if 0x5A in devices: - print("MLX90614 found at address 0x5A!") -else: - print("MLX90614 not found. Please check your wiring.") -mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c) - -# --- Launch background worker --- -_thread.start_new_thread(mqtt_background_thread, ()) -time.sleep(2) # Give the thread a moment to initial connect -mqtt_hello_sent_timestamp = -config.MQTT_HELLO_INTERVAL -mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS) - -while True: - # MQTT HELLO sent every x seconds until we get a response from the orchestrator - if (orchestrator_id == None and -(mqtt_hello_sent_timestamp - time.time()) > config.MQTT_HELLO_INTERVAL): - print("[Main] Attempting to send initial hello to orchestrator...") - queue_publish(config.MQTT_TOPIC_HELLO, payloads.mqtt_hello(DEVICE_ID)) - mqtt_hello_sent_timestamp = time.time() - pass +def on_mqtt_message(message): + global orchestrator_id, cooking_state + print("[MQTT] Received message on topic:", message.get("topic")) - # 1. Listen for incoming UART serial packets from the WROOM board - while uart_device.any(): - command = uart_device.read_as_command() - if command: - print(f"[Main] Received command from LoRa Board: {command.command_type}") - if command.command_type == UARTCommandType.COOKING_STATE_UPDATE: - # Handle cooking state update command - new_state = command.payload.get("state", None) - print(f"[Main] Cooking state update received: {new_state}") - on_received_cooking_state_update(new_state) - else: - print(f"[Main] Unknown command type received: {command.command_type}") + payload_data = None + try: + payload_data = json.loads(message["payload"]) + except Exception as e: + print("[MQTT] Payload parsing warning:", e) + + topic = message.get("topic") + + # 1. Orchestrator Hello Response + if ( + topic == config.MQTT_TOPIC_HELLO + and payload_data + and payload_data.get("id_microwave") == DEVICE_ID + ): + orchestrator_id = payload_data.get("id_orchestrator") + print("[MQTT] Hello response received from orchestrator:", orchestrator_id) + try: + mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO) + print("[MQTT] Unsubscribed from topic:", config.MQTT_TOPIC_HELLO) + except Exception as e: + print("[MQTT] Unsubscribe error:", e) + sys.print_exception(e) + + # 2. Cooking Parameters / Sensor Request + elif ( + topic == config.MQTT_TOPIC_COOKING + and payload_data + and payload_data.get("id_microwave") == DEVICE_ID + ): + if "cook_time" not in payload_data: + print("[MQTT] Sensor data requested by orchestrator.") + obj_temp = mlx_temperature_sensor.read_object_temp() if mlx_temperature_sensor else 0 + amb_temp = mlx_temperature_sensor.read_ambient_temp() if mlx_temperature_sensor else 0 + + sensor_payload = payloads.mqtt_sensor_data(DEVICE_ID, obj_temp, amb_temp) + mqtt_client.publish( + config.MQTT_TOPIC_SENSOR, sensor_payload, qos=config.MQTT_QOS + ) else: - # Fallback to reading as a raw string if parsing fails - raw_command = uart_device.read() - print(f"[Main] Received raw command from WiFi Board: {raw_command}") - - # 2. Example: Send data to the Heltec board every 5 seconds - # uart_device.send("Status Check: WiFi Active") - time.sleep(1) \ No newline at end of file + print("[MQTT] Cooking parameters received:", payload_data) + if cookingState: + cooking_state = cookingState.CookingState( + cook_time=payload_data["cook_time"], + power_level=payload_data["power_level"], + target_temp=payload_data["target_temp"], + ) + cooking_state.set_state_change_callback(on_cooking_state_change) + cooking_state.set_state(cookingState.CookingStates.IDLE) + + if uart_device and UARTCommand: + uart_device.send_as_command( + UARTCommand(UARTCommandType.COOKING_PARAMS, payload_data) + ) + print("[MQTT] Cooking parameters sent to LoRa board.") + + +mqtt_client.set_callback(on_mqtt_message) + + +async def connect_mqtt_async(): + """Connects to MQTT safely while memory is clean.""" + global mqtt_connected + mqtt_connected = False + + while True: + try: + print("[MQTT] Connecting to broker with TLS...") + gc.collect() + mqtt_client.connect() + print("[MQTT] Connected! Subscribing to topics...") + mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, qos=config.MQTT_QOS) + mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS) + print("[MQTT] Subscribed successfully!") + mqtt_connected = True + return + except Exception as e: + print("[MQTT] Connection failed:", e) + sys.print_exception(e) + try: + mqtt_client.close() + except Exception: + pass + print("[MQTT] Retrying connection in 5 seconds...") + await asyncio.sleep(5) + + +# --- CONCURRENT ASYNC TASKS --- + +async def mqtt_poll_task(): + global mqtt_connected + last_ping = time.time() + + while True: + if mqtt_connected: + try: + mqtt_client.poll() + now = time.time() + if now - last_ping >= 15: + if mqtt_client._client: + mqtt_client._client.ping() + last_ping = now + except OSError as e: + print("[MQTT Task] Socket error encountered during poll/ping:", e) + mqtt_connected = False + await connect_mqtt_async() + + await asyncio.sleep_ms(30) + + +async def orchestrator_hello_task(): + global mqtt_connected + while True: + if mqtt_connected and orchestrator_id is None: + print("[Hello Task] Sending initial hello to orchestrator...") + try: + mqtt_client.publish( + config.MQTT_TOPIC_HELLO, + payloads.mqtt_hello(DEVICE_ID), + qos=config.MQTT_QOS, + ) + except OSError as e: + print("[Hello Task] Hello publish failed:", e) + mqtt_connected = False + await connect_mqtt_async() + + await asyncio.sleep(config.MQTT_HELLO_INTERVAL) + + +async def uart_task(): + while True: + if uart_device is not None: + while uart_device.any(): + command = uart_device.read_as_command() + if command: + print(f"[UART Task] Received command: {command.command_type}") + if command.command_type == UARTCommandType.COOKING_STATE_UPDATE: + new_state = command.payload.get("state", None) + print(f"[UART Task] Cooking state update: {new_state}") + on_received_cooking_state_update(new_state) + else: + print(f"[UART Task] Unknown command type: {command.command_type}") + else: + raw_command = uart_device.read() + print(f"[UART Task] Received raw command: {raw_command}") + + await asyncio.sleep_ms(20) + + +async def memory_cleanup_task(): + while True: + gc.collect() + await asyncio.sleep(10) + + +# --- MAIN ENTRY POINT --- +async def main(): + print("[Main] Starting application...") + + # STEP 1: Connect MQTT FIRST (while RAM is unfragmented) + await connect_mqtt_async() + + # STEP 2: Initialize Hardware & Secondary Modules AFTER connection + init_hardware() + + # STEP 3: Launch tasks + asyncio.create_task(mqtt_poll_task()) + asyncio.create_task(orchestrator_hello_task()) + asyncio.create_task(uart_task()) + asyncio.create_task(memory_cleanup_task()) + + print("[Main] All tasks running concurrently!") + + while True: + await asyncio.sleep(3600) + + +try: + asyncio.run(main()) +except KeyboardInterrupt: + print("[Main] Program stopped by user.") \ No newline at end of file diff --git a/shared/lora_device.py b/shared/lora_device.py index b3bca03..699339f 100644 --- a/shared/lora_device.py +++ b/shared/lora_device.py @@ -1,5 +1,6 @@ import sys import time +import random IS_MICROPYTHON = sys.implementation.name == 'micropython' @@ -8,49 +9,253 @@ if IS_MICROPYTHON: from machine import Pin, SPI import ubinascii import ujson as json - +else: + import threading + import serial + import json + + +# --- BASE RELIABLE LORA DEVICE --- +class BaseLoraDevice: + """Base class providing automatic ACK generation, retries, and duplicate filtering.""" + + def __init__(self): + self.processed_msg_ids = set() + self.received_acks = set() + self.pending_rx_queue = [] + self.default_group = 2 + + def _generate_msg_id(self): + return random.getrandbits(16) + + def _send_ack(self, ack_id): + """Sends an immediate acknowledgement packet back to the sender.""" + print(f"[ReliableLoRa] -> Triggering ACK send for msg_id: {ack_id}") + if IS_MICROPYTHON: + time.sleep_ms(10) + else: + time.sleep(0.01) + + ack_payload = {"_type": "_ack", "_ack_id": ack_id} + self.send(ack_payload) + + def _process_incoming_packet(self, packet): + """Internal packet processor: handles ACKs and deduplication.""" + if not packet or packet.get("raw"): + return packet + + data = packet.get("data") + if isinstance(data, dict): + # 1. Handle incoming ACK response + if data.get("_type") == "_ack": + ack_id = data.get("_ack_id") + print(f"[ReliableLoRa] <- SUCCESSFULLY MATCHED ACK ID: {ack_id}") + if ack_id is not None: + self.received_acks.add(ack_id) + if len(self.received_acks) > 100: + self.received_acks.clear() + return None # Drop internal protocol message from user queue + + # 2. Handle incoming command expecting an ACK + msg_id = data.get("_msg_id") + if msg_id is not None: + print(f"[ReliableLoRa] <- Received packet with msg_id {msg_id}. Queuing ACK.") + self._send_ack(msg_id) + + if msg_id in self.processed_msg_ids: + print(f"[ReliableLoRa] Discarding duplicate retry for msg_id {msg_id}") + return None # Discard duplicate retry + + self.processed_msg_ids.add(msg_id) + if len(self.processed_msg_ids) > 100: + self.processed_msg_ids.clear() + + return packet + + def send_reliable(self, payload, max_retries=4, ack_timeout=2.5): + """Sends a payload and retries until an ACK is received or max retries are reached.""" + 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 + + print(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===") + + for attempt in range(max_retries): + print(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}") + self.send(payload) + start_time = time.time() + + while (time.time() - start_time) < ack_timeout: + if lock: lock.acquire() + try: + if msg_id in self.received_acks: + self.received_acks.remove(msg_id) + print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===") + return True + finally: + if lock: lock.release() + + packet = self.receive_packet(timeout_ms=500) + if packet: + print(f"[ReliableLoRa] Received raw packet while waiting for ACK: {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() + + if lock: lock.acquire() + try: + if msg_id in self.received_acks: + self.received_acks.remove(msg_id) + print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} after poll ===") + return True + finally: + if lock: lock.release() + + print(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 + + def receive_reliable(self, timeout_ms=1000): + """Receives a packet, automatically sending ACKs and filtering duplicate retries.""" + if len(self.pending_rx_queue) > 0: + return self.pending_rx_queue.pop(0) + + start_time = time.time() + timeout_s = timeout_ms / 1000.0 + + while True: + elapsed = time.time() - start_time + remaining_ms = int((timeout_s - elapsed) * 1000) + if remaining_ms <= 0: + break + + poll_time = max(50, min(remaining_ms, 300)) + packet = self.receive_packet(timeout_ms=poll_time) + if packet: + filtered_packet = self._process_incoming_packet(packet) + if filtered_packet: + return filtered_packet + + return None + + +if IS_MICROPYTHON: # --- PILOTE SPI DIRECT (ESP32 / Heltec V3) --- - class LoraHardwareSPI: + class LoraHardwareSPI(BaseLoraDevice): def __init__(self, spi_bus=1, clk=9, mosi=10, miso=11, cs=8, irq=14, rst=12, gpio=13): - from sx1262 import SX1262 - self.lora = SX1262( - spi_bus=spi_bus, clk=clk, mosi=mosi, miso=miso, - cs=cs, irq=irq, rst=rst, gpio=gpio - ) - self.default_group = 2 # On définit le groupe par défaut ici - self.lock = _thread.allocate_lock() # Création du verrou + super().__init__() + self._pins = { + "spi_bus": spi_bus, "clk": clk, "mosi": mosi, "miso": miso, + "cs": cs, "irq": irq, "rst": rst, "gpio": gpio + } + self._cfg = {"freq": 868.1, "bw": 125.0, "sf": 7, "cr": 5, "power": 14} + self.lock = _thread.allocate_lock() + self.lora = None + self.reset_hardware() + + def reset_hardware(self): + """Resets SX1262 hardware and recreates driver instance.""" + with self.lock: + try: + irq_pin = Pin(self._pins["irq"], Pin.IN) + irq_pin.irq(handler=None) + except Exception: + pass + + try: + rst_pin = Pin(self._pins["rst"], Pin.OUT) + rst_pin.value(0) + time.sleep_ms(30) + rst_pin.value(1) + time.sleep_ms(50) + except Exception: + pass + + self.lora = None + time.sleep_ms(50) + + try: + from sx1262 import SX1262 + new_instance = SX1262(**self._pins) + new_instance.begin( + freq=self._cfg["freq"], bw=self._cfg["bw"], sf=self._cfg["sf"], + cr=self._cfg["cr"], power=self._cfg["power"], + useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False + ) + # SyncWord 0x12 = Decimal 18 + new_instance.setSyncWord(0x12) + self.lora = new_instance + except Exception as e: + print(f"[LoRa SPI] Initialization error: {e}") def configure(self, freq=868.1, bw=125.0, sf=7, cr=5, power=14): - self.lora.begin( - freq=freq, bw=bw, sf=sf, cr=cr, power=power, - useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False - ) - self.lora.setSyncWord(0x14) + self._cfg = {"freq": freq, "bw": bw, "sf": sf, "cr": cr, "power": power} + if self.lora is None: + self.reset_hardware() + else: + with self.lock: + try: + self.lora.begin( + freq=freq, bw=bw, sf=sf, cr=cr, power=power, + useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False + ) + self.lora.setSyncWord(0x12) + except Exception: + self.reset_hardware() def send(self, payload, group=None): - """Encode la payload en JSON si nécessaire, et injecte automatiquement l'octet de groupe.""" + """Encodes payload into JSON and prepends group byte.""" with self.lock: + if self.lora is None: + return + if group is None: group = self.default_group - # Si c'est un dictionnaire ou une liste, on le convertit en JSON textuel if isinstance(payload, (dict, list)): payload = json.dumps(payload) if isinstance(payload, str): payload = payload.encode('utf-8') - # Insertion automatique de l'octet de groupe au tout début de la trame physique paquet_physique = bytes([group]) + payload - self.lora.send(paquet_physique) + try: + self.lora.send(paquet_physique) + except Exception as e: + print(f"[LoRa SPI] Send error: {e}") - def receive_packet(self, timeout_ms=1000): - """Écoute, nettoie, extrait le groupe, gère le HEX et parse le JSON.""" + def receive_packet(self, timeout_ms=500): + """Listens on SPI bus with auto-detection for JSON vs. Grouped headers.""" with self.lock: - data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms) - if state == 0 and len(data) > 1: - group = data[0] - payload_brute = data[1:].strip(b'\x00 \r\n\t') + if self.lora is None: + return None + + try: + data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms) + except Exception as e: + print(f"[LoRa SPI] Recv error caught: {e}") + return None + + if state == 0 and data is not None and len(data) > 0: + if data[0] in (0x7B, 0x5B): # Starts with '{' or '[' + group = self.default_group + payload_brute = data.strip(b'\x00 \r\n\t') + elif len(data) > 1: + group = data[0] + payload_brute = data[1:].strip(b'\x00 \r\n\t') + else: + return None try: text = payload_brute.decode('utf-8').strip('\x00 \r\n\t') @@ -76,13 +281,10 @@ if IS_MICROPYTHON: return None else: - import threading - import serial - import json - # --- PILOTE SÉRIE (Raspberry Pi / Dragino LA66) --- - class LoraSerialAT: + class LoraSerialAT(BaseLoraDevice): def __init__(self, port): + super().__init__() self.port = port self.ser = serial.Serial( port=self.port, @@ -95,37 +297,60 @@ else: self.ser.reset_input_buffer() self.ser.reset_output_buffer() self.lock = threading.Lock() - def configure(self, **kwargs): - pass + + # Initial configuration + self.configure(freq=868.1, sf=7, bw=125) - def send(self, payload): - """Encode automatiquement la payload en HEX pour l'envoi via la clé.""" + def _send_at_cmd(self, cmd, wait_time=0.15): + """Helper to send AT command and purge response buffer.""" + self.ser.write(f"{cmd}\r\n".encode('utf-8')) + time.sleep(wait_time) + resp = "" + while self.ser.in_waiting > 0: + resp += self.ser.readline().decode('utf-8', errors='ignore') + return resp + + def configure(self, freq=868.1, sf=7, bw=125): + """Configures LA66 frequency, SF, BW, SyncWord, CRC, and continuous RX mode.""" with self.lock: + freq_hz = int(freq * 1000000) + bw_code = 0 if bw == 125 else 1 + + # Parameters: Freq, SF, BW, CR(0=4/5), Preamble(8), Header(1=Explicit), CRC(1=ON), IQ(0=Standard), NetMode(0=P2P), Power(14), SyncWord(18=0x12), Format(0), Type(1) + at_cfg_cmd = f"AT+CFG={freq_hz},{sf},{bw_code},0,8,1,1,0,0,14,18,0,1" + self._send_at_cmd(at_cfg_cmd, wait_time=0.2) + + # Fallback standalone commands + self._send_at_cmd("AT+SYNCWORD=18", wait_time=0.1) + self._send_at_cmd("AT+PRECV=65535", wait_time=0.1) + self.ser.reset_input_buffer() + + def send(self, payload, group=None): + """Encodes payload into HEX AT command and re-enables continuous RX.""" + with self.lock: + if group is None: + group = self.default_group + if isinstance(payload, (dict, list)): payload = json.dumps(payload) if isinstance(payload, str): payload = payload.encode('utf-8') - hex_payload = payload.hex() + paquet_physique = bytes([group]) + payload + hex_payload = paquet_physique.hex() self.ser.reset_input_buffer() - # 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") - self.ser.write(cmd.encode('utf-8')) + print(f"[RPi LoRa Serial] Transmitting HEX payload: {hex_payload}") + cmd = f"AT+PSEND={hex_payload}" + resp = self._send_at_cmd(cmd, wait_time=0.25) # Wait for RF TX to finish + print(f"[RPi LoRa Serial] AT+PSEND response: {resp.strip().replace(chr(10), ' | ')}") - time.sleep(0.2) - response = "" - start_wait = time.time() - while (time.time() - start_wait) < 1.5: - if self.ser.in_waiting > 0: - response += self.ser.readline().decode('utf-8', errors='ignore') - time.sleep(0.05) - - # print(f"[RPI LA66 TX STATUS] :\n{response.strip()}") + # Re-enable continuous receive mode after transmission completes + self._send_at_cmd("AT+PRECV=65535", wait_time=0.05) - def receive_packet(self, timeout_ms=5000): + def receive_packet(self, timeout_ms=500): + """Reads incoming serial lines from LA66 stick with robust format parsing.""" with self.lock: start_time = time.time() timeout_s = timeout_ms / 1000.0 @@ -136,18 +361,38 @@ else: if line: payload_bytes = None - if "(HEX:)" in line: + # 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 "+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) - except ValueError: - pass + 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) > 1: - group = payload_bytes[0] - payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t') + 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') @@ -180,4 +425,10 @@ def get_lora_device(port_or_pins=None): return LoraHardwareSPI(**pins) else: port = port_or_pins if port_or_pins else "/dev/serial/by-id/usb-Silicon_Labs_CP2102_USB_to_UART_Bridge_Controller_0001-if00-port0" - return LoraSerialAT(port) \ No newline at end of file + return LoraSerialAT(port) + + +class LoraCommands: + PING = "ping" + COOKING_STATE_UPDATE = "cooking_state_update" + TOGGLE_PAUSE = "toggle_pause" \ No newline at end of file diff --git a/shared/mqtt.py b/shared/mqtt.py index 53ce32c..bfca8f2 100644 --- a/shared/mqtt.py +++ b/shared/mqtt.py @@ -11,13 +11,10 @@ try: except ImportError: try: from umqtt.simple import MQTTClient as _MQTTClient + import _thread + import gc BACKEND_NAME = "umqtt.simple" IS_MICROPYTHON = True - # except ImportError: - # try: - # from umqtt.robust import MQTTClient as _MQTTClient - # BACKEND_NAME = "umqtt.robust" - # IS_MICROPYTHON = True except ImportError as exc: raise ImportError("No MQTT client found. Expected paho.mqtt or umqtt.") from exc @@ -79,6 +76,11 @@ class BrokerClient: self._client = None self._callback = None self._messages = [] + self._cadata = None # Cache cert bytes to prevent heap fragmentation + + # Thread safety lock for MicroPython socket reads/writes + if IS_MICROPYTHON: + self._lock = _thread.allocate_lock() def set_callback(self, callback): self._callback = callback @@ -104,16 +106,24 @@ class BrokerClient: return self._client if IS_MICROPYTHON: + gc.collect() # Clean Python heap before importing/allocating SSL import ssl ssl_params = self.ssl_params 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. + # OPTION A: If broker uses 'require_certificate false' and self-signed certs: + # Do NOT pass cadata when cert_reqs is CERT_NONE to save ~20KB of C-DRAM ssl_params = { - "cert_reqs": ssl.CERT_NONE, # Temporarily change to NONE to test if validation is the culprit - "cadata": _read_file_bytes(self.cafile) + "cert_reqs": ssl.CERT_NONE, + "server_hostname": self.host } + + # OPTION B: If strict CA validation IS required, load cadata ONLY with CERT_REQUIRED: + # ssl_params = { + # "cert_reqs": ssl.CERT_REQUIRED, + # "cadata": _read_file_bytes(self.cafile), + # "server_hostname": self.host + # } client = _MQTTClient( self.client_id or "smartWave-client", @@ -150,19 +160,35 @@ class BrokerClient: return self._client def connect(self): - client = self.open() if IS_MICROPYTHON: - client.connect() - return client + gc.collect() # Force C & Python memory cleanup right before TLS handshake - client.connect(self.host, self.port, self.keepalive) - return client + if self._client is not None: + self.close() + + client = self.open() + + try: + if IS_MICROPYTHON: + gc.collect() # Sweep memory right before umqtt calls ssl.wrap_socket() + with self._lock: + client.connect() + return client + + client.connect(self.host, self.port, self.keepalive) + return client + except Exception as e: + print("MQTT connection failed, closing client and releasing memory.") + print("Exception:", e) + self.close() + raise def publish(self, topic, payload, qos=2, retain=False): client = self.open() payload_bytes = _ensure_bytes(payload) if IS_MICROPYTHON: - return client.publish(topic, payload_bytes, retain=retain, qos=qos) + with self._lock: + return client.publish(topic, payload_bytes, retain=retain, qos=qos) if isinstance(topic, bytes): topic = topic.decode('utf-8') @@ -172,8 +198,9 @@ class BrokerClient: def subscribe(self, topic, qos=2): client = self.open() if IS_MICROPYTHON: - client.set_callback(self._on_micropython_message) - return client.subscribe(topic, qos=qos) + with self._lock: + client.set_callback(self._on_micropython_message) + return client.subscribe(topic, qos=qos) if isinstance(topic, bytes): topic = topic.decode('utf-8') @@ -184,27 +211,43 @@ class BrokerClient: client = self.open() if IS_MICROPYTHON: import struct - # Ensure the topic is bytes for writing to the socket + import time topic_bytes = topic if isinstance(topic, bytes) else topic.encode('utf-8') - # 1. Build the MQTT unsubscribe packet header + # 1. Increment and lock the PID for THIS specific request + client.pid = (client.pid % 65535) + 1 + sent_pid = client.pid # <-- Store local copy + + # 2. Construct UNSUBSCRIBE packet + rem_len = 2 + 2 + len(topic_bytes) pkt = bytearray(b"\xa2\0\0\0") - client.pid += 1 + struct.pack_into("!BH", pkt, 1, rem_len, sent_pid) - # Packet length is: 2 bytes (PID) + 2 bytes (topic length indicator) + topic string length - struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic_bytes), client.pid) - - # 2. Write the packet to the socket + # 3. Write packet to socket client.sock.write(pkt) client._send_str(topic_bytes) - # 3. Wait for the UNSUBACK confirmation frame (0xB0) from the broker - while True: + # 4. Wait for UNSUBACK (0xB0) + start = time.time() + while time.time() - start < 3: op = client.wait_msg() if op == 0xB0: - resp = client.sock.read(3) - assert resp[1] == pkt[2] and resp[2] == pkt[3] + resp = bytearray(3) + read_bytes = 0 + while read_bytes < 3: + chunk = client.sock.read(3 - read_bytes) + if chunk: + resp[read_bytes:read_bytes + len(chunk)] = chunk + read_bytes += len(chunk) + else: + time.sleep_ms(10) + + # Compare against sent_pid instead of client.pid + resp_pid = (resp[1] << 8) | resp[2] + if resp_pid != sent_pid: + print(f"[MQTT] UNSUBACK PID mismatch (expected {sent_pid}, got {resp_pid})") return client + return client if isinstance(topic, bytes): @@ -219,14 +262,16 @@ class BrokerClient: if self._client is None: return None if IS_MICROPYTHON: - return self._client.check_msg() + with self._lock: + return self._client.check_msg() return self._client.loop(timeout=timeout) def wait(self): if self._client is None: return None if IS_MICROPYTHON: - return self._client.wait_msg() + with self._lock: + return self._client.wait_msg() return self._client.loop_forever() def get_message(self): @@ -235,13 +280,29 @@ class BrokerClient: return self._messages.pop(0) def close(self): + """Safely clean up socket context without causing ESP32 C panics.""" if self._client is None: return - try: - self._client.disconnect() - except Exception: - pass - self._client = None + + if IS_MICROPYTHON: + with self._lock: + try: + if hasattr(self._client, "sock") and self._client.sock: + self._client.sock.close() + except Exception: + pass + finally: + if hasattr(self._client, "sock"): + self._client.sock = None + self._client = None + gc.collect() # Immediately reclaim freed socket & mbedTLS RAM + else: + try: + self._client.disconnect() + except Exception: + pass + finally: + self._client = None def __enter__(self): self.connect()