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)