import _thread import select from machine import Pin from sensors import temperature_gun 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)) # --- INITIALIZE CAMERA --- try: # Pass your confirmed working SCL and SDA pins here temperature_gun.init_camera(scl_pin=21, sda_pin=22, freq=100000) except Exception as e: print("[Main] Critical: Camera setup failed!") sys.print_exception(e) # --- 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) 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 the gun sensor...") temp = temperature_gun.read_temperature() print(f"[Main] Temperature read: {temp}°C") # 2. Example: Send data to the Heltec board every 5 seconds # uart_device.send("Status Check: WiFi Active") time.sleep(1)