diff --git a/core.py b/core.py index 1c627ec..ff6c91b 100644 --- a/core.py +++ b/core.py @@ -98,7 +98,7 @@ def carica_rubrica(): UTENTI_SIP_DMR = carica_rubrica() # ==================================================================== -# DASHBOARD MANAGER (Motore Dati - Modificato Sincrono) +# DASHBOARD MANAGER (Motore Dati) # ==================================================================== class DashboardManager: def __init__(self): @@ -142,12 +142,26 @@ class DashboardManager: asyncio.create_task(self.broadcast({"evento": "chiamata_terminata", "dati": dati})) async def watchdog_rx(self): + global sip_manager while True: await asyncio.sleep(1.0) - ora = time.time() - for id_chiamata, dati in list(self.chiamate_attive.items()): - if dati["direzione"] == "RX" and ora - dati.get("ultimo_frame", ora) > 1.5: - self.termina_chiamata(id_chiamata) + try: + ora = time.time() + + # 1. WATCHDOG SIP (RTP Timeout): Se un'estensione non invia pacchetti RTP da oltre 5 secondi, è disconnessa! + if sip_manager: + for cid, sess in list(sip_manager.sessioni.items()): + if not sess.ptt_attivo and (ora - getattr(sess, 'ultimo_rtp', ora)) > 5.0: + sip_manager.termina_chiamata(cid) + + # 2. WATCHDOG TRAFFICO RADIO RX: 1.5s di silenzio radio o sessione SIP rimossa dalla RAM + for id_chiamata, dati in list(self.chiamate_attive.items()): + if dati["direzione"] == "RX": + ext_id = dati.get("ext_id") + if (ora - dati.get("ultimo_frame", ora) > 1.5) or (ext_id and sip_manager and ext_id not in sip_manager.sessioni): + self.termina_chiamata(id_chiamata) + except Exception: + pass # ==================================================================== # ENDPOINT WEB (aiohttp) @@ -195,7 +209,8 @@ async def index_dashboard(request): class SessioneSIP: def __init__(self, call_id, chiamante, destinatario_tg, rtp_addr, sip_addr, full_from, full_to, ip_vps): self.call_id = call_id - self.chiamante = chiamante + self.call_id_safe = "".join(c for c in str(call_id) if c.isalnum()) + self.chiamante = str(chiamante) self.tg_destinazione = str(destinatario_tg) self.dmr_id = int(UTENTI_SIP_DMR.get(str(chiamante), 2223619)) self.rtp_client_addr = rtp_addr @@ -204,6 +219,7 @@ class SessioneSIP: self.rtp_seq = 0 self.rtp_timestamp = 0 self.rtp_confermato = False + self.ultimo_rtp = time.time() # Timestamp per rilevare il riaggancio da RTP self.dialog_to = full_from self.dialog_from = full_to.split(';')[0] + ";tag=987654" @@ -348,21 +364,21 @@ class SessionManager: def termina_chiamata(self, call_id): if call_id in self.sessioni: - sess = self.sessioni[call_id] - if sess.ptt_attivo: - self.gestisci_ptt(sess, False) - self.vocoder.rilascia_porta(f"SIP_{call_id}") - - # Chiusura sincrona per evitare code - for id_dash in list(dashboard_app.chiamate_attive.keys()): - if id_dash.startswith("RX_") and id_dash.endswith(f"_{call_id}"): - dashboard_app.termina_chiamata(id_dash) - - del self.sessioni[call_id] - asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"})) + sess = self.sessioni.pop(call_id, None) + if sess: + if sess.ptt_attivo: + self.gestisci_ptt(sess, False) + self.vocoder.rilascia_porta(f"SIP_{call_id}") + + # Svuota immediatamente la dashboard per questa estensione + for id_dash, dati in list(dashboard_app.chiamate_attive.items()): + if dati.get("ext_id") == call_id or f"_{sess.chiamante}_" in id_dash or id_dash.endswith(f"_{sess.chiamante}"): + dashboard_app.termina_chiamata(id_dash) + + asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"})) def gestisci_ptt(self, sess, stato): - id_dash_tx = f"TX_{sess.call_id}" + id_dash_tx = f"TX_{sess.chiamante}_{sess.call_id_safe}" if stato is True and not sess.ptt_attivo: sess.ptt_attivo = True sess.buffer_ambe = [] @@ -373,6 +389,7 @@ class SessionManager: nome_base = get_nominativo(sess.dmr_id) mittente_display = f"{nome_base} (Est. {sess.chiamante})" dashboard_app.inizia_chiamata(id_dash_tx, "TX", mittente_display, sess.tg_destinazione) + dashboard_app.chiamate_attive[id_dash_tx]["ext_id"] = sess.call_id elif stato is False and sess.ptt_attivo: sess.ptt_attivo = False if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) @@ -420,11 +437,12 @@ class SessionManager: for sess in ascoltatori: tg_arrivo = sess.tg_destinazione - id_dash_rx = f"RX_{src_id}_{sess.call_id}" + id_dash_rx = f"RX_{src_id}_{sess.chiamante}_{sess.call_id_safe}" if id_dash_rx not in dashboard_app.chiamate_attive: dest_display = f"{sess.chiamante} (TG {tg_arrivo})" dashboard_app.inizia_chiamata(id_dash_rx, "RX", src_id, dest_display) + dashboard_app.chiamate_attive[id_dash_rx]["ext_id"] = sess.call_id else: dashboard_app.aggiorna_rx(id_dash_rx) @@ -521,6 +539,7 @@ class RtpProtocol(asyncio.DatagramProtocol): def datagram_received(self, data, addr): sess = self.manager.trova_sessione_per_rtp(addr) if not sess or len(data) < 12: return + sess.ultimo_rtp = time.time() # Mantiene in vita la sessione finché arriva RTP payload_type = data[1] & 0x7F if payload_type == 8: asyncio.create_task(self.manager.processa_audio_rtp(sess, data[12:])) elif payload_type == 101: @@ -584,7 +603,15 @@ class SipProtocol(asyncio.DatagramProtocol): f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '').split()[0]} ACK\r\nContent-Length: 0\r\n\r\n") self.manager.sip_transport.sendto(ack.encode('utf-8'), addr) elif msg.startswith("BYE") or msg.startswith("CANCEL"): - self.manager.termina_chiamata(call_id) + if call_id in self.manager.sessioni: + self.manager.termina_chiamata(call_id) + else: + chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0] if "sip:" in headers.get("FROM", "") else "" + destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] if "sip:" in headers.get("TO", "") else "" + for cid, s in list(self.manager.sessioni.items()): + if s.chiamante in (chiamante, destinatario) or s.call_id_safe in call_id: + self.manager.termina_chiamata(cid) + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContent-Length: 0\r\n\r\n") self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) diff --git a/core.py.quasi.OK b/core.py.quasi.OK new file mode 100644 index 0000000..a4f6d85 --- /dev/null +++ b/core.py.quasi.OK @@ -0,0 +1,632 @@ +import json +from bitstring import BitArray +import asyncio +import struct +import time +import os +import audioop +from collections import deque +from aiohttp import web +from vocoder_engine import VocoderEngine + +from openbridge_protocol import crea_pacchetto_ping, calcola_hmac +from dmr_utils3.ambe_utils import convert49BitTo72BitAMBE, convert72BitTo49BitAMBE + +# ==================================================================== +# CARICAMENTO CONFIGURAZIONI (JSON) E DATABASE DMR +# ==================================================================== +def carica_config(): + try: + with open("config.json", "r") as f: + return json.load(f) + except Exception as e: + print(f"[ERRORE CRITICO] File config.json non trovato ({e}).") + exit(1) + +CONFIG = carica_config() +HBLINK_IP = CONFIG.get("hblink_ip", "127.0.0.1") +HBLINK_PORT = CONFIG.get("hblink_port", 62047) +HBLINK_PASSWORD = CONFIG.get("hblink_password", "TestVOIP") +PEER_ID = CONFIG.get("peer_id", 22241) +SIP_PORT = CONFIG.get("sip_port", 55060) +RTP_PORT = CONFIG.get("rtp_port", 10000) +FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") +DASHBOARD_PORT = CONFIG.get("dashboard_port", 8081) + +DMR_DB = {} +def carica_database_dmr(): + global 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(" -") + if DMR_DB: return + except Exception: pass + + for dat_file in ["DMRIds.dat", "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"[Database] Caricati {count} nominativi da {dat_file}") + return + except Exception as e: + print(f"[Database] Errore lettura {dat_file}: {e}") + +carica_database_dmr() + +def get_nominativo(dmr_id): + try: + dmr_id = int(dmr_id) + return DMR_DB.get(dmr_id, str(dmr_id)) + except ValueError: + return str(dmr_id) + +def carica_rubrica(): + try: + with open("utenti.json", "r") as f: + return json.load(f) + except Exception: + return {} + +UTENTI_SIP_DMR = carica_rubrica() + +# ==================================================================== +# DASHBOARD MANAGER (Motore Dati) +# ==================================================================== +class DashboardManager: + def __init__(self): + self.storico = deque(maxlen=100) + self.chiamate_attive = {} + self.ws_clients = set() + + async def broadcast(self, messaggio_dict): + msg_str = json.dumps(messaggio_dict) + for ws in list(self.ws_clients): + try: await ws.send_str(msg_str) + except Exception: self.ws_clients.discard(ws) + + def inizia_chiamata(self, id_chiamata, direzione, mittente, destinatario): + nom_mittente = get_nominativo(mittente) + nom_dest = get_nominativo(destinatario) + + dati = { + "id": id_chiamata, + "direzione": direzione, + "mittente": nom_mittente, + "destinatario": nom_dest, + "inizio": time.time(), + "ultimo_frame": time.time(), + "stato": "ON" + } + self.chiamate_attive[id_chiamata] = dati + asyncio.create_task(self.broadcast({"evento": "chiamata_iniziata", "dati": dati})) + + def aggiorna_rx(self, id_chiamata): + if id_chiamata in self.chiamate_attive: + self.chiamate_attive[id_chiamata]["ultimo_frame"] = time.time() + + def termina_chiamata(self, id_chiamata): + if id_chiamata in self.chiamate_attive: + dati = self.chiamate_attive.pop(id_chiamata) + dati["fine"] = time.time() + dati["durata"] = round(dati["fine"] - dati["inizio"], 1) + dati["stato"] = "OFF" + self.storico.appendleft(dati) + asyncio.create_task(self.broadcast({"evento": "chiamata_terminata", "dati": dati})) + + async def watchdog_rx(self): + while True: + await asyncio.sleep(1.0) + try: + ora = time.time() + for id_chiamata, dati in list(self.chiamate_attive.items()): + if dati["direzione"] == "RX" and ora - dati.get("ultimo_frame", ora) > 1.5: + self.termina_chiamata(id_chiamata) + except Exception: + pass + +# ==================================================================== +# ENDPOINT WEB (aiohttp) +# ==================================================================== +dashboard_app = DashboardManager() +sip_manager = None + +async def api_storico(request): + estensioni_connesse = [] + if sip_manager: + for call_id, sess in sip_manager.sessioni.items(): + estensioni_connesse.append({ + "estensione": sess.chiamante, + "nome": get_nominativo(sess.dmr_id), + "tg": sess.tg_destinazione + }) + + dati = { + "attive": list(dashboard_app.chiamate_attive.values()), + "storico": list(dashboard_app.storico), + "estensioni": estensioni_connesse + } + return web.json_response(dati) + +async def ws_handler(request): + ws = web.WebSocketResponse(heartbeat=30.0) + await ws.prepare(request) + dashboard_app.ws_clients.add(ws) + try: + async for msg in ws: pass + finally: + dashboard_app.ws_clients.discard(ws) + return ws + +async def index_dashboard(request): + try: + with open("dashboard.html", "r", encoding="utf-8") as f: + return web.Response(text=f.read(), content_type="text/html") + except FileNotFoundError: + return web.Response(text="Errore: File dashboard.html non trovato nella cartella.", status=404) + +# ==================================================================== +# CLASSI SIP, RTP E OPENBRIDGE +# ==================================================================== +class SessioneSIP: + def __init__(self, call_id, chiamante, destinatario_tg, rtp_addr, sip_addr, full_from, full_to, ip_vps): + self.call_id = call_id + self.call_id_safe = "".join(c for c in str(call_id) if c.isalnum()) + self.chiamante = chiamante + self.tg_destinazione = str(destinatario_tg) + self.dmr_id = int(UTENTI_SIP_DMR.get(str(chiamante), 2223619)) + self.rtp_client_addr = rtp_addr + self.sip_client_addr = sip_addr + self.ip_vps = ip_vps + self.rtp_seq = 0 + self.rtp_timestamp = 0 + self.rtp_confermato = False + + self.dialog_to = full_from + self.dialog_from = full_to.split(';')[0] + ";tag=987654" + self.local_cseq = 10 + self.ptt_attivo = False + self.buffer_ambe = [] + self.seq_dmr = 0 + self.stream_id = os.urandom(4) + self.ultimo_src_id = None + + self.agc_rx = 1.0 + self.agc_tx = 1.0 + self.dtmf_buffer = "" + self.dtmf_task = None + self.ultimo_rtp_timestamp_dtmf = None + self.ultimo_tempo_rtp_dtmf = 0 + self.ultimo_evento_dtmf = None + +class OpenBridgeClient(asyncio.DatagramProtocol): + def __init__(self, manager): + self.transport = None + self.manager = manager + + def connection_made(self, transport): + self.transport = transport + 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')) + sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo] + if not sessioni_interessate: return + + 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.manager.ricevi_frame_dmr_multi(frame_uniti, sessioni_interessate, src_id) + + async def keepalive_loop(self): + while True: + if self.transport: + pacchetto_ping = crea_pacchetto_ping(PEER_ID, HBLINK_PASSWORD) + self.transport.sendto(pacchetto_ping) + await asyncio.sleep(10) + + def apri_flusso_chiamata(self, sess): + if not self.transport: return + flags = 0 + payload_base = struct.pack("!3sB I 4s B I I", b"OPB", 0x80, PEER_ID, sess.stream_id, flags, sess.dmr_id, int(sess.tg_destinazione)) + firma_hmac = calcola_hmac(HBLINK_PASSWORD, payload_base) + self.transport.sendto(payload_base + firma_hmac) + + def invia_dmrd_voce(self, sess, payload_ambe): + if not self.transport: return + try: + magic = b"DMRD" + seq_byte = struct.pack("B", sess.seq_dmr) + src_bytes = sess.dmr_id.to_bytes(3, 'big') + dst_bytes = int(sess.tg_destinazione).to_bytes(3, 'big') + peer_bytes = PEER_ID.to_bytes(4, 'big') + flags_byte = struct.pack("B", 0x00) + + header = magic + seq_byte + src_bytes + dst_bytes + peer_bytes + flags_byte + sess.stream_id + payload_ambe = payload_ambe.ljust(27, b'\x00') + + ambe1 = payload_ambe[0:9] + ambe2 = payload_ambe[9:18] + ambe3 = payload_ambe[18:27] + + payload_33 = bytearray(33) + payload_33[0:9] = ambe1 + payload_33[9:13] = ambe2[0:4] + payload_33[13] = ambe2[4] & 0xF0 + payload_33[19] = ambe2[4] & 0x0F + payload_33[20:24] = ambe2[5:9] + payload_33[24:33] = ambe3 + + pacchetto_base = header + bytes(payload_33) + firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) + self.transport.sendto(pacchetto_base + firma_hmac) + except Exception: pass + + def chiudi_flusso_chiamata(self, sess): + if not self.transport: return + magic = b"DMRD" + flags_byte = struct.pack("B", 0x02) + src_bytes = sess.dmr_id.to_bytes(3, 'big') + dst_bytes = int(sess.tg_destinazione).to_bytes(3, 'big') + peer_bytes = PEER_ID.to_bytes(4, 'big') + + for i in range(3): + seq_byte = struct.pack("B", (sess.seq_dmr + i) % 256) + header = magic + seq_byte + src_bytes + dst_bytes + peer_bytes + flags_byte + sess.stream_id + pacchetto_base = header + (b'\x00' * 33) + firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) + self.transport.sendto(pacchetto_base + firma_hmac) + +class SessionManager: + def __init__(self): + self.vocoder = VocoderEngine(port_start=2470, port_end=2474) + self.openbridge = None + self.rtp_transport = None + self.sip_transport = None + self.sessioni = {} + + async def avvia(self): + await self.vocoder.avvia() + loop = asyncio.get_running_loop() + _, self.openbridge = await loop.create_datagram_endpoint( + lambda: OpenBridgeClient(self), + local_addr=('0.0.0.0', HBLINK_PORT), + remote_addr=(HBLINK_IP, HBLINK_PORT) + ) + print(f"[Core] PBX avviato su {SIP_PORT} | Dashboard su {DASHBOARD_PORT}") + asyncio.create_task(dashboard_app.watchdog_rx()) + + def trova_sessione_per_rtp(self, addr): + for sess in self.sessioni.values(): + if sess.rtp_confermato and sess.rtp_client_addr == addr: return sess + for sess in self.sessioni.values(): + if not sess.rtp_confermato and sess.rtp_client_addr == addr: + sess.rtp_confermato = True; return sess + non_confermate = [s for s in self.sessioni.values() if s.rtp_client_addr and s.rtp_client_addr[0] == addr[0] and not s.rtp_confermato] + if non_confermate: + sess = non_confermate[0] + sess.rtp_client_addr = addr + sess.rtp_confermato = True + return sess + return None + + def gestisci_nuova_chiamata(self, call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps): + sess = SessioneSIP(call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps) + self.sessioni[call_id] = sess + asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"})) + + def termina_chiamata(self, call_id): + if call_id in self.sessioni: + # 1. Pop immediato: la sessione viene sradicata all'istante bloccando nuovi pacchetti DMR + sess = self.sessioni.pop(call_id) + if sess.ptt_attivo: + self.gestisci_ptt(sess, False) + self.vocoder.rilascia_porta(f"SIP_{call_id}") + + # 2. Chiudiamo eventuali righe RX bloccate + for id_dash in list(dashboard_app.chiamate_attive.keys()): + if id_dash.startswith("RX_") and f"_{sess.chiamante}_{sess.call_id_safe}" in id_dash: + dashboard_app.termina_chiamata(id_dash) + + asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"})) + + def gestisci_ptt(self, sess, stato): + id_dash_tx = f"TX_{sess.chiamante}_{sess.call_id_safe}" + if stato is True and not sess.ptt_attivo: + sess.ptt_attivo = True + sess.buffer_ambe = [] + sess.seq_dmr = 0 + sess.stream_id = os.urandom(4) + if self.openbridge: self.openbridge.apri_flusso_chiamata(sess) + + nome_base = get_nominativo(sess.dmr_id) + mittente_display = f"{nome_base} (Est. {sess.chiamante})" + dashboard_app.inizia_chiamata(id_dash_tx, "TX", mittente_display, sess.tg_destinazione) + elif stato is False and sess.ptt_attivo: + sess.ptt_attivo = False + if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) + dashboard_app.termina_chiamata(id_dash_tx) + + def gestisci_dtmf_digit(self, sess, digit): + if sess.ptt_attivo: return + sess.dtmf_buffer += str(digit) + if sess.dtmf_task: sess.dtmf_task.cancel() + sess.dtmf_task = asyncio.create_task(self.attendi_cambio_tg(sess)) + + async def attendi_cambio_tg(self, sess): + try: + await asyncio.sleep(2.5) + if sess.dtmf_buffer: + nuovo_tg = sess.dtmf_buffer + sess.dtmf_buffer = "" + sess.tg_destinazione = nuovo_tg + self.notifica_display(sess, f"TG: {nuovo_tg}") + asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"})) + except asyncio.CancelledError: pass + + def notifica_display(self, sess, testo_display): + if not self.sip_transport or not sess.sip_client_addr: return + sess.local_cseq += 1 + branch = f"z9hG4bK-{os.urandom(4).hex()}" + client_sip_ip, client_sip_port = sess.sip_client_addr + sdp = (f"v=0\r\no=- 12345 12345 IN IP4 {sess.ip_vps}\r\ns=Talk\r\n" + f"c=IN IP4 {sess.ip_vps}\r\nt=0 0\r\nm=audio {RTP_PORT} RTP/AVP 8 101\r\n" + f"a=rtpmap:8 PCMA/8000\r\na=rtpmap:101 telephone-event/8000\r\na=sendrecv\r\n") + sip_reinvite = ( + f"INVITE sip:{sess.chiamante}@{client_sip_ip}:{client_sip_port} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch={branch}\r\n" + f"Max-Forwards: 70\r\nFrom: {sess.dialog_from}\r\nTo: {sess.dialog_to}\r\n" + f"Call-ID: {sess.call_id}\r\nCSeq: {sess.local_cseq} INVITE\r\n" + f"Contact: \r\n" + f"Remote-Party-ID: \"{testo_display}\" ;party=called;screen=yes;privacy=off\r\n" + f"P-Asserted-Identity: \"{testo_display}\" \r\n" + f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}") + self.sip_transport.sendto(sip_reinvite.encode('utf-8'), sess.sip_client_addr) + + def ricevi_frame_dmr_multi(self, ambe_frames, sessioni_interessate, src_id): + ascoltatori = [s for s in sessioni_interessate if not s.ptt_attivo] + if not ascoltatori or not self.rtp_transport: return + + for sess in ascoltatori: + tg_arrivo = sess.tg_destinazione + id_dash_rx = f"RX_{src_id}_{sess.chiamante}_{sess.call_id_safe}" + + if id_dash_rx not in dashboard_app.chiamate_attive: + dest_display = f"{sess.chiamante} (TG {tg_arrivo})" + dashboard_app.inizia_chiamata(id_dash_rx, "RX", src_id, dest_display) + else: + dashboard_app.aggiorna_rx(id_dash_rx) + + if sess.ultimo_src_id != src_id: + sess.ultimo_src_id = src_id + self.notifica_display(sess, f"CID:{src_id}") + + asyncio.create_task(self.processa_audio_dmr(ambe_frames, ascoltatori)) + + async def processa_audio_dmr(self, ambe_27_bytes, ascoltatori): + if not ascoltatori: return + tg_arrivo = ascoltatori[0].tg_destinazione + try: + frames_pcm = [] + 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_bits = convert72BitTo49BitAMBE(BitArray(bytes=frame_9)) + ba_49 = ambe_49_bits if isinstance(ambe_49_bits, BitArray) else BitArray(ambe_49_bits) + 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 + + pcm_data = await self.vocoder.converti_ambe_in_pcm(f"TG_{tg_arrivo}", frame_7) + if pcm_data and len(pcm_data) >= 2: + if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1] + livello_rms = audioop.rms(pcm_data, 2) + picco_max = audioop.max(pcm_data, 2) + if livello_rms > 50: + target_mult = 4000.0 / livello_rms + if target_mult > 2.0: target_mult = 2.0 + elif target_mult < 0.3: target_mult = 0.3 + for sess in ascoltatori: + if target_mult < sess.agc_rx: sess.agc_rx = (sess.agc_rx * 0.90) + (target_mult * 0.10) + else: sess.agc_rx = (sess.agc_rx * 0.99) + (target_mult * 0.01) + moltiplicatore = ascoltatori[0].agc_rx + if picco_max > 0 and (picco_max * moltiplicatore) > 30000: + moltiplicatore_sicuro = 30000 / picco_max + for sess in ascoltatori: sess.agc_rx = (sess.agc_rx * 0.70) + (moltiplicatore_sicuro * 0.30) + moltiplicatore = sess.agc_rx + pcm_data = audioop.mul(pcm_data, 2, moltiplicatore) + pcma_data = audioop.lin2alaw(pcm_data, 2) + frames_pcm.append(pcma_data) + + for pcma_data in frames_pcm: + for sess in ascoltatori: + header = bytearray(12) + header[0] = 0x80; header[1] = 0x08 + header[2:4] = struct.pack("!H", sess.rtp_seq) + header[4:8] = struct.pack("!I", sess.rtp_timestamp) + header[8:12] = struct.pack("!I", 0x12345678) + if sess.rtp_client_addr: self.rtp_transport.sendto(header + pcma_data, sess.rtp_client_addr) + sess.rtp_seq = (sess.rtp_seq + 1) % 65536 + sess.rtp_timestamp = (sess.rtp_timestamp + 160) % 4294967296 + except Exception: pass + + async def processa_audio_rtp(self, sess, rtp_payload): + if not sess.ptt_attivo: return + try: + pcm_lineare = audioop.alaw2lin(rtp_payload, 2) + livello_rms = audioop.rms(pcm_lineare, 2) + if livello_rms > 50: + target_mult = 4000.0 / livello_rms + if target_mult > 2.5: target_mult = 2.5 + elif target_mult < 0.5: target_mult = 0.5 + sess.agc_tx = (sess.agc_tx * 0.90) + (target_mult * 0.10) + pcm_lineare = audioop.mul(pcm_lineare, 2, sess.agc_tx) + + rtp_payload = audioop.lin2alaw(pcm_lineare, 2) + ambe_bytes = await asyncio.wait_for(self.vocoder.converti_rtp_in_ambe(f"SIP_{sess.call_id}", rtp_payload), timeout=0.1) + if not ambe_bytes: return + + dim_frame = 7 if len(ambe_bytes) % 7 == 0 else 9 + for i in range(0, len(ambe_bytes), dim_frame): + frame = ambe_bytes[i:i+dim_frame] + if len(frame) == 7: + stringa_bit = BitArray(bytes=frame).bin + ambe_49_bits = [int(b) for b in stringa_bit[:49]] + sess.buffer_ambe.append(bytes(convert49BitTo72BitAMBE(ambe_49_bits))) + elif len(frame) == 9: + sess.buffer_ambe.append(frame) + + while len(sess.buffer_ambe) >= 3: + payload_vocale = b''.join(sess.buffer_ambe[:3]) + sess.buffer_ambe = sess.buffer_ambe[3:] + self.openbridge.invia_dmrd_voce(sess, payload_vocale) + sess.seq_dmr = (sess.seq_dmr + 1) % 256 + except Exception: pass + +class RtpProtocol(asyncio.DatagramProtocol): + def __init__(self, manager): self.manager = manager + def connection_made(self, transport): self.manager.rtp_transport = transport + def datagram_received(self, data, addr): + sess = self.manager.trova_sessione_per_rtp(addr) + if not sess or len(data) < 12: return + payload_type = data[1] & 0x7F + if payload_type == 8: asyncio.create_task(self.manager.processa_audio_rtp(sess, data[12:])) + elif payload_type == 101: + ora = time.time() + rtp_timestamp = struct.unpack("!I", data[4:8])[0] + evento_dtmf = data[12] + if rtp_timestamp == sess.ultimo_rtp_timestamp_dtmf or (ora - sess.ultimo_tempo_rtp_dtmf < 0.20 and sess.ultimo_evento_dtmf == evento_dtmf): return + sess.ultimo_rtp_timestamp_dtmf = rtp_timestamp + sess.ultimo_tempo_rtp_dtmf = ora + sess.ultimo_evento_dtmf = evento_dtmf + if evento_dtmf == 10: self.manager.gestisci_ptt(sess, True) + elif evento_dtmf == 11: self.manager.gestisci_ptt(sess, False) + elif 0 <= evento_dtmf <= 9: self.manager.gestisci_dtmf_digit(sess, evento_dtmf) + +class SipProtocol(asyncio.DatagramProtocol): + def __init__(self, manager): self.manager = manager + def connection_made(self, transport): self.manager.sip_transport = transport + def datagram_received(self, data, addr): + if addr[0] != FREEPBX_IP and FREEPBX_IP != "0.0.0.0": return + msg = data.decode('utf-8', errors='ignore').lstrip() + lines = msg.splitlines() + headers = {line.split(":", 1)[0].strip().upper(): line.split(":", 1)[1].strip() for line in lines[1:] if ":" in line} + call_id = headers.get('CALL-ID', '') + + if msg.startswith("OPTIONS"): + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContact: \r\nContent-Length: 0\r\n\r\n") + self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) + elif msg.startswith("INVITE"): + client_rtp_port = 4000 + for line in lines: + if line.strip().startswith("m=audio "): client_rtp_port = int(line.strip().split()[1]) + ip_vps = headers.get("TO", "").split("@")[1].split(">")[0] if "@" in headers.get("TO", "") else "127.0.0.1" + chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0] + destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] + self.manager.gestisci_nuova_chiamata(call_id, chiamante, destinatario, (addr[0], client_rtp_port), addr, headers.get('FROM', ''), headers.get('TO', ''), ip_vps) + sdp = (f"v=0\r\no=- 12345 12345 IN IP4 {ip_vps}\r\ns=Talk\r\n" + f"c=IN IP4 {ip_vps}\r\nt=0 0\r\nm=audio {RTP_PORT} RTP/AVP 8 101\r\n" + f"a=rtpmap:8 PCMA/8000\r\na=rtpmap:101 telephone-event/8000\r\na=sendrecv\r\n") + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')};tag=987654\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContact: \r\n" + f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}") + self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) + elif msg.startswith("INFO"): + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContent-Length: 0\r\n\r\n") + self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) + sess = self.manager.sessioni.get(call_id) + if sess: + for line in lines: + if line.upper().startswith("SIGNAL="): + signal = line.split("=")[1].strip() + evento = 10 if signal == "*" else 11 if signal == "#" else int(signal) if signal.isdigit() else None + if evento == 10: self.manager.gestisci_ptt(sess, True) + elif evento == 11: self.manager.gestisci_ptt(sess, False) + elif evento is not None and 0 <= evento <= 9: self.manager.gestisci_dtmf_digit(sess, evento) + elif msg.startswith("SIP/2.0 200 OK") and "INVITE" in headers.get('CSEQ', ''): + ack_uri = headers["CONTACT"].split("<")[1].split(">")[0] if "CONTACT" in headers and "<" in headers["CONTACT"] else f"sip:{addr[0]}:{addr[1]}" + ack = (f"ACK {ack_uri} SIP/2.0\r\nVia: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch=z9hG4bK-{os.urandom(4).hex()}\r\n" + f"Max-Forwards: 70\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '').split()[0]} ACK\r\nContent-Length: 0\r\n\r\n") + self.manager.sip_transport.sendto(ack.encode('utf-8'), addr) + elif msg.startswith("BYE") or msg.startswith("CANCEL"): + if call_id in self.manager.sessioni: + self.manager.termina_chiamata(call_id) + else: + # FALLBACK LETALE: Se il PBX altera il Call-ID, forza la chiusura usando l'estensione + for hdr in ["FROM", "TO"]: + val = headers.get(hdr, "") + if "sip:" in val: + ext = val.split("sip:")[1].split("@")[0] + for cid, s in list(self.manager.sessioni.items()): + if s.chiamante == ext: + self.manager.termina_chiamata(cid) + + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContent-Length: 0\r\n\r\n") + self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) + +async def main(): + global sip_manager + loop = asyncio.get_running_loop() + + # Motore SIP/RTP + manager = SessionManager() + sip_manager = manager + await manager.avvia() + await loop.create_datagram_endpoint(lambda: SipProtocol(manager), local_addr=('0.0.0.0', SIP_PORT)) + await loop.create_datagram_endpoint(lambda: RtpProtocol(manager), local_addr=('0.0.0.0', RTP_PORT)) + + # Motore Web API Dashboard + app = web.Application() + app.router.add_get("/", index_dashboard) + app.router.add_get("/api/storico", api_storico) + app.router.add_get("/ws", ws_handler) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, '0.0.0.0', DASHBOARD_PORT) + await site.start() + + try: await asyncio.Event().wait() + except KeyboardInterrupt: pass + +if __name__ == "__main__": + asyncio.run(main())