mirrors/hblink3
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
hblink3/dashboard/server.py
T
Cort BuffingtonandClaude Opus 4.8 405aad466b reporting: emit repeater_connected/_disconnected (not peer_*) in hblink3 
Consistency fix: hblink3 calls this entity a REPEATER everywhere -- the
REPEATERS dict, json_repeater(), and the dashboard UI -- and so does hblink4
(repeater_connected). The lone peer_* event name was the odd one out. Rename
the wire vocabulary to repeater_connected / repeater_disconnected so the same
thing isn't called "peer" in one place and "repeater" in another.

dmrlink3 keeps peer_* deliberately: IPSC genuinely has "peers" (PEERS dict,
_peer_snapshot, UI says Peer). The internal send_peer/report_peer_deltas method
names stay as-is in both daemons -- shared ReportServer plumbing kept symmetric
for the deferred Step-4 extraction; only the wire/UI vocabulary changed here.

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

449 lines
17 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.
###############################################################################
'''
HBlink3 dashboard backend.
Connects to HBlink3's NDJSON reporting feed (one JSON object per line), keeps the
authoritative display state, and serves a single-page UI that receives live JSON
deltas over a WebSocket. Run: python server.py (or via run_dashboard.py).
'''
import asyncio
import json
import logging
import os
import socket
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
# Ensure THIS directory's config.py wins over HBlink3's top-level config.py.
HERE = os.path.dirname(os.path.abspath(__file__))
STATIC = os.path.join(HERE, 'static')
sys.path.insert(0, HERE)
# ---- configuration -----------------------------------------------------------
try:
from config import (REPORT_NAME, HBLINK_IP, HBLINK_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.')
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 SERVER_REPEATERS
except ImportError:
SERVER_REPEATERS = 'open'
# Feed transport: 'tcp' (default, connect to HBLINK_IP:HBLINK_PORT) or 'unix'
# (connect to the daemon's local Unix socket HBLINK_SOCKET). Optional in config.
try:
from config import HBLINK_TRANSPORT
except ImportError:
HBLINK_TRANSPORT = 'tcp'
try:
from config import HBLINK_SOCKET
except ImportError:
HBLINK_SOCKET = ''
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger('hbdash')
# Drop an "active" stream this many seconds after its START if no END arrives.
# This is only a backstop for a lost/never-sent terminator while the feed stays
# connected -- a dropped feed already clears all active streams on disconnect,
# and HBlink3 itself synthesizes an END a couple seconds after audio stops
# (bridge.py STREAM_TIMEOUT). Keep it well above any real transmission (amateur
# TOT is ~3 min) so legitimate long calls are never clipped mid-stream.
STREAM_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' event payload
self.bridges = {} # last 'bridges' event payload (enriched)
self.streams = {} # key -> last START stream event (active calls)
self.log = deque(maxlen=LOG_LINES)
self.hblink = False
self.clients = set() # connected dashboard WebSockets
self.ping_time = 5 # PING_TIME from hblink global config
self.max_missed = 3 # MAX_MISSED from hblink global config
self.ping_loss_warn = 5 # PING_LOSS_WARN %: gold callsign at/above this
STATE = State()
def stream_key(evt):
return '{}|{}|{}'.format(evt['system'], evt['slot'], evt['stream_id'])
def enrich_stream(evt):
evt['src_alias'] = alias(evt['src'], SUBSCRIBER_IDS)
evt['peer_alias'] = alias(evt['peer'], PEER_IDS)
# A unit call's destination is a subscriber; a group call's is a talkgroup.
if evt.get('call_type') == 'UNIT VOICE':
evt['dst_alias'] = alias(evt['dst'], SUBSCRIBER_IDS)
else:
evt['dst_alias'] = alias(evt['dst'], TALKGROUP_IDS)
return evt
# Convert server-side absolute epochs to "seconds since/until" deltas so the
# browser can tick live counters without depending on the client's clock.
def enrich_config(systems):
now = time.time()
for sysview in systems.values():
if sysview['MODE'] == 'SERVER':
for p in sysview.get('REPEATERS', {}).values():
p['connected_secs'] = int(max(0, now - p.get('CONNECTED', now)))
last_ping = p.get('LAST_PING', 0)
p['last_ping_secs'] = int(max(0, now - last_ping)) if last_ping else None
elif sysview['MODE'] == 'OUTBOUND':
c = sysview.get('STATS', {}).get('CONNECTED')
sysview['STATS']['connected_secs'] = int(max(0, now - c)) if c 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['TGID'], TALKGROUP_IDS)
if m['TO_TYPE'] in ('ON', 'OFF'):
m['remaining'] = int(m['TIMER'] - 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_time = evt.get('ping_time', STATE.ping_time)
STATE.max_missed = evt.get('max_missed', STATE.max_missed)
STATE.ping_loss_warn = evt.get('ping_loss_warn', STATE.ping_loss_warn)
STATE.systems = enrich_config(evt['systems'])
await broadcast({'type': 'config', 'systems': STATE.systems,
'ping_time': STATE.ping_time, 'max_missed': STATE.max_missed,
'ping_loss_warn': STATE.ping_loss_warn})
elif t in ('repeater_connected', 'repeater_disconnected'):
# Granular repeater 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 full push. 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 and sysview.get('MODE') == 'SERVER':
reps = sysview.setdefault('REPEATERS', {})
rid = str(evt.get('radio_id'))
if t == 'repeater_connected' and evt.get('info') is not None:
reps[rid] = evt['info']
elif t == 'repeater_disconnected':
reps.pop(rid, None)
STATE.systems = enrich_config(STATE.systems)
await broadcast({'type': 'config', 'systems': STATE.systems,
'ping_time': STATE.ping_time, 'max_missed': STATE.max_missed,
'ping_loss_warn': STATE.ping_loss_warn})
elif t == 'bridges':
STATE.bridges = enrich_bridges(evt['bridges'])
await broadcast({'type': 'bridges', 'bridges': STATE.bridges})
elif t in ('stream_start', 'stream_end'):
enrich_stream(evt)
evt['_ts'] = time.time() # authoritative server time so "Last Heard" age is correct across reloads
key = stream_key(evt)
if evt['action'] == 'START':
evt['_seen'] = time.time()
STATE.streams[key] = evt
else:
STATE.streams.pop(key, None)
# Log ingress (RX) legs only, matching the original monitor's behavior.
if evt['trx'] == 'RX':
STATE.log.appendleft(evt)
await broadcast(evt)
elif t == 'ping':
pass # liveness heartbeat; receiving it already reset the feed read timeout
else:
logger.debug('ignoring unknown event type: %s', t)
# hblink3 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 HBlink3 feed: %s', e)
# ---- HBlink3 feed client (with reconnect) ------------------------------------
async def hblink_feed():
while True:
writer = None
try:
if HBLINK_TRANSPORT == 'unix':
reader, writer = await asyncio.open_unix_connection(HBLINK_SOCKET)
logger.info('connected to HBlink3 feed at unix socket %s', HBLINK_SOCKET)
else:
reader, writer = await asyncio.open_connection(HBLINK_IP, HBLINK_PORT)
_enable_tcp_keepalive(writer)
logger.info('connected to HBlink3 feed at %s:%s', HBLINK_IP, HBLINK_PORT)
STATE.hblink = True
await broadcast({'type': 'hblink', 'connected': True})
while True:
try:
line = await asyncio.wait_for(reader.readline(), FEED_READ_TIMEOUT)
except asyncio.TimeoutError:
logger.warning('no data from HBlink3 for %ss; assuming link dead, reconnecting',
FEED_READ_TIMEOUT)
break
if not line:
break # connection closed
try:
await handle_event(json.loads(line))
except json.JSONDecodeError:
logger.warning('bad JSON line from HBlink3: %r', line[:120])
except (ConnectionRefusedError, OSError) as e:
logger.warning('HBlink3 feed unavailable (%s); retrying', e)
finally:
if writer is not None:
try:
writer.close()
except Exception:
pass
if STATE.hblink:
STATE.hblink = False
STATE.systems = {}
STATE.bridges = {}
STATE.streams = {}
await broadcast({'type': 'hblink', 'connected': False})
await asyncio.sleep(3) # reconnect delay
async def reap_streams():
while True:
await asyncio.sleep(5)
now = time.time()
stale = [k for k, e in STATE.streams.items() if now - e.get('_seen', now) > STREAM_STALE]
for k in stale:
STATE.streams.pop(k, None)
if stale:
logger.debug('reaped %d stale stream(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(hblink_feed())
reaper = asyncio.create_task(reap_streams())
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('{{SERVER_REPEATERS}}', str(SERVER_REPEATERS)))
@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,
'hblink': STATE.hblink,
'systems': STATE.systems,
'bridges': STATE.bridges,
'streams': list(STATE.streams.values()),
'log': list(STATE.log),
'ping_time': STATE.ping_time,
'max_missed': STATE.max_missed,
'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() # ignore; keep the socket open
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