logan/drb-c2-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement heartbeat

Browse Source
This commit is contained in:
Logan Cusano
2025-12-28 14:23:34 -05:00
parent bfd3b5b2c5
commit 71775ff33a
1 changed files with 26 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
app/c2_main.py
View File

@@ -54,6 +54,14 @@ def on_message(client, userdata, msg):
if MAIN_LOOP: if MAIN_LOOP:
asyncio.run_coroutine_threadsafe(handle_message(msg), MAIN_LOOP) asyncio.run_coroutine_threadsafe(handle_message(msg), MAIN_LOOP)
async def update_last_seen(node_id):
"""Generic helper to update the timestamp on any contact."""
try:
doc_ref = db.collection("nodes").document(node_id)
await async_firestore(doc_ref.set, {"last_seen": datetime.utcnow()}, merge=True)
except Exception as e:
print(f"Failed to update heartbeat for {node_id}: {e}")
async def handle_message(msg): async def handle_message(msg):
topic_parts = msg.topic.split('/') topic_parts = msg.topic.split('/')
if len(topic_parts) < 3: return if len(topic_parts) < 3: return
@@ -65,37 +73,46 @@ async def handle_message(msg):
payload = json.loads(msg.payload.decode()) payload = json.loads(msg.payload.decode())
timestamp = datetime.utcnow() timestamp = datetime.utcnow()
# 1. ALWAYS update last_seen if we hear from the node
await update_last_seen(node_id)
if event_type == "checkin": if event_type == "checkin":
print(f"Processing checkin for {node_id}...") # This now receives the periodic heartbeat
print(f"Heartbeat/Checkin from {node_id}")
data = { data = {
"node_id": node_id, "node_id": node_id,
"last_seen": timestamp, "last_seen": timestamp,
"status": payload.get("status", "online"), "status": payload.get("status", "online"),
"active_system": payload.get("active_system"), "active_system": payload.get("active_system"),
"available_systems": payload.get("available_systems", []), "available_systems": payload.get("available_systems", []),
"config": payload,
"radio_state": "active" if payload.get("is_listening") else "idle" "radio_state": "active" if payload.get("is_listening") else "idle"
} }
doc_ref = db.collection("nodes").document(node_id) doc_ref = db.collection("nodes").document(node_id)
print(f"Writing to Firestore: {doc_ref.path} in DB {FIRESTORE_DB_ID}")
await async_firestore(doc_ref.set, data, merge=True) await async_firestore(doc_ref.set, data, merge=True)
print(f"Successfully updated checkin for {node_id}")
ACTIVE_NODES_CACHE[node_id] = data ACTIVE_NODES_CACHE[node_id] = data
elif event_type == "status": elif event_type == "status":
print(f"Processing status update for {node_id}...") # Handle explicit Offline messages (LWT or clean shutdown)
print(f"Status update for {node_id}: {payload.get('status')}")
status = payload.get("status") status = payload.get("status")
data = {"status": status, "last_seen": timestamp}
# If offline, maybe clear active system?
if status == "offline":
data["radio_state"] = "unknown"
doc_ref = db.collection("nodes").document(node_id) doc_ref = db.collection("nodes").document(node_id)
print(f"Writing to Firestore: {doc_ref.path} in DB {FIRESTORE_DB_ID}") await async_firestore(doc_ref.set, data, merge=True)
await async_firestore(doc_ref.set, {"status": status, "last_seen": timestamp}, merge=True)
print(f"Successfully updated status for {node_id}")
if node_id in ACTIVE_NODES_CACHE: if node_id in ACTIVE_NODES_CACHE:
ACTIVE_NODES_CACHE[node_id]["status"] = status ACTIVE_NODES_CACHE[node_id].update(data)
except Exception as e: except Exception as e:
print(f"Error processing MQTT message from {node_id}: {e}") print(f"Error processing MQTT message from {node_id}: {e}")
traceback.print_exc() traceback.print_exc()
# MQTT Setup # MQTT Setup
mqtt_client = mqtt.Client(client_id=C2_ID) mqtt_client = mqtt.Client(client_id=C2_ID)
mqtt_client.on_connect = on_connect mqtt_client.on_connect = on_connect