Implement Metadata Watcher #1

Merged
logan merged 9 commits from metadata-watcher into main 2025-12-29 19:04:07 -05:00
2 changed files with 34 additions and 29 deletions
Showing only changes of commit b8ee991192 - Show all commits

View File

@@ -186,12 +186,16 @@ async def mqtt_lifecycle_manager():
async def metadata_watcher(): async def metadata_watcher():
""" """
Polls OP25 HTTP terminal for metadata and publishes events to MQTT. Polls OP25 HTTP terminal for metadata and publishes events to MQTT.
Corrected to use the POST-based command API found in the HAR capture.
""" """
last_tgid = 0 last_tgid = 0
last_metadata = {} last_metadata = {}
potential_end_time = None potential_end_time = None
DEBOUNCE_SECONDS = 2.5 DEBOUNCE_SECONDS = 2.5
OP25_DATA_URL = "http://127.0.0.1:8081/data.json" OP25_DATA_URL = "http://127.0.0.1:8081/"
# This is the specific payload the OP25 web interface uses [cite: 45562, 45563]
COMMAND_PAYLOAD = [{"command": "update", "arg1": 0, "arg2": 0}]
while True: while True:
if not MQTT_CONNECTED: if not MQTT_CONNECTED:
@@ -199,9 +203,12 @@ async def mqtt_lifecycle_manager():
continue continue
try: try:
# Run blocking request in executor to avoid blocking the asyncio loop # Run blocking POST request in executor
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
response = await loop.run_in_executor(None, lambda: requests.get(OP25_DATA_URL, timeout=0.5)) response = await loop.run_in_executor(
None,
lambda: requests.post(OP25_DATA_URL, json=COMMAND_PAYLOAD, timeout=0.5)
)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
@@ -209,10 +216,17 @@ async def mqtt_lifecycle_manager():
current_tgid = 0 current_tgid = 0
current_meta = {} current_meta = {}
# Handle multi_rx list or single dict structure # The response is an array of update objects
if isinstance(data, list): for item in data:
for ch in data: if item.get("json_type") == "channel_update":
t = ch.get("tgid", 0) # The terminal provides channel info keyed by channel index (e.g., "0")
# We look for the first channel that has an active TGID
for key in item:
if key.isdigit():
ch = item[key]
t = ch.get("tgid")
# OP25 returns null or 0 when no talkgroup is active
if t and int(t) > 0: if t and int(t) > 0:
current_tgid = int(t) current_tgid = int(t)
current_meta = { current_meta = {
@@ -222,29 +236,19 @@ async def mqtt_lifecycle_manager():
"sysname": str(ch.get("system", "")).strip() "sysname": str(ch.get("system", "")).strip()
} }
break break
elif isinstance(data, dict): if current_tgid: break
t = data.get("tgid", 0)
if t and int(t) > 0:
current_tgid = int(t)
current_meta = {
"tgid": str(t),
"alpha_tag": str(data.get("tag", "")).strip(),
"frequency": str(data.get("freq", 0)),
"sysname": str(data.get("system", "")).strip()
}
now = datetime.now() now = datetime.now()
# Logic for handling call start/end events
if current_tgid != 0: if current_tgid != 0:
potential_end_time = None # Reset debounce potential_end_time = None
if current_tgid != last_tgid: if current_tgid != last_tgid:
if last_tgid != 0: if last_tgid != 0:
# End previous call immediately if switching channels
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)
# Start new call
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

View File

@@ -7,6 +7,7 @@ services:
restart: unless-stopped restart: unless-stopped
ports: ports:
- 8001:8001 - 8001:8001
- 8081:8081
devices: devices:
- "/dev/bus/usb:/dev/bus/usb" - "/dev/bus/usb:/dev/bus/usb"
volumes: volumes: