Add Dashboard

This commit is contained in:
2026-08-29 18:24:15 +02:00
parent 8b124751d5
commit fcd356da64
5 changed files with 322711 additions and 88 deletions
+322124
View File
File diff suppressed because it is too large Load Diff
+233 -88
View File
@@ -5,24 +5,25 @@ import struct
import time import time
import os import os
import audioop import audioop
from collections import deque
from aiohttp import web
from vocoder_engine import VocoderEngine from vocoder_engine import VocoderEngine
from openbridge_protocol import crea_pacchetto_ping, calcola_hmac from openbridge_protocol import crea_pacchetto_ping, calcola_hmac
from dmr_utils3.ambe_utils import convert49BitTo72BitAMBE, convert72BitTo49BitAMBE from dmr_utils3.ambe_utils import convert49BitTo72BitAMBE, convert72BitTo49BitAMBE
# ==================================================================== # ====================================================================
# CARICAMENTO CONFIGURAZIONI (JSON) # CARICAMENTO CONFIGURAZIONI (JSON) E DATABASE DMR
# ==================================================================== # ====================================================================
def carica_config(): def carica_config():
try: try:
with open("config.json", "r") as f: with open("config.json", "r") as f:
return json.load(f) return json.load(f)
except Exception as e: except Exception as e:
print(f"[ERRORE CRITICO] File config.json non trovato o errato ({e}). Chiusura.") print(f"[ERRORE CRITICO] File config.json non trovato ({e}).")
exit(1) exit(1)
CONFIG = carica_config() CONFIG = carica_config()
HBLINK_IP = CONFIG.get("hblink_ip", "127.0.0.1") HBLINK_IP = CONFIG.get("hblink_ip", "127.0.0.1")
HBLINK_PORT = CONFIG.get("hblink_port", 62047) HBLINK_PORT = CONFIG.get("hblink_port", 62047)
HBLINK_PASSWORD = CONFIG.get("hblink_password", "TestVOIP") HBLINK_PASSWORD = CONFIG.get("hblink_password", "TestVOIP")
@@ -30,16 +31,167 @@ PEER_ID = CONFIG.get("peer_id", 22241)
SIP_PORT = CONFIG.get("sip_port", 55060) SIP_PORT = CONFIG.get("sip_port", 55060)
RTP_PORT = CONFIG.get("rtp_port", 10000) RTP_PORT = CONFIG.get("rtp_port", 10000)
FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") 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(): def carica_rubrica():
try: try:
with open("utenti.json", "r") as f: with open("utenti.json", "r") as f:
return json.load(f) return json.load(f)
except Exception as e: except Exception:
return {} return {}
UTENTI_SIP_DMR = carica_rubrica() 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)
async 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
await self.broadcast({"evento": "chiamata_iniziata", "dati": dati})
async def aggiorna_rx(self, id_chiamata):
if id_chiamata in self.chiamate_attive:
self.chiamate_attive[id_chiamata]["ultimo_frame"] = time.time()
async 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)
await self.broadcast({"evento": "chiamata_terminata", "dati": dati})
async def watchdog_rx(self):
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:
await self.termina_chiamata(id_chiamata)
# ====================================================================
# 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: class SessioneSIP:
def __init__(self, call_id, chiamante, destinatario_tg, rtp_addr, sip_addr, full_from, full_to, ip_vps): 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 = call_id
@@ -62,10 +214,8 @@ class SessioneSIP:
self.stream_id = os.urandom(4) self.stream_id = os.urandom(4)
self.ultimo_src_id = None self.ultimo_src_id = None
# Variabili Inerzia per l'AGC
self.agc_rx = 1.0 self.agc_rx = 1.0
self.agc_tx = 1.0 self.agc_tx = 1.0
self.dtmf_buffer = "" self.dtmf_buffer = ""
self.dtmf_task = None self.dtmf_task = None
self.ultimo_rtp_timestamp_dtmf = None self.ultimo_rtp_timestamp_dtmf = None
@@ -79,7 +229,6 @@ class OpenBridgeClient(asyncio.DatagramProtocol):
def connection_made(self, transport): def connection_made(self, transport):
self.transport = transport self.transport = transport
print(f"[OpenBridge] Socket connesso a {HBLINK_IP}:{HBLINK_PORT}")
asyncio.create_task(self.keepalive_loop()) asyncio.create_task(self.keepalive_loop())
def datagram_received(self, data, addr): def datagram_received(self, data, addr):
@@ -90,7 +239,6 @@ class OpenBridgeClient(asyncio.DatagramProtocol):
sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo] sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo]
if not sessioni_interessate: return if not sessioni_interessate: return
print("D", end="", flush=True)
payload_33 = data[20:53] payload_33 = data[20:53]
ambe1 = payload_33[0:9] ambe1 = payload_33[0:9]
ambe2 = bytearray(9) ambe2 = bytearray(9)
@@ -143,8 +291,7 @@ class OpenBridgeClient(asyncio.DatagramProtocol):
pacchetto_base = header + bytes(payload_33) pacchetto_base = header + bytes(payload_33)
firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base)
self.transport.sendto(pacchetto_base + firma_hmac) self.transport.sendto(pacchetto_base + firma_hmac)
except Exception: except Exception: pass
pass
def chiudi_flusso_chiamata(self, sess): def chiudi_flusso_chiamata(self, sess):
if not self.transport: return if not self.transport: return
@@ -177,7 +324,8 @@ class SessionManager:
local_addr=('0.0.0.0', HBLINK_PORT), local_addr=('0.0.0.0', HBLINK_PORT),
remote_addr=(HBLINK_IP, HBLINK_PORT) remote_addr=(HBLINK_IP, HBLINK_PORT)
) )
print(f"[Core] PBX Multi-Utente avviato. SIP su porta {SIP_PORT}...\n") 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): def trova_sessione_per_rtp(self, addr):
for sess in self.sessioni.values(): for sess in self.sessioni.values():
@@ -196,26 +344,39 @@ class SessionManager:
def gestisci_nuova_chiamata(self, call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps): 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) sess = SessioneSIP(call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps)
self.sessioni[call_id] = sess self.sessioni[call_id] = sess
print(f"\n[Core] >>> SIP {chiamante} -> TG {destinatario}") asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"}))
def termina_chiamata(self, call_id): def termina_chiamata(self, call_id):
if call_id in self.sessioni: if call_id in self.sessioni:
sess = self.sessioni[call_id] sess = self.sessioni[call_id]
if sess.ptt_attivo and self.openbridge: if sess.ptt_attivo:
self.openbridge.chiudi_flusso_chiamata(sess) self.gestisci_ptt(sess, False)
self.vocoder.rilascia_porta(f"SIP_{call_id}") self.vocoder.rilascia_porta(f"SIP_{call_id}")
# PULIZIA FORZATA: Chiudiamo eventuali righe RX rimaste appese per questo interno
for id_dash in list(dashboard_app.chiamate_attive.keys()):
if id_dash.startswith("RX_") and id_dash.endswith(f"_{call_id}"):
asyncio.create_task(dashboard_app.termina_chiamata(id_dash))
del self.sessioni[call_id] del self.sessioni[call_id]
asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"}))
def gestisci_ptt(self, sess, stato): def gestisci_ptt(self, sess, stato):
id_dash_tx = f"TX_{sess.call_id}"
if stato is True and not sess.ptt_attivo: if stato is True and not sess.ptt_attivo:
sess.ptt_attivo = True sess.ptt_attivo = True
sess.buffer_ambe = [] sess.buffer_ambe = []
sess.seq_dmr = 0 sess.seq_dmr = 0
sess.stream_id = os.urandom(4) sess.stream_id = os.urandom(4)
if self.openbridge: self.openbridge.apri_flusso_chiamata(sess) if self.openbridge: self.openbridge.apri_flusso_chiamata(sess)
nome_base = get_nominativo(sess.dmr_id)
mittente_display = f"{nome_base} (Est. {sess.chiamante})"
asyncio.create_task(dashboard_app.inizia_chiamata(id_dash_tx, "TX", mittente_display, sess.tg_destinazione))
elif stato is False and sess.ptt_attivo: elif stato is False and sess.ptt_attivo:
sess.ptt_attivo = False sess.ptt_attivo = False
if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess)
asyncio.create_task(dashboard_app.termina_chiamata(id_dash_tx))
def gestisci_dtmf_digit(self, sess, digit): def gestisci_dtmf_digit(self, sess, digit):
if sess.ptt_attivo: return if sess.ptt_attivo: return
@@ -229,9 +390,9 @@ class SessionManager:
if sess.dtmf_buffer: if sess.dtmf_buffer:
nuovo_tg = sess.dtmf_buffer nuovo_tg = sess.dtmf_buffer
sess.dtmf_buffer = "" sess.dtmf_buffer = ""
print(f"\n[Core] *** CAMBIO TG: {sess.chiamante} è passato al Talkgroup {nuovo_tg} ***")
sess.tg_destinazione = nuovo_tg sess.tg_destinazione = nuovo_tg
self.notifica_display(sess, f"TG: {nuovo_tg}") self.notifica_display(sess, f"TG: {nuovo_tg}")
asyncio.create_task(dashboard_app.broadcast({"evento": "estensioni_update"}))
except asyncio.CancelledError: pass except asyncio.CancelledError: pass
def notifica_display(self, sess, testo_display): def notifica_display(self, sess, testo_display):
@@ -256,16 +417,26 @@ class SessionManager:
def ricevi_frame_dmr_multi(self, ambe_frames, sessioni_interessate, src_id): def ricevi_frame_dmr_multi(self, ambe_frames, sessioni_interessate, src_id):
ascoltatori = [s for s in sessioni_interessate if not s.ptt_attivo] ascoltatori = [s for s in sessioni_interessate if not s.ptt_attivo]
if not ascoltatori or not self.rtp_transport: return if not ascoltatori or not self.rtp_transport: return
for sess in ascoltatori: for sess in ascoltatori:
tg_arrivo = sess.tg_destinazione
id_dash_rx = f"RX_{src_id}_{sess.call_id}"
if id_dash_rx not in dashboard_app.chiamate_attive:
dest_display = f"{sess.chiamante} (TG {tg_arrivo})"
asyncio.create_task(dashboard_app.inizia_chiamata(id_dash_rx, "RX", src_id, dest_display))
else:
asyncio.create_task(dashboard_app.aggiorna_rx(id_dash_rx))
if sess.ultimo_src_id != src_id: if sess.ultimo_src_id != src_id:
sess.ultimo_src_id = src_id sess.ultimo_src_id = src_id
self.notifica_display(sess, f"CID:{src_id}") self.notifica_display(sess, f"CID:{src_id}")
asyncio.create_task(self.processa_audio_dmr(ambe_frames, ascoltatori)) asyncio.create_task(self.processa_audio_dmr(ambe_frames, ascoltatori))
async def processa_audio_dmr(self, ambe_27_bytes, ascoltatori): async def processa_audio_dmr(self, ambe_27_bytes, ascoltatori):
if not ascoltatori: return if not ascoltatori: return
tg_arrivo = ascoltatori[0].tg_destinazione tg_arrivo = ascoltatori[0].tg_destinazione
try: try:
frames_pcm = [] frames_pcm = []
for i in range(0, len(ambe_27_bytes), 9): for i in range(0, len(ambe_27_bytes), 9):
@@ -281,40 +452,24 @@ class SessionManager:
pcm_data = await self.vocoder.converti_ambe_in_pcm(f"TG_{tg_arrivo}", frame_7) pcm_data = await self.vocoder.converti_ambe_in_pcm(f"TG_{tg_arrivo}", frame_7)
if pcm_data and len(pcm_data) >= 2: 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]
# --- AGC DEFINITIVO (Zero Sleep, Soft-Limiting) ---
livello_rms = audioop.rms(pcm_data, 2) livello_rms = audioop.rms(pcm_data, 2)
picco_max = audioop.max(pcm_data, 2) picco_max = audioop.max(pcm_data, 2)
if livello_rms > 50: if livello_rms > 50:
target_rms = 4000.0 target_mult = 4000.0 / livello_rms
target_mult = target_rms / livello_rms
if target_mult > 2.0: target_mult = 2.0 if target_mult > 2.0: target_mult = 2.0
elif target_mult < 0.3: target_mult = 0.3 elif target_mult < 0.3: target_mult = 0.3
for sess in ascoltatori: for sess in ascoltatori:
if target_mult < sess.agc_rx: if target_mult < sess.agc_rx: sess.agc_rx = (sess.agc_rx * 0.90) + (target_mult * 0.10)
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)
else:
sess.agc_rx = (sess.agc_rx * 0.99) + (target_mult * 0.01)
moltiplicatore = ascoltatori[0].agc_rx moltiplicatore = ascoltatori[0].agc_rx
# Soft-Brickwall: previene scatti improvvisi del moltiplicatore
if picco_max > 0 and (picco_max * moltiplicatore) > 30000: if picco_max > 0 and (picco_max * moltiplicatore) > 30000:
moltiplicatore_sicuro = 30000 / picco_max moltiplicatore_sicuro = 30000 / picco_max
# Lasciamo scivolare l'inerzia verso il basso dolcemente, non di scatto for sess in ascoltatori: sess.agc_rx = (sess.agc_rx * 0.70) + (moltiplicatore_sicuro * 0.30)
for sess in ascoltatori:
sess.agc_rx = (sess.agc_rx * 0.70) + (moltiplicatore_sicuro * 0.30)
moltiplicatore = sess.agc_rx moltiplicatore = sess.agc_rx
pcm_data = audioop.mul(pcm_data, 2, moltiplicatore) pcm_data = audioop.mul(pcm_data, 2, moltiplicatore)
pcma_data = audioop.lin2alaw(pcm_data, 2) pcma_data = audioop.lin2alaw(pcm_data, 2)
frames_pcm.append(pcma_data) frames_pcm.append(pcma_data)
# --- INVIO RTP A RAFFICA (Gestito dal Jitter Buffer del SIP) ---
for pcma_data in frames_pcm: for pcma_data in frames_pcm:
for sess in ascoltatori: for sess in ascoltatori:
header = bytearray(12) header = bytearray(12)
@@ -322,41 +477,25 @@ class SessionManager:
header[2:4] = struct.pack("!H", sess.rtp_seq) header[2:4] = struct.pack("!H", sess.rtp_seq)
header[4:8] = struct.pack("!I", sess.rtp_timestamp) header[4:8] = struct.pack("!I", sess.rtp_timestamp)
header[8:12] = struct.pack("!I", 0x12345678) header[8:12] = struct.pack("!I", 0x12345678)
rtp_packet = header + pcma_data if sess.rtp_client_addr: self.rtp_transport.sendto(header + pcma_data, sess.rtp_client_addr)
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_seq = (sess.rtp_seq + 1) % 65536
# Il timestamp matematico è tutto ciò che serve al telefono per temporizzare l'audio
sess.rtp_timestamp = (sess.rtp_timestamp + 160) % 4294967296 sess.rtp_timestamp = (sess.rtp_timestamp + 160) % 4294967296
except Exception: pass except Exception: pass
async def processa_audio_rtp(self, sess, rtp_payload): async def processa_audio_rtp(self, sess, rtp_payload):
if not sess.ptt_attivo: return if not sess.ptt_attivo: return
try: try:
# --- AGC DINAMICO MORBIDO SIP -> DMR ---
pcm_lineare = audioop.alaw2lin(rtp_payload, 2) pcm_lineare = audioop.alaw2lin(rtp_payload, 2)
livello_rms = audioop.rms(pcm_lineare, 2) livello_rms = audioop.rms(pcm_lineare, 2)
if livello_rms > 50: if livello_rms > 50:
target_rms = 4000.0 target_mult = 4000.0 / livello_rms
target_mult = target_rms / livello_rms
if target_mult > 2.5: target_mult = 2.5 if target_mult > 2.5: target_mult = 2.5
elif target_mult < 0.5: target_mult = 0.5 elif target_mult < 0.5: target_mult = 0.5
# Inerzia per evitare scatti
sess.agc_tx = (sess.agc_tx * 0.90) + (target_mult * 0.10) sess.agc_tx = (sess.agc_tx * 0.90) + (target_mult * 0.10)
pcm_lineare = audioop.mul(pcm_lineare, 2, sess.agc_tx) pcm_lineare = audioop.mul(pcm_lineare, 2, sess.agc_tx)
rtp_payload = audioop.lin2alaw(pcm_lineare, 2) 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)
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 if not ambe_bytes: return
dim_frame = 7 if len(ambe_bytes) % 7 == 0 else 9 dim_frame = 7 if len(ambe_bytes) % 7 == 0 else 9
@@ -365,8 +504,7 @@ class SessionManager:
if len(frame) == 7: if len(frame) == 7:
stringa_bit = BitArray(bytes=frame).bin stringa_bit = BitArray(bytes=frame).bin
ambe_49_bits = [int(b) for b in stringa_bit[:49]] ambe_49_bits = [int(b) for b in stringa_bit[:49]]
fec_9_byte = convert49BitTo72BitAMBE(ambe_49_bits) sess.buffer_ambe.append(bytes(convert49BitTo72BitAMBE(ambe_49_bits)))
sess.buffer_ambe.append(bytes(fec_9_byte))
elif len(frame) == 9: elif len(frame) == 9:
sess.buffer_ambe.append(frame) sess.buffer_ambe.append(frame)
@@ -384,8 +522,7 @@ class RtpProtocol(asyncio.DatagramProtocol):
sess = self.manager.trova_sessione_per_rtp(addr) sess = self.manager.trova_sessione_per_rtp(addr)
if not sess or len(data) < 12: return if not sess or len(data) < 12: return
payload_type = data[1] & 0x7F payload_type = data[1] & 0x7F
if payload_type == 8: if payload_type == 8: asyncio.create_task(self.manager.processa_audio_rtp(sess, data[12:]))
asyncio.create_task(self.manager.processa_audio_rtp(sess, data[12:]))
elif payload_type == 101: elif payload_type == 101:
ora = time.time() ora = time.time()
rtp_timestamp = struct.unpack("!I", data[4:8])[0] rtp_timestamp = struct.unpack("!I", data[4:8])[0]
@@ -401,11 +538,13 @@ class RtpProtocol(asyncio.DatagramProtocol):
class SipProtocol(asyncio.DatagramProtocol): class SipProtocol(asyncio.DatagramProtocol):
def __init__(self, manager): self.manager = manager def __init__(self, manager): self.manager = manager
def connection_made(self, transport): self.manager.sip_transport = transport def connection_made(self, transport): self.manager.sip_transport = transport
def datagram_received(self, data, addr): 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') 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}
# FIX: splitlines() digerisce correttamente sia \r\n che \n
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', '') call_id = headers.get('CALL-ID', '')
if msg.startswith("OPTIONS"): if msg.startswith("OPTIONS"):
@@ -414,20 +553,16 @@ class SipProtocol(asyncio.DatagramProtocol):
self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
elif msg.startswith("INVITE"): elif msg.startswith("INVITE"):
client_rtp_port = 4000 client_rtp_port = 4000
for line in msg.split('\n'): for line in lines:
line = line.strip() if line.strip().startswith("m=audio "): client_rtp_port = int(line.strip().split()[1])
if line.startswith("m=audio "): client_rtp_port = int(line.split()[1]) ip_vps = headers.get("TO", "").split("@")[1].split(">")[0] if "@" in headers.get("TO", "") else "127.0.0.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] chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0]
destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0]
full_from = headers.get('FROM', '') self.manager.gestisci_nuova_chiamata(call_id, chiamante, destinatario, (addr[0], client_rtp_port), addr, headers.get('FROM', ''), headers.get('TO', ''), ip_vps)
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" 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"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") 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\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: <sip:gateway@{ip_vps}:{SIP_PORT}>\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\nContact: <sip:gateway@{ip_vps}:{SIP_PORT}>\r\n"
f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}") 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) self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
@@ -437,37 +572,47 @@ class SipProtocol(asyncio.DatagramProtocol):
self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
sess = self.manager.sessioni.get(call_id) sess = self.manager.sessioni.get(call_id)
if sess: if sess:
for line in msg.split('\r\n'): for line in lines:
if line.upper().startswith("SIGNAL="): if line.upper().startswith("SIGNAL="):
signal = line.split("=")[1].strip() signal = line.split("=")[1].strip()
if signal == "*": evento = 10 evento = 10 if signal == "*" else 11 if signal == "#" else int(signal) if signal.isdigit() else None
elif signal == "#": evento = 11
elif signal.isdigit(): evento = int(signal)
else: continue
if evento == 10: self.manager.gestisci_ptt(sess, True) if evento == 10: self.manager.gestisci_ptt(sess, True)
elif evento == 11: self.manager.gestisci_ptt(sess, False) elif evento == 11: self.manager.gestisci_ptt(sess, False)
elif 0 <= evento <= 9: self.manager.gestisci_dtmf_digit(sess, evento) elif evento is not None and 0 <= evento <= 9: self.manager.gestisci_dtmf_digit(sess, evento)
elif msg.startswith("SIP/2.0 200 OK"): elif msg.startswith("SIP/2.0 200 OK") and "INVITE" in headers.get('CSEQ', ''):
if "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]}"
cseq_num = headers.get('CSEQ', '').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"
ack_uri = f"sip:{addr[0]}:{addr[1]}" f"Max-Forwards: 70\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n"
if "CONTACT" in headers and "<" in headers["CONTACT"]: ack_uri = headers["CONTACT"].split("<")[1].split(">")[0] f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '').split()[0]} ACK\r\nContent-Length: 0\r\n\r\n")
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" self.manager.sip_transport.sendto(ack.encode('utf-8'), addr)
f"Max-Forwards: 70\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" # FIX: Aggiunto supporto al CANCEL e chiusura forzata in dashboard
f"Call-ID: {call_id}\r\nCSeq: {cseq_num} ACK\r\nContent-Length: 0\r\n\r\n") elif msg.startswith("BYE") or msg.startswith("CANCEL"):
self.manager.sip_transport.sendto(ack.encode('utf-8'), addr)
elif msg.startswith("BYE"):
self.manager.termina_chiamata(call_id) self.manager.termina_chiamata(call_id)
resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\nFrom: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\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"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\nContent-Length: 0\r\n\r\n")
self.manager.sip_transport.sendto(resp.encode('utf-8'), addr) self.manager.sip_transport.sendto(resp.encode('utf-8'), addr)
async def main(): async def main():
global sip_manager
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
# Motore SIP/RTP
manager = SessionManager() manager = SessionManager()
sip_manager = manager
await manager.avvia() 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: 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)) 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() try: await asyncio.Event().wait()
except KeyboardInterrupt: pass except KeyboardInterrupt: pass
+217
View File
@@ -0,0 +1,217 @@
<!DOCTYPE html>
<html lang="it" data-bs-theme="dark">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIP-DMR Live Dashboard</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<style>
body { background-color: #121212; font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; }
.badge-rx { background-color: #198754; color: white; }
.badge-tx { background-color: #dc3545; color: white; }
.pulsing { animation: pulse 1.5s infinite; }
@keyframes pulse { 0% { opacity: 1; } 50% { opacity: 0.5; } 100% { opacity: 1; } }
.table-dark { --bs-table-bg: #1e1e1e; }
/* Contenitore con scorrimento solo per lo storico */
.table-scrollable {
max-height: 75vh;
overflow-y: auto;
}
/* Intestazioni fisse in alto durante lo scorrimento */
thead th {
position: sticky;
top: 0;
background-color: #2c3034 !important;
z-index: 10;
box-shadow: 0 1px 2px rgba(0,0,0,0.5);
}
::-webkit-scrollbar { width: 8px; }
::-webkit-scrollbar-track { background: #1e1e1e; }
::-webkit-scrollbar-thumb { background: #555; border-radius: 4px; }
::-webkit-scrollbar-thumb:hover { background: #888; }
</style>
</head>
<body>
<div class="container-fluid py-4 px-4">
<!-- HEADER -->
<div class="d-flex justify-content-between align-items-center mb-4">
<h2 class="m-0 text-light">📡 SIP-DMR Gateway</h2>
<span id="conn-status" class="badge bg-warning text-dark fs-6">Connessione in corso...</span>
</div>
<div class="row">
<!-- TABELLA ESTENSIONI (SINISTRA - Altezza naturale fissa) -->
<div class="col-xl-3 col-lg-4 mb-4">
<div class="card bg-dark border-secondary shadow-lg">
<div class="card-header bg-secondary text-white fw-bold">
🔌 Estensioni Connesse
</div>
<!-- Rimossa la classe table-scrollable da qui -->
<div class="card-body p-0">
<table class="table table-dark table-hover m-0">
<thead>
<tr>
<th>Est.</th>
<th>Nome</th>
<th>In Ascolto</th>
</tr>
</thead>
<tbody id="tabella-estensioni">
<!-- Estensioni caricate da JS -->
</tbody>
</table>
</div>
</div>
</div>
<!-- TABELLA STORICO (DESTRA - Scorrimento dinamico) -->
<div class="col-xl-9 col-lg-8 mb-4">
<div class="card bg-dark border-secondary shadow-lg h-100">
<div class="card-header bg-secondary text-white fw-bold">
📻 Traffico Radio Live
</div>
<!-- Mantenuta la classe table-scrollable -->
<div class="card-body p-0 table-scrollable">
<table class="table table-dark table-hover m-0">
<thead>
<tr>
<th>Orario</th>
<th>Dir</th>
<th>Mittente</th>
<th>Destinatario (TG)</th>
<th>Durata</th>
<th>Stato</th>
</tr>
</thead>
<tbody id="tabella-storico">
<!-- Storico caricato da JS -->
</tbody>
</table>
</div>
</div>
</div>
</div>
</div>
<script>
const SERVER_IP = window.location.hostname === "" || window.location.hostname === "localhost"
? "127.0.0.1" : window.location.hostname;
const WS_PORT = "8081";
const tbodyStorico = document.getElementById('tabella-storico');
const tbodyEstensioni = document.getElementById('tabella-estensioni');
const statusBadge = document.getElementById('conn-status');
function formattaOrario(timestamp) {
const d = new Date(timestamp * 1000);
return d.toLocaleTimeString('it-IT', { hour12: false }) + '.' + String(d.getMilliseconds()).padStart(3, '0').substring(0,2);
}
function disegnaEstensioni(estensioni) {
tbodyEstensioni.innerHTML = "";
if (estensioni.length === 0) {
tbodyEstensioni.innerHTML = `<tr><td colspan="3" class="text-center text-muted py-3"><small>Nessun dispositivo collegato</small></td></tr>`;
return;
}
estensioni.forEach(est => {
const tr = document.createElement('tr');
tr.innerHTML = `
<td class="fw-bold text-light">${est.estensione}</td>
<td class="text-secondary" style="font-size: 0.9em;">${est.nome}</td>
<td><span class="badge bg-info text-dark">TG ${est.tg}</span></td>
`;
tbodyEstensioni.appendChild(tr);
});
}
async function caricaEstensioni() {
try {
const res = await fetch(`http://${SERVER_IP}:${WS_PORT}/api/storico`);
const data = await res.json();
disegnaEstensioni(data.estensioni);
} catch (err) { console.error("Errore aggiornamento estensioni:", err); }
}
function creaRiga(dati) {
const tr = document.createElement('tr');
tr.id = `row-${dati.id}`;
const badgeDir = dati.direzione === "RX" ? "badge-rx" : "badge-tx";
const statoHtml = dati.stato === "ON"
? `<span class="badge bg-primary pulsing">In Trasmissione</span>`
: `<span class="badge bg-secondary">Terminata</span>`;
const durata = dati.durata ? `${dati.durata}s` : `<span class="spinner-grow spinner-grow-sm text-light" role="status"></span>`;
tr.innerHTML = `
<td class="text-secondary">${formattaOrario(dati.inizio)}</td>
<td><span class="badge ${badgeDir} px-2 py-1">${dati.direzione}</span></td>
<td class="fw-bold text-light">${dati.mittente}</td>
<td class="text-info">${dati.destinatario}</td>
<td id="durata-${dati.id}">${durata}</td>
<td id="stato-${dati.id}">${statoHtml}</td>
`;
return tr;
}
async function caricaStorico() {
try {
const res = await fetch(`http://${SERVER_IP}:${WS_PORT}/api/storico`);
const data = await res.json();
tbodyStorico.innerHTML = "";
data.attive.forEach(call => tbodyStorico.appendChild(creaRiga(call)));
data.storico.forEach(call => tbodyStorico.appendChild(creaRiga(call)));
disegnaEstensioni(data.estensioni);
} catch (err) { console.error("Errore caricamento storico:", err); }
}
function avviaWebSocket() {
const ws = new WebSocket(`ws://${SERVER_IP}:${WS_PORT}/ws`);
ws.onopen = () => {
statusBadge.textContent = "Connesso (Live)";
statusBadge.className = "badge bg-success text-white fs-6";
caricaStorico();
};
ws.onclose = () => {
statusBadge.textContent = "Disconnesso - Riconnessione...";
statusBadge.className = "badge bg-danger text-white fs-6";
setTimeout(avviaWebSocket, 3000);
};
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.evento === "estensioni_update") {
caricaEstensioni();
return;
}
const dati = msg.dati;
if (msg.evento === "chiamata_iniziata") {
tbodyStorico.insertBefore(creaRiga(dati), tbodyStorico.firstChild);
if (tbodyStorico.children.length > 100) tbodyStorico.lastChild.remove();
}
else if (msg.evento === "chiamata_terminata") {
const rigaEx = document.getElementById(`row-${dati.id}`);
if (rigaEx) {
document.getElementById(`durata-${dati.id}`).textContent = `${dati.durata}s`;
document.getElementById(`stato-${dati.id}`).innerHTML = `<span class="badge bg-secondary">Terminata</span>`;
} else {
tbodyStorico.insertBefore(creaRiga(dati), tbodyStorico.firstChild);
}
}
};
}
avviaWebSocket();
</script>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
</body>
</html>
+1
View File
@@ -1,2 +1,3 @@
bitstring bitstring
dmr_utils3 dmr_utils3
aiohttp
+136
View File
@@ -0,0 +1,136 @@
<!DOCTYPE html>
<html lang="it" data-bs-theme="dark">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIP-DMR Live Dashboard</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<style>
body { background-color: #121212; font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; }
.badge-rx { background-color: #198754; color: white; }
.badge-tx { background-color: #dc3545; color: white; }
.pulsing { animation: pulse 1.5s infinite; }
@keyframes pulse { 0% { opacity: 1; } 50% { opacity: 0.5; } 100% { opacity: 1; } }
.table-dark { --bs-table-bg: #1e1e1e; }
</style>
</head>
<body>
<div class="container py-4">
<div class="d-flex justify-content-between align-items-center mb-4">
<h2 class="m-0 text-light">📡 SIP-DMR Gateway</h2>
<span id="conn-status" class="badge bg-warning text-dark fs-6">Connessione in corso...</span>
</div>
<div class="card bg-dark border-secondary shadow-lg">
<div class="card-body p-0">
<table class="table table-dark table-hover m-0">
<thead class="table-secondary">
<tr>
<th>Orario</th>
<th>Dir</th>
<th>Mittente</th>
<th>Destinatario (TG)</th>
<th>Durata</th>
<th>Stato</th>
</tr>
</thead>
<tbody id="tabella-storico">
<!-- Le righe verranno iniettate qui da JS -->
</tbody>
</table>
</div>
</div>
</div>
<script>
// SOSTITUISCI QUI l'IP se apri il file in locale, altrimenti usa l'hostname automatico
const SERVER_IP = window.location.hostname === "" || window.location.hostname === "localhost"
? "127.0.0.1" : window.location.hostname;
const WS_PORT = "8081";
const tbody = document.getElementById('tabella-storico');
const statusBadge = document.getElementById('conn-status');
function formattaOrario(timestamp) {
const d = new Date(timestamp * 1000);
return d.toLocaleTimeString('it-IT', { hour12: false }) + '.' + String(d.getMilliseconds()).padStart(3, '0').substring(0,2);
}
function creaRiga(dati) {
const tr = document.createElement('tr');
tr.id = `row-${dati.id}`;
const badgeDir = dati.direzione === "RX" ? "badge-rx" : "badge-tx";
const statoHtml = dati.stato === "ON"
? `<span class="badge bg-primary pulsing">In Trasmissione</span>`
: `<span class="badge bg-secondary">Terminata</span>`;
const durata = dati.durata ? `${dati.durata}s` : `<span class="spinner-grow spinner-grow-sm text-light" role="status"></span>`;
tr.innerHTML = `
<td class="text-secondary">${formattaOrario(dati.inizio)}</td>
<td><span class="badge ${badgeDir} px-2 py-1">${dati.direzione}</span></td>
<td class="fw-bold text-light">${dati.mittente}</td>
<td class="text-info">${dati.destinatario}</td>
<td id="durata-${dati.id}">${durata}</td>
<td id="stato-${dati.id}">${statoHtml}</td>
`;
return tr;
}
async function caricaStorico() {
try {
const res = await fetch(`http://${SERVER_IP}:${WS_PORT}/api/storico`);
const data = await res.json();
tbody.innerHTML = "";
// Inseriamo prima le attive, poi lo storico
data.attive.forEach(call => tbody.appendChild(creaRiga(call)));
data.storico.forEach(call => tbody.appendChild(creaRiga(call)));
} catch (err) {
console.error("Errore caricamento storico:", err);
}
}
function avviaWebSocket() {
const ws = new WebSocket(`ws://${SERVER_IP}:${WS_PORT}/ws`);
ws.onopen = () => {
statusBadge.textContent = "Connesso (Live)";
statusBadge.className = "badge bg-success text-white fs-6";
caricaStorico(); // Ricarica tutto quando la connessione si stabilisce
};
ws.onclose = () => {
statusBadge.textContent = "Disconnesso - Riconnessione...";
statusBadge.className = "badge bg-danger text-white fs-6";
setTimeout(avviaWebSocket, 3000);
};
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
const dati = msg.dati;
if (msg.evento === "chiamata_iniziata") {
tbody.insertBefore(creaRiga(dati), tbody.firstChild);
// Rimuove eventuali righe in eccesso per non appesantire la pagina
if (tbody.children.length > 100) tbody.lastChild.remove();
}
else if (msg.evento === "chiamata_terminata") {
const rigaEx = document.getElementById(`row-${dati.id}`);
if (rigaEx) {
document.getElementById(`durata-${dati.id}`).textContent = `${dati.durata}s`;
document.getElementById(`stato-${dati.id}`).innerHTML = `<span class="badge bg-secondary">Terminata</span>`;
} else {
// Se la riga non esisteva (es. connessione lenta), la aggiungiamo
tbody.insertBefore(creaRiga(dati), tbody.firstChild);
}
}
};
}
avviaWebSocket();
</script>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
</body>
</html>