Fix definitivo chiusura sessioni SIP, watchdog RTP
This commit is contained in:
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user