Prima release ufficiale di Web-DMR
This commit is contained in:
@@ -0,0 +1,319 @@
|
||||
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'))
|
||||
|
||||
# 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 = {}
|
||||
|
||||
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"""
|
||||
while True:
|
||||
await asyncio.sleep(0.4)
|
||||
ora = time.time()
|
||||
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
|
||||
})
|
||||
|
||||
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:
|
||||
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)
|
||||
Reference in New Issue
Block a user