logan/drb-edge-node
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Implement Metadata Watcher #1

Merged
logan merged 9 commits from metadata-watcher into main 2025-12-29 19:04:07 -05:00
Conversation 0 Commits 9 Files Changed 6 +9 -2
1 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit ca984be293 - Show all commits

11
app/node_main.py
View File

@@ -212,6 +212,7 @@ async def mqtt_lifecycle_manager():
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
LOGGER.debug(f"Response from OP25 API: {data}")
current_tgid = 0 current_tgid = 0
current_meta = {} current_meta = {}
@@ -246,9 +247,11 @@ async def mqtt_lifecycle_manager():
if current_tgid != last_tgid: if current_tgid != last_tgid:
if last_tgid != 0: if last_tgid != 0:
LOGGER.debug(f"Switching TGID: {last_tgid} -> {current_tgid}")
payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata} payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata}
client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0)
LOGGER.debug(f"Call Start: TGID {current_tgid} ({current_meta.get('alpha_tag')})")
payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_start", "metadata": current_meta} payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_start", "metadata": current_meta}
client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0)
last_tgid = current_tgid last_tgid = current_tgid
@@ -256,16 +259,20 @@ async def mqtt_lifecycle_manager():
elif last_tgid != 0: elif last_tgid != 0:
if potential_end_time is None: if potential_end_time is None:
LOGGER.debug(f"Signal lost for TGID {last_tgid}. Starting debounce.")
potential_end_time = now potential_end_time = now
elif (now - potential_end_time).total_seconds() > DEBOUNCE_SECONDS: elif (now - potential_end_time).total_seconds() > DEBOUNCE_SECONDS:
LOGGER.debug(f"Call End (Debounce expired): TGID {last_tgid}")
payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata} payload = {"node_id": NODE_ID, "timestamp": now.isoformat(), "event": "call_end", "metadata": last_metadata}
client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0) client.publish(f"nodes/{NODE_ID}/metadata", json.dumps(payload), qos=0)
last_tgid = 0 last_tgid = 0
last_metadata = {} last_metadata = {}
potential_end_time = None potential_end_time = None
else:
LOGGER.debug(f"OP25 API returned status: {response.status_code}")
except Exception: except Exception as e:
pass # OP25 might be restarting or busy LOGGER.warning(f"Metadata watcher error: {e}")
await asyncio.sleep(0.25) await asyncio.sleep(0.25)