Working MQTT back !
This commit is contained in:
@@ -5,16 +5,50 @@ esp.osdebug(True)
|
||||
#import webrepl
|
||||
#webrepl.start()
|
||||
|
||||
def do_connect(ssid, pwd):
|
||||
import network
|
||||
sta_if = network.WLAN(network.STA_IF)
|
||||
if not sta_if.isconnected():
|
||||
print('connecting to network...')
|
||||
sta_if.active(True)
|
||||
sta_if.connect(ssid, pwd)
|
||||
while not sta_if.isconnected():
|
||||
pass
|
||||
print('network config:', sta_if.ifconfig())
|
||||
# def do_connect(ssid, pwd):
|
||||
# import network
|
||||
# sta_if = network.WLAN(network.STA_IF)
|
||||
# sta_if.config(pm=sta_if.PM_NONE)
|
||||
# if not sta_if.isconnected():
|
||||
# print('connecting to network...')
|
||||
# sta_if.active(True)
|
||||
# sta_if.connect(ssid, pwd)
|
||||
# while not sta_if.isconnected():
|
||||
# pass
|
||||
# print('network config:', sta_if.ifconfig())
|
||||
|
||||
import network
|
||||
import time
|
||||
|
||||
def do_connect(ssid, password):
|
||||
wlan = network.WLAN(network.STA_IF)
|
||||
|
||||
# 1. ALWAYS activate the interface FIRST
|
||||
if not wlan.active():
|
||||
wlan.active(True)
|
||||
|
||||
# 2. Configure Wi-Fi options AFTER activation
|
||||
try:
|
||||
# Disable Wi-Fi modem sleep (0 = PM_NONE)
|
||||
wlan.config(pm=0)
|
||||
except Exception as e:
|
||||
print("[Wi-Fi] Warning: Failed to set power management:", e)
|
||||
|
||||
# 3. Connect to the access point
|
||||
if not wlan.isconnected():
|
||||
print(f"[Wi-Fi] Connecting to {ssid}...")
|
||||
wlan.connect(ssid, password)
|
||||
|
||||
timeout = 15
|
||||
start_time = time.time()
|
||||
while not wlan.isconnected():
|
||||
if time.time() - start_time > timeout:
|
||||
print("[Wi-Fi] Connection timed out!")
|
||||
return False
|
||||
time.sleep(0.5)
|
||||
|
||||
print("[Wi-Fi] Connected! Network config:", wlan.ifconfig())
|
||||
return True
|
||||
|
||||
# Attempt to connect to WiFi network
|
||||
do_connect("Smartwave-1", 'Smartwave-prot-1')
|
||||
|
||||
+268
-203
@@ -1,23 +1,12 @@
|
||||
import _thread
|
||||
import select
|
||||
from machine import Pin, I2C
|
||||
from sensors import temperature_sensor
|
||||
from shared import get_mqtt_client, get_uart, config, payloads, cookingState
|
||||
from shared.uart_comm import UARTCommand, UARTCommandType
|
||||
from shared.sensors import RGBLED
|
||||
from shared.logging import log
|
||||
import gc
|
||||
import sys
|
||||
import time
|
||||
import ujson as json
|
||||
import sys
|
||||
import uasyncio as asyncio
|
||||
from machine import Pin, I2C
|
||||
|
||||
# Simple thread-safe queue list
|
||||
msg_queue = []
|
||||
queue_lock = _thread.allocate_lock()
|
||||
|
||||
def queue_publish(topic, payload):
|
||||
"""Safely queues a message from the main thread."""
|
||||
with queue_lock:
|
||||
msg_queue.append((topic, payload))
|
||||
# 1. Clean memory immediately
|
||||
gc.collect()
|
||||
|
||||
# --- READ DEVICE ID ---
|
||||
try:
|
||||
@@ -26,214 +15,290 @@ try:
|
||||
except Exception:
|
||||
DEVICE_ID = "ESP32_Inconnu"
|
||||
|
||||
# --- Cooking State ---
|
||||
cooking_state = None # This will hold the current cooking state if any
|
||||
# --- GLOBAL APP STATE ---
|
||||
orchestrator_id = None
|
||||
cooking_state = None
|
||||
mqtt_connected = False
|
||||
|
||||
# --- MQTT SETUP (Initialized First!) ---
|
||||
from shared import get_mqtt_client, config, payloads
|
||||
|
||||
# --- MQTT SETUP ---
|
||||
MQTT_CA_FILE = "/certs/ca.crt"
|
||||
|
||||
mqtt_client = get_mqtt_client(
|
||||
host=config.MQTT_BROKER_HOST,
|
||||
client_id="smartwave-esp32-" + DEVICE_ID,
|
||||
use_tls=config.USE_TLS,
|
||||
host="192.168.50.1",
|
||||
client_id="smartwave-esp32-demo",
|
||||
use_tls=True,
|
||||
cafile=MQTT_CA_FILE,
|
||||
keepalive=config.MQTT_KEEPALIVE,
|
||||
keepalive=30,
|
||||
)
|
||||
|
||||
global orchestrator_id
|
||||
orchestrator_id = None
|
||||
def on_mqtt_message(message):
|
||||
print("[MQTT Thread] Received message:", message)
|
||||
# --- HARDWARE & MODULE DEFERRED IMPORTS ---
|
||||
# We declare variables here, but initialize them AFTER MQTT connects
|
||||
status_led = None
|
||||
uart_device = None
|
||||
mlx_temperature_sensor = None
|
||||
cookingState = None
|
||||
log = None
|
||||
UARTCommand = None
|
||||
UARTCommandType = None
|
||||
|
||||
|
||||
def init_hardware():
|
||||
"""Initializes hardware peripherals AFTER MQTT TLS has reserved its memory."""
|
||||
global status_led, uart_device, mlx_temperature_sensor
|
||||
global cookingState, log, UARTCommand, UARTCommandType
|
||||
|
||||
# Try and parse the payload as json, but if it fails, just print the raw payload
|
||||
payload_data=None
|
||||
try:
|
||||
payload_data = json.loads(message['payload'])
|
||||
except Exception as e:
|
||||
print("[MQTT Thread] Error parsing JSON:", e)
|
||||
sys.print_exception(e)
|
||||
pass # Maybe it's not JSON
|
||||
print("[Main] Initializing hardware peripherals...")
|
||||
|
||||
if message['topic'] == config.MQTT_TOPIC_HELLO and payload_data and "id_orchestrator" in payload_data and payload_data["id_microwave"] == DEVICE_ID:
|
||||
print("[MQTT Thread] Hello response received from orchestrator:", payload_data["id_orchestrator"])
|
||||
global orchestrator_id
|
||||
orchestrator_id = payload_data["id_orchestrator"]
|
||||
# Unsubscribe from the hello topic since we got a response
|
||||
mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO)
|
||||
print("[MQTT Thread] Unsubscribed from topic:", config.MQTT_TOPIC_HELLO)
|
||||
|
||||
# Handle cooking messages
|
||||
elif message['topic'] == config.MQTT_TOPIC_COOKING and payload_data and payload_data["id_microwave"] == DEVICE_ID:
|
||||
# Cooking sensors init request
|
||||
if not "cook_time" in payload_data:
|
||||
print("[MQTT Thread] Cooking sensors init received from the orchestrator")
|
||||
obj_temp = mlx_temperature_sensor.read_object_temp()
|
||||
amb_temp = mlx_temperature_sensor.read_ambient_temp()
|
||||
queue_publish(config.MQTT_TOPIC_SENSOR, payloads.mqtt_sensor_data(DEVICE_ID, obj_temp, amb_temp))
|
||||
# Received cooking parameters from the orchestrator
|
||||
else:
|
||||
print("[MQTT Thread] Cooking parameters received from the orchestrator:", payload_data)
|
||||
global cooking_state
|
||||
cooking_state = cookingState.CookingState(
|
||||
cook_time=payload_data["cook_time"],
|
||||
power_level=payload_data["power_level"],
|
||||
target_temp=payload_data["target_temp"]
|
||||
)
|
||||
cooking_state.set_state_change_callback(on_cooking_state_change)
|
||||
cooking_state.set_state(cookingState.CookingStates.IDLE) # Set initial state to IDLE
|
||||
# Send to the LoRa board the cooking parameters
|
||||
uart_device.send_as_command(UARTCommand(UARTCommandType.COOKING_PARAMS, payload_data))
|
||||
print("[MQTT Thread] Cooking parameters sent to LoRa board.")
|
||||
|
||||
print("[MQTT Thread] Message processing complete.")
|
||||
# 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
|
||||
from sensors import temperature_sensor
|
||||
|
||||
mqtt_client.set_callback(on_mqtt_message)
|
||||
cookingState = cs
|
||||
log = logging.log
|
||||
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),
|
||||
sda=Pin(26, Pin.IN, Pin.PULL_UP),
|
||||
freq=100000,
|
||||
)
|
||||
devices = temperature_sensor_i2c.scan()
|
||||
if 0x5A in devices:
|
||||
print("[Main] MLX90614 found at address 0x5A!")
|
||||
else:
|
||||
print("[Main] MLX90614 not found on I2C bus.")
|
||||
mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c)
|
||||
|
||||
|
||||
def mqtt_background_thread():
|
||||
"""Background MQTT worker handling ALL socket operations safely."""
|
||||
print("[Thread] Background MQTT worker started.")
|
||||
# --- CALLBACKS ---
|
||||
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
|
||||
|
||||
while True:
|
||||
try:
|
||||
print("[Thread] Attempting connection to MQTT broker...")
|
||||
mqtt_client.connect()
|
||||
print("[Thread] Connected! Subscribing to topic...")
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, qos=config.MQTT_QOS)
|
||||
print("[Thread] Successfully subscribed. Setting up poller...")
|
||||
|
||||
poller = select.poll()
|
||||
poller.register(mqtt_client._client.sock, select.POLLIN)
|
||||
|
||||
last_check = time.time()
|
||||
|
||||
while True:
|
||||
# 1. Process outbound messages queued by the main thread
|
||||
while len(msg_queue) > 0:
|
||||
with queue_lock:
|
||||
topic, payload = msg_queue.pop(0)
|
||||
print(f"[Thread] Safely publishing queued message to {topic}...")
|
||||
mqtt_client.publish(topic, payload, qos=config.MQTT_QOS)
|
||||
|
||||
# 2. Check for incoming messages (non-blocking poll)
|
||||
events = poller.poll(200)
|
||||
if events:
|
||||
mqtt_client.wait()
|
||||
|
||||
# 3. Handle Keepalive tracking manually
|
||||
if time.time() - last_check >= 15:
|
||||
# print("[Thread] Sending keepalive ping...")
|
||||
mqtt_client._client.ping()
|
||||
last_check = time.time()
|
||||
|
||||
# Small breathe room for the CPU core
|
||||
time.sleep_ms(50)
|
||||
|
||||
except Exception as e:
|
||||
print("[Thread] Connection dropped or error encountered:", e)
|
||||
sys.print_exception(e)
|
||||
print("[Thread] Cleaning up socket context. Retrying in 5 seconds...")
|
||||
|
||||
# --- FIX FOR ERROR 23 (SOCKET LEAK) ---
|
||||
# Manually force-kill the underlying socket file descriptor if it exists
|
||||
try:
|
||||
if mqtt_client._client and hasattr(mqtt_client._client, "sock"):
|
||||
if mqtt_client._client.sock is not None:
|
||||
mqtt_client._client.sock.close()
|
||||
except Exception:
|
||||
pass # Already dead or closed
|
||||
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)
|
||||
|
||||
# Now we let the wrapper do its normal cleanup safely
|
||||
try:
|
||||
mqtt_client.close()
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(5)
|
||||
|
||||
# UART
|
||||
uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16)
|
||||
|
||||
# Cooking Cycle
|
||||
cooking_state = None
|
||||
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
|
||||
|
||||
log(f"[Main] Updating cooking state to: {new_state}")
|
||||
if log:
|
||||
log(f"[Main] Updating cooking state to: {new_state}")
|
||||
cooking_state.set_state(new_state)
|
||||
|
||||
# Status LED
|
||||
status_led = RGBLED(red_pin=21, green_pin=19, blue_pin=18)
|
||||
def on_cooking_state_change(state):
|
||||
print(f"[Main] Cooking state changed to: {state.state}")
|
||||
|
||||
# === STATUS LED UPDATE ===
|
||||
BLINK_INTERVAL_MS = 500 # Blink every 500ms
|
||||
|
||||
if state.state == cookingState.CookingStates.IDLE:
|
||||
status_led.color = RGBLED.OFF
|
||||
status_led.blink_off()
|
||||
if state.state == cookingState.CookingStates.COOKING:
|
||||
status_led.color = RGBLED.YELLOW
|
||||
status_led.blink_off()
|
||||
if state.state == cookingState.CookingStates.STIRRING_REQUIRED:
|
||||
status_led.color = RGBLED.ORANGE
|
||||
status_led.blink_on(BLINK_INTERVAL_MS)
|
||||
if state.state == cookingState.CookingStates.DONE:
|
||||
status_led.color = RGBLED.GREEN
|
||||
status_led.blink_off()
|
||||
if state.state == cookingState.CookingStates.ALERT:
|
||||
status_led.color = RGBLED.RED
|
||||
status_led.blink_on(BLINK_INTERVAL_MS)
|
||||
|
||||
|
||||
# --- MAIN APPLICATION THREAD (Core 0) ---
|
||||
print("[Main] Main execution path active.")
|
||||
|
||||
# Temperature sensor setup
|
||||
temperature_sensor_i2c = I2C(0, scl=Pin(25, Pin.IN, Pin.PULL_UP), sda=Pin(26, Pin.IN, Pin.PULL_UP), freq=100000)
|
||||
# Scan to verify the sensor is connected and detected
|
||||
print("Scanning I2C bus...")
|
||||
devices = temperature_sensor_i2c.scan()
|
||||
if 0x5A in devices:
|
||||
print("MLX90614 found at address 0x5A!")
|
||||
else:
|
||||
print("MLX90614 not found. Please check your wiring.")
|
||||
mlx_temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c)
|
||||
|
||||
# --- Launch background worker ---
|
||||
_thread.start_new_thread(mqtt_background_thread, ())
|
||||
time.sleep(2) # Give the thread a moment to initial connect
|
||||
mqtt_hello_sent_timestamp = -config.MQTT_HELLO_INTERVAL
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS)
|
||||
|
||||
while True:
|
||||
# MQTT HELLO sent every x seconds until we get a response from the orchestrator
|
||||
if (orchestrator_id == None and -(mqtt_hello_sent_timestamp - time.time()) > config.MQTT_HELLO_INTERVAL):
|
||||
print("[Main] Attempting to send initial hello to orchestrator...")
|
||||
queue_publish(config.MQTT_TOPIC_HELLO, payloads.mqtt_hello(DEVICE_ID))
|
||||
mqtt_hello_sent_timestamp = time.time()
|
||||
pass
|
||||
def on_mqtt_message(message):
|
||||
global orchestrator_id, cooking_state
|
||||
print("[MQTT] Received message on topic:", message.get("topic"))
|
||||
|
||||
# 1. Listen for incoming UART serial packets from the WROOM board
|
||||
while uart_device.any():
|
||||
command = uart_device.read_as_command()
|
||||
if command:
|
||||
print(f"[Main] Received command from LoRa Board: {command.command_type}")
|
||||
if command.command_type == UARTCommandType.COOKING_STATE_UPDATE:
|
||||
# Handle cooking state update command
|
||||
new_state = command.payload.get("state", None)
|
||||
print(f"[Main] Cooking state update received: {new_state}")
|
||||
on_received_cooking_state_update(new_state)
|
||||
else:
|
||||
print(f"[Main] Unknown command type received: {command.command_type}")
|
||||
payload_data = None
|
||||
try:
|
||||
payload_data = json.loads(message["payload"])
|
||||
except Exception as e:
|
||||
print("[MQTT] Payload parsing warning:", e)
|
||||
|
||||
topic = message.get("topic")
|
||||
|
||||
# 1. Orchestrator Hello Response
|
||||
if (
|
||||
topic == config.MQTT_TOPIC_HELLO
|
||||
and payload_data
|
||||
and payload_data.get("id_microwave") == DEVICE_ID
|
||||
):
|
||||
orchestrator_id = payload_data.get("id_orchestrator")
|
||||
print("[MQTT] Hello response received from orchestrator:", orchestrator_id)
|
||||
try:
|
||||
mqtt_client.unsubscribe(config.MQTT_TOPIC_HELLO)
|
||||
print("[MQTT] 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 (
|
||||
topic == config.MQTT_TOPIC_COOKING
|
||||
and payload_data
|
||||
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
|
||||
)
|
||||
else:
|
||||
# Fallback to reading as a raw string if parsing fails
|
||||
raw_command = uart_device.read()
|
||||
print(f"[Main] Received raw command from WiFi Board: {raw_command}")
|
||||
|
||||
# 2. Example: Send data to the Heltec board every 5 seconds
|
||||
# uart_device.send("Status Check: WiFi Active")
|
||||
time.sleep(1)
|
||||
print("[MQTT] Cooking parameters received:", payload_data)
|
||||
if cookingState:
|
||||
cooking_state = cookingState.CookingState(
|
||||
cook_time=payload_data["cook_time"],
|
||||
power_level=payload_data["power_level"],
|
||||
target_temp=payload_data["target_temp"],
|
||||
)
|
||||
cooking_state.set_state_change_callback(on_cooking_state_change)
|
||||
cooking_state.set_state(cookingState.CookingStates.IDLE)
|
||||
|
||||
if uart_device and UARTCommand:
|
||||
uart_device.send_as_command(
|
||||
UARTCommand(UARTCommandType.COOKING_PARAMS, payload_data)
|
||||
)
|
||||
print("[MQTT] Cooking parameters sent to LoRa board.")
|
||||
|
||||
|
||||
mqtt_client.set_callback(on_mqtt_message)
|
||||
|
||||
|
||||
async def connect_mqtt_async():
|
||||
"""Connects to MQTT safely while memory is clean."""
|
||||
global mqtt_connected
|
||||
mqtt_connected = False
|
||||
|
||||
while True:
|
||||
try:
|
||||
print("[MQTT] Connecting to broker with TLS...")
|
||||
gc.collect()
|
||||
mqtt_client.connect()
|
||||
print("[MQTT] Connected! Subscribing to topics...")
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_COOKING, qos=config.MQTT_QOS)
|
||||
mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS)
|
||||
print("[MQTT] Subscribed successfully!")
|
||||
mqtt_connected = True
|
||||
return
|
||||
except Exception as e:
|
||||
print("[MQTT] Connection failed:", e)
|
||||
sys.print_exception(e)
|
||||
try:
|
||||
mqtt_client.close()
|
||||
except Exception:
|
||||
pass
|
||||
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()
|
||||
|
||||
while True:
|
||||
if mqtt_connected:
|
||||
try:
|
||||
mqtt_client.poll()
|
||||
now = time.time()
|
||||
if now - last_ping >= 15:
|
||||
if mqtt_client._client:
|
||||
mqtt_client._client.ping()
|
||||
last_ping = now
|
||||
except OSError as e:
|
||||
print("[MQTT Task] Socket error encountered during poll/ping:", e)
|
||||
mqtt_connected = False
|
||||
await connect_mqtt_async()
|
||||
|
||||
await asyncio.sleep_ms(30)
|
||||
|
||||
|
||||
async def orchestrator_hello_task():
|
||||
global mqtt_connected
|
||||
while True:
|
||||
if mqtt_connected and orchestrator_id is None:
|
||||
print("[Hello Task] Sending initial hello to orchestrator...")
|
||||
try:
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_HELLO,
|
||||
payloads.mqtt_hello(DEVICE_ID),
|
||||
qos=config.MQTT_QOS,
|
||||
)
|
||||
except OSError as e:
|
||||
print("[Hello Task] Hello publish failed:", e)
|
||||
mqtt_connected = False
|
||||
await connect_mqtt_async()
|
||||
|
||||
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()
|
||||
await asyncio.sleep(10)
|
||||
|
||||
|
||||
# --- MAIN ENTRY POINT ---
|
||||
async def main():
|
||||
print("[Main] Starting application...")
|
||||
|
||||
# STEP 1: Connect MQTT FIRST (while RAM is unfragmented)
|
||||
await connect_mqtt_async()
|
||||
|
||||
# STEP 2: Initialize Hardware & Secondary Modules AFTER connection
|
||||
init_hardware()
|
||||
|
||||
# STEP 3: Launch tasks
|
||||
asyncio.create_task(mqtt_poll_task())
|
||||
asyncio.create_task(orchestrator_hello_task())
|
||||
asyncio.create_task(uart_task())
|
||||
asyncio.create_task(memory_cleanup_task())
|
||||
|
||||
print("[Main] All tasks running concurrently!")
|
||||
|
||||
while True:
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
print("[Main] Program stopped by user.")
|
||||
+308
-57
@@ -1,5 +1,6 @@
|
||||
import sys
|
||||
import time
|
||||
import random
|
||||
|
||||
IS_MICROPYTHON = sys.implementation.name == 'micropython'
|
||||
|
||||
@@ -8,49 +9,253 @@ if IS_MICROPYTHON:
|
||||
from machine import Pin, SPI
|
||||
import ubinascii
|
||||
import ujson as json
|
||||
|
||||
else:
|
||||
import threading
|
||||
import serial
|
||||
import json
|
||||
|
||||
|
||||
# --- BASE RELIABLE LORA DEVICE ---
|
||||
class BaseLoraDevice:
|
||||
"""Base class providing automatic ACK generation, retries, and duplicate filtering."""
|
||||
|
||||
def __init__(self):
|
||||
self.processed_msg_ids = set()
|
||||
self.received_acks = set()
|
||||
self.pending_rx_queue = []
|
||||
self.default_group = 2
|
||||
|
||||
def _generate_msg_id(self):
|
||||
return random.getrandbits(16)
|
||||
|
||||
def _send_ack(self, ack_id):
|
||||
"""Sends an immediate acknowledgement packet back to the sender."""
|
||||
print(f"[ReliableLoRa] -> Triggering ACK send for msg_id: {ack_id}")
|
||||
if IS_MICROPYTHON:
|
||||
time.sleep_ms(10)
|
||||
else:
|
||||
time.sleep(0.01)
|
||||
|
||||
ack_payload = {"_type": "_ack", "_ack_id": ack_id}
|
||||
self.send(ack_payload)
|
||||
|
||||
def _process_incoming_packet(self, packet):
|
||||
"""Internal packet processor: handles ACKs and deduplication."""
|
||||
if not packet or packet.get("raw"):
|
||||
return packet
|
||||
|
||||
data = packet.get("data")
|
||||
if isinstance(data, dict):
|
||||
# 1. Handle incoming ACK response
|
||||
if data.get("_type") == "_ack":
|
||||
ack_id = data.get("_ack_id")
|
||||
print(f"[ReliableLoRa] <- SUCCESSFULLY MATCHED ACK ID: {ack_id}")
|
||||
if ack_id is not None:
|
||||
self.received_acks.add(ack_id)
|
||||
if len(self.received_acks) > 100:
|
||||
self.received_acks.clear()
|
||||
return None # Drop internal protocol message from user queue
|
||||
|
||||
# 2. Handle incoming command expecting an ACK
|
||||
msg_id = data.get("_msg_id")
|
||||
if msg_id is not None:
|
||||
print(f"[ReliableLoRa] <- Received packet with msg_id {msg_id}. Queuing ACK.")
|
||||
self._send_ack(msg_id)
|
||||
|
||||
if msg_id in self.processed_msg_ids:
|
||||
print(f"[ReliableLoRa] Discarding duplicate retry for msg_id {msg_id}")
|
||||
return None # Discard duplicate retry
|
||||
|
||||
self.processed_msg_ids.add(msg_id)
|
||||
if len(self.processed_msg_ids) > 100:
|
||||
self.processed_msg_ids.clear()
|
||||
|
||||
return packet
|
||||
|
||||
def send_reliable(self, payload, max_retries=4, ack_timeout=2.5):
|
||||
"""Sends a payload and retries until an ACK is received or max retries are reached."""
|
||||
lock = getattr(self, 'lock', None)
|
||||
|
||||
if isinstance(payload, dict):
|
||||
payload = dict(payload)
|
||||
else:
|
||||
payload = {"data": payload}
|
||||
|
||||
msg_id = self._generate_msg_id()
|
||||
payload["_msg_id"] = msg_id
|
||||
|
||||
print(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===")
|
||||
|
||||
for attempt in range(max_retries):
|
||||
print(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}")
|
||||
self.send(payload)
|
||||
start_time = time.time()
|
||||
|
||||
while (time.time() - start_time) < ack_timeout:
|
||||
if lock: lock.acquire()
|
||||
try:
|
||||
if msg_id in self.received_acks:
|
||||
self.received_acks.remove(msg_id)
|
||||
print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===")
|
||||
return True
|
||||
finally:
|
||||
if lock: lock.release()
|
||||
|
||||
packet = self.receive_packet(timeout_ms=500)
|
||||
if packet:
|
||||
print(f"[ReliableLoRa] Received raw packet while waiting for ACK: {packet}")
|
||||
if lock: lock.acquire()
|
||||
try:
|
||||
filtered_packet = self._process_incoming_packet(packet)
|
||||
if filtered_packet:
|
||||
self.pending_rx_queue.append(filtered_packet)
|
||||
finally:
|
||||
if lock: lock.release()
|
||||
|
||||
if lock: lock.acquire()
|
||||
try:
|
||||
if msg_id in self.received_acks:
|
||||
self.received_acks.remove(msg_id)
|
||||
print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} after poll ===")
|
||||
return True
|
||||
finally:
|
||||
if lock: lock.release()
|
||||
|
||||
print(f"[ReliableLoRa] Attempt {attempt + 1} timed out waiting for ACK for msg_id {msg_id}")
|
||||
|
||||
print(f"[ReliableLoRa] ERROR: Failed to receive ACK for msg_id {msg_id} after {max_retries} attempts.")
|
||||
return False
|
||||
|
||||
def receive_reliable(self, timeout_ms=1000):
|
||||
"""Receives a packet, automatically sending ACKs and filtering duplicate retries."""
|
||||
if len(self.pending_rx_queue) > 0:
|
||||
return self.pending_rx_queue.pop(0)
|
||||
|
||||
start_time = time.time()
|
||||
timeout_s = timeout_ms / 1000.0
|
||||
|
||||
while True:
|
||||
elapsed = time.time() - start_time
|
||||
remaining_ms = int((timeout_s - elapsed) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
break
|
||||
|
||||
poll_time = max(50, min(remaining_ms, 300))
|
||||
packet = self.receive_packet(timeout_ms=poll_time)
|
||||
if packet:
|
||||
filtered_packet = self._process_incoming_packet(packet)
|
||||
if filtered_packet:
|
||||
return filtered_packet
|
||||
|
||||
return None
|
||||
|
||||
|
||||
if IS_MICROPYTHON:
|
||||
# --- PILOTE SPI DIRECT (ESP32 / Heltec V3) ---
|
||||
class LoraHardwareSPI:
|
||||
class LoraHardwareSPI(BaseLoraDevice):
|
||||
def __init__(self, spi_bus=1, clk=9, mosi=10, miso=11, cs=8, irq=14, rst=12, gpio=13):
|
||||
from sx1262 import SX1262
|
||||
self.lora = SX1262(
|
||||
spi_bus=spi_bus, clk=clk, mosi=mosi, miso=miso,
|
||||
cs=cs, irq=irq, rst=rst, gpio=gpio
|
||||
)
|
||||
self.default_group = 2 # On définit le groupe par défaut ici
|
||||
self.lock = _thread.allocate_lock() # Création du verrou
|
||||
super().__init__()
|
||||
self._pins = {
|
||||
"spi_bus": spi_bus, "clk": clk, "mosi": mosi, "miso": miso,
|
||||
"cs": cs, "irq": irq, "rst": rst, "gpio": gpio
|
||||
}
|
||||
self._cfg = {"freq": 868.1, "bw": 125.0, "sf": 7, "cr": 5, "power": 14}
|
||||
self.lock = _thread.allocate_lock()
|
||||
self.lora = None
|
||||
self.reset_hardware()
|
||||
|
||||
def reset_hardware(self):
|
||||
"""Resets SX1262 hardware and recreates driver instance."""
|
||||
with self.lock:
|
||||
try:
|
||||
irq_pin = Pin(self._pins["irq"], Pin.IN)
|
||||
irq_pin.irq(handler=None)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
rst_pin = Pin(self._pins["rst"], Pin.OUT)
|
||||
rst_pin.value(0)
|
||||
time.sleep_ms(30)
|
||||
rst_pin.value(1)
|
||||
time.sleep_ms(50)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self.lora = None
|
||||
time.sleep_ms(50)
|
||||
|
||||
try:
|
||||
from sx1262 import SX1262
|
||||
new_instance = SX1262(**self._pins)
|
||||
new_instance.begin(
|
||||
freq=self._cfg["freq"], bw=self._cfg["bw"], sf=self._cfg["sf"],
|
||||
cr=self._cfg["cr"], power=self._cfg["power"],
|
||||
useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False
|
||||
)
|
||||
# SyncWord 0x12 = Decimal 18
|
||||
new_instance.setSyncWord(0x12)
|
||||
self.lora = new_instance
|
||||
except Exception as e:
|
||||
print(f"[LoRa SPI] Initialization error: {e}")
|
||||
|
||||
def configure(self, freq=868.1, bw=125.0, sf=7, cr=5, power=14):
|
||||
self.lora.begin(
|
||||
freq=freq, bw=bw, sf=sf, cr=cr, power=power,
|
||||
useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False
|
||||
)
|
||||
self.lora.setSyncWord(0x14)
|
||||
self._cfg = {"freq": freq, "bw": bw, "sf": sf, "cr": cr, "power": power}
|
||||
if self.lora is None:
|
||||
self.reset_hardware()
|
||||
else:
|
||||
with self.lock:
|
||||
try:
|
||||
self.lora.begin(
|
||||
freq=freq, bw=bw, sf=sf, cr=cr, power=power,
|
||||
useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False
|
||||
)
|
||||
self.lora.setSyncWord(0x12)
|
||||
except Exception:
|
||||
self.reset_hardware()
|
||||
|
||||
def send(self, payload, group=None):
|
||||
"""Encode la payload en JSON si nécessaire, et injecte automatiquement l'octet de groupe."""
|
||||
"""Encodes payload into JSON and prepends group byte."""
|
||||
with self.lock:
|
||||
if self.lora is None:
|
||||
return
|
||||
|
||||
if group is None:
|
||||
group = self.default_group
|
||||
|
||||
# Si c'est un dictionnaire ou une liste, on le convertit en JSON textuel
|
||||
if isinstance(payload, (dict, list)):
|
||||
payload = json.dumps(payload)
|
||||
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode('utf-8')
|
||||
|
||||
# Insertion automatique de l'octet de groupe au tout début de la trame physique
|
||||
paquet_physique = bytes([group]) + payload
|
||||
self.lora.send(paquet_physique)
|
||||
try:
|
||||
self.lora.send(paquet_physique)
|
||||
except Exception as e:
|
||||
print(f"[LoRa SPI] Send error: {e}")
|
||||
|
||||
def receive_packet(self, timeout_ms=1000):
|
||||
"""Écoute, nettoie, extrait le groupe, gère le HEX et parse le JSON."""
|
||||
def receive_packet(self, timeout_ms=500):
|
||||
"""Listens on SPI bus with auto-detection for JSON vs. Grouped headers."""
|
||||
with self.lock:
|
||||
data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms)
|
||||
if state == 0 and len(data) > 1:
|
||||
group = data[0]
|
||||
payload_brute = data[1:].strip(b'\x00 \r\n\t')
|
||||
if self.lora is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms)
|
||||
except Exception as e:
|
||||
print(f"[LoRa SPI] Recv error caught: {e}")
|
||||
return None
|
||||
|
||||
if state == 0 and data is not None and len(data) > 0:
|
||||
if data[0] in (0x7B, 0x5B): # Starts with '{' or '['
|
||||
group = self.default_group
|
||||
payload_brute = data.strip(b'\x00 \r\n\t')
|
||||
elif len(data) > 1:
|
||||
group = data[0]
|
||||
payload_brute = data[1:].strip(b'\x00 \r\n\t')
|
||||
else:
|
||||
return None
|
||||
|
||||
try:
|
||||
text = payload_brute.decode('utf-8').strip('\x00 \r\n\t')
|
||||
@@ -76,13 +281,10 @@ if IS_MICROPYTHON:
|
||||
return None
|
||||
|
||||
else:
|
||||
import threading
|
||||
import serial
|
||||
import json
|
||||
|
||||
# --- PILOTE SÉRIE (Raspberry Pi / Dragino LA66) ---
|
||||
class LoraSerialAT:
|
||||
class LoraSerialAT(BaseLoraDevice):
|
||||
def __init__(self, port):
|
||||
super().__init__()
|
||||
self.port = port
|
||||
self.ser = serial.Serial(
|
||||
port=self.port,
|
||||
@@ -95,37 +297,60 @@ else:
|
||||
self.ser.reset_input_buffer()
|
||||
self.ser.reset_output_buffer()
|
||||
self.lock = threading.Lock()
|
||||
def configure(self, **kwargs):
|
||||
pass
|
||||
|
||||
# Initial configuration
|
||||
self.configure(freq=868.1, sf=7, bw=125)
|
||||
|
||||
def send(self, payload):
|
||||
"""Encode automatiquement la payload en HEX pour l'envoi via la clé."""
|
||||
def _send_at_cmd(self, cmd, wait_time=0.15):
|
||||
"""Helper to send AT command and purge response buffer."""
|
||||
self.ser.write(f"{cmd}\r\n".encode('utf-8'))
|
||||
time.sleep(wait_time)
|
||||
resp = ""
|
||||
while self.ser.in_waiting > 0:
|
||||
resp += self.ser.readline().decode('utf-8', errors='ignore')
|
||||
return resp
|
||||
|
||||
def configure(self, freq=868.1, sf=7, bw=125):
|
||||
"""Configures LA66 frequency, SF, BW, SyncWord, CRC, and continuous RX mode."""
|
||||
with self.lock:
|
||||
freq_hz = int(freq * 1000000)
|
||||
bw_code = 0 if bw == 125 else 1
|
||||
|
||||
# Parameters: Freq, SF, BW, CR(0=4/5), Preamble(8), Header(1=Explicit), CRC(1=ON), IQ(0=Standard), NetMode(0=P2P), Power(14), SyncWord(18=0x12), Format(0), Type(1)
|
||||
at_cfg_cmd = f"AT+CFG={freq_hz},{sf},{bw_code},0,8,1,1,0,0,14,18,0,1"
|
||||
self._send_at_cmd(at_cfg_cmd, wait_time=0.2)
|
||||
|
||||
# Fallback standalone commands
|
||||
self._send_at_cmd("AT+SYNCWORD=18", wait_time=0.1)
|
||||
self._send_at_cmd("AT+PRECV=65535", wait_time=0.1)
|
||||
self.ser.reset_input_buffer()
|
||||
|
||||
def send(self, payload, group=None):
|
||||
"""Encodes payload into HEX AT command and re-enables continuous RX."""
|
||||
with self.lock:
|
||||
if group is None:
|
||||
group = self.default_group
|
||||
|
||||
if isinstance(payload, (dict, list)):
|
||||
payload = json.dumps(payload)
|
||||
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode('utf-8')
|
||||
|
||||
hex_payload = payload.hex()
|
||||
paquet_physique = bytes([group]) + payload
|
||||
hex_payload = paquet_physique.hex()
|
||||
self.ser.reset_input_buffer()
|
||||
|
||||
# La clé ajoute d'elle-même l'octet de groupe configuré dans ses registres
|
||||
cmd = f"AT+SEND=1,{hex_payload},1,3\r\n"
|
||||
# print(f"RPI : Envoi de la commande HEX -> AT+SEND=1,[HEX_DATA],1,3")
|
||||
self.ser.write(cmd.encode('utf-8'))
|
||||
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), ' | ')}")
|
||||
|
||||
time.sleep(0.2)
|
||||
response = ""
|
||||
start_wait = time.time()
|
||||
while (time.time() - start_wait) < 1.5:
|
||||
if self.ser.in_waiting > 0:
|
||||
response += self.ser.readline().decode('utf-8', errors='ignore')
|
||||
time.sleep(0.05)
|
||||
|
||||
# print(f"[RPI LA66 TX STATUS] :\n{response.strip()}")
|
||||
# Re-enable continuous receive mode after transmission completes
|
||||
self._send_at_cmd("AT+PRECV=65535", wait_time=0.05)
|
||||
|
||||
def receive_packet(self, timeout_ms=5000):
|
||||
def receive_packet(self, timeout_ms=500):
|
||||
"""Reads incoming serial lines from LA66 stick with robust format parsing."""
|
||||
with self.lock:
|
||||
start_time = time.time()
|
||||
timeout_s = timeout_ms / 1000.0
|
||||
@@ -136,18 +361,38 @@ else:
|
||||
if line:
|
||||
payload_bytes = None
|
||||
|
||||
if "(HEX:)" in line:
|
||||
# Robust parsing for LA66 response variants (+RECV:, +RCV=, +DRX:, HEX:, Data:)
|
||||
if "+RECV:" in line:
|
||||
parts = line.split("+RECV:")[1].strip().split(",")
|
||||
hex_str = parts[2].strip() if len(parts) >= 3 else parts[0].strip()
|
||||
try: payload_bytes = bytes.fromhex(hex_str)
|
||||
except ValueError: pass
|
||||
elif "+RCV=" in line:
|
||||
parts = line.split("+RCV=")[1].strip().split(",")
|
||||
if len(parts) >= 4:
|
||||
try: payload_bytes = bytes.fromhex(parts[3].strip())
|
||||
except ValueError: pass
|
||||
elif "+DRX:" in line:
|
||||
parts = line.split("+DRX:")[1].strip().split(",")
|
||||
if len(parts) >= 2:
|
||||
try: payload_bytes = bytes.fromhex(parts[1].strip())
|
||||
except ValueError: pass
|
||||
elif "(HEX:)" in line:
|
||||
hex_part = line.split("(HEX:)")[1].strip().replace(" ", "")
|
||||
try:
|
||||
payload_bytes = bytes.fromhex(hex_part)
|
||||
except ValueError:
|
||||
pass
|
||||
try: payload_bytes = bytes.fromhex(hex_part)
|
||||
except ValueError: pass
|
||||
elif "Data:" in line:
|
||||
payload_bytes = line.split("Data:")[1].strip().encode('utf-8')
|
||||
|
||||
if payload_bytes and len(payload_bytes) > 1:
|
||||
group = payload_bytes[0]
|
||||
payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t')
|
||||
if payload_bytes and len(payload_bytes) > 0:
|
||||
if payload_bytes[0] in (0x7B, 0x5B):
|
||||
group = self.default_group
|
||||
payload_clean = payload_bytes.strip(b'\x00 \r\n\t')
|
||||
elif len(payload_bytes) > 1:
|
||||
group = payload_bytes[0]
|
||||
payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t')
|
||||
else:
|
||||
continue
|
||||
|
||||
try:
|
||||
text = payload_clean.decode('utf-8').strip('\x00 \r\n\t')
|
||||
@@ -180,4 +425,10 @@ def get_lora_device(port_or_pins=None):
|
||||
return LoraHardwareSPI(**pins)
|
||||
else:
|
||||
port = port_or_pins if port_or_pins else "/dev/serial/by-id/usb-Silicon_Labs_CP2102_USB_to_UART_Bridge_Controller_0001-if00-port0"
|
||||
return LoraSerialAT(port)
|
||||
return LoraSerialAT(port)
|
||||
|
||||
|
||||
class LoraCommands:
|
||||
PING = "ping"
|
||||
COOKING_STATE_UPDATE = "cooking_state_update"
|
||||
TOGGLE_PAUSE = "toggle_pause"
|
||||
+96
-35
@@ -11,13 +11,10 @@ try:
|
||||
except ImportError:
|
||||
try:
|
||||
from umqtt.simple import MQTTClient as _MQTTClient
|
||||
import _thread
|
||||
import gc
|
||||
BACKEND_NAME = "umqtt.simple"
|
||||
IS_MICROPYTHON = True
|
||||
# except ImportError:
|
||||
# try:
|
||||
# from umqtt.robust import MQTTClient as _MQTTClient
|
||||
# BACKEND_NAME = "umqtt.robust"
|
||||
# IS_MICROPYTHON = True
|
||||
except ImportError as exc:
|
||||
raise ImportError("No MQTT client found. Expected paho.mqtt or umqtt.") from exc
|
||||
|
||||
@@ -79,6 +76,11 @@ class BrokerClient:
|
||||
self._client = None
|
||||
self._callback = None
|
||||
self._messages = []
|
||||
self._cadata = None # Cache cert bytes to prevent heap fragmentation
|
||||
|
||||
# Thread safety lock for MicroPython socket reads/writes
|
||||
if IS_MICROPYTHON:
|
||||
self._lock = _thread.allocate_lock()
|
||||
|
||||
def set_callback(self, callback):
|
||||
self._callback = callback
|
||||
@@ -104,16 +106,24 @@ class BrokerClient:
|
||||
return self._client
|
||||
|
||||
if IS_MICROPYTHON:
|
||||
gc.collect() # Clean Python heap before importing/allocating SSL
|
||||
import ssl
|
||||
ssl_params = self.ssl_params
|
||||
|
||||
if self.use_tls and ssl_params is None:
|
||||
# MicroPython uses context-less structures.
|
||||
# If your CA is self-signed, validation can fail without a valid hostname match.
|
||||
# OPTION A: If broker uses 'require_certificate false' and self-signed certs:
|
||||
# Do NOT pass cadata when cert_reqs is CERT_NONE to save ~20KB of C-DRAM
|
||||
ssl_params = {
|
||||
"cert_reqs": ssl.CERT_NONE, # Temporarily change to NONE to test if validation is the culprit
|
||||
"cadata": _read_file_bytes(self.cafile)
|
||||
"cert_reqs": ssl.CERT_NONE,
|
||||
"server_hostname": self.host
|
||||
}
|
||||
|
||||
# OPTION B: If strict CA validation IS required, load cadata ONLY with CERT_REQUIRED:
|
||||
# ssl_params = {
|
||||
# "cert_reqs": ssl.CERT_REQUIRED,
|
||||
# "cadata": _read_file_bytes(self.cafile),
|
||||
# "server_hostname": self.host
|
||||
# }
|
||||
|
||||
client = _MQTTClient(
|
||||
self.client_id or "smartWave-client",
|
||||
@@ -150,19 +160,35 @@ class BrokerClient:
|
||||
return self._client
|
||||
|
||||
def connect(self):
|
||||
client = self.open()
|
||||
if IS_MICROPYTHON:
|
||||
client.connect()
|
||||
return client
|
||||
gc.collect() # Force C & Python memory cleanup right before TLS handshake
|
||||
|
||||
client.connect(self.host, self.port, self.keepalive)
|
||||
return client
|
||||
if self._client is not None:
|
||||
self.close()
|
||||
|
||||
client = self.open()
|
||||
|
||||
try:
|
||||
if IS_MICROPYTHON:
|
||||
gc.collect() # Sweep memory right before umqtt calls ssl.wrap_socket()
|
||||
with self._lock:
|
||||
client.connect()
|
||||
return client
|
||||
|
||||
client.connect(self.host, self.port, self.keepalive)
|
||||
return client
|
||||
except Exception as e:
|
||||
print("MQTT connection failed, closing client and releasing memory.")
|
||||
print("Exception:", e)
|
||||
self.close()
|
||||
raise
|
||||
|
||||
def publish(self, topic, payload, qos=2, retain=False):
|
||||
client = self.open()
|
||||
payload_bytes = _ensure_bytes(payload)
|
||||
if IS_MICROPYTHON:
|
||||
return client.publish(topic, payload_bytes, retain=retain, qos=qos)
|
||||
with self._lock:
|
||||
return client.publish(topic, payload_bytes, retain=retain, qos=qos)
|
||||
|
||||
if isinstance(topic, bytes):
|
||||
topic = topic.decode('utf-8')
|
||||
@@ -172,8 +198,9 @@ class BrokerClient:
|
||||
def subscribe(self, topic, qos=2):
|
||||
client = self.open()
|
||||
if IS_MICROPYTHON:
|
||||
client.set_callback(self._on_micropython_message)
|
||||
return client.subscribe(topic, qos=qos)
|
||||
with self._lock:
|
||||
client.set_callback(self._on_micropython_message)
|
||||
return client.subscribe(topic, qos=qos)
|
||||
|
||||
if isinstance(topic, bytes):
|
||||
topic = topic.decode('utf-8')
|
||||
@@ -184,27 +211,43 @@ class BrokerClient:
|
||||
client = self.open()
|
||||
if IS_MICROPYTHON:
|
||||
import struct
|
||||
# Ensure the topic is bytes for writing to the socket
|
||||
import time
|
||||
topic_bytes = topic if isinstance(topic, bytes) else topic.encode('utf-8')
|
||||
|
||||
# 1. Build the MQTT unsubscribe packet header
|
||||
# 1. Increment and lock the PID for THIS specific request
|
||||
client.pid = (client.pid % 65535) + 1
|
||||
sent_pid = client.pid # <-- Store local copy
|
||||
|
||||
# 2. Construct UNSUBSCRIBE packet
|
||||
rem_len = 2 + 2 + len(topic_bytes)
|
||||
pkt = bytearray(b"\xa2\0\0\0")
|
||||
client.pid += 1
|
||||
struct.pack_into("!BH", pkt, 1, rem_len, sent_pid)
|
||||
|
||||
# Packet length is: 2 bytes (PID) + 2 bytes (topic length indicator) + topic string length
|
||||
struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic_bytes), client.pid)
|
||||
|
||||
# 2. Write the packet to the socket
|
||||
# 3. Write packet to socket
|
||||
client.sock.write(pkt)
|
||||
client._send_str(topic_bytes)
|
||||
|
||||
# 3. Wait for the UNSUBACK confirmation frame (0xB0) from the broker
|
||||
while True:
|
||||
# 4. Wait for UNSUBACK (0xB0)
|
||||
start = time.time()
|
||||
while time.time() - start < 3:
|
||||
op = client.wait_msg()
|
||||
if op == 0xB0:
|
||||
resp = client.sock.read(3)
|
||||
assert resp[1] == pkt[2] and resp[2] == pkt[3]
|
||||
resp = bytearray(3)
|
||||
read_bytes = 0
|
||||
while read_bytes < 3:
|
||||
chunk = client.sock.read(3 - read_bytes)
|
||||
if chunk:
|
||||
resp[read_bytes:read_bytes + len(chunk)] = chunk
|
||||
read_bytes += len(chunk)
|
||||
else:
|
||||
time.sleep_ms(10)
|
||||
|
||||
# Compare against sent_pid instead of client.pid
|
||||
resp_pid = (resp[1] << 8) | resp[2]
|
||||
if resp_pid != sent_pid:
|
||||
print(f"[MQTT] UNSUBACK PID mismatch (expected {sent_pid}, got {resp_pid})")
|
||||
return client
|
||||
|
||||
return client
|
||||
|
||||
if isinstance(topic, bytes):
|
||||
@@ -219,14 +262,16 @@ class BrokerClient:
|
||||
if self._client is None:
|
||||
return None
|
||||
if IS_MICROPYTHON:
|
||||
return self._client.check_msg()
|
||||
with self._lock:
|
||||
return self._client.check_msg()
|
||||
return self._client.loop(timeout=timeout)
|
||||
|
||||
def wait(self):
|
||||
if self._client is None:
|
||||
return None
|
||||
if IS_MICROPYTHON:
|
||||
return self._client.wait_msg()
|
||||
with self._lock:
|
||||
return self._client.wait_msg()
|
||||
return self._client.loop_forever()
|
||||
|
||||
def get_message(self):
|
||||
@@ -235,13 +280,29 @@ class BrokerClient:
|
||||
return self._messages.pop(0)
|
||||
|
||||
def close(self):
|
||||
"""Safely clean up socket context without causing ESP32 C panics."""
|
||||
if self._client is None:
|
||||
return
|
||||
try:
|
||||
self._client.disconnect()
|
||||
except Exception:
|
||||
pass
|
||||
self._client = None
|
||||
|
||||
if IS_MICROPYTHON:
|
||||
with self._lock:
|
||||
try:
|
||||
if hasattr(self._client, "sock") and self._client.sock:
|
||||
self._client.sock.close()
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
if hasattr(self._client, "sock"):
|
||||
self._client.sock = None
|
||||
self._client = None
|
||||
gc.collect() # Immediately reclaim freed socket & mbedTLS RAM
|
||||
else:
|
||||
try:
|
||||
self._client.disconnect()
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
self._client = None
|
||||
|
||||
def __enter__(self):
|
||||
self.connect()
|
||||
|
||||
Reference in New Issue
Block a user