358 lines
15 KiB
Python
358 lines
15 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
from aiohttp import web
|
|
from bitstring import BitArray
|
|
|
|
from vocoder_engine import VocoderEngine
|
|
from openbridge_protocol import crea_pacchetto_ping
|
|
from dmr_utils3.ambe_utils import convert72BitTo49BitAMBE
|
|
|
|
# Importazione compatibilità audioop per Python 3.13+
|
|
try:
|
|
import audioop
|
|
except ImportError:
|
|
import audioop_lts as audioop
|
|
|
|
# ====================================================================
|
|
# CONFIGURAZIONE
|
|
# ====================================================================
|
|
MAX_UTENTI_WEB = 5 # Limite massimo di utenti in ascolto contemporaneo
|
|
|
|
def carica_config():
|
|
try:
|
|
with open("config.json", "r") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {
|
|
"hblink_ip": "127.0.0.1",
|
|
"hblink_port": 62047,
|
|
"hblink_password": "TestVOIP",
|
|
"peer_id": 22242
|
|
}
|
|
|
|
CONFIG = carica_config()
|
|
|
|
# ====================================================================
|
|
# DATABASE NOMINATIVI DMR
|
|
# ====================================================================
|
|
DMR_DB = {}
|
|
|
|
def carica_database_dmr():
|
|
global DMR_DB
|
|
DMR_DB = {}
|
|
|
|
if os.path.exists("DMRIds.json"):
|
|
try:
|
|
with open("DMRIds.json", "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if isinstance(data, dict) and "users" in data:
|
|
for entry in data["users"]:
|
|
uid = int(entry.get("id", 0))
|
|
call = entry.get("callsign", "")
|
|
fname = entry.get("fname", "")
|
|
if uid and call:
|
|
DMR_DB[uid] = f"{call} - {fname}".strip(" -")
|
|
elif isinstance(data, dict):
|
|
for k, v in data.items():
|
|
try:
|
|
uid = int(k)
|
|
if isinstance(v, str): DMR_DB[uid] = v
|
|
elif isinstance(v, dict):
|
|
DMR_DB[uid] = f"{v.get('callsign', '')} - {v.get('fname', '')}".strip(" -")
|
|
except ValueError: continue
|
|
print(f"[DMR-DB] Caricati con successo {len(DMR_DB)} nominativi da DMRIds.json!")
|
|
return
|
|
except Exception as e:
|
|
print(f"[DMR-DB] Errore lettura DMRIds.json: {e}")
|
|
|
|
for dat_file in ["DMRIds.dat", "StripDMRIDs.dat"]:
|
|
if os.path.exists(dat_file):
|
|
try:
|
|
count = 0
|
|
with open(dat_file, "r", encoding="utf-8", errors="ignore") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"): continue
|
|
if ";" in line: parts = [p.strip() for p in line.split(";") if p.strip()]
|
|
elif "," in line: parts = [p.strip() for p in line.split(",") if p.strip()]
|
|
else: parts = line.split()
|
|
|
|
if len(parts) >= 2:
|
|
try:
|
|
uid = int(parts[0])
|
|
call = parts[1]
|
|
fname = parts[2] if len(parts) > 2 else ""
|
|
lname = parts[3] if len(parts) > 3 else ""
|
|
nome_completo = f"{fname} {lname}".strip()
|
|
DMR_DB[uid] = f"{call} - {nome_completo}" if nome_completo else call
|
|
count += 1
|
|
except ValueError: continue
|
|
print(f"[DMR-DB] Caricati con successo {count} nominativi da {dat_file}!")
|
|
return
|
|
except Exception as e:
|
|
print(f"[DMR-DB] Errore lettura {dat_file}: {e}")
|
|
|
|
print("[DMR-DB] Nessun file DMRIds valido trovato.")
|
|
|
|
def cerca_nominativo(dmr_id):
|
|
if dmr_id in DMR_DB:
|
|
return f"{DMR_DB[dmr_id]} ({dmr_id})"
|
|
return f"ID: {dmr_id}"
|
|
|
|
# ====================================================================
|
|
# PROTOCOLLO OPENBRIDGE (Solo Ricezione)
|
|
# ====================================================================
|
|
class OpenBridgeRxClient(asyncio.DatagramProtocol):
|
|
def __init__(self, server_app):
|
|
self.transport = None
|
|
self.server_app = server_app
|
|
|
|
def connection_made(self, transport):
|
|
self.transport = transport
|
|
print(f"[OpenBridge] Client RX connesso a {CONFIG['hblink_ip']}:{CONFIG['hblink_port']}")
|
|
asyncio.create_task(self.keepalive_loop())
|
|
|
|
def datagram_received(self, data, addr):
|
|
if len(data) < 53: return
|
|
if data.startswith(b"DMRD") and len(data) >= 73:
|
|
src_id = int.from_bytes(data[5:8], 'big')
|
|
tg_arrivo = str(int.from_bytes(data[8:11], 'big'))
|
|
|
|
# Traccia il traffico a prescindere da chi lo ascolta
|
|
self.server_app.aggiorna_traffico_globale(src_id, tg_arrivo)
|
|
|
|
# Ricava l'elenco dei TG attualmente richiesti da almeno un utente Web
|
|
tg_attivi = {client_data["tg"] for client_data in self.server_app.clients.values()}
|
|
|
|
# Processa l'audio solo se c'è almeno un utente in ascolto su quel TG
|
|
if tg_arrivo in tg_attivi:
|
|
payload_33 = data[20:53]
|
|
ambe1 = payload_33[0:9]
|
|
ambe2 = bytearray(9)
|
|
ambe2[0:4] = payload_33[9:13]
|
|
ambe2[4] = (payload_33[13] & 0xF0) | (payload_33[19] & 0x0F)
|
|
ambe2[5:9] = payload_33[20:24]
|
|
ambe3 = payload_33[24:33]
|
|
frame_uniti = ambe1 + bytes(ambe2) + ambe3
|
|
|
|
self.server_app.ricevi_frame_dmr(frame_uniti, src_id, tg_arrivo)
|
|
|
|
async def keepalive_loop(self):
|
|
while True:
|
|
if self.transport:
|
|
pacchetto_ping = crea_pacchetto_ping(CONFIG['peer_id'], CONFIG['hblink_password'])
|
|
self.transport.sendto(pacchetto_ping)
|
|
await asyncio.sleep(10)
|
|
|
|
# ====================================================================
|
|
# GESTORE DEL SERVER MULTI-UTENTE
|
|
# ====================================================================
|
|
class WebDmrServer:
|
|
def __init__(self):
|
|
self.vocoder = VocoderEngine(port_start=2475, port_end=2479)
|
|
self.openbridge = None
|
|
|
|
# Dizionario client: ws -> {"tg": "222"}
|
|
self.clients = {}
|
|
|
|
# Dizionario stati indipendenti per TG: "222" -> {"rx_in_corso": False, ...}
|
|
self.tg_states = {}
|
|
|
|
# Dizionario globale traffico
|
|
self.traffico_globale = {}
|
|
|
|
def aggiorna_traffico_globale(self, src_id, tg):
|
|
"""Aggiorna il monitor globale e avvisa i client se c'è un nuovo TX"""
|
|
ora = time.time()
|
|
nuovo = tg not in self.traffico_globale or self.traffico_globale[tg]["src_id"] != src_id
|
|
self.traffico_globale[tg] = {"src_id": src_id, "time": ora}
|
|
|
|
if nuovo:
|
|
print(f"[Traffico Globale] Rilevato TX su TG {tg} da ID {src_id}") # <--- AGGIUNGI QUESTA RIGA
|
|
asyncio.create_task(self.invia_stato_traffico())
|
|
|
|
async def invia_stato_traffico(self):
|
|
"""Invia la lista dei TG attivi a tutti i WebSocket connessi"""
|
|
attivi = [{"tg": k, "src": cerca_nominativo(v["src_id"])} for k, v in self.traffico_globale.items()]
|
|
msg_str = json.dumps({"action": "global_traffic", "data": attivi})
|
|
for ws in list(self.clients.keys()):
|
|
try: await ws.send_str(msg_str)
|
|
except Exception: pass
|
|
|
|
def ricevi_frame_dmr(self, ambe_frames, src_id, tg):
|
|
if tg not in self.tg_states:
|
|
self.tg_states[tg] = {"rx_in_corso": False, "ultimo_src_id": None, "ultimo_frame_time": 0}
|
|
|
|
stato_tg = self.tg_states[tg]
|
|
stato_tg["ultimo_frame_time"] = time.time()
|
|
|
|
# Notifica i client in ascolto su questo TG se inizia una nuova trasmissione
|
|
if not stato_tg["rx_in_corso"] or stato_tg["ultimo_src_id"] != src_id:
|
|
stato_tg["rx_in_corso"] = True
|
|
stato_tg["ultimo_src_id"] = src_id
|
|
|
|
info_utente = cerca_nominativo(src_id)
|
|
print(f"[DMR-RX] Inizio trasmissione su TG {tg}: {info_utente}")
|
|
|
|
asyncio.create_task(self.broadcast_json_to_tg(tg, {
|
|
"action": "display",
|
|
"text": info_utente,
|
|
"active": True
|
|
}))
|
|
|
|
asyncio.create_task(self.processa_audio_dmr(ambe_frames, tg))
|
|
|
|
async def watchdog_rx(self):
|
|
"""Monitora il silenzio e ripristina lo stato per i singoli TG e per il traffico globale"""
|
|
while True:
|
|
await asyncio.sleep(0.4)
|
|
ora = time.time()
|
|
|
|
# Watchdog TG sintonizzati
|
|
for tg, stato_tg in list(self.tg_states.items()):
|
|
if stato_tg["rx_in_corso"] and (ora - stato_tg["ultimo_frame_time"] > 1.2):
|
|
stato_tg["rx_in_corso"] = False
|
|
stato_tg["ultimo_src_id"] = None
|
|
print(f"[DMR-RX] Chiamata terminata su TG {tg}")
|
|
await self.broadcast_json_to_tg(tg, {
|
|
"action": "display",
|
|
"text": "IN ASCOLTO",
|
|
"active": False
|
|
})
|
|
|
|
# Watchdog traffico globale
|
|
cambiamento = False
|
|
for tg in list(self.traffico_globale.keys()):
|
|
if ora - self.traffico_globale[tg]["time"] > 1.5:
|
|
del self.traffico_globale[tg]
|
|
cambiamento = True
|
|
if cambiamento:
|
|
await self.invia_stato_traffico()
|
|
|
|
async def processa_audio_dmr(self, ambe_27_bytes, tg):
|
|
try:
|
|
for i in range(0, len(ambe_27_bytes), 9):
|
|
frame_9 = ambe_27_bytes[i:i+9]
|
|
if len(frame_9) < 9: continue
|
|
try:
|
|
ambe_49 = convert72BitTo49BitAMBE(BitArray(bytes=frame_9))
|
|
ba_49 = ambe_49 if isinstance(ambe_49, BitArray) else BitArray(ambe_49)
|
|
if len(ba_49) < 56: ba_49.append(BitArray(uint=0, length=56 - len(ba_49)))
|
|
frame_7 = ba_49.bytes[:7]
|
|
except Exception: continue
|
|
|
|
# Identificatore Univoco del Vocoder in base al TG (es. WEB_RX_222)
|
|
pcm_data = await self.vocoder.converti_ambe_in_pcm(f"WEB_RX_{tg}", frame_7)
|
|
if pcm_data and len(pcm_data) >= 2:
|
|
if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1]
|
|
|
|
pcm_data = audioop.mul(pcm_data, 2, 1.4)
|
|
|
|
# Spediamo i byte PCM SOLO agli utenti in ascolto su quel TG
|
|
for ws, client_data in list(self.clients.items()):
|
|
if client_data["tg"] == tg:
|
|
try:
|
|
await ws.send_bytes(pcm_data)
|
|
except Exception:
|
|
self.clients.pop(ws, None)
|
|
|
|
except Exception as e:
|
|
print(f"[Audio RX Error su TG {tg}]: {e}")
|
|
|
|
async def broadcast_json_to_tg(self, tg, messaggio_dict):
|
|
"""Invia un messaggio di testo JSON solo agli utenti su un TG specifico"""
|
|
msg_str = json.dumps(messaggio_dict)
|
|
for ws, client_data in list(self.clients.items()):
|
|
if client_data["tg"] == tg:
|
|
try:
|
|
await ws.send_str(msg_str)
|
|
except Exception:
|
|
self.clients.pop(ws, None)
|
|
|
|
# ====================================================================
|
|
# ENDPOINT
|
|
# ====================================================================
|
|
server_app = WebDmrServer()
|
|
|
|
async def index(request):
|
|
content = open("index.html", "r").read()
|
|
return web.Response(content_type="text/html", text=content)
|
|
|
|
async def websocket_handler(request):
|
|
ws = web.WebSocketResponse(heartbeat=30.0)
|
|
await ws.prepare(request)
|
|
|
|
# --- CONTROLLO LIMITE UTENTI ---
|
|
if len(server_app.clients) >= MAX_UTENTI_WEB:
|
|
print(f"[WebSocket] Connessione rifiutata: limite di {MAX_UTENTI_WEB} utenti raggiunto.")
|
|
await ws.send_str(json.dumps({
|
|
"action": "error",
|
|
"text": "Server pieno! Raggiunto il limite massimo di 5 ascoltatori contemporanei."
|
|
}))
|
|
await ws.close()
|
|
return ws
|
|
|
|
# Aggiungi il nuovo client con TG di default "222"
|
|
server_app.clients[ws] = {"tg": "222"}
|
|
print(f"[WebSocket] Nuovo client connesso (Totali: {len(server_app.clients)}/{MAX_UTENTI_WEB})")
|
|
|
|
try:
|
|
# Invia subito lo stato del traffico globale al nuovo connesso
|
|
await server_app.invia_stato_traffico()
|
|
|
|
async for msg in ws:
|
|
if msg.type == web.WSMsgType.TEXT:
|
|
dati = json.loads(msg.data)
|
|
|
|
if dati.get("action") == "cambia_tg":
|
|
nuovo_tg = str(dati.get("tg", "222"))
|
|
server_app.clients[ws]["tg"] = nuovo_tg
|
|
print(f"[Cambio TG Web]: Un utente si è sintonizzato sul TG {nuovo_tg}")
|
|
|
|
# --- SINCRONIZZAZIONE LATE-JOINER ---
|
|
# Se c'è già una conversazione attiva su questo TG, invia subito l'ID
|
|
stato_tg = server_app.tg_states.get(nuovo_tg)
|
|
if stato_tg and stato_tg.get("rx_in_corso"):
|
|
info_utente = cerca_nominativo(stato_tg["ultimo_src_id"])
|
|
asyncio.create_task(ws.send_str(json.dumps({
|
|
"action": "display",
|
|
"text": info_utente,
|
|
"active": True
|
|
})))
|
|
|
|
finally:
|
|
server_app.clients.pop(ws, None)
|
|
print(f"[WebSocket] Client disconnesso (Rimanenti: {len(server_app.clients)}/{MAX_UTENTI_WEB})")
|
|
return ws
|
|
|
|
# ====================================================================
|
|
# AVVIO SERVER
|
|
# ====================================================================
|
|
async def on_startup(app):
|
|
carica_database_dmr()
|
|
await server_app.vocoder.avvia()
|
|
asyncio.create_task(server_app.watchdog_rx())
|
|
|
|
loop = asyncio.get_running_loop()
|
|
_, server_app.openbridge = await loop.create_datagram_endpoint(
|
|
lambda: OpenBridgeRxClient(server_app),
|
|
local_addr=('0.0.0.0', CONFIG["hblink_port"]),
|
|
remote_addr=(CONFIG["hblink_ip"], CONFIG["hblink_port"])
|
|
)
|
|
print("\n[Init] Ricevitore Web-DMR Multi-Utente avviato!")
|
|
|
|
if __name__ == "__main__":
|
|
app = web.Application()
|
|
app.on_startup.append(on_startup)
|
|
|
|
app.router.add_get("/", index)
|
|
app.router.add_get("/ws", websocket_handler)
|
|
app.router.add_static("/static/", path="static", name="static")
|
|
|
|
print("=========================================")
|
|
print(" DVSwitch Web-Audio Receiver in ascolto...")
|
|
print("=========================================")
|
|
web.run_app(app, port=8080)
|