mirrors/dmrlink3
1
0 Fork 0
mirror of https://github.com/n0mjs710/dmrlink3 synced 2026-11-08 04:19:24 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
dmrlink3/dashboard/server.py
T
Cort BuffingtonandClaude Opus 4.8 d727041259 reporting: normalize peer vocabulary + add Unix-socket transport (Step 5) 
Mirrors the hblink3 Step 5 work for the IPSC side:

  * peer -> peer_connected / peer_disconnected (the message type now carries
    the lifecycle; the dashboard consumes the new names). dmrlink3 has no
    'stream' event -- its bridge_event/ping/bridge are IPSC-specific and are
    deliberately left as-is (see the hblink/dmrlink dashboard-parity note:
    IPSC ping is a firewall timer, not an app-level keepalive).

  * Unix-socket transport for a same-box dashboard: REPORT_TRANSPORT=unix binds
    start_unix_server on REPORT_SOCKET (stale socket file cleared first); the
    dashboard connects via open_unix_connection(DMRLINK_SOCKET) when
    DMRLINK_TRANSPORT=unix. TCP stays the default for remote dashboards. A local
    socket can't NIC-flap or be conntrack-evicted, retiring the silent-severance
    class for local dashboards (filesystem perms replace the IP ACL).

Both REPORT_TRANSPORT/REPORT_SOCKET and DMRLINK_TRANSPORT/DMRLINK_SOCKET are
optional (fallback to tcp), so existing configs keep working unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-12 12:19:02 -05:00

