feat: fix CallerID dinamico su PBX e aggiunto AGC audio bidirezionale

README.md

@@ -10,6 +10,9 @@ Il sistema gestisce l'handshake SIP, la transcodifica audio in tempo reale tra P
 * **Cambio Canale al Volo:** Digita il numero del Talkgroup sul telefono e il gateway ti sposterà di stanza senza far cadere la chiamata.
 * **Sicurezza IP:** Rifiuta nativamente le connessioni SIP da IP non autorizzati.
 * **Aggiornamento Display:** Usa il protocollo `re-INVITE` per mostrare Caller ID (DMR ID) e Talkgroup attivi sui telefoni hardware compatibili.
+* **Transcoding 100% Software:** Nessun hardware dedicato richiesto (es. DVStick). Il sistema usa l'emulatore nativo `md380-emu` elaborando l'audio crudo (RAW) direttamente tramite CPU.
+* **Multi-Tenancy Dinamica:** Gestisce chiamate simultanee multiple. Il motore alloca e dealloca dinamicamente le porte UDP instradando gli stream audio in tempo reale sui vocoder liberi.
+* **Audio Bidirezionale G.711 / AMBE:** Converte al volo i pacchetti SIP A-Law in PCM lineare 16-bit (tramite modulo `audioop`) per la rete DMR, e viceversa.
 
 ---
 
@@ -26,31 +29,36 @@ sudo apt install -y git build-essential python3 python3-venv python3-pip
 
 ---
 
-## 📻 2. Installazione Motore Audio (md380-emu)
+## 📻 2. Installazione Motore Audio Scalabile (md380-emu)
 
-Il cuore della conversione audio è l'emulatore open-source `md380-emu`. Va scaricato, compilato e lasciato in esecuzione in background.
+Il cuore della conversione audio è l'emulatore open-source `md380-emu`. Va scaricato, compilato e configurato come "template" Systemd per supportare un pool di porte.
 
 ```bash
-cd /opt
+# 1. Crea la cartella di destinazione
-sudo git clone https://github.com/nostar/md380-emu.git
+sudo mkdir -p /opt/md380-emu
-cd md380-emu
+
-sudo make
+# 2. Estrai i binari inclusi nel repository
+sudo tar -xzvf bin/md380-emu-bin.tar.gz -C /opt/md380-emu/
+
+# 3. Assicurati che abbiano i permessi di esecuzione
+sudo chmod +x /opt/md380-emu/md380-emu /opt/md380-emu/qemu-arm-static
 ```
 
-**Creazione del demone Systemd per l'emulatore:**
+**Creazione del demone Systemd Template per le istanze multiple:**
 
-Crea il file `/etc/systemd/system/md380-emu.service`:
+Crea il file `/etc/systemd/system/md380-emu@.service`:
 
 ```ini
 [Unit]
-Description=MD380 AMBE Emulator
+Description=MD380 Emulator Vocoder (Port %i)
 After=network.target
 
 [Service]
 Type=simple
 User=root
 WorkingDirectory=/opt/md380-emu
-ExecStart=/opt/md380-emu/emulator -p 2470
+# Il %i verrà sostituito automaticamente con il numero di porta
+ExecStart=/opt/md380-emu/emulator -p %i
 Restart=always
 RestartSec=3
 
@@ -58,12 +66,19 @@ RestartSec=3
 WantedBy=multi-user.target
 ```
 
-Abilita e avvia il servizio:
+**Attivazione delle istanze (es. per 10 canali simultanei):**
+
+Ricarica i demoni e avvia il loop di porte (es. dalla 2470 alla 2479):
 
 ```bash
-sudo systemctl enable md380-emu --now
+sudo systemctl daemon-reload
+for port in {2470..2479}; do
+    sudo systemctl enable --now md380-emu@$port.service
+done
 ```
 
+*(Per aumentare la capacità futura, ti basterà avviare nuovi servizi systemctl ed espandere il range di porte generate in `vocoder_engine.py`).*
+
 ---
 
 ## 💻 3. Installazione del Gateway Python
@@ -72,7 +87,7 @@ Clona questo repository e configura l'ambiente virtuale:
 
 ```bash
 cd /opt
-git clone https://git.arifvg.it/iv3jdv/sip-dmr-gateway.git
+git clone git://git.arifvg.it/iv3jdv/sip-dmr-gateway.git
 cd sip-dmr-gateway
 
 # Crea e attiva il Virtual Environment
@@ -126,7 +141,7 @@ Per far avviare il Gateway automaticamente al boot, crea il file `/etc/systemd/s
 ```ini
 [Unit]
 Description=Gateway SIP-DMR Core
-After=network.target md380-emu.service
+After=network.target
 
 [Service]
 Type=simple
@@ -153,4 +168,4 @@ Per visualizzare i log in tempo reale e monitorare il traffico radio:
 
 ```bash
 journalctl -u sipdmr -f
-```
+```

core.py

@@ -29,9 +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)
-# --- NUOVA VARIABILE PER LA SICUREZZA IP ---
 FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1")
 
 def carica_rubrica():
