diff --git a/core.py b/core.py index 59853eb..aaf757a 100644 --- a/core.py +++ b/core.py @@ -29,8 +29,6 @@ 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) -VOCODER_HOST = CONFIG.get("vocoder_host", "127.0.0.1") -VOCODER_PORT = CONFIG.get("vocoder_port", 2470) FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") def carica_rubrica(): @@ -42,37 +40,28 @@ def carica_rubrica(): UTENTI_SIP_DMR = carica_rubrica() -# ==================================================================== -# CLASSE SESSIONE SIP -# ==================================================================== 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.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 - - # Flag per la gestione NAT delle porte RTP 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 - # --- VARIABILI DTMF --- self.dtmf_buffer = "" self.dtmf_task = None self.ultimo_rtp_timestamp_dtmf = None @@ -90,26 +79,21 @@ class OpenBridgeClient(asyncio.DatagramProtocol): asyncio.create_task(self.keepalive_loop()) def datagram_received(self, data, addr): - if len(data) < 53: - return - + 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 print("D", end="", flush=True) 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) @@ -173,12 +157,9 @@ class OpenBridgeClient(asyncio.DatagramProtocol): firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) self.transport.sendto(pacchetto_base + firma_hmac) -# ==================================================================== -# GESTORE CENTRALE DELLE SESSIONI -# ==================================================================== class SessionManager: def __init__(self): - self.vocoder = VocoderEngine(mode="emulator", host=VOCODER_HOST, port=VOCODER_PORT) + self.vocoder = VocoderEngine() self.openbridge = None self.rtp_transport = None self.sip_transport = None @@ -193,41 +174,33 @@ class SessionManager: remote_addr=(HBLINK_IP, HBLINK_PORT) ) print(f"[Core] PBX Multi-Utente avviato. SIP su porta {SIP_PORT}...\n") - print(f"[Security] Accetto connessioni SIP SOLO da: {FREEPBX_IP}\n") def trova_sessione_per_rtp(self, addr): - # 1. Match tra le sessioni già confermate (IP + Porta NAT reale) for sess in self.sessioni.values(): - if sess.rtp_confermato and sess.rtp_client_addr == addr: - return sess - - # 2. Match esatto per sessioni non ancora confermate (es. senza NAT) + 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 - - # 3. Gestione NAT: Assegna la porta NAT alla prima sessione ancora in attesa per quell'IP + 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 - print(f"[RTP] Flusso Audio/DTMF agganciato per l'utente {sess.chiamante} ({addr[0]}:{addr[1]})") 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 - print(f"\n[Core] >>> SIP {chiamante} -> TG {destinatario} | ID: {sess.dmr_id}") + print(f"\n[Core] >>> SIP {chiamante} -> TG {destinatario}") def termina_chiamata(self, call_id): if call_id in self.sessioni: sess = self.sessioni[call_id] if sess.ptt_attivo and self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) + # Rilascia la porta del pool associata a questa chiamata + self.vocoder.rilascia_porta(f"SIP_{call_id}") del self.sessioni[call_id] def gestisci_ptt(self, sess, stato): @@ -243,13 +216,8 @@ class SessionManager: def gestisci_dtmf_digit(self, sess, digit): if sess.ptt_attivo: return - - print(f"[DTMF] Utente {sess.chiamante} ha premuto: '{digit}'") sess.dtmf_buffer += str(digit) - - if sess.dtmf_task: - sess.dtmf_task.cancel() - + 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): @@ -258,60 +226,49 @@ class SessionManager: if sess.dtmf_buffer: nuovo_tg = sess.dtmf_buffer sess.dtmf_buffer = "" - print(f"\n[Core] *** CAMBIO TG: {sess.chiamante} è passato al Talkgroup {nuovo_tg} ***") sess.tg_destinazione = nuovo_tg self.notifica_display(sess, f"TG: {nuovo_tg}") - except asyncio.CancelledError: - pass + except asyncio.CancelledError: pass def notifica_display(self, sess, testo_display): - if not self.sip_transport or not sess.sip_client_addr: - return - + 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\n" - f"From: {sess.dialog_from}\r\n" - f"To: {sess.dialog_to}\r\n" - f"Call-ID: {sess.call_id}\r\n" - f"CSeq: {sess.local_cseq} INVITE\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\n" - f"Content-Length: {len(sdp)}\r\n" - f"\r\n" - f"{sdp}" - ) + 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: if sess.ultimo_src_id != src_id: sess.ultimo_src_id = src_id self.notifica_display(sess, f"RX: {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: + # 1. Decodifichiamo prima tutti e 3 i frame del superframe DMR + 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) @@ -319,31 +276,41 @@ class SessionManager: frame_7 = ba_49.bytes[:7] except Exception: continue - pcm_data = await self.vocoder.converti_ambe_in_pcm(frame_7) - if not pcm_data or len(pcm_data) < 2: continue - if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1] - pcma_data = audioop.lin2alaw(pcm_data, 2) - + 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] + pcma_data = audioop.lin2alaw(pcm_data, 2) + frames_pcm.append(pcma_data) + + # 2. Spediamo i pacchetti RTP cadenzandoli a ritmo di 19ms (sgranatura) + for idx, pcma_data in enumerate(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) - rtp_packet = header + pcma_data + if sess.rtp_client_addr: self.rtp_transport.sendto(rtp_packet, sess.rtp_client_addr) - + sess.rtp_seq = (sess.rtp_seq + 1) % 65536 sess.rtp_timestamp = (sess.rtp_timestamp + 160) % 4294967296 - except Exception: - pass + + # Attesa di ~19ms tra un frame audio e il successivo (tranne l'ultimo) + if idx < len(frames_pcm) - 1: + await asyncio.sleep(0.019) + + except Exception: pass async def processa_audio_rtp(self, sess, rtp_payload): if not sess.ptt_attivo: return try: - ambe_bytes = await asyncio.wait_for(self.vocoder.converti_rtp_in_ambe(rtp_payload), timeout=0.1) + # *** PASSIAMO IL CALL ID PER IL POOL *** + 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 @@ -362,19 +329,14 @@ class SessionManager: 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 + except Exception: pass -# ==================================================================== -# PROTOCOLLI DI RETE SIP/RTP -# ==================================================================== 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:])) @@ -382,78 +344,51 @@ class RtpProtocol(asyncio.DatagramProtocol): 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 - + 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) + 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.transport = transport - self.manager.sip_transport = transport + 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 - + if addr[0] != FREEPBX_IP and FREEPBX_IP != "0.0.0.0": return msg = data.decode('utf-8', errors='ignore') headers = {line.split(":", 1)[0].strip().upper(): line.split(":", 1)[1].strip() for line in msg.split('\r\n')[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\n" - f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" - f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" - f"Contact: \r\n" - f"Content-Length: 0\r\n\r\n") - self.transport.sendto(resp.encode('utf-8'), addr) - + 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 msg.split('\n'): line = line.strip() if line.startswith("m=audio "): client_rtp_port = int(line.split()[1]) - ip_vps = "127.0.0.1" if "@" in headers.get("TO", ""): ip_vps = headers.get("TO", "").split("@")[1].split(">")[0] - chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0] destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] - full_from = headers.get('FROM', '') full_to = headers.get('TO', '') - self.manager.gestisci_nuova_chiamata(call_id, chiamante, destinatario, (addr[0], client_rtp_port), addr, full_from, full_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\n" - f"From: {full_from}\r\nTo: {full_to};tag=987654\r\n" - f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" - f"Contact: \r\nContent-Type: application/sdp\r\n" - f"Content-Length: {len(sdp)}\r\n\r\n{sdp}") - self.transport.sendto(resp.encode('utf-8'), addr) - + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {full_from}\r\nTo: {full_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\n" - f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" - f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" - f"Content-Length: 0\r\n\r\n") - self.transport.sendto(resp.encode('utf-8'), addr) - + 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 msg.split('\r\n'): @@ -463,50 +398,32 @@ class SipProtocol(asyncio.DatagramProtocol): elif signal == "#": evento = 11 elif signal.isdigit(): evento = int(signal) else: continue - if evento == 10: self.manager.gestisci_ptt(sess, True) elif evento == 11: self.manager.gestisci_ptt(sess, False) elif 0 <= evento <= 9: self.manager.gestisci_dtmf_digit(sess, evento) - elif msg.startswith("SIP/2.0 200 OK"): if "INVITE" in headers.get('CSEQ', ''): cseq_num = headers.get('CSEQ', '').split()[0] ack_uri = f"sip:{addr[0]}:{addr[1]}" - if "CONTACT" in headers: - contact_hdr = headers["CONTACT"] - if "<" in contact_hdr: - ack_uri = contact_hdr.split("<")[1].split(">")[0] - - ack = (f"ACK {ack_uri} SIP/2.0\r\n" - f"Via: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch=z9hG4bK-{os.urandom(4).hex()}\r\n" - f"Max-Forwards: 70\r\n" - f"From: {headers.get('FROM', '')}\r\n" - f"To: {headers.get('TO', '')}\r\n" - f"Call-ID: {call_id}\r\n" - f"CSeq: {cseq_num} ACK\r\n" - f"Content-Length: 0\r\n\r\n") - self.transport.sendto(ack.encode('utf-8'), addr) - + if "CONTACT" in headers and "<" in headers["CONTACT"]: ack_uri = headers["CONTACT"].split("<")[1].split(">")[0] + 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: {cseq_num} ACK\r\nContent-Length: 0\r\n\r\n") + self.manager.sip_transport.sendto(ack.encode('utf-8'), addr) elif msg.startswith("BYE"): self.manager.termina_chiamata(call_id) - resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" - f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" - f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" - f"Content-Length: 0\r\n\r\n") - self.transport.sendto(resp.encode('utf-8'), addr) + 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(): loop = asyncio.get_running_loop() manager = SessionManager() 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)) - - try: - await asyncio.Event().wait() - except KeyboardInterrupt: - pass + try: await asyncio.Event().wait() + except KeyboardInterrupt: pass if __name__ == "__main__": asyncio.run(main()) diff --git a/core.py.con.bug.multiutenza b/core.py.con.bug.multiutenza new file mode 100644 index 0000000..efedf8a --- /dev/null +++ b/core.py.con.bug.multiutenza @@ -0,0 +1,480 @@ +import json +from bitstring import BitArray +import asyncio +import struct +import time +import os +import audioop +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) +# ==================================================================== +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 o errato ({e}). Chiusura.") + 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) +VOCODER_HOST = CONFIG.get("vocoder_host", "127.0.0.1") +VOCODER_PORT = CONFIG.get("vocoder_port", 2470) +# --- NUOVA VARIABILE PER LA SICUREZZA IP --- +FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") + +def carica_rubrica(): + try: + with open("utenti.json", "r") as f: + return json.load(f) + except Exception as e: + return {} + +UTENTI_SIP_DMR = carica_rubrica() + +# ==================================================================== +# CLASSE SESSIONE SIP (Con Buffer DTMF per il cambio TG) +# ==================================================================== +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.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.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 + + # --- VARIABILI PER IL CAMBIO TG TRAMITE DTMF --- + self.dtmf_buffer = "" + self.dtmf_task = 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 + print(f"[OpenBridge] Socket connesso a {HBLINK_IP}:{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')) + + # Il routing è dinamico: filtra in base al tg_destinazione attuale della sessione + sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo] + if not sessioni_interessate: return + + print("D", end="", flush=True) + 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) + +# ==================================================================== +# GESTORE CENTRALE DELLE SESSIONI +# ==================================================================== +class SessionManager: + def __init__(self): + self.vocoder = VocoderEngine(mode="emulator", host=VOCODER_HOST, port=VOCODER_PORT) + 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 Multi-Utente avviato. SIP su porta {SIP_PORT}...\n") + print(f"[Security] Accetto connessioni SIP SOLO da: {FREEPBX_IP}\n") + + def trova_sessione_per_rtp(self, addr): + for sess in self.sessioni.values(): + if sess.rtp_client_addr and sess.rtp_client_addr[0] == addr[0]: + sess.rtp_client_addr = addr + 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 + print(f"\n[Core] >>> SIP {chiamante} -> TG {destinatario} | ID: {sess.dmr_id}") + + def termina_chiamata(self, call_id): + if call_id in self.sessioni: + sess = self.sessioni[call_id] + if sess.ptt_attivo and self.openbridge: + self.openbridge.chiudi_flusso_chiamata(sess) + del self.sessioni[call_id] + + def gestisci_ptt(self, sess, stato): + 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) + elif stato is False and sess.ptt_attivo: + sess.ptt_attivo = False + if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) + + # --- LOGICA CAMBIO TG (DTMF 0-9) --- + def gestisci_dtmf_digit(self, sess, digit): + if sess.ptt_attivo: return # Evita di cambiare TG mentre sei in TX + + sess.dtmf_buffer += str(digit) + + # Resetta il task esistente se si preme un altro numero velocemente + 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: + # Attende 2.5 secondi dalla pressione dell'ultimo tasto + await asyncio.sleep(2.5) + if sess.dtmf_buffer: + nuovo_tg = sess.dtmf_buffer + sess.dtmf_buffer = "" # Svuota il buffer + + print(f"\n[Core] *** CAMBIO TG: {sess.chiamante} è passato al Talkgroup {nuovo_tg} ***") + sess.tg_destinazione = nuovo_tg + + # Notifica il telefono del cambio avvenuto + self.notifica_display(sess, f"TG: {nuovo_tg}") + except asyncio.CancelledError: + pass # Il task è stato annullato perché l'utente ha premuto un altro tasto + + # --- FUNZIONE DISPLAY GENERICA (Ora accetta stringhe personalizzate) --- + 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\n" + f"From: {sess.dialog_from}\r\n" + f"To: {sess.dialog_to}\r\n" + f"Call-ID: {sess.call_id}\r\n" + f"CSeq: {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\n" + f"Content-Length: {len(sdp)}\r\n" + f"\r\n" + f"{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: + if sess.ultimo_src_id != src_id: + sess.ultimo_src_id = src_id + self.notifica_display(sess, f"RX: {src_id}") + + asyncio.create_task(self.processa_audio_dmr(ambe_frames, ascoltatori)) + + async def processa_audio_dmr(self, ambe_27_bytes, ascoltatori): + 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_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(frame_7) + if not pcm_data or len(pcm_data) < 2: continue + if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1] + pcma_data = audioop.lin2alaw(pcm_data, 2) + + 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) + + rtp_packet = header + pcma_data + if sess.rtp_client_addr: + self.rtp_transport.sendto(rtp_packet, 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: + ambe_bytes = await asyncio.wait_for(self.vocoder.converti_rtp_in_ambe(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]] + fec_9_byte = convert49BitTo72BitAMBE(ambe_49_bits) + sess.buffer_ambe.append(bytes(fec_9_byte)) + 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 + +# ==================================================================== +# PROTOCOLLI DI RETE SIP/RTP +# ==================================================================== +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: + evento_dtmf = data[12] + + # Meccanismo anti-rimbalzo (Debounce) per ignorare i frame duplicati + ora = time.time() + if ora - sess.ultimo_tempo_rtp_dtmf < 0.2 and sess.ultimo_evento_dtmf == evento_dtmf: + return + sess.ultimo_tempo_rtp_dtmf = ora + sess.ultimo_evento_dtmf = evento_dtmf + + # Esecuzione dei Comandi + 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.transport = transport + self.manager.sip_transport = transport + + def datagram_received(self, data, addr): + # --- SICUREZZA: BLOCCO IP --- + if addr[0] != FREEPBX_IP and FREEPBX_IP != "0.0.0.0": + # Decommenta la riga sotto se vuoi vedere i log dei tentativi bloccati + # print(f"[SECURITY] Rifiutato pacchetto SIP da IP non autorizzato: {addr[0]}") + return + # ---------------------------- + + msg = data.decode('utf-8', errors='ignore') + headers = {line.split(":", 1)[0].strip().upper(): line.split(":", 1)[1].strip() for line in msg.split('\r\n')[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\n" + f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" + f"Contact: \r\n" + f"Content-Length: 0\r\n\r\n") + self.transport.sendto(resp.encode('utf-8'), addr) + + elif msg.startswith("INVITE"): + client_rtp_port = 4000 + for line in msg.split('\n'): + line = line.strip() + if line.startswith("m=audio "): client_rtp_port = int(line.split()[1]) + + ip_vps = "127.0.0.1" + if "@" in headers.get("TO", ""): ip_vps = headers.get("TO", "").split("@")[1].split(">")[0] + + chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0] + destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] + + full_from = headers.get('FROM', '') + full_to = headers.get('TO', '') + + self.manager.gestisci_nuova_chiamata(call_id, chiamante, destinatario, (addr[0], client_rtp_port), addr, full_from, full_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\n" + f"From: {full_from}\r\nTo: {full_to};tag=987654\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" + f"Contact: \r\nContent-Type: application/sdp\r\n" + f"Content-Length: {len(sdp)}\r\n\r\n{sdp}") + self.transport.sendto(resp.encode('utf-8'), addr) + + elif msg.startswith("SIP/2.0 200 OK"): + if "INVITE" in headers.get('CSEQ', ''): + cseq_num = headers.get('CSEQ', '').split()[0] + ack_uri = f"sip:{addr[0]}:{addr[1]}" + if "CONTACT" in headers: + contact_hdr = headers["CONTACT"] + if "<" in contact_hdr: + ack_uri = contact_hdr.split("<")[1].split(">")[0] + + ack = (f"ACK {ack_uri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch=z9hG4bK-{os.urandom(4).hex()}\r\n" + f"Max-Forwards: 70\r\n" + f"From: {headers.get('FROM', '')}\r\n" + f"To: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\n" + f"CSeq: {cseq_num} ACK\r\n" + f"Content-Length: 0\r\n\r\n") + self.transport.sendto(ack.encode('utf-8'), addr) + + elif msg.startswith("BYE"): + self.manager.termina_chiamata(call_id) + resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" + f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" + f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" + f"Content-Length: 0\r\n\r\n") + self.transport.sendto(resp.encode('utf-8'), addr) + +async def main(): + loop = asyncio.get_running_loop() + manager = SessionManager() + 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)) + + try: + await asyncio.Event().wait() + except KeyboardInterrupt: + pass + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/multi_vocoder.py b/multi_vocoder.py new file mode 100644 index 0000000..815c20c --- /dev/null +++ b/multi_vocoder.py @@ -0,0 +1,47 @@ +import socket +import subprocess +import os +import threading +import time + +BIN_PATH = "/opt/md380-emu/md380-emu" +SHM_DIR = "/dev/shm" + +def gestisci_emulatore(porta): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', porta)) + print(f"[*] Vocoder attivo in ascolto sulla porta UDP {porta}") + + while True: + try: + data, addr = sock.recvfrom(1024) + if not data: continue + + # Percorsi file univoci per ogni porta direttamente nella RAM + f_in = f"{SHM_DIR}/in_{porta}.bin" + f_out = f"{SHM_DIR}/out_{porta}.bin" + + with open(f_in, "wb") as f: + f.write(data) + + # Se riceviamo 7 byte, è AMBE da decodificare in PCM. Altrimenti è PCM da codificare. + if len(data) == 7 or len(data) == 9: + subprocess.run([BIN_PATH, "-d", "-i", f_in, "-o", f_out], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + else: + subprocess.run([BIN_PATH, "-e", "-i", f_in, "-o", f_out], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + if os.path.exists(f_out): + with open(f_out, "rb") as f: + sock.sendto(f.read(), addr) + except Exception: + pass + +# Avviamo 3 vocoder indipendenti in background +for p in [2470, 2471, 2472]: + threading.Thread(target=gestisci_emulatore, args=(p,), daemon=True).start() + +try: + while True: + time.sleep(1) +except KeyboardInterrupt: + print("Chiusura del pool...") diff --git a/test_ambe.py b/test_ambe.py new file mode 100644 index 0000000..77381ae --- /dev/null +++ b/test_ambe.py @@ -0,0 +1,37 @@ +import socket +import time + +# Questo è uno dei frame AMBE reali estratti dal tuo log +ambe_frame = bytes.fromhex("88 1D 2E 13 85 CD 00") +server = ('127.0.0.1', 2470) + +sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +sock.settimeout(2.0) + +test_cases = { + "1_Busta_Little_Endian": b'\x00\x07\x00' + ambe_frame, + "2_Busta_Big_Endian": b'\x00\x00\x07' + ambe_frame, + "3_Dati_Raw_Senza_Busta": ambe_frame, + "4_Busta_Tipo_1": b'\x01\x07\x00' + ambe_frame, + "5_Busta_Tipo_1_Big": b'\x01\x00\x07' + ambe_frame, +} + +print("Inizio test diagnostico UDP verso AMBEServer3003 sulla porta 2470...\n") + +for name, pkt in test_cases.items(): + print(f"---> Eseguo test: [{name}]") + + # Inviamo 3 frame identici in rapida successione. + # (Alcuni chip AMBE hardware bufferizzano 2-3 frame prima di sputare fuori il primo pacchetto PCM) + for _ in range(3): + sock.sendto(pkt, server) + time.sleep(0.02) + + try: + # Attendiamo la risposta + resp, addr = sock.recvfrom(2048) + print(f" BINGO! Risposta ricevuta ({len(resp)} bytes): {resp[:12].hex(' ')}...\n") + except socket.timeout: + print(" FALLITO: Nessuna risposta dal server.\n") + +sock.close() diff --git a/vocoder_engine.py b/vocoder_engine.py index 94fb872..73fdf62 100644 --- a/vocoder_engine.py +++ b/vocoder_engine.py @@ -1,47 +1,99 @@ import asyncio +import time import audioop -class VocoderEngine: - def __init__(self, mode="emulator", host="127.0.0.1", port=2470): - self.mode = mode - self.host = host - self.port = port +class UdpClientProtocol(asyncio.DatagramProtocol): + def __init__(self): self.transport = None - self.coda_risposte = asyncio.Queue() - # IL SEMAFORO ASINCRONO PER IL MULTI-UTENTE - self.lock = asyncio.Lock() + self.future = None + + def connection_made(self, transport): + self.transport = transport + + def datagram_received(self, data, addr): + if self.future and not self.future.done(): + self.future.set_result(data) + +class VocoderEngine: + def __init__(self, mode="emulator", host="127.0.0.1"): + self.host = host + self.ports = list(range(2470, 2480)) + self.pool = {} # porta -> (transport, protocol, lock) + self.stream_alloc = {} # stream_id -> porta + self.last_seen = {} # stream_id -> timestamp ultimo pacchetto async def avvia(self): loop = asyncio.get_running_loop() - class AmbedProtocol(asyncio.DatagramProtocol): - def __init__(self, coda): self.coda = coda - def connection_made(self, transport): pass - def datagram_received(self, data, addr): self.coda.put_nowait(data) - def error_received(self, exc): pass + for p in self.ports: + transport, protocol = await loop.create_datagram_endpoint( + lambda: UdpClientProtocol(), + remote_addr=(self.host, p) + ) + # Aggiungiamo un Lock esclusivo per ogni porta UDP + self.pool[p] = (transport, protocol, asyncio.Lock()) + print(f"[Vocoder] Pool UDP Multi-Porta Sincronizzato: {self.ports}") - self.transport, _ = await loop.create_datagram_endpoint( - lambda: AmbedProtocol(self.coda_risposte), - remote_addr=(self.host, self.port) - ) - print(f"[Vocoder] Motore avviato. Modalità: {self.mode.upper()}") - print(f"[Vocoder] Connesso a {self.host}:{self.port}") + def pulisci_stream_inattivi(self): + """Libera automaticamente le porte dei Talkgroup/SIP muti da più di 2.5 secondi""" + ora = time.time() + inattivi = [s for s, t in self.last_seen.items() if ora - t > 2.5] + for s in inattivi: + if s in self.stream_alloc: + del self.stream_alloc[s] + del self.last_seen[s] - async def converti_rtp_in_ambe(self, rtp_payload_alaw): - pcm_data = audioop.alaw2lin(rtp_payload_alaw, 2) - if self.mode == "emulator": - async with self.lock: # <--- Aspetta il suo turno se occupato - self.transport.sendto(pcm_data) - try: return await asyncio.wait_for(self.coda_risposte.get(), timeout=0.1) - except asyncio.TimeoutError: return None + def assegna_porta(self, stream_id): + self.pulisci_stream_inattivi() + self.last_seen[stream_id] = time.time() - async def converti_ambe_in_pcm(self, ambe_bytes): - if self.mode == "emulator": - async with self.lock: # <--- Aspetta il suo turno se occupato - self.transport.sendto(ambe_bytes) - try: return await asyncio.wait_for(self.coda_risposte.get(), timeout=0.1) - except asyncio.TimeoutError: return None + if stream_id in self.stream_alloc: + return self.stream_alloc[stream_id] - def ferma(self): - if self.transport: - self.transport.close() - print("[Vocoder] Motore spento.") + porte_usate = set(self.stream_alloc.values()) + porte_libere = [p for p in self.ports if p not in porte_usate] + + if porte_libere: + porta_scelta = porte_libere[0] + else: + # Se tutte le 3 porte sono saturate contemporaneamente, usiamo la prima come fallback + porta_scelta = self.ports[0] + + self.stream_alloc[stream_id] = porta_scelta + return porta_scelta + + def rilascia_porta(self, stream_id): + if stream_id in self.stream_alloc: + del self.stream_alloc[stream_id] + if stream_id in self.last_seen: + del self.last_seen[stream_id] + + async def _scambia_dati(self, porta, dati): + transport, protocol, lock = self.pool[porta] + + # Garantisce che un solo pacchetto alla volta usi questa porta UDP! + async with lock: + loop = asyncio.get_running_loop() + protocol.future = loop.create_future() + transport.sendto(dati) + try: + return await asyncio.wait_for(protocol.future, timeout=0.08) + except asyncio.TimeoutError: + return b"" + finally: + protocol.future = None + + async def converti_ambe_in_pcm(self, stream_id, ambe_bytes): + porta = self.assegna_porta(stream_id) + # Invio dati RAW + return await self._scambia_dati(porta, ambe_bytes) + + async def converti_rtp_in_ambe(self, stream_id, pcm_bytes): + porta = self.assegna_porta(stream_id) + + # Se riceviamo 160 byte dal SIP, li convertiamo in PCM lineare 16-bit (320 byte) + if len(pcm_bytes) == 160: + # Uso alaw2lin per PCMA (A-Law, standard europeo). + # (Se il tuo centralino usa u-Law, ti basterà cambiare in ulaw2lin) + pcm_bytes = audioop.alaw2lin(pcm_bytes, 2) + + return await self._scambia_dati(porta, pcm_bytes)