193 lines
7.6 KiB
Python
193 lines
7.6 KiB
Python
import _thread
|
|
import select
|
|
from machine import Pin, I2C
|
|
from sensors import temperature_sensor
|
|
from shared import get_mqtt_client, get_uart, config, payloads
|
|
import time
|
|
import ujson as json
|
|
import sys
|
|
|
|
# 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))
|
|
|
|
# --- READ DEVICE ID ---
|
|
try:
|
|
with open("device_id.txt", "r") as f:
|
|
DEVICE_ID = f.read().strip()
|
|
except Exception:
|
|
DEVICE_ID = "ESP32_Inconnu"
|
|
|
|
# --- 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,
|
|
cafile=MQTT_CA_FILE,
|
|
keepalive=config.MQTT_KEEPALIVE,
|
|
)
|
|
|
|
global orchestrator_id
|
|
orchestrator_id = None
|
|
def on_mqtt_message(message):
|
|
print("[MQTT Thread] Received message:", message)
|
|
|
|
# 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
|
|
|
|
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_seconds" in payload_data:
|
|
print("[MQTT Thread] Cooking sensors init received from the orchestrator")
|
|
obj_temp = temperature_sensor.read_object_temp()
|
|
amb_temp = 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)
|
|
# Here you would handle the cooking parameters, e.g., start a cooking process
|
|
# For now, we just print them
|
|
|
|
print("[MQTT Thread] Message processing complete.")
|
|
|
|
mqtt_client.set_callback(on_mqtt_message)
|
|
|
|
|
|
def mqtt_background_thread():
|
|
"""Background MQTT worker handling ALL socket operations safely."""
|
|
print("[Thread] Background MQTT worker started.")
|
|
|
|
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
|
|
|
|
# Now we let the wrapper do its normal cleanup safely
|
|
try:
|
|
mqtt_client.close()
|
|
except Exception:
|
|
pass
|
|
time.sleep(5)
|
|
|
|
# --- UART BACKGROUND THREAD ---
|
|
def uart_background_thread():
|
|
"""Background UART worker handling all serial operations safely."""
|
|
print("[Thread] Background UART worker started.")
|
|
|
|
uart_device = get_uart(uart_id=2, tx_pin=17, rx_pin=16)
|
|
|
|
while True:
|
|
try:
|
|
# 1. Check for incoming messages from the Heltec board
|
|
while uart_device.any():
|
|
incoming_msg = uart_device.read()
|
|
print(f"[Thread] Received from esp-lora over UART: {incoming_msg}")
|
|
|
|
# 2. Example: Send data to the Heltec board every 5 seconds
|
|
# uart_device.send("Status Check: WiFi Active")
|
|
|
|
time.sleep(5) # Fast responsive polling loop for local UART
|
|
|
|
except Exception as e:
|
|
print("[Thread] UART error encountered:", e)
|
|
time.sleep(5)
|
|
|
|
# --- Launch background worker ---
|
|
_thread.start_new_thread(mqtt_background_thread, ())
|
|
_thread.start_new_thread(uart_background_thread, ())
|
|
|
|
|
|
# --- MAIN APPLICATION THREAD (Core 0) ---
|
|
print("[Main] Main execution path active.")
|
|
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)
|
|
|
|
# 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.")
|
|
temperature_sensor = temperature_sensor.MLX90614(temperature_sensor_i2c)
|
|
|
|
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
|
|
|
|
# 2. Example: Send data to the Heltec board every 5 seconds
|
|
# uart_device.send("Status Check: WiFi Active")
|
|
time.sleep(1) |