@@ -43,36 +40,31 @@ def carica_rubrica():
 
 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.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 PER IL CAMBIO TG TRAMITE DTMF ---
         self.dtmf_buffer = ""
         self.dtmf_task = None
+        self.ultimo_rtp_timestamp_dtmf = None
         self.ultimo_tempo_rtp_dtmf = 0
         self.ultimo_evento_dtmf = None
 
@@ -87,27 +79,21 @@ class OpenBridgeClient(asyncio.DatagramProtocol):
         asyncio.create_task(self.keepalive_loop())
 
     def datagram_received(self, data, addr):
-        if len(data) < 53: 
+        if len(data) < 53: return
-            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)
 
@@ -171,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 
@@ -191,25 +174,32 @@ 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):
         for sess in self.sessioni.values():
-            if sess.rtp_client_addr and sess.rtp_client_addr[0] == addr[0]:
+            if sess.rtp_confermato and sess.rtp_client_addr == addr: return sess
-                sess.rtp_client_addr = addr 
+        for sess in self.sessioni.values():
-                return sess
+            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
-        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)
+            self.vocoder.rilascia_porta(f"SIP_{call_id}")
             del self.sessioni[call_id]
 
     def gestisci_ptt(self, sess, stato):
@@ -223,82 +213,60 @@ class SessionManager:
             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
+        if sess.ptt_attivo: return
-        
         sess.dtmf_buffer += str(digit)
-        
+        if sess.dtmf_task: sess.dtmf_task.cancel()
-        # 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
+                sess.dtmf_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:
+        except asyncio.CancelledError: pass
-            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:
+        if not self.sip_transport or not sess.sip_client_addr: return
-            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"Max-Forwards: 70\r\nFrom: {sess.dialog_from}\r\nTo: {sess.dialog_to}\r\n"
-            f"From: {sess.dialog_from}\r\n"
+            f"Call-ID: {sess.call_id}\r\nCSeq: {sess.local_cseq} INVITE\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: <sip:gateway@127.0.0.1:{SIP_PORT}>\r\n"
-            f"Remote-Party-ID: \"{testo_display}\" <sip:000@127.0.0.1>;party=called;screen=yes;privacy=off\r\n"
+            f"Remote-Party-ID: \"{testo_display}\" <sip:{sess.tg_destinazione}@{sess.ip_vps}>;party=called;screen=yes;privacy=off\r\n"
-            f"P-Asserted-Identity: \"{testo_display}\" <sip:000@127.0.0.1>\r\n"
+            f"P-Asserted-Identity: \"{testo_display}\" <sip:{sess.tg_destinazione}@{sess.ip_vps}>\r\n"
-            f"Content-Type: application/sdp\r\n"
+            f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}")
-            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}")
+                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)
@@ -306,31 +274,69 @@ class SessionManager:
                     frame_7 = ba_49.bytes[:7]
                 except Exception: continue
 
-                pcm_data = await self.vocoder.converti_ambe_in_pcm(frame_7)
+                pcm_data = await self.vocoder.converti_ambe_in_pcm(f"TG_{tg_arrivo}", frame_7)
-                if not pcm_data or len(pcm_data) < 2: continue
+                if pcm_data and len(pcm_data) >= 2:
-                if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1]
+                    if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1]
-                pcma_data = audioop.lin2alaw(pcm_data, 2)
+                    
-                
+                    # --- INIZIO AGC DMR -> SIP ---
+                    livello_rms = audioop.rms(pcm_data, 2)
+                    if livello_rms > 18000:
+                        pcm_data = audioop.mul(pcm_data, 2, 0.4)
+                    elif livello_rms > 12000:
+                        pcm_data = audioop.mul(pcm_data, 2, 0.7)
+                    elif livello_rms < 300:
+                        pcm_data = audioop.mul(pcm_data, 2, 0.0) 
+                    # --- FINE AGC ---
+
+                    pcma_data = audioop.lin2alaw(pcm_data, 2)
+                    frames_pcm.append(pcma_data)
+
+            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
+                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)
+            # --- INIZIO AGC SIP -> DMR ---
+            # Decodifichiamo G.711a in PCM lineare temporaneamente per calcolare il volume
+            pcm_lineare = audioop.alaw2lin(rtp_payload, 2)
+            livello_rms = audioop.rms(pcm_lineare, 2)
+            
+            if livello_rms > 18000:
+                pcm_lineare = audioop.mul(pcm_lineare, 2, 0.4)
+            elif livello_rms > 12000:
+                pcm_lineare = audioop.mul(pcm_lineare, 2, 0.7)
+            elif livello_rms > 500 and livello_rms < 3000:
+                # Piccolo booster se il microfono SIP è basso
+                pcm_lineare = audioop.mul(pcm_lineare, 2, 1.5)
+            elif livello_rms < 150:
+                # Noise gate
+                pcm_lineare = audioop.mul(pcm_lineare, 2, 0.0)
+            
+            # Ricodifichiamo in formato PCMA/A-Law prima di passarlo al vocoder
+            rtp_payload = audioop.lin2alaw(pcm_lineare, 2)
+            # --- FINE AGC ---
+
+            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
@@ -349,12 +355,8 @@ 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:
+        except Exception: pass
-            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 
@@ -365,116 +367,89 @@ class RtpProtocol(asyncio.DatagramProtocol):
         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:
