import json import threading import queue import time import traceback from orchestrateur.sensors import gps from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads from shared.logging import log 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" # Thread-safe queue for application messages data_queue = queue.Queue() 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, 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) mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS) print(f"Subscribed to topic: {config.MQTT_TOPIC_SENSOR}") # --- 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.") def mqtt_listener(): """Background Thread: Constantly inspects incoming MQTT message cache.""" print("Thread MQTT démarré.") 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 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) # Button button_state = False def button_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) # 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.start_button_monitoring_thread() print("Orchestrateur prêt. Le main loop est libre.") # 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 # 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 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["button_state"] = button_state return sensor_data # --- MAIN EXECUTION LOOP --- while True: try: # Check for non-heartbeat data 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": 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 print(f"\n[Main Loop] MQTT : Données traitées : {msg['data']}") except queue.Empty: pass # 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()