443 lines
16 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2026 Cortney T. Buffington, N0MJS <n0mjs@me.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
###############################################################################
'''
DMRlink3 dashboard backend.
Connects to dmrlink3's NDJSON reporting feed (one JSON object per line), keeps
authoritative display state, and serves a single-page UI that receives live
JSON events over a WebSocket.
Run: python server.py (from the dashboard/ directory, or set PYTHONPATH)
'''
import asyncio
import socket
import json
import logging
import os
import ssl
import sys
import time
from collections import deque
from contextlib import asynccontextmanager
from urllib.request import urlopen
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
import ijson
import uvicorn
from dmr_utils3.utils import mk_id_dict, get_alias
HERE = os.path.dirname(os.path.abspath(__file__))
STATIC = os.path.join(HERE, 'static')
sys.path.insert(0, HERE)
try:
from config import (REPORT_NAME, DMRLINK_IP, DMRLINK_PORT,
WEB_HOST, WEB_PORT, LOG_LINES,
PATH, PEER_FILE, SUBSCRIBER_FILE, TGID_FILE,
LOCAL_SUB_FILE, LOCAL_PEER_FILE)
except ImportError:
sys.exit('No config.py found — copy config_sample.py to config.py and edit it.')
# Feed transport: 'tcp' (default, connect to DMRLINK_IP:DMRLINK_PORT) or 'unix'
# (connect to the daemon's local Unix socket DMRLINK_SOCKET). Optional in config.
try:
from config import DMRLINK_TRANSPORT
except ImportError:
DMRLINK_TRANSPORT = 'tcp'
try:
from config import DMRLINK_SOCKET
except ImportError:
DMRLINK_SOCKET = ''
try:
from config import TRY_DOWNLOAD, PEER_URL, SUBSCRIBER_URL, STALE_DAYS
except ImportError:
TRY_DOWNLOAD = False
PEER_URL = 'https://www.radioid.net/static/rptrs.json'
SUBSCRIBER_URL = 'https://www.radioid.net/static/users.json'
STALE_DAYS = 7
try:
from config import FILTER_COUNTRIES
except ImportError:
FILTER_COUNTRIES = None
try:
from config import LAST_HEARD, LAST_HEARD_COUNT
except ImportError:
LAST_HEARD = 'open'
LAST_HEARD_COUNT = 10
try:
from config import SYSTEM_PEERS
except ImportError:
SYSTEM_PEERS = 'open'
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger('dmrdash')
# Drop stale "active" calls this many seconds after START if no END arrives.
# Backstop only: a dropped feed already clears active calls on disconnect, so
# this just guards against a lost/never-sent terminator on a still-connected
# feed. Keep it well above any real transmission (amateur TOT is ~3 min) so
# legitimate long calls are never clipped mid-stream.
CALL_STALE = 300
# ---- alias resolution --------------------------------------------------------
def _abs(p):
return p if os.path.isabs(p) else os.path.join(HERE, p)
def _stream_id_file(url, path, json_key, countries, stale_secs):
now = time.time()
if os.path.isfile(path) and (os.path.getmtime(path) + stale_secs) >= now:
logger.info('ID ALIAS MAPPER: %s is current, not downloaded', os.path.basename(path))
return
no_verify = ssl._create_unverified_context()
tmp = path + '.tmp'
try:
with urlopen(url, context=no_verify) as response, \
open(tmp, 'w', encoding='utf-8') as out:
out.write('{"%s":[' % json_key)
first = True
for record in ijson.items(response, json_key + '.item'):
if not countries or record.get('country') in countries:
if not first:
out.write(',')
json.dump({'id': record['id'], 'callsign': record['callsign']}, out)
first = False
out.write(']}')
os.replace(tmp, path)
label = ', '.join(sorted(countries)) if countries else 'all countries'
logger.info('ID ALIAS MAPPER: %s downloaded (%s)', os.path.basename(path), label)
except IOError as e:
logger.error('ID ALIAS MAPPER: download of %s failed: %s', os.path.basename(path), e)
if os.path.exists(tmp):
os.remove(tmp)
def _download_aliases():
if not TRY_DOWNLOAD:
return
base = _abs(PATH)
stale_secs = int(STALE_DAYS) * 86400
countries = set(FILTER_COUNTRIES) if FILTER_COUNTRIES else None
_stream_id_file(PEER_URL, base + PEER_FILE, 'rptrs', countries, stale_secs)
_stream_id_file(SUBSCRIBER_URL, base + SUBSCRIBER_FILE, 'users', countries, stale_secs)
def _reload_aliases():
global PEER_IDS, SUBSCRIBER_IDS, TALKGROUP_IDS
base = _abs(PATH)
PEER_IDS = mk_id_dict(base, PEER_FILE)
SUBSCRIBER_IDS = mk_id_dict(base, SUBSCRIBER_FILE)
TALKGROUP_IDS = mk_id_dict(base, TGID_FILE)
if LOCAL_PEER_FILE:
PEER_IDS.update(mk_id_dict(base, LOCAL_PEER_FILE))
if LOCAL_SUB_FILE:
SUBSCRIBER_IDS.update(mk_id_dict(base, LOCAL_SUB_FILE))
logger.info('aliases loaded: %d peers, %d subscribers, %d talkgroups',
len(PEER_IDS), len(SUBSCRIBER_IDS), len(TALKGROUP_IDS))
PEER_IDS = {}
SUBSCRIBER_IDS = {}
TALKGROUP_IDS = {}
async def _alias_refresh_loop():
while True:
await asyncio.sleep(86400)
logger.info('aliases: starting daily refresh')
await asyncio.to_thread(_download_aliases)
_reload_aliases()
# Logo
try:
from config import LOGO_FILE
except ImportError:
LOGO_FILE = ''
_logo_path = _abs(LOGO_FILE) if LOGO_FILE else ''
_logo_exists = bool(_logo_path and os.path.isfile(_logo_path))
_logo_media_type = {
'.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
'.gif': 'image/gif', '.svg': 'image/svg+xml', '.webp': 'image/webp',
}.get(os.path.splitext(_logo_path)[1].lower(), 'image/png') if _logo_exists else 'image/png'
LOGO_HTML = '<img src="/logo" alt="" class="logo">' if _logo_exists else ''
def alias(_id, _dict):
a = get_alias(_id, _dict)
return None if a == _id else a
# ---- shared state ------------------------------------------------------------
class State:
def __init__(self):
self.systems = {} # last 'config' payload
self.bridges = {} # last 'bridge' payload (enriched)
self.active = {} # call_key -> START bridge_event
self.log = deque(maxlen=LOG_LINES)
self.dmrlink = False
self.clients = set() # connected WebSockets
self.ping_loss_warn = 5 # PING_LOSS_WARN %: dashboard flags peer/master gold at/above this
STATE = State()
def call_key(evt):
return '{}|{}|{}'.format(evt.get('system', ''), evt.get('ts', 0), evt.get('call_id', 0))
def enrich_call(evt):
evt['src_alias'] = alias(evt.get('src'), SUBSCRIBER_IDS)
evt['peer_alias'] = alias(evt.get('peer'), PEER_IDS)
evt['tgid_alias'] = alias(evt.get('tgid'), TALKGROUP_IDS)
return evt
def enrich_systems(systems):
now = time.time()
for sys in systems.values():
m = sys.get('MASTER', {})
ct = m.get('CONNECT_TIME', 0)
m['connected_secs'] = int(max(0, now - ct)) if ct else None
for p in sys.get('PEERS', {}).values():
ct = p.get('CONNECT_TIME', 0)
p['connected_secs'] = int(max(0, now - ct)) if ct else None
return systems
def enrich_bridges(bridges):
now = time.time()
for members in bridges.values():
for m in members:
m['TGID_NAME'] = alias(m.get('TGID'), TALKGROUP_IDS)
if m.get('TO_TYPE') in ('ON', 'OFF'):
m['remaining'] = int(m.get('TIMER', now) - now)
else:
m['remaining'] = None
return bridges
async def broadcast(obj):
dead = set()
for ws in STATE.clients:
try:
await ws.send_json(obj)
except Exception:
dead.add(ws)
STATE.clients -= dead
async def handle_event(evt):
global FEED_READ_TIMEOUT
t = evt.get('type')
if t == 'config':
interval = evt.get('report_interval')
if interval:
FEED_READ_TIMEOUT = max(float(interval) * 3, 30.0)
STATE.ping_loss_warn = evt.get('ping_loss_warn', STATE.ping_loss_warn)
STATE.systems = enrich_systems(evt.get('systems', {}))
await broadcast({'type': 'config', 'systems': STATE.systems,
'ping_loss_warn': STATE.ping_loss_warn})
elif t in ('peer_connected', 'peer_disconnected'):
# Granular IPSC peer connect/disconnect delta. Apply it to the in-memory
# systems view and re-broadcast the (enriched) config so the browser
# renders it without waiting for the next slow resync. Ignored if the
# system isn't known yet -- the on-connect snapshot will carry it.
sysview = STATE.systems.get(evt.get('system'))
if sysview is not None:
peers = sysview.setdefault('PEERS', {})
rid = str(evt.get('radio_id'))
if t == 'peer_connected' and evt.get('info') is not None:
peers[rid] = evt['info']
elif t == 'peer_disconnected':
peers.pop(rid, None)
STATE.systems = enrich_systems(STATE.systems)
await broadcast({'type': 'config', 'systems': STATE.systems,
'ping_loss_warn': STATE.ping_loss_warn})
elif t == 'ping':
pass # liveness heartbeat; receiving it already reset the feed read timeout
elif t == 'bridge':
STATE.bridges = enrich_bridges(evt.get('bridges', {}))
await broadcast({'type': 'bridge', 'bridges': STATE.bridges})
elif t == 'bridge_event':
data = evt.get('data', {})
enrich_call(data)
data['_arrived'] = time.time()
key = call_key(data)
event_name = data.get('event', '')
if 'START' in event_name:
STATE.active[key] = data
else:
STATE.active.pop(key, None)
STATE.log.appendleft(data)
await broadcast({'type': 'bridge_event', 'data': data})
else:
logger.debug('ignoring unknown event type: %s', t)
# dmrlink3 pushes a config snapshot at least every REPORT_INTERVAL, so if we go
# dmrlink3 pushes a config snapshot every REPORT_INTERVAL seconds -- that push is
# the de-facto heartbeat. If we go long enough with no line at all, the link is
# dead (whether or not a clean FIN arrived); without this, a silently-severed
# connection leaves readline() blocked forever and the reconnect loop never runs.
# The timeout must be safely LARGER than the push interval or it false-trips on a
# healthy idle link -- so we size it to 3x the interval the daemon advertises (see
# handle_event), not a fixed value. TCP keepalive (above) stays the primary
# detector of a truly dead socket. The default applies only until the first
# config arrives (which the daemon sends immediately on connect).
FEED_READ_TIMEOUT = 180.0
def _enable_tcp_keepalive(writer):
sock = writer.get_extra_info('socket')
if sock is None:
return
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
if hasattr(socket, 'TCP_KEEPIDLE'):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 15)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 4)
except OSError as e:
logger.warning('could not set TCP keepalive on dmrlink3 feed: %s', e)
# ---- dmrlink3 feed client (with reconnect) -----------------------------------
async def dmrlink_feed():
while True:
writer = None
try:
if DMRLINK_TRANSPORT == 'unix':
reader, writer = await asyncio.open_unix_connection(DMRLINK_SOCKET)
logger.info('connected to dmrlink3 feed at unix socket %s', DMRLINK_SOCKET)
else:
reader, writer = await asyncio.open_connection(DMRLINK_IP, DMRLINK_PORT)
_enable_tcp_keepalive(writer)
logger.info('connected to dmrlink3 feed at %s:%s', DMRLINK_IP, DMRLINK_PORT)
STATE.dmrlink = True
await broadcast({'type': 'dmrlink', 'connected': True})
while True:
try:
line = await asyncio.wait_for(reader.readline(), FEED_READ_TIMEOUT)
except asyncio.TimeoutError:
logger.warning('no data from dmrlink3 for %ss; assuming link dead, reconnecting',
FEED_READ_TIMEOUT)
break
if not line:
break
try:
await handle_event(json.loads(line))
except json.JSONDecodeError:
logger.warning('bad JSON line from dmrlink3: %r', line[:120])
except (ConnectionRefusedError, OSError) as e:
logger.warning('dmrlink3 feed unavailable (%s); retrying in 3s', e)
finally:
if writer is not None:
try:
writer.close()
except Exception:
pass
if STATE.dmrlink:
STATE.dmrlink = False
STATE.systems = {}
STATE.bridges = {}
STATE.active = {}
await broadcast({'type': 'dmrlink', 'connected': False})
await asyncio.sleep(3)
async def reap_calls():
while True:
await asyncio.sleep(10)
now = time.time()
stale = [k for k, e in STATE.active.items()
if now - e.get('_arrived', now) > CALL_STALE]
for k in stale:
STATE.active.pop(k, None)
if stale:
logger.debug('reaped %d stale active call(s)', len(stale))
@asynccontextmanager
async def lifespan(app):
await asyncio.to_thread(_download_aliases)
_reload_aliases()
refresher = asyncio.create_task(_alias_refresh_loop())
feed = asyncio.create_task(dmrlink_feed())
reaper = asyncio.create_task(reap_calls())
yield
feed.cancel()
reaper.cancel()
refresher.cancel()
app = FastAPI(lifespan=lifespan)
# ---- HTTP + WebSocket --------------------------------------------------------
@app.get('/', response_class=HTMLResponse)
async def index():
with open(os.path.join(STATIC, 'dashboard.html'), encoding='utf-8') as f:
html = f.read()
return (html.replace('{{REPORT_NAME}}', REPORT_NAME)
.replace('{{LOGO_HTML}}', LOGO_HTML)
.replace('{{LAST_HEARD}}', str(LAST_HEARD))
.replace('{{LAST_HEARD_COUNT}}', str(LAST_HEARD_COUNT))
.replace('{{SYSTEM_PEERS}}', str(SYSTEM_PEERS)))
@app.get('/logo')
async def serve_logo():
if not _logo_exists:
raise HTTPException(status_code=404)
return FileResponse(_logo_path, media_type=_logo_media_type)
@app.get('/api/state')
async def api_state():
return JSONResponse(current_state())
def current_state():
return {
'report_name': REPORT_NAME,
'dmrlink': STATE.dmrlink,
'systems': STATE.systems,
'bridges': STATE.bridges,
'active': list(STATE.active.values()),
'log': list(STATE.log),
'ping_loss_warn': STATE.ping_loss_warn,
}
@app.websocket('/ws')
async def ws_endpoint(ws: WebSocket):
await ws.accept()
STATE.clients.add(ws)
try:
await ws.send_json({'type': 'initial', **current_state()})
while True:
await ws.receive_text()
except WebSocketDisconnect:
pass
finally:
STATE.clients.discard(ws)
def main():
uvicorn.run(app, host=WEB_HOST, port=WEB_PORT, log_level='warning')
if __name__ == '__main__':
main()
Reference in new issue View Git Blame Copy Permalink