+            rtp_timestamp = struct.unpack("!I", data[4:8])[0]
-                return
+            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)
-            # Esecuzione dei Comandi
+            elif evento_dtmf == 11: self.manager.gestisci_ptt(sess, False)
-            if evento_dtmf == 10: 
+            elif 0 <= evento_dtmf <= 9: self.manager.gestisci_dtmf_digit(sess, evento_dtmf)
-                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):
+    def connection_made(self, transport): self.manager.sip_transport = 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": return  
-        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"
+            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"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n"
+                    f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContact: <sip:gateway@127.0.0.1:{SIP_PORT}>\r\nContent-Length: 0\r\n\r\n")
-                    f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n"
+            self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
-                    f"Contact: <sip:gateway@127.0.0.1:{SIP_PORT}>\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\nFrom: {full_from}\r\nTo: {full_to};tag=987654\r\n"
-            resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n"
+                    f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContact: <sip:gateway@{ip_vps}:{SIP_PORT}>\r\n"
-                    f"From: {full_from}\r\nTo: {full_to};tag=987654\r\n"
+                    f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}")
-                    f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n"
+            self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
-                    f"Contact: <sip:gateway@{ip_vps}:{SIP_PORT}>\r\nContent-Type: application/sdp\r\n"
+        elif msg.startswith("INFO"):
-                    f"Content-Length: {len(sdp)}\r\n\r\n{sdp}")
+            resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n"
-            self.transport.sendto(resp.encode('utf-8'), addr)
+                    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'):
+                    if line.upper().startswith("SIGNAL="):
+                        signal = line.split("=")[1].strip()
+                        if signal == "*": evento = 10
+                        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:
+                if "CONTACT" in headers and "<" in headers["CONTACT"]: ack_uri = headers["CONTACT"].split("<")[1].split(">")[0]
-                    contact_hdr = headers["CONTACT"]
+                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"
-                    if "<" in contact_hdr:
+                       f"Max-Forwards: 70\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n"
-                        ack_uri = contact_hdr.split("<")[1].split(">")[0]
+                       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)
-                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"
+            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"From: {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")
-                    f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n"
+            self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
-                    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()
-    try:
+    except KeyboardInterrupt: pass
-        await asyncio.Event().wait()
-    except KeyboardInterrupt:
-        pass
 
 if __name__ == "__main__":
     asyncio.run(main())

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...")

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()

vocoder_engine.py

@@ -1,47 +1,99 @@
 import asyncio
+import time
 import audioop
 
-class VocoderEngine:
+class UdpClientProtocol(asyncio.DatagramProtocol):
-    def __init__(self, mode="emulator", host="127.0.0.1", port=2470):
+    def __init__(self):
-        self.mode = mode
-        self.host = host
-        self.port = port
         self.transport = None
-        self.coda_risposte = asyncio.Queue()
+        self.future = None
-        # IL SEMAFORO ASINCRONO PER IL MULTI-UTENTE
+
-        self.lock = asyncio.Lock()
+    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):
+        for p in self.ports:
-            def __init__(self, coda): self.coda = coda
+            transport, protocol = await loop.create_datagram_endpoint(
-            def connection_made(self, transport): pass
+                lambda: UdpClientProtocol(),
-            def datagram_received(self, data, addr): self.coda.put_nowait(data)
+                remote_addr=(self.host, p)
-            def error_received(self, exc): pass
+            )
+            # 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(
+    def pulisci_stream_inattivi(self):
-            lambda: AmbedProtocol(self.coda_risposte),
+        """Libera automaticamente le porte dei Talkgroup/SIP muti da più di 2.5 secondi"""
-            remote_addr=(self.host, self.port)
+        ora = time.time()
-        )
+        inattivi = [s for s, t in self.last_seen.items() if ora - t > 2.5]
-        print(f"[Vocoder] Motore avviato. Modalità: {self.mode.upper()}")
+        for s in inattivi:
-        print(f"[Vocoder] Connesso a {self.host}:{self.port}")
+            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):
+    def assegna_porta(self, stream_id):
-        pcm_data = audioop.alaw2lin(rtp_payload_alaw, 2)
+        self.pulisci_stream_inattivi()
-        if self.mode == "emulator":
+        self.last_seen[stream_id] = time.time()
-            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
 
-    async def converti_ambe_in_pcm(self, ambe_bytes):
+        if stream_id in self.stream_alloc:
-        if self.mode == "emulator":
+            return self.stream_alloc[stream_id]
-            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
 
-    def ferma(self):
+        porte_usate = set(self.stream_alloc.values())
-        if self.transport:
+        porte_libere = [p for p in self.ports if p not in porte_usate]
-            self.transport.close()
+
-            print("[Vocoder] Motore spento.")
+        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)