116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
import threading
|
|
import queue
|
|
import time
|
|
from shared import get_lora, get_mqtt_client, deviceTypes, config
|
|
|
|
# --- 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)
|
|
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:
|
|
print(f"\n[Thread MQTT] Message reçu : {message}")
|
|
data_queue.put({"source": "MQTT", "data": message})
|
|
|
|
# --- THE CPU FIX ---
|
|
# Sleep for 100ms. Prevents the thread from turning into an infinite 100% CPU hog.
|
|
time.sleep(0.1)
|
|
|
|
|
|
# Launch background monitoring workers
|
|
threading.Thread(target=lora_listener, daemon=True).start()
|
|
threading.Thread(target=mqtt_listener, daemon=True).start()
|
|
|
|
print("Orchestrateur prêt. Le main loop est libre.")
|
|
|
|
# --- MAIN EXECUTION LOOP ---
|
|
while True:
|
|
try:
|
|
# Check for non-heartbeat data safely
|
|
try:
|
|
msg = data_queue.get(block=False)
|
|
print(f"\n[Main Loop] Données traitées : {msg['data']}")
|
|
except queue.Empty:
|
|
pass
|
|
|
|
time.sleep(1)
|
|
|
|
# Publish debug telemetry message
|
|
print("[Main Loop] Envoi d'un message de debug sur MQTT...")
|
|
response = mqtt_client.publish(
|
|
config.MQTT_TOPIC_COOKING,
|
|
f"Orchestrateur actif, ID: {DEVICE_ID}",
|
|
qos=config.MQTT_QOS
|
|
)
|
|
|
|
# This will now unblock instantly because loop_start() handles the delivery confirmation!
|
|
response.wait_for_publish()
|
|
print("[Main Loop] Message de debug publié avec succès.")
|
|
|
|
time.sleep(9)
|
|
|
|
|
|
except KeyboardInterrupt:
|
|
break
|
|
|
|
# Clean termination
|
|
if hasattr(mqtt_client._client, "loop_stop"):
|
|
mqtt_client._client.loop_stop()
|
|
mqtt_client.close() |