import _thread import select from machine import Pin, I2C from sensors import temperature_sensor from shared import get_mqtt_client, get_uart, config, payloads import time import ujson as json import sys # 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)) # --- READ DEVICE ID --- try: with open("device_id.txt", "r") as f: DEVICE_ID = f.read().strip() except Exception: DEVICE_ID = "ESP32_Inconnu" # --- 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, cafile=MQTT_CA_FILE, keepalive=config.MQTT_KEEPALIVE, ) global orchestrator_id orchestrator_id = None def on_mqtt_message(message): print("[MQTT Thread] Received message:", message) # 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 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) print("[MQTT Thread] Message processing complete.") mqtt_client.set_callback(on_mqtt_message) def mqtt_background_thread(): """Background MQTT worker handling ALL socket operations safely.""" print("[Thread] Background MQTT worker started.") 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) # Shortened timeout to keep the queue responsive events = poller.poll(200) if events: mqtt_client.wait() # 3. Handle Keepalive tracking manually if time.time() - last_check >= 15: # print("[Thread] Sending keepalive ping...") mqtt_client._client.ping() last_check = time.time() # Small breathe room for the CPU core time.sleep_ms(50) except Exception as e: print("[Thread] Connection dropped or error encountered:", e) 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 # Now we let the wrapper do its normal cleanup safely try: mqtt_client.close() except Exception: pass time.sleep(5) # --- UART BACKGROUND THREAD --- def uart_background_thread(): """Background UART worker handling all serial operations safely.""" print("[Thread] Background UART worker started.") uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16) while True: try: # 1. Check for incoming messages from the Heltec board while uart_device.any(): incoming_msg = uart_device.read() print(f"[Thread] Received from esp-lora over UART: {incoming_msg}") # 2. Example: Send data to the Heltec board every 5 seconds # uart_device.send("Status Check: WiFi Active") time.sleep(5) # Fast responsive polling loop for local UART except Exception as e: print("[Thread] UART error encountered:", e) time.sleep(5) # --- Launch background worker --- # _thread.start_new_thread(mqtt_background_thread, ()) # _thread.start_new_thread(uart_background_thread, ()) # --- MAIN APPLICATION THREAD (Core 0) --- print("[Main] Main execution path active.") 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) # Temperature sensor setup temperature_sensor_i2c = I2C(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.") temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c) 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 # Sensors print(f"[Main] Reading temperature from sensor...") obj_temp = temperature_sensor.read_object_temp() amb_temp = temperature_sensor.read_ambient_temp() if obj_temp is not None and amb_temp is not None: print(f"Object: {obj_temp:.2f}°C | Ambient: {amb_temp:.2f}°C") # 2. Example: Send data to the Heltec board every 5 seconds # uart_device.send("Status Check: WiFi Active") time.sleep(1)