import _thread from machine import Pin, SoftI2C from shared.safeQueue import SafeQueue from shared import get_lora, get_uart, deviceTypes, config, cookingState from shared.uart_comm import UARTCommand, UARTCommandType from shared.sensors import RGBLED from shared.logging import log from shared.lora_device import LoraCommands import framebuf import ssd1306 import time # --- Configuration Matérielle --- vext = Pin(19, Pin.OUT) vext.value(0) time.sleep_ms(100) # --- Lecture de l'ID unique de l'ESP --- try: with open("device_id.txt", "r") as f: DEVICE_ID = f.read().strip() except Exception: DEVICE_ID = "ESP32_Inconnu" # --- Initialisation LoRa --- lora = get_lora() lora.configure(freq=868.1, sf=7) data_queue = SafeQueue() # --- Création des lEDs RGB --- magnetron_led = RGBLED(red_pin=48, green_pin=47, blue_pin=33) magnetron_led.color = RGBLED.WHITE_YELLOW magnetron_led.off() # --- Création de l'écran OLED --- scl_pin = Pin(18, Pin.OUT, pull=Pin.PULL_UP) sda_pin = Pin(17, Pin.OUT, pull=Pin.PULL_UP) display_i2c = SoftI2C(scl=scl_pin, sda=sda_pin, freq=100000) display = ssd1306.SSD1306_I2C(128, 64, display_i2c, addr=0x3C) display.text("Booting...", 1, 2, 1) display.show() print(f"ESP32 initialisé avec l'ID : '{DEVICE_ID}' (Type : {deviceTypes.DEVICE_TYPES['MICROWAVE']})") PING_PAYLOAD = { "id": DEVICE_ID, "type": deviceTypes.DEVICE_TYPES["MICROWAVE"] } def heartbeat_loop(): last_heartbeat_time = 0 while True: now = time.time() # 1. Send periodic heartbeat if now - last_heartbeat_time >= config.LORA_HEARTBEAT_INTERVAL: last_heartbeat_time = now print("\nESP32 : Envoi du Heartbeat...") lora.send(PING_PAYLOAD) # 2. Increase listen window to 300ms so radio stays active in RX mode paquet = lora.receive_reliable(timeout_ms=300) if paquet is not None: log(f"[LoRa Thread] New Packet Received: {paquet}") data_queue.put(paquet) time.sleep_ms(10) # UART uart_device = get_uart(uart_id=1, tx_pin=46, rx_pin=45) # Lancer la boucle de heartbeat dans un thread séparé try: _thread.stack_size(16 * 1024) except Exception: pass _thread.start_new_thread(heartbeat_loop, ()) # Cooking parameters cooking_state = None def cooking_state_temperature_provider(): return 22.0, 29.0 # TODO Remplacer par la lecture réelle de la température du plat et de l'air ambiant def cooking_state_on_state_change(state): print(f"[Main] Cooking state changed to: {state.state}") # Send to the Wifi board the current state uart_device.send_as_command(UARTCommand(UARTCommandType.COOKING_STATE_UPDATE, {"state": state.state})) # Send to the orchestrator the current state lora.send_reliable({"id": DEVICE_ID, "new_cooking_state": state.state}) display.text(cookingState.CookingStates.get_state_name(state.state), 1, 2, 1) display.show() if state.paused or state.state == cookingState.CookingStates.DONE or state.state == cookingState.CookingStates.IDLE: magnetron_led.off() else: magnetron_led.on() if state.state == cookingState.CookingStates.COOKING: pass if state.state == cookingState.CookingStates.STIRRING_REQUIRED: pass if state.state == cookingState.CookingStates.DONE: pass if state.state == cookingState.CookingStates.ALERT: pass def cooking_state_on_refresh(state): # TODO Show screen information pass def cooking_state_on_pause(state): # If the cooking is unpaused and was in STIRRING_REQUIRED or ALERT state, we set the state back to COOKING. if not state.paused and (state.state == cookingState.CookingStates.STIRRING_REQUIRED or state.state == cookingState.CookingStates.ALERT): state.set_state(cookingState.CookingStates.COOKING) # TODO send_reliable lora message to orchestrator about pause/resume state # --- MAIN APPLICATION THREAD --- print("[Main] Main execution path active.") while True: # 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 WiFi Board: {command.command_type}") if command.command_type == UARTCommandType.COOKING_PARAMS: # Handle cooking parameters command params = command.payload print(f"[Main] Cooking parameters received: {params}") cooking_state = cookingState.CookingState( cook_time=params["cook_time"], power_level=params["power_level"], target_temp=params["target_temp"] ) cooking_state.set_temperature_provider(cooking_state_temperature_provider) cooking_state.set_state_change_callback(cooking_state_on_state_change) cooking_state.set_refresh_callback(cooking_state_on_refresh) cooking_state.set_pause_callback(cooking_state_on_pause) time.sleep_ms(20) # Before sending back right away cooking_state_on_state_change(cooking_state) else: print(f"[Main] Unknown command type received: {command.command_type}") # 2. Listen for incoming LoRa packets from the orchestrator while not data_queue.empty(): paquet = data_queue.get() if paquet and not paquet["raw"]: data = paquet["data"] # Commands if "action" in data: if data["action"] == LoraCommands.TOGGLE_PAUSE: if cooking_state != None: if (cooking_state.state == cookingState.CookingStates.DONE): print("[Main] Cooking is done. We reset the microwave for the next cooking session.") cooking_state.set_state(cookingState.CookingStates.IDLE) time.sleep_ms(20) # Before sending back right away cooking_state = None else: cooking_state.toggle_pause() if cooking_state.paused: print("[Main] Cooking paused via orchestrator command.") else: print("[Main] Cooking resumed via orchestrator command.") else: log("[Main] No active cooking state to toggle pause/resume.") # uart_device.send(f"Hello from esp-32 lora ID {DEVICE_ID}") # Cooking State Update if cooking_state != None: cooking_state.update_tick() print(f"[Main] Cooking state : State : {cooking_state.state}, Temperature: {cooking_state.current_dish_temp}, Paused: {cooking_state.paused}, Remaining Time: {cooking_state.get_remaining_time():.2f}s, Estimated Remaining Time: {cooking_state.get_remaining_time_estimation():.2f}s") time.sleep_ms(500)