diff --git a/micro_ondes/esp_wifi/main.py b/micro_ondes/esp_wifi/main.py index 21d046f..31f8396 100644 --- a/micro_ondes/esp_wifi/main.py +++ b/micro_ondes/esp_wifi/main.py @@ -5,7 +5,7 @@ import ujson as json import uasyncio as asyncio from machine import Pin, I2C -# 1. Clean memory immediately +# 1. Clean memory immediately before performing any operations gc.collect() # --- READ DEVICE ID --- @@ -19,8 +19,13 @@ except Exception: orchestrator_id = None cooking_state = None mqtt_connected = False +unsubscribed_hello = False -# --- MQTT SETUP (Initialized First!) --- +# --- ASYNC SIGNALS & QUEUES --- +# Event to signal when orchestrator requests sensor data (prevents MQTT lock deadlock) +sensor_request_event = None + +# --- MQTT SETUP --- from shared import get_mqtt_client, config, payloads MQTT_CA_FILE = "/certs/ca.crt" @@ -34,7 +39,6 @@ mqtt_client = get_mqtt_client( ) # --- HARDWARE & MODULE DEFERRED IMPORTS --- -# We declare variables here, but initialize them AFTER MQTT connects status_led = None uart_device = None mlx_temperature_sensor = None @@ -45,13 +49,12 @@ UARTCommandType = None def init_hardware(): - """Initializes hardware peripherals AFTER MQTT TLS has reserved its memory.""" + """Initializes hardware peripherals AFTER MQTT TLS has reserved its RAM.""" global status_led, uart_device, mlx_temperature_sensor global cookingState, log, UARTCommand, UARTCommandType - + print("[Main] Initializing hardware peripherals...") - - # 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 @@ -62,13 +65,9 @@ def init_hardware(): 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), @@ -83,51 +82,46 @@ def init_hardware(): mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c) -# --- CALLBACKS --- +def on_received_cooking_state_update(state, is_error=False, is_terminated=False): + """Callback executed when state changes are received from the LoRa board over UART.""" + if cooking_state: + if is_error: + cooking_state.set_state(cookingState.CookingStates.ERROR) + elif is_terminated: + cooking_state.set_state(cookingState.CookingStates.ABORTED) + else: + cooking_state.set_state(state) + + 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 - - 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) - - -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 - if log: - log(f"[Main] Updating cooking state to: {new_state}") - cooking_state.set_state(new_state) + """Callback executed whenever local cooking state transitions.""" + if status_led and cookingState: + if state == cookingState.CookingStates.IDLE: + status_led.set_color(0, 0, 0) # Off + elif state == cookingState.CookingStates.PREHEATING: + status_led.set_color(255, 165, 0) # Orange + elif state == cookingState.CookingStates.COOKING: + status_led.set_color(255, 0, 0) # Red + elif state == cookingState.CookingStates.DONE: + status_led.set_color(0, 255, 0) # Green + elif state in ( + cookingState.CookingStates.ERROR, + cookingState.CookingStates.ABORTED, + ): + status_led.set_color(255, 0, 255) # Magenta/Purple def on_mqtt_message(message): - global orchestrator_id, cooking_state + """Sync callback: Lightweight! Only updates variables or triggers async signals.""" + global orchestrator_id, cooking_state, unsubscribed_hello print("[MQTT] Received message on topic:", message.get("topic")) - + 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 @@ -138,12 +132,13 @@ def on_mqtt_message(message): ): 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) + if not unsubscribed_hello: + unsubscribed_hello = True + try: + mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO) + print("[MQTT] Successfully unsubscribed from topic:", config.MQTT_TOPIC_HELLO) + except Exception as e: + print("[MQTT] Unsubscribe error:", e) # 2. Cooking Parameters / Sensor Request elif ( @@ -152,14 +147,9 @@ def on_mqtt_message(message): 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 - ) + print("[MQTT] Sensor data requested! Triggering async publisher...") + # Trigger async event instead of calling publish() directly inside lock context! + sensor_request_event.set() else: print("[MQTT] Cooking parameters received:", payload_data) if cookingState: @@ -170,26 +160,89 @@ def on_mqtt_message(message): ) 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.") + print("[MQTT] Cooking parameters sent to LoRa board over UART.") -mqtt_client.set_callback(on_mqtt_message) +# --- DEDICATED ASYNC TASK FOR SENSOR PUBLISHING --- +async def sensor_publisher_task(): + """Waits for sensor_request_event, reads hardware, and publishes outside the MQTT lock.""" + while True: + await sensor_request_event.wait() + sensor_request_event.clear() + + print("[Sensor Task] Reading temperature sensors...") + 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) + + try: + print("[Sensor Task] Publishing sensor data to MQTT...") + mqtt_client.publish( + config.MQTT_TOPIC_SENSOR, sensor_payload, qos=config.MQTT_QOS + ) + print("[Sensor Task] Sensor data successfully published:", sensor_payload) + except Exception as e: + print("[Sensor Task] Failed to publish sensor data:", e) + + +async def uart_task(): + """Polls incoming UART messages from the LoRa board using dynamic method fallback.""" + while True: + if uart_device: + try: + cmd = uart_device.read_as_command() + + if cmd: + print("[UART] Command received from LoRa board:", cmd) + if ( + hasattr(cmd, "command_type") + and cmd.command_type == UARTCommandType.STATE_UPDATE + and on_received_cooking_state_update + ): + on_received_cooking_state_update( + cmd.payload.get("state"), + cmd.payload.get("is_error", False), + cmd.payload.get("is_terminated", False), + ) + except Exception as e: + print("[UART Task] Error reading command:", e) + + await asyncio.sleep_ms(50) async def connect_mqtt_async(): - """Connects to MQTT safely while memory is clean.""" - global mqtt_connected + global mqtt_connected, mqtt_client mqtt_connected = False - + while True: try: print("[MQTT] Connecting to broker with TLS...") + # Re-instantiate client to clear old socket buffers gc.collect() + mqtt_client = get_mqtt_client( + host="192.168.50.1", # TODO : Use config.MQTT_BROKER_HOST instead of hardcoding + port=8884, + client_id="smartwave-esp32-demo", + use_tls=True, + cafile=MQTT_CA_FILE, + keepalive=30, + ) + mqtt_client.set_callback(on_mqtt_message) + mqtt_client.connect() print("[MQTT] Connected! Subscribing to topics...") mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, qos=config.MQTT_QOS) @@ -204,12 +257,14 @@ async def connect_mqtt_async(): mqtt_client.close() except Exception: pass + + # Force heap cleanup before sleeping + del mqtt_client + gc.collect() + print(f"[MQTT] Free RAM after cleanup: {gc.mem_free()} bytes") 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() @@ -234,9 +289,18 @@ async def mqtt_poll_task(): async def orchestrator_hello_task(): global mqtt_connected while True: - if mqtt_connected and orchestrator_id is None: + if orchestrator_id is not None: + # Hello successfully acknowledged! Stop looping this task. + print("[Hello Task] Orchestrator acknowledged. Stopping hello task.") + break + + if mqtt_connected: print("[Hello Task] Sending initial hello to orchestrator...") try: + if mqtt_client is None: + print("[Hello Task] MQTT client is None. Attempting to reconnect...") + await connect_mqtt_async() + mqtt_client.publish( config.MQTT_TOPIC_HELLO, payloads.mqtt_hello(DEVICE_ID), @@ -244,32 +308,11 @@ async def orchestrator_hello_task(): ) except OSError as e: print("[Hello Task] Hello publish failed:", e) - mqtt_connected = False - await connect_mqtt_async() + # mqtt_connected = False 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() @@ -278,17 +321,19 @@ async def memory_cleanup_task(): # --- MAIN ENTRY POINT --- async def main(): + global sensor_request_event print("[Main] Starting application...") + + # Initialize loop-bound events + sensor_request_event = asyncio.Event() - # 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 + # Launch background tasks asyncio.create_task(mqtt_poll_task()) asyncio.create_task(orchestrator_hello_task()) + asyncio.create_task(sensor_publisher_task()) asyncio.create_task(uart_task()) asyncio.create_task(memory_cleanup_task()) diff --git a/orchestrateur/main.py b/orchestrateur/main.py index 9d8ffd4..1b26fc6 100644 --- a/orchestrateur/main.py +++ b/orchestrateur/main.py @@ -1,394 +1,329 @@ import base64 import json -import threading -import queue import time import traceback - +import asyncio import requests + from orchestrateur.sensors import gps from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads from shared.logging import log +from shared.cookingState import CookingStates +from shared.lora_device import LoraCommands from sensors import ultrasonicRanger, temp_hum, button, camera # --- Read Unique Device ID --- -try: - with open("device_id.txt", "r") as f: - DEVICE_ID = f.read().strip() -except Exception: - 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" +def get_device_id(): + for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]: + try: + with open(path, "r") as f: + return f.read().strip() + except Exception: + pass + return "RPI_Orchestrateur_Default" -# Thread-safe queue for application messages -data_queue = queue.Queue() -cooking_queue = {} -active_cooks = {} -active_cooks_lock = threading.Lock() +DEVICE_ID = get_device_id() +# --- STATE MACHINE DEFINITIONS --- +class MicrowaveState: + IDLE = "IDLE" # Microwave is empty + ANALYZING = "ANALYZING" # Reading sensors & waiting for IR + WAITING_FOR_CLOUD = "WAITING_FOR_CLOUD" # Waiting for API parameters + COOKING = "COOKING" # Microwave is active + DONE = "DONE" # Finished/Stopped, waiting for dish removal + +# Global state trackers +microwave_states = {"2": MicrowaveState.IDLE} +cooking_data_cache = {} # Replaces cooking_queue +button_state = False +async_event_queue = None + +# --- HARDWARE SETUP --- lora = get_lora() lora.configure() -def lora_listener(): - """Background Thread: Listens to LoRa traffic and responds to Heartbeats.""" - print("Thread Écouteur LoRa démarré.") - while True: - paquet = lora.receive_packet(timeout_ms=1000) - if paquet: - donnees = paquet["data"] - expediteur_type = donnees.get("type") - - if expediteur_type == deviceTypes.DEVICE_TYPES["MICROWAVE"]: - print(f"\n[Thread LoRa] Heartbeat reçu de {donnees.get('id')}") - reponse = { - "id": DEVICE_ID, - "type": deviceTypes.DEVICE_TYPES["ORCHESTRATOR"] - } - lora.send(reponse) - else: - data_queue.put({"source": "LoRa", "data": paquet}) - -# --- 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, + host="192.168.50.1", + 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() mqtt_client.subscribe(config.MQTT_TOPIC_SENSOR, qos=config.MQTT_QOS) -print(f"Subscribed to topic: {config.MQTT_TOPIC_SENSOR}") mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS) -print(f"Subscribed to topic: {config.MQTT_TOPIC_HELLO}") -# --- 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.") + print("[MQTT] Paho background loop started.") +# --- BACKGROUND TASKS (PRODUCERS) --- +async def lora_listener_task(): + """Polls LoRa and pushes to the async queue.""" + print("[LoRa] Async listener started.") + while True: + # Run blocking lora receive in a thread to not block asyncio loop + paquet = await asyncio.to_thread(lora.receive_reliable, timeout_ms=100) + if paquet: + await async_event_queue.put({"source": "LoRa", "data": paquet}) + await asyncio.sleep(0.05) -def mqtt_listener(): - """Background Thread: Constantly inspects incoming MQTT message cache.""" - print("Thread MQTT démarré.") +async def mqtt_listener_task(): + """Polls MQTT cache and pushes to the async queue.""" + print("[MQTT] Async listener started.") while True: message = mqtt_client.get_message() - if message: - # Try to parse the payload as a python dictionary, but if it fails, just print the raw payload try: payload = json.loads(message['payload']) - except Exception as e: - print(f"Error parsing MQTT payload: {e}") - payload = message['payload'] # Fallback to raw payload if parsing fails + except Exception: + payload = message['payload'] - print(f"\n[Thread MQTT] Message reçu : {message}") - data_queue.put({"source": "MQTT", "topic": message['topic'] ,"data": payload}) - - # Sleep for 100ms. Prevents the thread from turning into an infinite 100% CPU hog. - time.sleep(0.2) + # --- SAFE TOPIC DECODING --- + topic = message['topic'] + if isinstance(topic, bytes): + topic = topic.decode('utf-8') + + await async_event_queue.put({ + "source": "MQTT", + "topic": topic, + "data": payload + }) + await asyncio.sleep(0.1) -# Button -button_state = False def button_callback(): + """Button physical interrupt callback.""" global button_state - button_state = not button_state - print(f"\n[Thread Button] Button state changed to: {button_state}") - -button.set_callback(button_callback) + if microwave_states.get("2") == MicrowaveState.COOKING: + print("[Button] Toggling pause/resume for microwave '2'.") + lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_PAUSE}) + else: + button_state = not button_state + print(f"[Button] Defrost state toggled to: {button_state}") -# Launch background monitoring workers -threading.Thread(target=lora_listener, daemon=True).start() -threading.Thread(target=mqtt_listener, daemon=True).start() -# Launch button monitoring thread +button.set_callback(button_callback) button.start_button_monitoring_thread() -print("Orchestrateur prêt. Le main loop est libre.") - -def _tryReadSensorsWithRetries(func, exception=True, max_retries=3, delay=1): - """ - Tries to read the sensor max_retries times until the return value of func is not None. - It will then return the value of func. If it fails max_retries times, it will fail if exception is True, otherwise it will return None. - """ - - for attempt in range(max_retries): - result = func() - if result is not None: - return result - else: - log(f"Attempt {attempt + 1} failed. Retrying in {delay} seconds...") - time.sleep(delay) - - if exception: - raise Exception(f"Failed to read sensor after {max_retries} attempts.") - else: - return None - -def read_sensors_for_cooking(microwave_id): - """Read all sensors and return a dictionary of their values, including the microwave ID.""" - log("\nLecture des capteurs...") - sensor_data = {} - sensor_data["microwave_id"] = microwave_id - - # === Notify the microwave of needed sensor readings === - mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_init(microwave_id), qos=config.MQTT_QOS) - - # Read Ultrasonic Ranger - sensor_data["ultrasonic_distance"] = _tryReadSensorsWithRetries(ultrasonicRanger.get_dish_height) - log(f"\nLecture du capteur Ultrason : {sensor_data['ultrasonic_distance']}") - - # Read Temperature and Humidity - temperature, humidity = temp_hum.get_temperature_and_humidity_with_retry() - if temperature is not None and humidity is not None: - log(f"\nLecture du capteur Temp/Hum : {temperature}, {humidity}") - sensor_data["temperature"] = temperature - sensor_data["humidity"] = humidity - - # Camera - def _getPicture(): - picture_bytes = None - try: - picture_bytes = camera.get_picture() - return picture_bytes - except Exception as e: - log(f"Error reading camera data: {e}") - return None - sensor_data["camera_image"] = _tryReadSensorsWithRetries(_getPicture, exception=True) - log(f"\nPhoto de la Caméra : {len(sensor_data['camera_image'])} bytes") - - # Read Button State (last because he can still change state while reading other sensors) - sensor_data["defrost_mode"] = button_state - - cooking_queue[microwave_id] = sensor_data - - +# --- HARDWARE CONTROLLERS --- def _stop_hardware(microwave_id: str): - """ - Hardware driver stop — halts magnetron/turntable immediately. - """ - print(f"[{microwave_id}] 🛑 Emergency stop issued to hardware.") - # TODO: Add physical hardware stop command here - # e.g., gpio_controller.stop() + print(f"[{microwave_id}] /!\ Emergency stop issued to hardware.") + # TODO: Add LoRa STOP command here - -def _send_params_to_microwave(microwave_id: str, cook_time: int, power_level: int, target_temp: float, cancel_event: threading.Event): - """ - Triggers physical microwave execution. - """ - if cancel_event.is_set(): - return - - print(f"[{microwave_id}] Sending microwave {microwave_id} cooking parameters : {cook_time}s @ {power_level}W power, target temp {target_temp}°C.") - mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_config(microwave_id, cook_time, power_level, target_temp), qos=config.MQTT_QOS) - -def _cooking_worker(microwave_id: str, sensors_data: dict, cancel_event: threading.Event): - """Worker function executing cloud API calls and hardware triggers.""" - URL = "https://smartwave.matthiasg.dev/cooking-params" +# --- ASYNC COOKING LOGIC --- +def read_local_sensors(microwave_id, initial_dish_height): + """Blocking function to read local I2C/SPI sensors. Runs in a thread.""" + print(f"[{microwave_id}] Reading local physical sensors...") + sensor_data = { + "microwave_id": microwave_id, + "defrost_mode": button_state, + "ultrasonic_distance": initial_dish_height # Reuse height from trigger + } + # Temp / Hum (handles DHT error safely) try: - # Check cancellation before network call - if cancel_event.is_set(): - print(f"[{microwave_id}] Job canceled before starting API call.") - return - - log(f"[{microwave_id}] Sending sensor data to cloud API...") - - # Ensure camera_image is encoded to Base64 string if it's currently raw bytes - if isinstance(sensors_data.get("camera_image"), bytes): - sensors_data["camera_image"] = base64.b64encode(sensors_data["camera_image"]).decode("utf-8") - - # 1. HTTP Request (15-second timeout) - response = requests.post(URL, json=sensors_data, timeout=360) # timeout for long-running requests - - # Check cancellation right after network call returns - if cancel_event.is_set(): - print(f"[{microwave_id}] Job was canceled while waiting for cloud response. Discarding result.") - return - - log(f"[{microwave_id}] Cloud API responded with : {response.json()}") - if response.status_code != 200 and response.status_code != 201: - print(f"[{microwave_id}] Cloud API returned error {response.status_code}: {response.json()}") - response.raise_for_status() - - # 2. Extract Response Parameters - response_json = response.json() - cook_plan = response_json.get("cook_plan", {}) - - cook_time = cook_plan.get("cook_time_seconds") - power_level = cook_plan.get("effective_power_watts") - target_temp = cook_plan.get("target_temp") - dish_name = response_json.get("dish_name", "Unknown Dish") - - if cook_time is None or power_level is None or target_temp is None: - print(f"[{microwave_id}] Cloud returned incomplete plan: {response_json}") - return - - # Check cancellation before starting physical microwave - if cancel_event.is_set(): - print(f"[{microwave_id}] Job was canceled before starting hardware execution.") - return - - print(f"[{microwave_id}] Received plan for '{dish_name}': {cook_time}s @ {power_level}W power, target temp {target_temp}°C.") - - # 3. Start Hardware Execution - _send_params_to_microwave(microwave_id, cook_time, power_level, target_temp, cancel_event) - - except requests.exceptions.Timeout: - print(f"[{microwave_id}] Request timed out waiting for cloud response.") - except requests.exceptions.RequestException as e: - print(f"[{microwave_id}] HTTP error reaching cloud API: {e}") + temp, hum = temp_hum.get_temperature_and_humidity_with_retry() + if temp is not None: + sensor_data["temperature"] = temp + sensor_data["humidity"] = hum except Exception as e: - print(f"[{microwave_id}] Unexpected error in worker thread: {e}") - traceback.print_exc() - finally: - # Clean up registry entry if this worker was the active one - with active_cooks_lock: - if active_cooks.get(microwave_id) == cancel_event: - del active_cooks[microwave_id] - - -def start_cooking_for_microwave(microwave_id: str, sensors_data: dict): - """ - Sends sensors data to the cloud and starts cooking in a separate thread. - If a worker is already running for the given microwave_id, it cancels - the previous process and stops the hardware before starting the new one. - """ - with active_cooks_lock: - # 1. If an active job exists for this microwave, cancel it - if microwave_id in active_cooks: - print(f"[{microwave_id}] Existing cooking job detected! Canceling old worker...") - active_cooks[microwave_id].set() # Signal existing thread to abort - _stop_hardware(microwave_id) # Stop hardware immediately - - # 2. Register a new cancellation event for this microwave - cancel_event = threading.Event() - active_cooks[microwave_id] = cancel_event - - # 3. Start the new background worker thread - thread = threading.Thread( - target=_cooking_worker, - args=(microwave_id, sensors_data, cancel_event), - daemon=True - ) - thread.start() - -# Sensor reading -def read_sensors(): - """Read all sensors and return a dictionary of their values.""" - log("\nLecture des capteurs...") - sensor_data = {} - - # Read Ultrasonic Ranger - distance = ultrasonicRanger.get_dish_height() - if distance is not None: - log(f"\nLecture du capteur Ultrason : {distance}") - sensor_data["ultrasonic_distance"] = distance - - # Read Temperature and Humidity - temperature, humidity = temp_hum.get_temperature_and_humidity() - if temperature is not None and humidity is not None: - log(f"\nLecture du capteur Temp/Hum : {temperature}, {humidity}") - sensor_data["temperature"] = temperature - sensor_data["humidity"] = humidity - - - # Read GPS Data - gps_data = gps.get_gps_data() - if gps_data: - log(f"\nLecture du capteur GPS : {gps_data}") - sensor_data["gps"] = gps_data + log(f"[{microwave_id}] DHT read warning: {e}") # Camera - picture_bytes = None try: - picture_bytes = camera.get_picture() - log(f"\nLecture du capteur Caméra : {len(picture_bytes)} bytes") - sensor_data["camera_image"] = picture_bytes + sensor_data["camera_image"] = camera.get_picture() except Exception as e: - log(f"Error reading camera data: {e}") - - # Read Button State (last because he can still change state while reading other sensors) - sensor_data["defrost_state"] = button_state - + log(f"[{microwave_id}] Camera read failed: {e}") + return sensor_data -# --- MAIN EXECUTION LOOP --- -while True: - try: - # === TREAT MESSAGE QUEUE === - try: - msg = data_queue.get(block=False) - - # print(msg) - - if msg["source"] == "LoRa": - print(f"\n[Main Loop] LoRa : Données traitées : {msg['data']}") - elif msg["source"] == "MQTT": - # MQTT HELLO - if (msg["topic"] == config.MQTT_TOPIC_HELLO.decode('utf-8')): - if ("id_orchestrator" in msg["data"] and msg["data"]["id_orchestrator"] == DEVICE_ID): - # Do not answer to messages coming from me - continue - microwave_id = msg["data"]["id_microwave"] - print(f"\n[Main Loop] MQTT : Hello reçu de {microwave_id}.") - # Responds - mqtt_client.publish(config.MQTT_TOPIC_HELLO, payloads.mqtt_hello_ack(DEVICE_ID, microwave_id), qos=config.MQTT_QOS) - print(f"[Main Loop] MQTT : Réponse Hello envoyée à {microwave_id}.") - # TODO : Save in database - # MQTT SENSOR DATA - elif (msg["topic"] == config.MQTT_TOPIC_SENSOR.decode('utf-8')): - print(f"\n[Main Loop] MQTT : Données capteurs reçues du micro-ondes : {msg['data']}") - microwave_id = msg["data"].get("id_microwave") - if not microwave_id: - print("[Main Loop] MQTT : Données capteurs reçues sans ID micro-ondes. Ignoré.") - continue - # Get the already existing cooking data for this microwave - sensors_data = cooking_queue.get(microwave_id) - if sensors_data is None: - print(f"[Main Loop] MQTT : Données capteurs reçues pour {microwave_id} mais aucune donnée de cuisson en cours. Ignoré.") - continue - # Merge the received sensor data into the existing cooking data - sensors_data["ir_initial_temp"] = msg["data"].get("dish_temp") - sensors_data["ir_ambient_temp"] = msg["data"].get("ambient_temp") - start_cooking_for_microwave(microwave_id, sensors_data) - # Remove the cooking data from the queue since it's now being processed - del cooking_queue[microwave_id] - - print(f"\n[Main Loop] MQTT : Données traitées : {msg['data']}") - except queue.Empty: - pass - - # === CHECK FOR DISH INSERTED === - # Read the dish height from the ultrasonic sensor. If it's below a certain threshold, we assume a dish has been inserted. - dish_height = ultrasonicRanger.get_dish_height() - if dish_height is not None and dish_height > 2.0: # Threshold in cm for detecting a dish - print(f"\n[Main Loop] Dish detected at height: {dish_height} cm. Initiating sensor read...") - # Read all sensors and store the data in the cooking queue for this microwave - read_sensors_for_cooking("2") - print(f"[Main Loop] Sensor data collected and queued for cooking.") - - # DEBUG : Read sensors - # sensor_values = read_sensors() - # if sensor_values: - # sensor_values_print = sensor_values.copy() - # if "camera_image" in sensor_values_print: - # sensor_values_print["camera_image"] = f"<{len(sensor_values_print['camera_image'])} bytes>" - # print(f"\nCapteurs Données lues : {sensor_values_print}") - - time.sleep(3) - - - except KeyboardInterrupt: - break - except Exception as e: - traceback.print_exc() - time.sleep(1) # Prevents rapid error logging in case of persistent issues -# Clean termination -if hasattr(mqtt_client._client, "loop_stop"): - mqtt_client._client.loop_stop() -mqtt_client.close() +async def handle_new_dish(microwave_id, detected_height): + """Triggered when a new dish is placed inside.""" + microwave_states[microwave_id] = MicrowaveState.ANALYZING + print(f"\n[{microwave_id}] 🍽️ Dish detected at {detected_height:.1f} cm! Requesting IR from microwave...") + + # 1. Ask microwave for IR temp via MQTT + mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_init(microwave_id), qos=config.MQTT_QOS) + + # 2. Read local sensors (passing detected_height to prevent GPIO collision) + sensors = await asyncio.to_thread(read_local_sensors, microwave_id, detected_height) + + # Check if state changed while taking photos + if microwave_states[microwave_id] != MicrowaveState.ANALYZING: + print(f"[{microwave_id}] Dish removed during sensor read. Aborting.") + return + + cooking_data_cache[microwave_id] = sensors + print(f"[{microwave_id}] Local sensors cached. Waiting for MQTT IR data...") + +async def request_cloud_cooking_plan(microwave_id, sensors_data): + """Sends all data to the cloud and starts the microwave if successful.""" + microwave_states[microwave_id] = MicrowaveState.WAITING_FOR_CLOUD + URL = "https://smartwave.matthiasg.dev/cooking-params" + + # Format image + if isinstance(sensors_data.get("camera_image"), bytes): + sensors_data["camera_image"] = base64.b64encode(sensors_data["camera_image"]).decode("utf-8") + + print(f"[{microwave_id}] Requesting cooking plan from cloud app...") + try: + response = await asyncio.to_thread(requests.post, URL, json=sensors_data, timeout=30) + + # Abort if state changed (e.g. user removed dish while waiting for wifi) + if microwave_states[microwave_id] != MicrowaveState.WAITING_FOR_CLOUD: + print(f"[{microwave_id}] Dish removed during API request. Discarding API plan.") + return + + response.raise_for_status() + plan = response.json().get("cook_plan", {}) + c_time = plan.get("cook_time_seconds") + c_power = plan.get("effective_power_watts") + c_temp = plan.get("target_temp") + + if c_time is None or c_power is None or c_temp is None: + print(f"[{microwave_id}] ❌ Invalid plan received: {response.json()}") + microwave_states[microwave_id] = MicrowaveState.DONE # Fail safe + return + + print(f"[{microwave_id}] Cloud Plan Received! Starting microwave: {c_time}s @ {c_power}W") + microwave_states[microwave_id] = MicrowaveState.COOKING + mqtt_client.publish( + config.MQTT_TOPIC_COOKING, + payloads.mqtt_cooking_config(microwave_id, c_time, c_power, c_temp), + qos=config.MQTT_QOS + ) + + except Exception as e: + print(f"[{microwave_id}] Cloud API Error: {e}") + microwave_states[microwave_id] = MicrowaveState.DONE + +# --- MAIN LOGIC TASKS --- +async def process_messages_task(): + """Consumes the unified event queue.""" + while True: + msg = await async_event_queue.get() + source = msg["source"] + data = msg["data"] + + if source == "LoRa": + if "new_cooking_state" in data.get("data", {}): + mw_id = data["data"].get("id") + n_state = data["data"].get("new_cooking_state") + print(f"[LoRa] Microwave {mw_id} state changed to: {n_state}") + + if n_state == CookingStates.IDLE and microwave_states.get(mw_id) == MicrowaveState.COOKING: + microwave_states[mw_id] = MicrowaveState.DONE + print(f"[{mw_id}] Cooking finished. Waiting for user to remove dish.") + + elif source == "MQTT": + topic = msg["topic"] + + # Helper to normalize config topics to str + def to_str(val): + return val.decode('utf-8') if isinstance(val, bytes) else val + + hello_topic = to_str(config.MQTT_TOPIC_HELLO) + sensor_topic = to_str(config.MQTT_TOPIC_SENSOR) + + if topic == hello_topic: + if data.get("id_orchestrator") != DEVICE_ID: + mw_id = data.get("id_microwave") + print(f"[MQTT] Hello from {mw_id}. Sending ACK.") + mqtt_client.publish( + config.MQTT_TOPIC_HELLO, + payloads.mqtt_hello_ack(DEVICE_ID, mw_id), + qos=config.MQTT_QOS + ) + + elif topic == sensor_topic: + mw_id = data.get("id_microwave") + + if mw_id and microwave_states.get(mw_id) == MicrowaveState.ANALYZING: + sensors = cooking_data_cache.get(mw_id) + if sensors: + sensors["ir_initial_temp"] = data.get("dish_temp") + sensors["ir_ambient_temp"] = data.get("ambient_temp") + asyncio.create_task(request_cloud_cooking_plan(mw_id, sensors)) + +async def get_filtered_dish_height(samples=3, delay=0.04): + """Reads ultrasonic sensor multiple times and returns the median, discarding invalid zeros.""" + valid_samples = [] + for _ in range(samples): + h = await asyncio.to_thread(ultrasonicRanger.get_dish_height) + # Discard 0.0 or near-zero timeout glitches + if h is not None and h > 0.5: + valid_samples.append(h) + await asyncio.sleep(delay) + + if valid_samples: + valid_samples.sort() + return valid_samples[len(valid_samples) // 2] # Median sample + return None # All reads failed or out of range + + +async def monitor_dish_height_task(): + """Monitors presence of dish with hysteresis and debouncing.""" + mw_id = "2" + consecutive_present = 0 + consecutive_absent = 0 + REQUIRED_STABLE_READS = 3 # Must see 3 stable states in a row (~1 second) + + while True: + dist = await get_filtered_dish_height() + current_state = microwave_states.get(mw_id, MicrowaveState.IDLE) + + if dist is not None: + # Hysteresis Thresholds: + # - Must be > 2.5 cm to detect dish insertion + # - Must be < 1.2 cm to detect dish removal + if dist > 2.5: + consecutive_present += 1 + consecutive_absent = 0 + elif dist < 1.2: + consecutive_absent += 1 + consecutive_present = 0 + else: + # Dead-zone (1.2cm to 2.5cm) -> Noise buffer + consecutive_present = 0 + consecutive_absent = 0 + + # --- DISH INSERTED CONFIRMED --- + if consecutive_present >= REQUIRED_STABLE_READS and current_state == MicrowaveState.IDLE: + consecutive_present = 0 + asyncio.create_task(handle_new_dish(mw_id, dist)) + + # --- DISH REMOVED CONFIRMED --- + elif consecutive_absent >= REQUIRED_STABLE_READS and current_state != MicrowaveState.IDLE: + consecutive_absent = 0 + print(f"\n[{mw_id}] Dish Removed! Resetting state to IDLE.") + microwave_states[mw_id] = MicrowaveState.IDLE + if current_state == MicrowaveState.COOKING: + _stop_hardware(mw_id) + if mw_id in cooking_data_cache: + del cooking_data_cache[mw_id] + + await asyncio.sleep(0.3) + +# --- BOOTSTRAP --- +async def main(): + global async_event_queue + print("🚀 Orchestrateur Asyncio prêt. Lancement des tâches...") + + async_event_queue = asyncio.Queue() + + await asyncio.gather( + lora_listener_task(), + mqtt_listener_task(), + process_messages_task(), + monitor_dish_height_task() + ) + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\nArrêt manuel.") + finally: + if hasattr(mqtt_client._client, "loop_stop"): + mqtt_client._client.loop_stop() + mqtt_client.close() \ No newline at end of file