359 lines
14 KiB
Python
359 lines
14 KiB
Python
import base64
|
|
import json
|
|
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(path, "r") as f:
|
|
return f.read().strip()
|
|
except Exception:
|
|
pass
|
|
return "RPI_Orchestrateur_Default"
|
|
|
|
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}
|
|
button_state = False
|
|
async_event_queue = None
|
|
# Async synchronization trackers for MQTT IR sensors responses
|
|
ir_data_cache = {} # mw_id -> dict of IR readings
|
|
ir_data_events = {} # mw_id -> asyncio.Event()
|
|
|
|
|
|
# --- HARDWARE SETUP ---
|
|
lora = get_lora()
|
|
lora.configure()
|
|
|
|
mqtt_client = get_mqtt_client(
|
|
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",
|
|
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)
|
|
|
|
if hasattr(mqtt_client._client, "loop_start"):
|
|
mqtt_client._client.loop_start()
|
|
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)
|
|
|
|
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:
|
|
payload = json.loads(message['payload'])
|
|
except Exception:
|
|
payload = message['payload']
|
|
|
|
# --- SAFE TOPIC DECODING ---
|
|
topic = message['topic']
|
|
if isinstance(topic, bytes):
|
|
topic = topic.decode('utf-8')
|
|
|
|
await async_event_queue.put({
|
|
"source": "MQTT",
|
|
"topic": topic,
|
|
"data": payload
|
|
})
|
|
await asyncio.sleep(0.1)
|
|
|
|
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"[Button] Defrost state toggled to: {button_state}")
|
|
|
|
button.set_callback(button_callback)
|
|
button.start_button_monitoring_thread()
|
|
|
|
# --- HARDWARE CONTROLLERS ---
|
|
def _stop_hardware(microwave_id: str):
|
|
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
|
|
}
|
|
|
|
# Temp / Hum (handles DHT error safely)
|
|
try:
|
|
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:
|
|
log(f"[{microwave_id}] DHT read warning: {e}")
|
|
|
|
# Camera
|
|
try:
|
|
sensor_data["camera_image"] = camera.get_picture()
|
|
except Exception as e:
|
|
log(f"[{microwave_id}] Camera read failed: {e}")
|
|
|
|
return sensor_data
|
|
|
|
|
|
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. Setup synchronization event and clear previous cache for this microwave
|
|
event = asyncio.Event()
|
|
ir_data_events[microwave_id] = event
|
|
ir_data_cache.pop(microwave_id, None)
|
|
|
|
# 2. Send IR request to ESP32 via MQTT immediately
|
|
mqtt_client.publish(
|
|
config.MQTT_TOPIC_COOKING,
|
|
payloads.mqtt_cooking_init(microwave_id),
|
|
qos=config.MQTT_QOS
|
|
)
|
|
|
|
# 3. Start local sensor reading in parallel
|
|
sensor_task = asyncio.create_task(asyncio.to_thread(read_local_sensors, microwave_id, detected_height))
|
|
|
|
# 4. Wait for local sensors to finish reading
|
|
sensors_data = await sensor_task
|
|
|
|
# Check if dish was removed while reading sensors
|
|
if microwave_states.get(microwave_id) != MicrowaveState.ANALYZING:
|
|
print(f"[{microwave_id}] Dish removed during sensor read. Aborting.")
|
|
ir_data_events.pop(microwave_id, None)
|
|
return
|
|
|
|
# 5. Wait for MQTT IR data (if it already arrived, event.wait() returns instantly)
|
|
try:
|
|
await asyncio.wait_for(event.wait(), timeout=10.0)
|
|
ir_payload = ir_data_cache.get(microwave_id, {})
|
|
sensors_data["ir_initial_temp"] = ir_payload.get("dish_temp")
|
|
sensors_data["ir_ambient_temp"] = ir_payload.get("ambient_temp")
|
|
print(f"[{microwave_id}] IR data synchronized successfully: {ir_payload}")
|
|
except asyncio.TimeoutError:
|
|
print(f"[{microwave_id}] ⚠️ Timeout waiting for MQTT IR data from ESP32.")
|
|
sensors_data["ir_initial_temp"] = None
|
|
sensors_data["ir_ambient_temp"] = None
|
|
finally:
|
|
ir_data_events.pop(microwave_id, None)
|
|
|
|
# 6. Dispatch cloud request task
|
|
asyncio.create_task(request_cloud_cooking_plan(microwave_id, sensors_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:
|
|
response = await asyncio.to_thread(requests.post, URL, json=sensors_data, timeout=30)
|
|
|
|
# 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
|
|
|
|
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")
|
|
|
|
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
|
|
|
|
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
|
|
)
|
|
|
|
except Exception as e:
|
|
print(f"[{microwave_id}] Cloud API Error: {e}")
|
|
microwave_states[microwave_id] = MicrowaveState.DONE
|
|
|
|
# --- 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 = str(data.get("id_microwave"))
|
|
print(f"[MQTT] Sensor data received for microwave {mw_id}: {data}")
|
|
|
|
# Store IR data and notify the waiting dish handler
|
|
ir_data_cache[mw_id] = data
|
|
if mw_id in ir_data_events:
|
|
ir_data_events[mw_id].set()
|
|
|
|
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)
|
|
# Remove from IR cache and events
|
|
ir_data_cache.pop(mw_id, None)
|
|
ir_data_events.pop(mw_id, None)
|
|
|
|
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() |