Asyncio refactor for wifi and rpi
This commit is contained in:
+131
-86
@@ -5,7 +5,7 @@ import ujson as json
|
||||
import uasyncio as asyncio
|
||||
from machine import Pin, I2C
|
||||
|
||||
# 1. Clean memory immediately
|
||||
# 1. Clean memory immediately before performing any operations
|
||||
gc.collect()
|
||||
|
||||
# --- READ DEVICE ID ---
|
||||
@@ -19,8 +19,13 @@ except Exception:
|
||||
orchestrator_id = None
|
||||
cooking_state = None
|
||||
mqtt_connected = False
|
||||
unsubscribed_hello = False
|
||||
|
||||
# --- MQTT SETUP (Initialized First!) ---
|
||||
# --- ASYNC SIGNALS & QUEUES ---
|
||||
# Event to signal when orchestrator requests sensor data (prevents MQTT lock deadlock)
|
||||
sensor_request_event = None
|
||||
|
||||
# --- MQTT SETUP ---
|
||||
from shared import get_mqtt_client, config, payloads
|
||||
|
||||
MQTT_CA_FILE = "/certs/ca.crt"
|
||||
@@ -34,7 +39,6 @@ mqtt_client = get_mqtt_client(
|
||||
)
|
||||
|
||||
# --- HARDWARE & MODULE DEFERRED IMPORTS ---
|
||||
# We declare variables here, but initialize them AFTER MQTT connects
|
||||
status_led = None
|
||||
uart_device = None
|
||||
mlx_temperature_sensor = None
|
||||
@@ -45,13 +49,12 @@ UARTCommandType = None
|
||||
|
||||
|
||||
def init_hardware():
|
||||
"""Initializes hardware peripherals AFTER MQTT TLS has reserved its memory."""
|
||||
"""Initializes hardware peripherals AFTER MQTT TLS has reserved its RAM."""
|
||||
global status_led, uart_device, mlx_temperature_sensor
|
||||
global cookingState, log, UARTCommand, UARTCommandType
|
||||
|
||||
print("[Main] Initializing hardware peripherals...")
|
||||
|
||||
# Deferred module imports
|
||||
from shared import get_uart, cookingState as cs, logging
|
||||
from shared.uart_comm import UARTCommand as UC, UARTCommandType as UCT
|
||||
from shared.sensors import RGBLED
|
||||
@@ -62,13 +65,9 @@ def init_hardware():
|
||||
UARTCommand = UC
|
||||
UARTCommandType = UCT
|
||||
|
||||
# Status LED
|
||||
status_led = RGBLED(red_pin=21, green_pin=19, blue_pin=18)
|
||||
|
||||
# Hardware UART 2
|
||||
uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16)
|
||||
|
||||
# I2C Temperature Sensor
|
||||
temperature_sensor_i2c = I2C(
|
||||
0,
|
||||
scl=Pin(25, Pin.IN, Pin.PULL_UP),
|
||||
@@ -83,43 +82,38 @@ def init_hardware():
|
||||
mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c)
|
||||
|
||||
|
||||
# --- CALLBACKS ---
|
||||
def on_received_cooking_state_update(state, is_error=False, is_terminated=False):
|
||||
"""Callback executed when state changes are received from the LoRa board over UART."""
|
||||
if cooking_state:
|
||||
if is_error:
|
||||
cooking_state.set_state(cookingState.CookingStates.ERROR)
|
||||
elif is_terminated:
|
||||
cooking_state.set_state(cookingState.CookingStates.ABORTED)
|
||||
else:
|
||||
cooking_state.set_state(state)
|
||||
|
||||
|
||||
def on_cooking_state_change(state):
|
||||
if status_led is None:
|
||||
return
|
||||
print(f"[Main] Cooking state changed to: {state.state}")
|
||||
BLINK_INTERVAL_MS = 500
|
||||
|
||||
from shared.sensors import RGBLED
|
||||
if state.state == cookingState.CookingStates.IDLE:
|
||||
status_led.color = RGBLED.OFF
|
||||
status_led.blink_off()
|
||||
elif state.state == cookingState.CookingStates.COOKING:
|
||||
status_led.color = RGBLED.YELLOW
|
||||
status_led.blink_off()
|
||||
elif state.state == cookingState.CookingStates.STIRRING_REQUIRED:
|
||||
status_led.color = RGBLED.ORANGE
|
||||
status_led.blink_on(BLINK_INTERVAL_MS)
|
||||
elif state.state == cookingState.CookingStates.DONE:
|
||||
status_led.color = RGBLED.GREEN
|
||||
status_led.blink_off()
|
||||
elif state.state == cookingState.CookingStates.ALERT:
|
||||
status_led.color = RGBLED.RED
|
||||
status_led.blink_on(BLINK_INTERVAL_MS)
|
||||
|
||||
|
||||
def on_received_cooking_state_update(new_state):
|
||||
global cooking_state
|
||||
if cooking_state is None:
|
||||
print("[Main] No active cooking state to update.")
|
||||
return
|
||||
if log:
|
||||
log(f"[Main] Updating cooking state to: {new_state}")
|
||||
cooking_state.set_state(new_state)
|
||||
"""Callback executed whenever local cooking state transitions."""
|
||||
if status_led and cookingState:
|
||||
if state == cookingState.CookingStates.IDLE:
|
||||
status_led.set_color(0, 0, 0) # Off
|
||||
elif state == cookingState.CookingStates.PREHEATING:
|
||||
status_led.set_color(255, 165, 0) # Orange
|
||||
elif state == cookingState.CookingStates.COOKING:
|
||||
status_led.set_color(255, 0, 0) # Red
|
||||
elif state == cookingState.CookingStates.DONE:
|
||||
status_led.set_color(0, 255, 0) # Green
|
||||
elif state in (
|
||||
cookingState.CookingStates.ERROR,
|
||||
cookingState.CookingStates.ABORTED,
|
||||
):
|
||||
status_led.set_color(255, 0, 255) # Magenta/Purple
|
||||
|
||||
|
||||
def on_mqtt_message(message):
|
||||
global orchestrator_id, cooking_state
|
||||
"""Sync callback: Lightweight! Only updates variables or triggers async signals."""
|
||||
global orchestrator_id, cooking_state, unsubscribed_hello
|
||||
print("[MQTT] Received message on topic:", message.get("topic"))
|
||||
|
||||
payload_data = None
|
||||
@@ -138,12 +132,13 @@ def on_mqtt_message(message):
|
||||
):
|
||||
orchestrator_id = payload_data.get("id_orchestrator")
|
||||
print("[MQTT] Hello response received from orchestrator:", orchestrator_id)
|
||||
if not unsubscribed_hello:
|
||||
unsubscribed_hello = True
|
||||
try:
|
||||
mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO)
|
||||
print("[MQTT] Unsubscribed from topic:", config.MQTT_TOPIC_HELLO)
|
||||
print("[MQTT] Successfully unsubscribed from topic:", config.MQTT_TOPIC_HELLO)
|
||||
except Exception as e:
|
||||
print("[MQTT] Unsubscribe error:", e)
|
||||
sys.print_exception(e)
|
||||
|
||||
# 2. Cooking Parameters / Sensor Request
|
||||
elif (
|
||||
@@ -152,14 +147,9 @@ def on_mqtt_message(message):
|
||||
and payload_data.get("id_microwave") == DEVICE_ID
|
||||
):
|
||||
if "cook_time" not in payload_data:
|
||||
print("[MQTT] Sensor data requested by orchestrator.")
|
||||
obj_temp = mlx_temperature_sensor.read_object_temp() if mlx_temperature_sensor else 0
|
||||
amb_temp = mlx_temperature_sensor.read_ambient_temp() if mlx_temperature_sensor else 0
|
||||
|
||||
sensor_payload = payloads.mqtt_sensor_data(DEVICE_ID, obj_temp, amb_temp)
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_SENSOR, sensor_payload, qos=config.MQTT_QOS
|
||||
)
|
||||
print("[MQTT] Sensor data requested! Triggering async publisher...")
|
||||
# Trigger async event instead of calling publish() directly inside lock context!
|
||||
sensor_request_event.set()
|
||||
else:
|
||||
print("[MQTT] Cooking parameters received:", payload_data)
|
||||
if cookingState:
|
||||
@@ -175,21 +165,84 @@ def on_mqtt_message(message):
|
||||
uart_device.send_as_command(
|
||||
UARTCommand(UARTCommandType.COOKING_PARAMS, payload_data)
|
||||
)
|
||||
print("[MQTT] Cooking parameters sent to LoRa board.")
|
||||
print("[MQTT] Cooking parameters sent to LoRa board over UART.")
|
||||
|
||||
|
||||
mqtt_client.set_callback(on_mqtt_message)
|
||||
# --- DEDICATED ASYNC TASK FOR SENSOR PUBLISHING ---
|
||||
async def sensor_publisher_task():
|
||||
"""Waits for sensor_request_event, reads hardware, and publishes outside the MQTT lock."""
|
||||
while True:
|
||||
await sensor_request_event.wait()
|
||||
sensor_request_event.clear()
|
||||
|
||||
print("[Sensor Task] Reading temperature sensors...")
|
||||
obj_temp = (
|
||||
mlx_temperature_sensor.read_object_temp()
|
||||
if mlx_temperature_sensor
|
||||
else 0
|
||||
)
|
||||
amb_temp = (
|
||||
mlx_temperature_sensor.read_ambient_temp()
|
||||
if mlx_temperature_sensor
|
||||
else 0
|
||||
)
|
||||
|
||||
sensor_payload = payloads.mqtt_sensor_data(DEVICE_ID, obj_temp, amb_temp)
|
||||
|
||||
try:
|
||||
print("[Sensor Task] Publishing sensor data to MQTT...")
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_SENSOR, sensor_payload, qos=config.MQTT_QOS
|
||||
)
|
||||
print("[Sensor Task] Sensor data successfully published:", sensor_payload)
|
||||
except Exception as e:
|
||||
print("[Sensor Task] Failed to publish sensor data:", e)
|
||||
|
||||
|
||||
async def uart_task():
|
||||
"""Polls incoming UART messages from the LoRa board using dynamic method fallback."""
|
||||
while True:
|
||||
if uart_device:
|
||||
try:
|
||||
cmd = uart_device.read_as_command()
|
||||
|
||||
if cmd:
|
||||
print("[UART] Command received from LoRa board:", cmd)
|
||||
if (
|
||||
hasattr(cmd, "command_type")
|
||||
and cmd.command_type == UARTCommandType.STATE_UPDATE
|
||||
and on_received_cooking_state_update
|
||||
):
|
||||
on_received_cooking_state_update(
|
||||
cmd.payload.get("state"),
|
||||
cmd.payload.get("is_error", False),
|
||||
cmd.payload.get("is_terminated", False),
|
||||
)
|
||||
except Exception as e:
|
||||
print("[UART Task] Error reading command:", e)
|
||||
|
||||
await asyncio.sleep_ms(50)
|
||||
|
||||
|
||||
async def connect_mqtt_async():
|
||||
"""Connects to MQTT safely while memory is clean."""
|
||||
global mqtt_connected
|
||||
global mqtt_connected, mqtt_client
|
||||
mqtt_connected = False
|
||||
|
||||
while True:
|
||||
try:
|
||||
print("[MQTT] Connecting to broker with TLS...")
|
||||
# Re-instantiate client to clear old socket buffers
|
||||
gc.collect()
|
||||
mqtt_client = get_mqtt_client(
|
||||
host="192.168.50.1", # TODO : Use config.MQTT_BROKER_HOST instead of hardcoding
|
||||
port=8884,
|
||||
client_id="smartwave-esp32-demo",
|
||||
use_tls=True,
|
||||
cafile=MQTT_CA_FILE,
|
||||
keepalive=30,
|
||||
)
|
||||
mqtt_client.set_callback(on_mqtt_message)
|
||||
|
||||
mqtt_client.connect()
|
||||
print("[MQTT] Connected! Subscribing to topics...")
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, qos=config.MQTT_QOS)
|
||||
@@ -204,12 +257,14 @@ async def connect_mqtt_async():
|
||||
mqtt_client.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Force heap cleanup before sleeping
|
||||
del mqtt_client
|
||||
gc.collect()
|
||||
print(f"[MQTT] Free RAM after cleanup: {gc.mem_free()} bytes")
|
||||
print("[MQTT] Retrying connection in 5 seconds...")
|
||||
await asyncio.sleep(5)
|
||||
|
||||
|
||||
# --- CONCURRENT ASYNC TASKS ---
|
||||
|
||||
async def mqtt_poll_task():
|
||||
global mqtt_connected
|
||||
last_ping = time.time()
|
||||
@@ -234,9 +289,18 @@ async def mqtt_poll_task():
|
||||
async def orchestrator_hello_task():
|
||||
global mqtt_connected
|
||||
while True:
|
||||
if mqtt_connected and orchestrator_id is None:
|
||||
if orchestrator_id is not None:
|
||||
# Hello successfully acknowledged! Stop looping this task.
|
||||
print("[Hello Task] Orchestrator acknowledged. Stopping hello task.")
|
||||
break
|
||||
|
||||
if mqtt_connected:
|
||||
print("[Hello Task] Sending initial hello to orchestrator...")
|
||||
try:
|
||||
if mqtt_client is None:
|
||||
print("[Hello Task] MQTT client is None. Attempting to reconnect...")
|
||||
await connect_mqtt_async()
|
||||
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_HELLO,
|
||||
payloads.mqtt_hello(DEVICE_ID),
|
||||
@@ -244,32 +308,11 @@ async def orchestrator_hello_task():
|
||||
)
|
||||
except OSError as e:
|
||||
print("[Hello Task] Hello publish failed:", e)
|
||||
mqtt_connected = False
|
||||
await connect_mqtt_async()
|
||||
# mqtt_connected = False
|
||||
|
||||
await asyncio.sleep(config.MQTT_HELLO_INTERVAL)
|
||||
|
||||
|
||||
async def uart_task():
|
||||
while True:
|
||||
if uart_device is not None:
|
||||
while uart_device.any():
|
||||
command = uart_device.read_as_command()
|
||||
if command:
|
||||
print(f"[UART Task] Received command: {command.command_type}")
|
||||
if command.command_type == UARTCommandType.COOKING_STATE_UPDATE:
|
||||
new_state = command.payload.get("state", None)
|
||||
print(f"[UART Task] Cooking state update: {new_state}")
|
||||
on_received_cooking_state_update(new_state)
|
||||
else:
|
||||
print(f"[UART Task] Unknown command type: {command.command_type}")
|
||||
else:
|
||||
raw_command = uart_device.read()
|
||||
print(f"[UART Task] Received raw command: {raw_command}")
|
||||
|
||||
await asyncio.sleep_ms(20)
|
||||
|
||||
|
||||
async def memory_cleanup_task():
|
||||
while True:
|
||||
gc.collect()
|
||||
@@ -278,17 +321,19 @@ async def memory_cleanup_task():
|
||||
|
||||
# --- MAIN ENTRY POINT ---
|
||||
async def main():
|
||||
global sensor_request_event
|
||||
print("[Main] Starting application...")
|
||||
|
||||
# STEP 1: Connect MQTT FIRST (while RAM is unfragmented)
|
||||
await connect_mqtt_async()
|
||||
# Initialize loop-bound events
|
||||
sensor_request_event = asyncio.Event()
|
||||
|
||||
# STEP 2: Initialize Hardware & Secondary Modules AFTER connection
|
||||
await connect_mqtt_async()
|
||||
init_hardware()
|
||||
|
||||
# STEP 3: Launch tasks
|
||||
# Launch background tasks
|
||||
asyncio.create_task(mqtt_poll_task())
|
||||
asyncio.create_task(orchestrator_hello_task())
|
||||
asyncio.create_task(sensor_publisher_task())
|
||||
asyncio.create_task(uart_task())
|
||||
asyncio.create_task(memory_cleanup_task())
|
||||
|
||||
|
||||
+259
-324
@@ -1,58 +1,49 @@
|
||||
import base64
|
||||
import json
|
||||
import threading
|
||||
import queue
|
||||
import time
|
||||
import traceback
|
||||
|
||||
import asyncio
|
||||
import requests
|
||||
|
||||
from orchestrateur.sensors import gps
|
||||
from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads
|
||||
from shared.logging import log
|
||||
from shared.cookingState import CookingStates
|
||||
from shared.lora_device import LoraCommands
|
||||
from sensors import ultrasonicRanger, temp_hum, button, camera
|
||||
|
||||
# --- Read Unique Device ID ---
|
||||
def get_device_id():
|
||||
for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]:
|
||||
try:
|
||||
with open("device_id.txt", "r") as f:
|
||||
DEVICE_ID = f.read().strip()
|
||||
with open(path, "r") as f:
|
||||
return 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"
|
||||
pass
|
||||
return "RPI_Orchestrateur_Default"
|
||||
|
||||
# Thread-safe queue for application messages
|
||||
data_queue = queue.Queue()
|
||||
cooking_queue = {}
|
||||
active_cooks = {}
|
||||
active_cooks_lock = threading.Lock()
|
||||
DEVICE_ID = get_device_id()
|
||||
|
||||
# --- STATE MACHINE DEFINITIONS ---
|
||||
class MicrowaveState:
|
||||
IDLE = "IDLE" # Microwave is empty
|
||||
ANALYZING = "ANALYZING" # Reading sensors & waiting for IR
|
||||
WAITING_FOR_CLOUD = "WAITING_FOR_CLOUD" # Waiting for API parameters
|
||||
COOKING = "COOKING" # Microwave is active
|
||||
DONE = "DONE" # Finished/Stopped, waiting for dish removal
|
||||
|
||||
# Global state trackers
|
||||
microwave_states = {"2": MicrowaveState.IDLE}
|
||||
cooking_data_cache = {} # Replaces cooking_queue
|
||||
button_state = False
|
||||
async_event_queue = None
|
||||
|
||||
# --- HARDWARE SETUP ---
|
||||
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
|
||||
host="192.168.50.1",
|
||||
client_id="smartwave-orchestrateur-" + DEVICE_ID,
|
||||
use_tls=config.USE_TLS,
|
||||
cafile="/home/pi/SmartWave/orchestrateur/mqtt/certs/ca.crt",
|
||||
@@ -60,335 +51,279 @@ mqtt_client = get_mqtt_client(
|
||||
)
|
||||
mqtt_client.connect()
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_SENSOR, qos=config.MQTT_QOS)
|
||||
print(f"Subscribed to topic: {config.MQTT_TOPIC_SENSOR}")
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS)
|
||||
print(f"Subscribed to topic: {config.MQTT_TOPIC_HELLO}")
|
||||
|
||||
# --- 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.")
|
||||
print("[MQTT] Paho background loop started.")
|
||||
|
||||
# --- BACKGROUND TASKS (PRODUCERS) ---
|
||||
async def lora_listener_task():
|
||||
"""Polls LoRa and pushes to the async queue."""
|
||||
print("[LoRa] Async listener started.")
|
||||
while True:
|
||||
# Run blocking lora receive in a thread to not block asyncio loop
|
||||
paquet = await asyncio.to_thread(lora.receive_reliable, timeout_ms=100)
|
||||
if paquet:
|
||||
await async_event_queue.put({"source": "LoRa", "data": paquet})
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
def mqtt_listener():
|
||||
"""Background Thread: Constantly inspects incoming MQTT message cache."""
|
||||
print("Thread MQTT démarré.")
|
||||
async def mqtt_listener_task():
|
||||
"""Polls MQTT cache and pushes to the async queue."""
|
||||
print("[MQTT] Async listener started.")
|
||||
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
|
||||
except Exception:
|
||||
payload = message['payload']
|
||||
|
||||
print(f"\n[Thread MQTT] Message reçu : {message}")
|
||||
data_queue.put({"source": "MQTT", "topic": message['topic'] ,"data": payload})
|
||||
# --- SAFE TOPIC DECODING ---
|
||||
topic = message['topic']
|
||||
if isinstance(topic, bytes):
|
||||
topic = topic.decode('utf-8')
|
||||
|
||||
# Sleep for 100ms. Prevents the thread from turning into an infinite 100% CPU hog.
|
||||
time.sleep(0.2)
|
||||
await async_event_queue.put({
|
||||
"source": "MQTT",
|
||||
"topic": topic,
|
||||
"data": payload
|
||||
})
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Button
|
||||
button_state = False
|
||||
def button_callback():
|
||||
"""Button physical interrupt callback."""
|
||||
global button_state
|
||||
if microwave_states.get("2") == MicrowaveState.COOKING:
|
||||
print("[Button] Toggling pause/resume for microwave '2'.")
|
||||
lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_PAUSE})
|
||||
else:
|
||||
button_state = not button_state
|
||||
print(f"\n[Thread Button] Button state changed to: {button_state}")
|
||||
print(f"[Button] Defrost state toggled 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.")
|
||||
|
||||
def _tryReadSensorsWithRetries(func, exception=True, max_retries=3, delay=1):
|
||||
"""
|
||||
Tries to read the sensor max_retries times until the return value of func is not None.
|
||||
It will then return the value of func. If it fails max_retries times, it will fail if exception is True, otherwise it will return None.
|
||||
"""
|
||||
|
||||
for attempt in range(max_retries):
|
||||
result = func()
|
||||
if result is not None:
|
||||
return result
|
||||
else:
|
||||
log(f"Attempt {attempt + 1} failed. Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
|
||||
if exception:
|
||||
raise Exception(f"Failed to read sensor after {max_retries} attempts.")
|
||||
else:
|
||||
return None
|
||||
|
||||
def read_sensors_for_cooking(microwave_id):
|
||||
"""Read all sensors and return a dictionary of their values, including the microwave ID."""
|
||||
log("\nLecture des capteurs...")
|
||||
sensor_data = {}
|
||||
sensor_data["microwave_id"] = microwave_id
|
||||
|
||||
# === Notify the microwave of needed sensor readings ===
|
||||
mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_init(microwave_id), qos=config.MQTT_QOS)
|
||||
|
||||
# Read Ultrasonic Ranger
|
||||
sensor_data["ultrasonic_distance"] = _tryReadSensorsWithRetries(ultrasonicRanger.get_dish_height)
|
||||
log(f"\nLecture du capteur Ultrason : {sensor_data['ultrasonic_distance']}")
|
||||
|
||||
# Read Temperature and Humidity
|
||||
temperature, humidity = temp_hum.get_temperature_and_humidity_with_retry()
|
||||
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
|
||||
|
||||
# Camera
|
||||
def _getPicture():
|
||||
picture_bytes = None
|
||||
try:
|
||||
picture_bytes = camera.get_picture()
|
||||
return picture_bytes
|
||||
except Exception as e:
|
||||
log(f"Error reading camera data: {e}")
|
||||
return None
|
||||
sensor_data["camera_image"] = _tryReadSensorsWithRetries(_getPicture, exception=True)
|
||||
log(f"\nPhoto de la Caméra : {len(sensor_data['camera_image'])} bytes")
|
||||
|
||||
# Read Button State (last because he can still change state while reading other sensors)
|
||||
sensor_data["defrost_mode"] = button_state
|
||||
|
||||
cooking_queue[microwave_id] = sensor_data
|
||||
|
||||
|
||||
# --- HARDWARE CONTROLLERS ---
|
||||
def _stop_hardware(microwave_id: str):
|
||||
"""
|
||||
Hardware driver stop — halts magnetron/turntable immediately.
|
||||
"""
|
||||
print(f"[{microwave_id}] 🛑 Emergency stop issued to hardware.")
|
||||
# TODO: Add physical hardware stop command here
|
||||
# e.g., gpio_controller.stop()
|
||||
print(f"[{microwave_id}] /!\ Emergency stop issued to hardware.")
|
||||
# TODO: Add LoRa STOP command here
|
||||
|
||||
# --- ASYNC COOKING LOGIC ---
|
||||
def read_local_sensors(microwave_id, initial_dish_height):
|
||||
"""Blocking function to read local I2C/SPI sensors. Runs in a thread."""
|
||||
print(f"[{microwave_id}] Reading local physical sensors...")
|
||||
sensor_data = {
|
||||
"microwave_id": microwave_id,
|
||||
"defrost_mode": button_state,
|
||||
"ultrasonic_distance": initial_dish_height # Reuse height from trigger
|
||||
}
|
||||
|
||||
def _send_params_to_microwave(microwave_id: str, cook_time: int, power_level: int, target_temp: float, cancel_event: threading.Event):
|
||||
"""
|
||||
Triggers physical microwave execution.
|
||||
"""
|
||||
if cancel_event.is_set():
|
||||
return
|
||||
|
||||
print(f"[{microwave_id}] Sending microwave {microwave_id} cooking parameters : {cook_time}s @ {power_level}W power, target temp {target_temp}°C.")
|
||||
mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_config(microwave_id, cook_time, power_level, target_temp), qos=config.MQTT_QOS)
|
||||
|
||||
def _cooking_worker(microwave_id: str, sensors_data: dict, cancel_event: threading.Event):
|
||||
"""Worker function executing cloud API calls and hardware triggers."""
|
||||
URL = "https://smartwave.matthiasg.dev/cooking-params"
|
||||
|
||||
# Temp / Hum (handles DHT error safely)
|
||||
try:
|
||||
# Check cancellation before network call
|
||||
if cancel_event.is_set():
|
||||
print(f"[{microwave_id}] Job canceled before starting API call.")
|
||||
return
|
||||
|
||||
log(f"[{microwave_id}] Sending sensor data to cloud API...")
|
||||
|
||||
# Ensure camera_image is encoded to Base64 string if it's currently raw bytes
|
||||
if isinstance(sensors_data.get("camera_image"), bytes):
|
||||
sensors_data["camera_image"] = base64.b64encode(sensors_data["camera_image"]).decode("utf-8")
|
||||
|
||||
# 1. HTTP Request (15-second timeout)
|
||||
response = requests.post(URL, json=sensors_data, timeout=360) # timeout for long-running requests
|
||||
|
||||
# Check cancellation right after network call returns
|
||||
if cancel_event.is_set():
|
||||
print(f"[{microwave_id}] Job was canceled while waiting for cloud response. Discarding result.")
|
||||
return
|
||||
|
||||
log(f"[{microwave_id}] Cloud API responded with : {response.json()}")
|
||||
if response.status_code != 200 and response.status_code != 201:
|
||||
print(f"[{microwave_id}] Cloud API returned error {response.status_code}: {response.json()}")
|
||||
response.raise_for_status()
|
||||
|
||||
# 2. Extract Response Parameters
|
||||
response_json = response.json()
|
||||
cook_plan = response_json.get("cook_plan", {})
|
||||
|
||||
cook_time = cook_plan.get("cook_time_seconds")
|
||||
power_level = cook_plan.get("effective_power_watts")
|
||||
target_temp = cook_plan.get("target_temp")
|
||||
dish_name = response_json.get("dish_name", "Unknown Dish")
|
||||
|
||||
if cook_time is None or power_level is None or target_temp is None:
|
||||
print(f"[{microwave_id}] Cloud returned incomplete plan: {response_json}")
|
||||
return
|
||||
|
||||
# Check cancellation before starting physical microwave
|
||||
if cancel_event.is_set():
|
||||
print(f"[{microwave_id}] Job was canceled before starting hardware execution.")
|
||||
return
|
||||
|
||||
print(f"[{microwave_id}] Received plan for '{dish_name}': {cook_time}s @ {power_level}W power, target temp {target_temp}°C.")
|
||||
|
||||
# 3. Start Hardware Execution
|
||||
_send_params_to_microwave(microwave_id, cook_time, power_level, target_temp, cancel_event)
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
print(f"[{microwave_id}] Request timed out waiting for cloud response.")
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"[{microwave_id}] HTTP error reaching cloud API: {e}")
|
||||
temp, hum = temp_hum.get_temperature_and_humidity_with_retry()
|
||||
if temp is not None:
|
||||
sensor_data["temperature"] = temp
|
||||
sensor_data["humidity"] = hum
|
||||
except Exception as e:
|
||||
print(f"[{microwave_id}] Unexpected error in worker thread: {e}")
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
# Clean up registry entry if this worker was the active one
|
||||
with active_cooks_lock:
|
||||
if active_cooks.get(microwave_id) == cancel_event:
|
||||
del active_cooks[microwave_id]
|
||||
|
||||
|
||||
def start_cooking_for_microwave(microwave_id: str, sensors_data: dict):
|
||||
"""
|
||||
Sends sensors data to the cloud and starts cooking in a separate thread.
|
||||
If a worker is already running for the given microwave_id, it cancels
|
||||
the previous process and stops the hardware before starting the new one.
|
||||
"""
|
||||
with active_cooks_lock:
|
||||
# 1. If an active job exists for this microwave, cancel it
|
||||
if microwave_id in active_cooks:
|
||||
print(f"[{microwave_id}] Existing cooking job detected! Canceling old worker...")
|
||||
active_cooks[microwave_id].set() # Signal existing thread to abort
|
||||
_stop_hardware(microwave_id) # Stop hardware immediately
|
||||
|
||||
# 2. Register a new cancellation event for this microwave
|
||||
cancel_event = threading.Event()
|
||||
active_cooks[microwave_id] = cancel_event
|
||||
|
||||
# 3. Start the new background worker thread
|
||||
thread = threading.Thread(
|
||||
target=_cooking_worker,
|
||||
args=(microwave_id, sensors_data, cancel_event),
|
||||
daemon=True
|
||||
)
|
||||
thread.start()
|
||||
|
||||
# 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
|
||||
log(f"[{microwave_id}] DHT read warning: {e}")
|
||||
|
||||
# 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
|
||||
sensor_data["camera_image"] = camera.get_picture()
|
||||
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["defrost_state"] = button_state
|
||||
log(f"[{microwave_id}] Camera read failed: {e}")
|
||||
|
||||
return sensor_data
|
||||
|
||||
# --- MAIN EXECUTION LOOP ---
|
||||
while True:
|
||||
|
||||
async def handle_new_dish(microwave_id, detected_height):
|
||||
"""Triggered when a new dish is placed inside."""
|
||||
microwave_states[microwave_id] = MicrowaveState.ANALYZING
|
||||
print(f"\n[{microwave_id}] 🍽️ Dish detected at {detected_height:.1f} cm! Requesting IR from microwave...")
|
||||
|
||||
# 1. Ask microwave for IR temp via MQTT
|
||||
mqtt_client.publish(config.MQTT_TOPIC_COOKING, payloads.mqtt_cooking_init(microwave_id), qos=config.MQTT_QOS)
|
||||
|
||||
# 2. Read local sensors (passing detected_height to prevent GPIO collision)
|
||||
sensors = await asyncio.to_thread(read_local_sensors, microwave_id, detected_height)
|
||||
|
||||
# Check if state changed while taking photos
|
||||
if microwave_states[microwave_id] != MicrowaveState.ANALYZING:
|
||||
print(f"[{microwave_id}] Dish removed during sensor read. Aborting.")
|
||||
return
|
||||
|
||||
cooking_data_cache[microwave_id] = sensors
|
||||
print(f"[{microwave_id}] Local sensors cached. Waiting for MQTT IR data...")
|
||||
|
||||
async def request_cloud_cooking_plan(microwave_id, sensors_data):
|
||||
"""Sends all data to the cloud and starts the microwave if successful."""
|
||||
microwave_states[microwave_id] = MicrowaveState.WAITING_FOR_CLOUD
|
||||
URL = "https://smartwave.matthiasg.dev/cooking-params"
|
||||
|
||||
# Format image
|
||||
if isinstance(sensors_data.get("camera_image"), bytes):
|
||||
sensors_data["camera_image"] = base64.b64encode(sensors_data["camera_image"]).decode("utf-8")
|
||||
|
||||
print(f"[{microwave_id}] Requesting cooking plan from cloud app...")
|
||||
try:
|
||||
# === TREAT MESSAGE QUEUE ===
|
||||
try:
|
||||
msg = data_queue.get(block=False)
|
||||
response = await asyncio.to_thread(requests.post, URL, json=sensors_data, timeout=30)
|
||||
|
||||
# print(msg)
|
||||
# Abort if state changed (e.g. user removed dish while waiting for wifi)
|
||||
if microwave_states[microwave_id] != MicrowaveState.WAITING_FOR_CLOUD:
|
||||
print(f"[{microwave_id}] Dish removed during API request. Discarding API plan.")
|
||||
return
|
||||
|
||||
if msg["source"] == "LoRa":
|
||||
print(f"\n[Main Loop] LoRa : Données traitées : {msg['data']}")
|
||||
elif msg["source"] == "MQTT":
|
||||
# MQTT HELLO
|
||||
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
|
||||
# MQTT SENSOR DATA
|
||||
elif (msg["topic"] == config.MQTT_TOPIC_SENSOR.decode('utf-8')):
|
||||
print(f"\n[Main Loop] MQTT : Données capteurs reçues du micro-ondes : {msg['data']}")
|
||||
microwave_id = msg["data"].get("id_microwave")
|
||||
if not microwave_id:
|
||||
print("[Main Loop] MQTT : Données capteurs reçues sans ID micro-ondes. Ignoré.")
|
||||
continue
|
||||
# Get the already existing cooking data for this microwave
|
||||
sensors_data = cooking_queue.get(microwave_id)
|
||||
if sensors_data is None:
|
||||
print(f"[Main Loop] MQTT : Données capteurs reçues pour {microwave_id} mais aucune donnée de cuisson en cours. Ignoré.")
|
||||
continue
|
||||
# Merge the received sensor data into the existing cooking data
|
||||
sensors_data["ir_initial_temp"] = msg["data"].get("dish_temp")
|
||||
sensors_data["ir_ambient_temp"] = msg["data"].get("ambient_temp")
|
||||
start_cooking_for_microwave(microwave_id, sensors_data)
|
||||
# Remove the cooking data from the queue since it's now being processed
|
||||
del cooking_queue[microwave_id]
|
||||
response.raise_for_status()
|
||||
plan = response.json().get("cook_plan", {})
|
||||
c_time = plan.get("cook_time_seconds")
|
||||
c_power = plan.get("effective_power_watts")
|
||||
c_temp = plan.get("target_temp")
|
||||
|
||||
print(f"\n[Main Loop] MQTT : Données traitées : {msg['data']}")
|
||||
except queue.Empty:
|
||||
pass
|
||||
if c_time is None or c_power is None or c_temp is None:
|
||||
print(f"[{microwave_id}] ❌ Invalid plan received: {response.json()}")
|
||||
microwave_states[microwave_id] = MicrowaveState.DONE # Fail safe
|
||||
return
|
||||
|
||||
# === CHECK FOR DISH INSERTED ===
|
||||
# Read the dish height from the ultrasonic sensor. If it's below a certain threshold, we assume a dish has been inserted.
|
||||
dish_height = ultrasonicRanger.get_dish_height()
|
||||
if dish_height is not None and dish_height > 2.0: # Threshold in cm for detecting a dish
|
||||
print(f"\n[Main Loop] Dish detected at height: {dish_height} cm. Initiating sensor read...")
|
||||
# Read all sensors and store the data in the cooking queue for this microwave
|
||||
read_sensors_for_cooking("2")
|
||||
print(f"[Main Loop] Sensor data collected and queued for cooking.")
|
||||
print(f"[{microwave_id}] Cloud Plan Received! Starting microwave: {c_time}s @ {c_power}W")
|
||||
microwave_states[microwave_id] = MicrowaveState.COOKING
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_COOKING,
|
||||
payloads.mqtt_cooking_config(microwave_id, c_time, c_power, c_temp),
|
||||
qos=config.MQTT_QOS
|
||||
)
|
||||
|
||||
# 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
|
||||
print(f"[{microwave_id}] Cloud API Error: {e}")
|
||||
microwave_states[microwave_id] = MicrowaveState.DONE
|
||||
|
||||
# Clean termination
|
||||
# --- MAIN LOGIC TASKS ---
|
||||
async def process_messages_task():
|
||||
"""Consumes the unified event queue."""
|
||||
while True:
|
||||
msg = await async_event_queue.get()
|
||||
source = msg["source"]
|
||||
data = msg["data"]
|
||||
|
||||
if source == "LoRa":
|
||||
if "new_cooking_state" in data.get("data", {}):
|
||||
mw_id = data["data"].get("id")
|
||||
n_state = data["data"].get("new_cooking_state")
|
||||
print(f"[LoRa] Microwave {mw_id} state changed to: {n_state}")
|
||||
|
||||
if n_state == CookingStates.IDLE and microwave_states.get(mw_id) == MicrowaveState.COOKING:
|
||||
microwave_states[mw_id] = MicrowaveState.DONE
|
||||
print(f"[{mw_id}] Cooking finished. Waiting for user to remove dish.")
|
||||
|
||||
elif source == "MQTT":
|
||||
topic = msg["topic"]
|
||||
|
||||
# Helper to normalize config topics to str
|
||||
def to_str(val):
|
||||
return val.decode('utf-8') if isinstance(val, bytes) else val
|
||||
|
||||
hello_topic = to_str(config.MQTT_TOPIC_HELLO)
|
||||
sensor_topic = to_str(config.MQTT_TOPIC_SENSOR)
|
||||
|
||||
if topic == hello_topic:
|
||||
if data.get("id_orchestrator") != DEVICE_ID:
|
||||
mw_id = data.get("id_microwave")
|
||||
print(f"[MQTT] Hello from {mw_id}. Sending ACK.")
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_HELLO,
|
||||
payloads.mqtt_hello_ack(DEVICE_ID, mw_id),
|
||||
qos=config.MQTT_QOS
|
||||
)
|
||||
|
||||
elif topic == sensor_topic:
|
||||
mw_id = data.get("id_microwave")
|
||||
|
||||
if mw_id and microwave_states.get(mw_id) == MicrowaveState.ANALYZING:
|
||||
sensors = cooking_data_cache.get(mw_id)
|
||||
if sensors:
|
||||
sensors["ir_initial_temp"] = data.get("dish_temp")
|
||||
sensors["ir_ambient_temp"] = data.get("ambient_temp")
|
||||
asyncio.create_task(request_cloud_cooking_plan(mw_id, sensors))
|
||||
|
||||
async def get_filtered_dish_height(samples=3, delay=0.04):
|
||||
"""Reads ultrasonic sensor multiple times and returns the median, discarding invalid zeros."""
|
||||
valid_samples = []
|
||||
for _ in range(samples):
|
||||
h = await asyncio.to_thread(ultrasonicRanger.get_dish_height)
|
||||
# Discard 0.0 or near-zero timeout glitches
|
||||
if h is not None and h > 0.5:
|
||||
valid_samples.append(h)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
if valid_samples:
|
||||
valid_samples.sort()
|
||||
return valid_samples[len(valid_samples) // 2] # Median sample
|
||||
return None # All reads failed or out of range
|
||||
|
||||
|
||||
async def monitor_dish_height_task():
|
||||
"""Monitors presence of dish with hysteresis and debouncing."""
|
||||
mw_id = "2"
|
||||
consecutive_present = 0
|
||||
consecutive_absent = 0
|
||||
REQUIRED_STABLE_READS = 3 # Must see 3 stable states in a row (~1 second)
|
||||
|
||||
while True:
|
||||
dist = await get_filtered_dish_height()
|
||||
current_state = microwave_states.get(mw_id, MicrowaveState.IDLE)
|
||||
|
||||
if dist is not None:
|
||||
# Hysteresis Thresholds:
|
||||
# - Must be > 2.5 cm to detect dish insertion
|
||||
# - Must be < 1.2 cm to detect dish removal
|
||||
if dist > 2.5:
|
||||
consecutive_present += 1
|
||||
consecutive_absent = 0
|
||||
elif dist < 1.2:
|
||||
consecutive_absent += 1
|
||||
consecutive_present = 0
|
||||
else:
|
||||
# Dead-zone (1.2cm to 2.5cm) -> Noise buffer
|
||||
consecutive_present = 0
|
||||
consecutive_absent = 0
|
||||
|
||||
# --- DISH INSERTED CONFIRMED ---
|
||||
if consecutive_present >= REQUIRED_STABLE_READS and current_state == MicrowaveState.IDLE:
|
||||
consecutive_present = 0
|
||||
asyncio.create_task(handle_new_dish(mw_id, dist))
|
||||
|
||||
# --- DISH REMOVED CONFIRMED ---
|
||||
elif consecutive_absent >= REQUIRED_STABLE_READS and current_state != MicrowaveState.IDLE:
|
||||
consecutive_absent = 0
|
||||
print(f"\n[{mw_id}] Dish Removed! Resetting state to IDLE.")
|
||||
microwave_states[mw_id] = MicrowaveState.IDLE
|
||||
if current_state == MicrowaveState.COOKING:
|
||||
_stop_hardware(mw_id)
|
||||
if mw_id in cooking_data_cache:
|
||||
del cooking_data_cache[mw_id]
|
||||
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# --- BOOTSTRAP ---
|
||||
async def main():
|
||||
global async_event_queue
|
||||
print("🚀 Orchestrateur Asyncio prêt. Lancement des tâches...")
|
||||
|
||||
async_event_queue = asyncio.Queue()
|
||||
|
||||
await asyncio.gather(
|
||||
lora_listener_task(),
|
||||
mqtt_listener_task(),
|
||||
process_messages_task(),
|
||||
monitor_dish_height_task()
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
print("\nArrêt manuel.")
|
||||
finally:
|
||||
if hasattr(mqtt_client._client, "loop_stop"):
|
||||
mqtt_client._client.loop_stop()
|
||||
mqtt_client.close()
|
||||
Reference in New Issue
Block a user