diff --git a/micro_ondes/esp_lora/main.py b/micro_ondes/esp_lora/main.py index 3fb37df..f5c95b2 100644 --- a/micro_ondes/esp_lora/main.py +++ b/micro_ondes/esp_lora/main.py @@ -1,9 +1,13 @@ import _thread -from machine import Pin +from machine import Pin, SoftI2C +from shared.safeQueue import SafeQueue from shared import get_lora, get_uart, deviceTypes, config, cookingState from shared.uart_comm import UARTCommand, UARTCommandType from shared.sensors import RGBLED from shared.logging import log +from shared.lora_device import LoraCommands +import framebuf +import ssd1306 import time # --- Configuration Matérielle --- @@ -21,44 +25,56 @@ except Exception: # --- Initialisation LoRa --- lora = get_lora() lora.configure(freq=868.1, sf=7) +data_queue = SafeQueue() # --- Création des lEDs RGB --- magnetron_led = RGBLED(red_pin=48, green_pin=47, blue_pin=33) magnetron_led.color = RGBLED.WHITE_YELLOW magnetron_led.off() +# --- Création de l'écran OLED --- +scl_pin = Pin(18, Pin.OUT, pull=Pin.PULL_UP) +sda_pin = Pin(17, Pin.OUT, pull=Pin.PULL_UP) +display_i2c = SoftI2C(scl=scl_pin, sda=sda_pin, freq=100000) +display = ssd1306.SSD1306_I2C(128, 64, display_i2c, addr=0x3C) +display.text("Booting...", 1, 2, 1) +display.show() + print(f"ESP32 initialisé avec l'ID : '{DEVICE_ID}' (Type : {deviceTypes.DEVICE_TYPES['MICROWAVE']})") +PING_PAYLOAD = { + "id": DEVICE_ID, + "type": deviceTypes.DEVICE_TYPES["MICROWAVE"] +} + def heartbeat_loop(): + last_heartbeat_time = 0 while True: - print(f"\nESP32 : Envoi du Heartbeat...") - # Envoi périodique - ping_payload = { - "id": DEVICE_ID, - "type": deviceTypes.DEVICE_TYPES["MICROWAVE"] - } - lora.send(ping_payload) + now = time.time() - # Le receive_packet est maintenant protégé par le lock dans lora_device - # Si le main thread utilise la radio, ce thread attendra son tour - paquet = lora.receive_packet(timeout_ms=2000) + # 1. Send periodic heartbeat + if now - last_heartbeat_time >= config.LORA_HEARTBEAT_INTERVAL: + last_heartbeat_time = now + print("\nESP32 : Envoi du Heartbeat...") + lora.send(PING_PAYLOAD) - if paquet and not paquet["raw"]: - donnees = paquet["data"] - # Vérification si le paquet reçu est bien la réponse attendue de l'orchestrateur - if donnees.get("type") == deviceTypes.DEVICE_TYPES["ORCHESTRATOR"]: - print(f"ESP32 : Réponse reçue de l'orchestrateur '{donnees.get('id')}' ! [Statut: ALIVE]") - else: - print(f"ESP32 : Paquet reçu d'un type inattendu : {donnees.get('type')}") - else: - print("ESP32 : Pas de réponse de l'orchestrateur (Le RPI est-il éteint ?)") + # 2. Increase listen window to 300ms so radio stays active in RX mode + paquet = lora.receive_reliable(timeout_ms=300) - time.sleep(config.HEARTBEAT_INTERVAL) + if paquet is not None: + log(f"[LoRa Thread] New Packet Received: {paquet}") + data_queue.put(paquet) + + time.sleep_ms(10) # UART uart_device = get_uart(uart_id=1, tx_pin=46, rx_pin=45) # Lancer la boucle de heartbeat dans un thread séparé +try: + _thread.stack_size(16 * 1024) +except Exception: + pass _thread.start_new_thread(heartbeat_loop, ()) # Cooking parameters @@ -71,8 +87,13 @@ def cooking_state_on_state_change(state): # Send to the Wifi board the current state uart_device.send_as_command(UARTCommand(UARTCommandType.COOKING_STATE_UPDATE, {"state": state.state})) + # Send to the orchestrator the current state + lora.send_reliable({"id": DEVICE_ID, "new_cooking_state": state.state}) - if state.paused or state.state == cookingState.CookingStates.DONE: + display.text(cookingState.CookingStates.get_state_name(state.state), 1, 2, 1) + display.show() + + if state.paused or state.state == cookingState.CookingStates.DONE or state.state == cookingState.CookingStates.IDLE: magnetron_led.off() else: magnetron_led.on() @@ -83,8 +104,7 @@ def cooking_state_on_state_change(state): if state.state == cookingState.CookingStates.STIRRING_REQUIRED: pass if state.state == cookingState.CookingStates.DONE: - global cooking_state - cooking_state = None + pass if state.state == cookingState.CookingStates.ALERT: pass @@ -92,6 +112,13 @@ def cooking_state_on_refresh(state): # TODO Show screen information pass +def cooking_state_on_pause(state): + # If the cooking is unpaused and was in STIRRING_REQUIRED or ALERT state, we set the state back to COOKING. + if not state.paused and (state.state == cookingState.CookingStates.STIRRING_REQUIRED or state.state == cookingState.CookingStates.ALERT): + state.set_state(cookingState.CookingStates.COOKING) + + # TODO send_reliable lora message to orchestrator about pause/resume state + # --- MAIN APPLICATION THREAD --- print("[Main] Main execution path active.") @@ -113,17 +140,40 @@ while True: cooking_state.set_temperature_provider(cooking_state_temperature_provider) cooking_state.set_state_change_callback(cooking_state_on_state_change) cooking_state.set_refresh_callback(cooking_state_on_refresh) + cooking_state.set_pause_callback(cooking_state_on_pause) time.sleep_ms(20) # Before sending back right away cooking_state_on_state_change(cooking_state) else: print(f"[Main] Unknown command type received: {command.command_type}") + # 2. Listen for incoming LoRa packets from the orchestrator + while not data_queue.empty(): + paquet = data_queue.get() + if paquet and not paquet["raw"]: + data = paquet["data"] + # Commands + if "action" in data: + if data["action"] == LoraCommands.TOGGLE_PAUSE: + if cooking_state != None: + if (cooking_state.state == cookingState.CookingStates.DONE): + print("[Main] Cooking is done. We reset the microwave for the next cooking session.") + cooking_state.set_state(cookingState.CookingStates.IDLE) + time.sleep_ms(20) # Before sending back right away + cooking_state = None + else: + cooking_state.toggle_pause() + if cooking_state.paused: + print("[Main] Cooking paused via orchestrator command.") + else: + print("[Main] Cooking resumed via orchestrator command.") + else: + log("[Main] No active cooking state to toggle pause/resume.") # uart_device.send(f"Hello from esp-32 lora ID {DEVICE_ID}") # Cooking State Update - if cooking_state: + if cooking_state != None: cooking_state.update_tick() print(f"[Main] Cooking state : State : {cooking_state.state}, Temperature: {cooking_state.current_dish_temp}, Paused: {cooking_state.paused}, Remaining Time: {cooking_state.get_remaining_time():.2f}s, Estimated Remaining Time: {cooking_state.get_remaining_time_estimation():.2f}s") - time.sleep_ms(200) \ No newline at end of file + time.sleep_ms(500) \ No newline at end of file diff --git a/micro_ondes/esp_lora/tests/main_lora.py b/micro_ondes/esp_lora/tests/main_lora.py index 0c0f4ac..dedfcf2 100644 --- a/micro_ondes/esp_lora/tests/main_lora.py +++ b/micro_ondes/esp_lora/tests/main_lora.py @@ -14,10 +14,10 @@ while True: mesures = {"id": "ESP32_Salon", "temp": 22.4, "hum": 55.2} # Envoi direct (le pilote s'occupe de mettre le groupe \x02) - lora.send(b'\x02' + lora.send_json_bytes_helper if False else bytes([2]) + lora.send_helper if False else b'\x02' + __import__('ujson').dumps(mesures).encode('utf-8')) + lora.send_reliable(b'\x02' + lora.send_json_bytes_helper if False else bytes([2]) + lora.send_helper if False else b'\x02' + __import__('ujson').dumps(mesures).encode('utf-8')) # Réception propre - paquet = lora.receive_packet(3000) + paquet = lora.receive_reliable(3000) if paquet: # paquet est un dict : {"group": 2, "data": {...}, "raw": False} print(f"ESP32 : Message reçu du groupe {paquet['group']}") diff --git a/micro_ondes/esp_wifi/main.py b/micro_ondes/esp_wifi/main.py index 31f8396..9b038c9 100644 --- a/micro_ondes/esp_wifi/main.py +++ b/micro_ondes/esp_wifi/main.py @@ -19,7 +19,7 @@ except Exception: orchestrator_id = None cooking_state = None mqtt_connected = False -unsubscribed_hello = False +should_unsubscribe_hello = False # --- ASYNC SIGNALS & QUEUES --- # Event to signal when orchestrator requests sensor data (prevents MQTT lock deadlock) @@ -95,25 +95,29 @@ def on_received_cooking_state_update(state, is_error=False, is_terminated=False) def on_cooking_state_change(state): """Callback executed whenever local cooking state transitions.""" + BLINK_INTERVAL_MS = 500 + 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 + status_led.color = status_led.OFF + status_led.blink_off() elif state == cookingState.CookingStates.COOKING: - status_led.set_color(255, 0, 0) # Red + status_led.color = status_led.YELLOW + status_led.blink_off() + elif state == cookingState.CookingStates.STIRRING_REQUIRED: + status_led.color = status_led.ORANGE + status_led.blink_on(BLINK_INTERVAL_MS) + elif state == cookingState.CookingStates.ALERT: + status_led.color = status_led.RED + status_led.blink_on(BLINK_INTERVAL_MS) 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 + status_led.color = status_led.GREEN + status_led.blink_off() def on_mqtt_message(message): """Sync callback: Lightweight! Only updates variables or triggers async signals.""" - global orchestrator_id, cooking_state, unsubscribed_hello + global orchestrator_id, cooking_state, should_unsubscribe_hello print("[MQTT] Received message on topic:", message.get("topic")) payload_data = None @@ -132,13 +136,7 @@ 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] Successfully unsubscribed from topic:", config.MQTT_TOPIC_HELLO) - except Exception as e: - print("[MQTT] Unsubscribe error:", e) + should_unsubscribe_hello = True # 2. Cooking Parameters / Sensor Request elif ( @@ -275,8 +273,7 @@ async def mqtt_poll_task(): mqtt_client.poll() now = time.time() if now - last_ping >= 15: - if mqtt_client._client: - mqtt_client._client.ping() + mqtt_client.ping() last_ping = now except OSError as e: print("[MQTT Task] Socket error encountered during poll/ping:", e) @@ -287,9 +284,17 @@ async def mqtt_poll_task(): async def orchestrator_hello_task(): - global mqtt_connected + global mqtt_connected, should_unsubscribe_hello while True: if orchestrator_id is not None: + if should_unsubscribe_hello: + try: + mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO) + should_unsubscribe_hello = False + print("[MQTT] Successfully unsubscribed from hello topic.") + except Exception as e: + print("[MQTT] Unsubscribe error:", e) + # Hello successfully acknowledged! Stop looping this task. print("[Hello Task] Orchestrator acknowledged. Stopping hello task.") break diff --git a/orchestrateur/main.py b/orchestrateur/main.py index 1b26fc6..8556634 100644 --- a/orchestrateur/main.py +++ b/orchestrateur/main.py @@ -34,9 +34,12 @@ class MicrowaveState: # Global state trackers microwave_states = {"2": MicrowaveState.IDLE} -cooking_data_cache = {} # Replaces cooking_queue 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() @@ -141,20 +144,47 @@ 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: + + # 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 - cooking_data_cache[microwave_id] = sensors - print(f"[{microwave_id}] Local sensors cached. Waiting for MQTT IR data...") + # 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.""" @@ -236,14 +266,13 @@ async def process_messages_task(): ) elif topic == sensor_topic: - mw_id = data.get("id_microwave") + mw_id = str(data.get("id_microwave")) + print(f"[MQTT] Sensor data received for microwave {mw_id}: {data}") - 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)) + # 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.""" @@ -299,8 +328,9 @@ async def monitor_dish_height_task(): 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] + # 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) diff --git a/orchestrateur/sensors/button.py b/orchestrateur/sensors/button.py index 703b882..f383547 100644 --- a/orchestrateur/sensors/button.py +++ b/orchestrateur/sensors/button.py @@ -11,7 +11,8 @@ grovepi.pinMode(button, "INPUT") button_callback = None def read_button_state(): - if not grove_lock.acquire(timeout=0.05): + # Increase timeout slightly so the button thread can wait for long I2C sensor reads to finish + if not grove_lock.acquire(timeout=0.2): return None try: return grovepi.digitalRead(button) @@ -26,15 +27,18 @@ def monitor_button(): last_button_state = button_switch_state while True: - time.sleep(0.04) - current_state = read_button_state() if current_state is not None: + # Rising edge detection (0 -> 1 transition) if current_state == 1 and last_button_state == 0: if button_callback: button_callback() last_button_state = current_state + time.sleep(0.02) # Fast 20ms poll when lock is clear + else: + # Lock was busy; retry quickly without updating last_button_state + time.sleep(0.01) def start_button_monitoring_thread(): threading.Thread(target=monitor_button, daemon=True).start() diff --git a/orchestrateur/tests/main_lora.py b/orchestrateur/tests/main_lora.py index 835cfe2..5292b7f 100644 --- a/orchestrateur/tests/main_lora.py +++ b/orchestrateur/tests/main_lora.py @@ -7,7 +7,7 @@ lora.configure() print("Raspberry Pi : En attente active de JSON...") while True: - paquet = lora.receive_packet(timeout_ms=5000) + paquet = lora.receive_reliable(timeout_ms=5000) if paquet: # Plus besoin de décoder du HEX ou de parser du JSON manuellement ! groupe = paquet['group'] diff --git a/shared/__init__.py b/shared/__init__.py index 9bb5d8e..df1e3fd 100644 --- a/shared/__init__.py +++ b/shared/__init__.py @@ -5,6 +5,11 @@ import shared.deviceTypes as deviceTypes import shared.config as config import shared.payloads as payloads import shared.cookingState as cookingState +import shared.safeQueue as safeQueue +try: + import shared.lora_device as lora_device +except ImportError: + pass # No need try: import shared.uart_comm as uart_comm except ImportError: diff --git a/shared/config.py b/shared/config.py index 71ebd68..be1e467 100644 --- a/shared/config.py +++ b/shared/config.py @@ -1,7 +1,7 @@ DEBUG=True # LoRa -HEARTBEAT_INTERVAL = 30 +LORA_HEARTBEAT_INTERVAL = 30 # MQTT MQTT_BROKER_HOST = "192.168.50.1" diff --git a/shared/cookingState.py b/shared/cookingState.py index 57ce827..d5637af 100644 --- a/shared/cookingState.py +++ b/shared/cookingState.py @@ -13,6 +13,7 @@ class CookingState: self.temperature_provider = temperature_provider self.on_state_change = on_state_change self.on_refresh = on_refresh + self.on_pause = None self.state = CookingStates.COOKING self.paused = False @@ -24,6 +25,7 @@ class CookingState: self.estimated_remaining_time = float(cook_time) self._last_temperature_sample = None self._last_refresh_signature = None + self._stirred = False def set_temperature_provider(self, temperature_provider): self.temperature_provider = temperature_provider @@ -33,6 +35,9 @@ class CookingState: def set_refresh_callback(self, callback): self.on_refresh = callback + + def set_pause_callback(self, callback): + self.on_pause = callback def pause(self): if self.paused: @@ -40,8 +45,9 @@ class CookingState: self.paused = True self._pause_started_at = time.time() - self._notify_refresh(force=True) - + # self._notify_refresh(force=True) + if self.on_pause: + self.on_pause(self) def unpause(self): if not self.paused: return @@ -50,15 +56,16 @@ class CookingState: if self._pause_started_at is not None: self._paused_duration += now - self._pause_started_at - self._pause_started_at = None + # self._pause_started_at = None self.paused = False - self._notify_refresh(force=True) + # self._notify_refresh(force=True) def toggle_pause(self): if self.paused: self.unpause() else: self.pause() + self.on_pause(self) def set_state(self, state): if self.state == state: @@ -154,6 +161,8 @@ class CookingState: self.on_refresh(self) def update_tick(self): + if self.state == CookingStates.IDLE: + return self.state if self.paused: self._notify_refresh() return self.state @@ -169,18 +178,20 @@ class CookingState: now = time.time() elapsed_time = self.get_elapsed_time() self.estimated_remaining_time = self.get_remaining_time_estimation() + + print(elapsed_time, self.cook_time, self.current_dish_temp, self.target_temp, self._paused_duration, self._pause_started_at, now) if self.current_dish_temp is not None: - if elapsed_time < (self.cook_time / 2.0) and self.current_dish_temp >= self.target_temp: + if elapsed_time < (self.cook_time / 2.0) and self.current_dish_temp >= self.target_temp and (self._paused_duration == None or self._paused_duration < 5): # If the dish is heating too fast, we require stirring self.state = CookingStates.STIRRING_REQUIRED self.pause() elif elapsed_time >= self.cook_time and self.current_dish_temp >= (self.target_temp - self.TEMPERATURE_TOLERANCE): self.state = CookingStates.DONE - elif self.state == CookingStates.DONE and self.current_dish_temp < (self.target_temp - self.TEMPERATURE_TOLERANCE): - self.state = CookingStates.COOKING - elif elapsed_time >= self.cook_time * 1.25: # If the dish is not heating up + elif elapsed_time >= self.cook_time * 1.25 and (self._paused_duration == None or self._paused_duration < 5): # If the dish is not heating up self.state = CookingStates.STIRRING_REQUIRED self.pause() + elif self._pause_started_at != None and (self._pause_started_at + self._paused_duration) < (now - (self.cook_time * 0.75)): # If the dish had to be pause and it's been a long time, we stop the cooking + self.state = CookingStates.DONE self._last_temperature_sample = (now, self.current_dish_temp) @@ -196,4 +207,11 @@ class CookingStates: STIRRING_REQUIRED = 1 DONE = 2 ALERT = 3 # Microwave is too hot internally or other alerts - IDLE = 4 # Waiting for cooking parameters to be set, or after cooking is done \ No newline at end of file + IDLE = 4 # Waiting for cooking parameters to be set, or after cooking is done + + @staticmethod + def get_state_name(state_val): + for key, value in CookingStates.__dict__.items(): + if value == state_val and not key.startswith('__'): + return key + return "UNKNOWN" \ No newline at end of file diff --git a/shared/lora_device.py b/shared/lora_device.py index 699339f..0198abd 100644 --- a/shared/lora_device.py +++ b/shared/lora_device.py @@ -344,7 +344,7 @@ else: print(f"[RPi LoRa Serial] Transmitting HEX payload: {hex_payload}") cmd = f"AT+PSEND={hex_payload}" resp = self._send_at_cmd(cmd, wait_time=0.25) # Wait for RF TX to finish - print(f"[RPi LoRa Serial] AT+PSEND response: {resp.strip().replace(chr(10), ' | ')}") + print(f"[RPi LoRa Serial] AT+PSEND response: {resp}") # Re-enable continuous receive mode after transmission completes self._send_at_cmd("AT+PRECV=65535", wait_time=0.05) diff --git a/shared/mqtt.py b/shared/mqtt.py index bfca8f2..e81381f 100644 --- a/shared/mqtt.py +++ b/shared/mqtt.py @@ -303,6 +303,17 @@ class BrokerClient: pass finally: self._client = None + + def ping(self): + """Thread-safe PINGREQ wrapper for MicroPython.""" + if self._client is None: + return + if IS_MICROPYTHON: + with self._lock: + return self._client.ping() + else: + # Paho handles keepalives automatically via loop_start/loop + pass def __enter__(self): self.connect() diff --git a/shared/safeQueue.py b/shared/safeQueue.py new file mode 100644 index 0000000..f1f0cec --- /dev/null +++ b/shared/safeQueue.py @@ -0,0 +1,37 @@ +import _thread + +class SafeQueue: + """A lightweight, thread-safe FIFO queue for MicroPython.""" + def __init__(self, maxsize=20): + self._queue = [] + self._lock = _thread.allocate_lock() + self.maxsize = maxsize + + def put(self, item) -> bool: + """Push an item to the end of the queue. Returns False if queue is full.""" + with self._lock: + if len(self._queue) < self.maxsize: + self._queue.append(item) + return True + else: + print("[Queue Warning] Buffer full, dropping oldest message.") + self._queue.pop(0) # Drop oldest to make room + self._queue.append(item) + return False + + def get(self): + """Pop and return the oldest item from the queue, or None if empty.""" + with self._lock: + if self._queue: + return self._queue.pop(0) + return None + + def empty(self) -> bool: + """Check if the queue has no items.""" + with self._lock: + return len(self._queue) == 0 + + def size(self) -> int: + """Return current number of queued items.""" + with self._lock: + return len(self._queue) \ No newline at end of file diff --git a/tests/mqtt/micropython_example.py b/tests/mqtt/micropython_example.py index 40e1e3a..e3fd664 100644 --- a/tests/mqtt/micropython_example.py +++ b/tests/mqtt/micropython_example.py @@ -4,7 +4,7 @@ Edit BROKER_HOST so it points to the broker machine IP address. Do not use localhost from the ESP32. """ -from shared.mqtt import BrokerClient +import shared BROKER_HOST = "192.168.50.1" @@ -17,7 +17,7 @@ def on_message(message): def main(): - client = BrokerClient( + client = shared.get_mqtt_client( host=BROKER_HOST, client_id="smartwave-esp32-demo", use_tls=True, @@ -28,7 +28,7 @@ def main(): client.set_callback(on_message) client.connect() client.subscribe(TOPIC, qos=2) - client.publish(TOPIC, b"hello from MicroPython", qos=2, retain=False) + client.publish(TOPIC, b"hello from MicroPython", qos=1, retain=False) for _ in range(30): client.poll()