MQTT hello is saved in database
This commit is contained in:
+45
-8
@@ -6,12 +6,40 @@ import asyncio
|
|||||||
import requests
|
import requests
|
||||||
|
|
||||||
from orchestrateur.sensors import gps
|
from orchestrateur.sensors import gps
|
||||||
from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads
|
from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads, db
|
||||||
from shared.logging import log
|
from shared.logging import log
|
||||||
from shared.cookingState import CookingStates
|
from shared.cookingState import CookingStates
|
||||||
from shared.lora_device import LoraCommands
|
from shared.lora_device import LoraCommands
|
||||||
from sensors import ultrasonicRanger, temp_hum, button, camera
|
from sensors import ultrasonicRanger, temp_hum, button, camera
|
||||||
|
|
||||||
|
# --- DB SETUP ---
|
||||||
|
DB_PATH = "orchestrateur/db.sqlite"
|
||||||
|
|
||||||
|
def init_db():
|
||||||
|
"""Ensures the connected_components table exists on startup."""
|
||||||
|
sql = """
|
||||||
|
CREATE TABLE IF NOT EXISTS connected_components (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
type TEXT,
|
||||||
|
timestamp INTEGER
|
||||||
|
);
|
||||||
|
"""
|
||||||
|
db.execute(DB_PATH, sql)
|
||||||
|
print(f"[DB] Initialized database table at {DB_PATH}")
|
||||||
|
|
||||||
|
def save_connected_component(component_id: str, component_type: str):
|
||||||
|
"""Upserts component information into the database (blocking sync worker)."""
|
||||||
|
current_time = int(time.time())
|
||||||
|
sql = """
|
||||||
|
INSERT INTO connected_components (id, type, timestamp)
|
||||||
|
VALUES (?, ?, ?)
|
||||||
|
ON CONFLICT(id) DO UPDATE SET
|
||||||
|
type = excluded.type,
|
||||||
|
timestamp = excluded.timestamp;
|
||||||
|
"""
|
||||||
|
db.execute(DB_PATH, sql, (str(component_id), str(component_type), current_time))
|
||||||
|
print(f"[DB] Component saved/updated -> ID: {component_id}, Type: {component_type}, Timestamp: {current_time}")
|
||||||
|
|
||||||
# --- Read Unique Device ID ---
|
# --- Read Unique Device ID ---
|
||||||
def get_device_id():
|
def get_device_id():
|
||||||
for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]:
|
for path in ["device_id.txt", "/home/pi/SmartWave/orchestrateur/device_id.txt"]:
|
||||||
@@ -287,13 +315,19 @@ async def process_messages_task():
|
|||||||
|
|
||||||
if topic == hello_topic:
|
if topic == hello_topic:
|
||||||
if data.get("id_orchestrator") != DEVICE_ID:
|
if data.get("id_orchestrator") != DEVICE_ID:
|
||||||
mw_id = data.get("id_microwave")
|
component_id = data.get("id_microwave")
|
||||||
print(f"[MQTT] Hello from {mw_id}. Sending ACK.")
|
component_type = data.get("type", deviceTypes.DEVICE_TYPES["MICROWAVE"])
|
||||||
mqtt_client.publish(
|
|
||||||
config.MQTT_TOPIC_HELLO,
|
if component_id:
|
||||||
payloads.mqtt_hello_ack(DEVICE_ID, mw_id),
|
print(f"[MQTT] Hello received from '{component_id}' ({component_type}). Updating DB & sending ACK.")
|
||||||
qos=config.MQTT_QOS
|
# Offload DB insertion to async thread execution pool
|
||||||
)
|
await asyncio.to_thread(save_connected_component, component_id, component_type)
|
||||||
|
|
||||||
|
mqtt_client.publish(
|
||||||
|
config.MQTT_TOPIC_HELLO,
|
||||||
|
payloads.mqtt_hello_ack(DEVICE_ID, component_id),
|
||||||
|
qos=config.MQTT_QOS
|
||||||
|
)
|
||||||
|
|
||||||
elif topic == sensor_topic:
|
elif topic == sensor_topic:
|
||||||
mw_id = str(data.get("id_microwave"))
|
mw_id = str(data.get("id_microwave"))
|
||||||
@@ -372,6 +406,9 @@ async def main():
|
|||||||
global async_event_queue
|
global async_event_queue
|
||||||
print("🚀 Orchestrateur Asyncio prêt. Lancement des tâches...")
|
print("🚀 Orchestrateur Asyncio prêt. Lancement des tâches...")
|
||||||
|
|
||||||
|
# Initialize SQLite database table
|
||||||
|
init_db()
|
||||||
|
|
||||||
async_event_queue = asyncio.Queue()
|
async_event_queue = asyncio.Queue()
|
||||||
|
|
||||||
await asyncio.gather(
|
await asyncio.gather(
|
||||||
|
|||||||
+3
-1
@@ -1,4 +1,5 @@
|
|||||||
from time import time
|
from time import time
|
||||||
|
from shared.deviceTypes import DEVICE_TYPES
|
||||||
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -16,7 +17,8 @@ def as_json(data):
|
|||||||
|
|
||||||
def mqtt_hello(id_microwave):
|
def mqtt_hello(id_microwave):
|
||||||
return as_json({
|
return as_json({
|
||||||
"id_microwave": id_microwave
|
"id_microwave": id_microwave,
|
||||||
|
"type": DEVICE_TYPES["MICROWAVE"]
|
||||||
})
|
})
|
||||||
|
|
||||||
def mqtt_hello_ack(id_orchestrator, id_microwave):
|
def mqtt_hello_ack(id_orchestrator, id_microwave):
|
||||||
|
|||||||
Reference in New Issue
Block a user