Skip to content

Instantly share code, notes, and snippets.

@rene-d

rene-d/rtp_ana.py

Created September 18, 2026 15:43
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save rene-d/a4f1f35a93676a6bf800cb4d50759dca to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save rene-d/a4f1f35a93676a6bf800cb4d50759dca to your computer and use it in GitHub Desktop.
Download ZIP
Raw
rtp_ana.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Analyse des trames RTP (G.711) d'une capture .pcap / .pcapng.
Aucune dependance externe : les formats pcap (classique) et pcapng sont
decodes directement, ainsi que les couches Ethernet/VLAN/IPv4/IPv6/UDP.
Hypothese de travail (cf. cahier des charges) : tout paquet UDP de la capture
est cense transporter une trame RTP. L'entete RTP est malgre tout validee ;
un paquet qui ne passe pas la validation est ignore et signale.
Usage :
python3 rtp_ana.py capture.pcap
python3 rtp_ana.py capture.pcap --seuil 15 --limite 30
python3 rtp_ana.py capture.pcap --graphe-svg niveaux.svg --wav piste
PROMPT:
écris un script Python qui va lire un .pcap (sans module externe)
et qui va analyser les trames RTP :
- un paquet UDP de la capture est une trame RTP. vérifie quand même l'entête RTP. sinon ignore le paquet et signale le
- les trames RTP sont en G.711
- nombre de trames
- espacement temporel minimal, maximal
- pour un espace temporel < 15ms, nombre de "paquets" de trames, et nombre de trames mini, maxi et moyen dans ces "paquets"
- vérifie que le seqnum est continu par SSRC
- vérifie que le timestamp est croissant par SSRC
- durée (au sens capture)
- durée (au sens timestamp RTP)
ajoute la création d'un graphe des niveaux sonores, comme Audacity
(ou wireshark) le ferait
"""
import argparse
import contextlib
import math
import os
import shutil
import struct
import sys
import wave
from array import array
from collections import Counter
# ---------------------------------------------------------------------------
# Constantes
# ---------------------------------------------------------------------------
G711_CLOCK = 8000 # Hz, horloge RTP de G.711 (PCMU/PCMA)
G711_PT = {0: "PCMU (G.711 u-law)", 8: "PCMA (G.711 A-law)"}
DEFAULT_BURST_MS = 15.0 # seuil de regroupement en "paquets" de trames
DEFAULT_BLOC_MS = 10.0 # resolution temporelle des graphes de niveau
DEFAULT_PLANCHER_DB = -60.0 # bas de l'echelle du graphe des niveaux
DEFAULT_SILENCE_DB = -50.0 # en dessous, un bloc est compte comme silencieux
DEFAULT_MAX_DUREE = 3600.0 # duree maximale reconstruite par flux (s)
LINKTYPE_NAMES = {
0: "NULL/Loopback", 1: "Ethernet", 9: "PPP", 12: "RAW IP", 101: "RAW IP",
105: "IEEE 802.11", 113: "Linux SLL", 127: "802.11 radiotap",
228: "IPv4 brut", 229: "IPv6 brut", 276: "Linux SLL2",
}
# ---------------------------------------------------------------------------
# Lecture des fichiers de capture
# ---------------------------------------------------------------------------
class CaptureError(Exception):
pass
def iter_pcap(f):
"""Itere sur un fichier pcap classique : (ts, linktype, octets, longueur_reelle)."""
hdr = f.read(24)
if len(hdr) < 24:
raise CaptureError("fichier pcap tronque (entete globale incomplete)")
magic = hdr[:4]
if magic == b"\xd4\xc3\xb2\xa1":
endian, div = "<", 1_000_000
elif magic == b"\xa1\xb2\xc3\xd4":
endian, div = ">", 1_000_000
elif magic == b"\x4d\x3c\xb2\xa1":
endian, div = "<", 1_000_000_000
elif magic == b"\xa1\xb2\x3c\x4d":
endian, div = ">", 1_000_000_000
else:
raise CaptureError("nombre magique pcap inconnu : %s" % magic.hex())
_vmaj, _vmin, _tz, _sig, _snap, linktype = struct.unpack(endian + "HHiIII", hdr[4:24])
linktype &= 0xFFFF # les bits hauts portent des drapeaux (FCS)
while True:
ph = f.read(16)
if len(ph) < 16:
break
ts_s, ts_frac, caplen, origlen = struct.unpack(endian + "IIII", ph)
data = f.read(caplen)
if len(data) < caplen:
break # dernier paquet tronque
yield ts_s + ts_frac / div, linktype, data, origlen
def _ng_options(buf, endian):
"""Itere sur les options d'un bloc pcapng : (code, valeur)."""
off = 0
while off + 4 <= len(buf):
code, olen = struct.unpack(endian + "HH", buf[off:off + 4])
off += 4
if code == 0: # opt_endofopt
break
val = buf[off:off + olen]
off += (olen + 3) & ~3 # padding sur 4 octets
yield code, val
def iter_pcapng(f):
"""Itere sur un fichier pcapng : (ts, linktype, octets, longueur_reelle)."""
endian = "<"
linktypes, tsdiv = {}, {}
if_index = 0
f.seek(0)
while True:
head = f.read(8)
if len(head) < 8:
break
# 0x0A0D0D0A est un palindrome d'octets : reconnu quel que soit le boutisme
btype = struct.unpack(endian + "I", head[:4])[0]
if btype == 0x0A0D0D0A: # Section Header Block
bom = f.read(4)
if bom == b"\x4d\x3c\x2b\x1a": # 0x1A2B3C4D ecrit en petit-boutiste
endian = "<"
elif bom == b"\x1a\x2b\x3c\x4d": # ... en gros-boutiste
endian = ">"
else:
raise CaptureError("byte-order magic pcapng invalide")
blen = struct.unpack(endian + "I", head[4:8])[0]
if blen < 16:
raise CaptureError("bloc SHB invalide")
f.read(blen - 12) # reste du bloc
linktypes, tsdiv, if_index = {}, {}, 0 # nouvelle section
continue
blen = struct.unpack(endian + "I", head[4:8])[0]
if blen < 12:
raise CaptureError("bloc pcapng de taille invalide (%d)" % blen)
rest = f.read(blen - 8) # corps + longueur totale repetee (4 o)
if len(rest) < blen - 8:
break
body = rest[:-4]
if btype == 0x00000001: # Interface Description Block
lt = struct.unpack(endian + "H", body[0:2])[0]
res = 6
for code, val in _ng_options(body[8:], endian):
if code == 9 and val: # if_tsresol
res = val[0]
linktypes[if_index] = lt
tsdiv[if_index] = (1 << (res & 0x7F)) if (res & 0x80) else 10 ** res
if_index += 1
elif btype == 0x00000006: # Enhanced Packet Block
iface, tsh, tsl, caplen, origlen = struct.unpack(endian + "IIIII", body[:20])
data = body[20:20 + caplen]
ts = ((tsh << 32) | tsl) / tsdiv.get(iface, 1_000_000)
yield ts, linktypes.get(iface, 1), data, origlen
# Les autres blocs (SPB sans horodatage, NRB, ISB...) sont ignores.
def iter_capture(path):
"""Ouvre la capture et renvoie (generateur, format)."""
f = open(path, "rb")
magic = f.read(4)
f.seek(0)
if magic == b"\x0a\x0d\x0d\x0a":
return iter_pcapng(f), "pcapng"
return iter_pcap(f), "pcap"
# ---------------------------------------------------------------------------
# Decodage des couches basses
# ---------------------------------------------------------------------------
def decode_link(linktype, data):
"""Renvoie (payload_ip, version_ip) ou (None, raison)."""
if linktype == 1: # Ethernet
if len(data) < 14:
return None, "trame Ethernet trop courte"
etype = struct.unpack("!H", data[12:14])[0]
off = 14
while etype in (0x8100, 0x88A8, 0x9100, 0x9200): # VLAN empilees
if len(data) < off + 4:
return None, "en-tete VLAN tronquee"
etype = struct.unpack("!H", data[off + 2:off + 4])[0]
off += 4
if etype == 0x0800:
return data[off:], 4
if etype == 0x86DD:
return data[off:], 6
return None, "ethertype 0x%04x (non IP)" % etype
if linktype == 0: # NULL / Loopback (BSD)
if len(data) < 4:
return None, "en-tete loopback tronquee"
fam = struct.unpack("<I", data[:4])[0]
if fam > 0xFFFF:
fam = struct.unpack(">I", data[:4])[0]
if fam == 2:
return data[4:], 4
if fam in (24, 28, 30):
return data[4:], 6
return None, "famille loopback %d (non IP)" % fam
if linktype == 113: # Linux cooked v1
if len(data) < 16:
return None, "en-tete SLL tronquee"
etype = struct.unpack("!H", data[14:16])[0]
if etype == 0x0800:
return data[16:], 4
if etype == 0x86DD:
return data[16:], 6
return None, "SLL protocole 0x%04x (non IP)" % etype
if linktype == 276: # Linux cooked v2
if len(data) < 20:
return None, "en-tete SLL2 tronquee"
etype = struct.unpack("!H", data[0:2])[0]
if etype == 0x0800:
return data[20:], 4
if etype == 0x86DD:
return data[20:], 6
return None, "SLL2 protocole 0x%04x (non IP)" % etype
if linktype in (12, 101, 228, 229): # IP brute
if not data:
return None, "paquet vide"
ver = data[0] >> 4
if ver in (4, 6):
return data, ver
return None, "IP brute de version %d" % ver
return None, "couche liaison non geree (linktype %d)" % linktype
def ip_str(raw):
if len(raw) == 4:
return ".".join(str(b) for b in raw)
parts = [raw[i:i + 2].hex() for i in range(0, 16, 2)]
return ":".join(p.lstrip("0") or "0" for p in parts)
IPV6_EXT = {0, 43, 44, 51, 60, 135}
def decode_ip_udp(payload, version):
"""Renvoie (src, dst, sport, dport, donnees_udp) ou (None, raison)."""
if version == 4:
if len(payload) < 20:
return None, "en-tete IPv4 tronquee"
ihl = (payload[0] & 0x0F) * 4
if ihl < 20 or len(payload) < ihl:
return None, "IHL IPv4 invalide"
total_len = struct.unpack("!H", payload[2:4])[0]
proto = payload[9]
frag = struct.unpack("!H", payload[6:8])[0]
src, dst = ip_str(payload[12:16]), ip_str(payload[16:20])
if (frag & 0x1FFF) != 0:
return None, "fragment IPv4 non initial"
if proto != 17:
return None, "protocole IP %d (non UDP)" % proto
end = total_len if 0 < total_len <= len(payload) else len(payload)
body = payload[ihl:end]
else:
if len(payload) < 40:
return None, "en-tete IPv6 tronquee"
plen = struct.unpack("!H", payload[4:6])[0]
nxt = payload[6]
src, dst = ip_str(payload[8:24]), ip_str(payload[24:40])
off = 40
end = min(40 + plen, len(payload)) if plen else len(payload)
while nxt in IPV6_EXT:
if nxt == 44: # fragment
return None, "fragment IPv6"
if len(payload) < off + 8:
return None, "en-tete d'extension IPv6 tronquee"
ext_len = (payload[off + 1] + 1) * 8
nxt = payload[off]
off += ext_len
if nxt != 17:
return None, "next header IPv6 %d (non UDP)" % nxt
body = payload[off:end]
if len(body) < 8:
return None, "en-tete UDP tronquee"
sport, dport, ulen, _cksum = struct.unpack("!HHHH", body[:8])
data = body[8:8 + ulen - 8] if 8 <= ulen <= len(body) else body[8:]
return (src, dst, sport, dport, data), None
# ---------------------------------------------------------------------------
# Decodage RTP
# ---------------------------------------------------------------------------
def parse_rtp(buf):
"""Valide et decode une entete RTP. Renvoie (dict, None) ou (None, raison)."""
if len(buf) < 12:
return None, "charge utile UDP < 12 octets (entete RTP impossible)"
b0, b1 = buf[0], buf[1]
version = b0 >> 6
if version != 2:
return None, "version RTP = %d (attendu 2)" % version
padding = (b0 >> 5) & 1
extension = (b0 >> 4) & 1
cc = b0 & 0x0F
marker = b1 >> 7
pt = b1 & 0x7F
if 72 <= pt <= 76:
return None, "paquet RTCP (PT=%d)" % pt
seq, ts, ssrc = struct.unpack("!HII", buf[2:12])
hlen = 12 + 4 * cc
if len(buf) < hlen:
return None, "entete tronquee (CC=%d)" % cc
if extension:
if len(buf) < hlen + 4:
return None, "extension RTP tronquee"
ext_words = struct.unpack("!H", buf[hlen + 2:hlen + 4])[0]
hlen += 4 + 4 * ext_words
if len(buf) < hlen:
return None, "extension RTP declaree plus longue que le paquet"
plen = len(buf) - hlen
pad_len = 0
if padding:
if plen < 1:
return None, "bit P positionne sans octet de bourrage"
pad_len = buf[-1]
if pad_len == 0 or pad_len > plen:
return None, "longueur de bourrage RTP invalide (%d)" % pad_len
plen -= pad_len
return {
"seq": seq, "ts": ts, "ssrc": ssrc, "pt": pt, "marker": marker,
"cc": cc, "ext": extension, "hlen": hlen, "plen": plen,
"payload": buf[hlen:hlen + plen],
}, None
# ---------------------------------------------------------------------------
# Decodage G.711 -> PCM 16 bits
# ---------------------------------------------------------------------------
PLEINE_ECHELLE = 32768.0 # reference 0 dBov pour du PCM 16 bits
DB_MINIMUM = -90.0 # valeur rendue pour un silence numerique parfait
def _table_ulaw():
"""Table de decodage G.711 u-law (PT 0) -> PCM 16 bits signe."""
table = []
for octet in range(256):
u = ~octet & 0xFF
signe = u & 0x80
exposant = (u >> 4) & 0x07
mantisse = u & 0x0F
val = (((mantisse << 3) + 0x84) << exposant) - 0x84
table.append(-val if signe else val)
return table
def _table_alaw():
"""Table de decodage G.711 A-law (PT 8) -> PCM 16 bits signe."""
table = []
for octet in range(256):
a = octet ^ 0x55
signe = a & 0x80
exposant = (a >> 4) & 0x07
mantisse = a & 0x0F
if exposant == 0:
val = (mantisse << 4) + 8
elif exposant == 1:
val = (mantisse << 4) + 0x108
else:
val = ((mantisse << 4) + 0x108) << (exposant - 1)
# en A-law le bit de signe vaut 1 pour les valeurs positives
table.append(val if signe else -val)
return table
# Tables d'octets : b"".join(map(table.__getitem__, payload)) reconstruit
# directement un buffer exploitable par array("h").frombytes(). Les deux tables
# reproduisent les valeurs de reference de l'UIT-T G.711 (u-law 0xFF -> 0,
# 0x00 -> -32124 ; A-law 0xD5 -> +8, 0x2A -> -32256) et rendent environ 39 dB
# de rapport signal/bruit sur un aller-retour a pleine echelle.
ULAW_OCTETS = [struct.pack("h", v) for v in _table_ulaw()]
ALAW_OCTETS = [struct.pack("h", v) for v in _table_alaw()]
TABLES_G711 = {0: ULAW_OCTETS, 8: ALAW_OCTETS}
def db_ov(amplitude):
"""Amplitude lineaire (unites PCM) -> niveau en dBov."""
if amplitude <= 0:
return DB_MINIMUM
val = 20.0 * math.log10(amplitude / PLEINE_ECHELLE)
return val if val > DB_MINIMUM else DB_MINIMUM
# ---------------------------------------------------------------------------
# Reconstruction du signal : la "piste" audio d'un flux
# ---------------------------------------------------------------------------
class OptionsAudio:
"""Parametres de reconstruction / de rendu des niveaux sonores."""
def __init__(self, actif=True, bloc_ms=10.0, max_secondes=3600.0, pcm=False):
self.actif = actif
self.bloc_ms = bloc_ms
self.max_secondes = max_secondes
self.pcm = pcm # conserver les echantillons (export WAV)
class Piste:
"""Signal audio d'un flux, replace sur l'echelle de temps de l'horodatage RTP.
Les echantillons ne sont pas archives : chaque trame est decodee puis
reduite a des agregats par bloc de `bloc_ms` (mini, maxi, somme des carres,
nombre d'echantillons). C'est la resolution des graphes -- 10 ms par defaut,
soit deux blocs par trame G.711 de 20 ms.
Un bloc sans echantillon est un "trou" : trame perdue, flux en pause, ou
charge utile non decodable. Les graphes le distinguent d'un vrai silence.
"""
def __init__(self, t_origine, clock=G711_CLOCK, bloc_ms=10.0,
max_secondes=3600.0, garder_pcm=False):
self.t_origine = t_origine # date de capture de la 1re trame
self.clock = clock
self.bloc = max(1, int(round(clock * bloc_ms / 1000.0)))
self.bloc_ms = 1000.0 * self.bloc / clock
self.max_blocs = max(1, int(max_secondes * clock / self.bloc))
self.bmin = array("h")
self.bmax = array("h")
self.bsq = array("d")
self.bn = array("l")
self.n_blocs = 0
self.pcm = array("h") if garder_pcm else None
self.echantillons = 0
self.trames = 0
self.non_decodees = Counter() # par payload type
self.hors_plage = 0 # offset negatif (desequencement initial)
self.debordements = 0 # au-dela de --graphe-max-duree
# -- alimentation ------------------------------------------------------
def _etendre(self, n_blocs):
manque = n_blocs - self.n_blocs
if manque <= 0:
return
self.bmin.extend([0] * manque)
self.bmax.extend([0] * manque)
self.bsq.extend([0.0] * manque)
self.bn.extend([0] * manque)
self.n_blocs = n_blocs
def _placer_pcm(self, offset, ech):
besoin = offset + len(ech)
if len(self.pcm) < besoin:
self.pcm.frombytes(bytes(2 * (besoin - len(self.pcm)))) # silence
self.pcm[offset:besoin] = ech
def ajouter(self, offset, payload, pt):
"""Decode une trame et l'inscrit a `offset` echantillons de l'origine."""
table = TABLES_G711.get(pt)
if table is None:
self.non_decodees[pt] += 1
return
if not payload:
return
if offset < 0:
self.hors_plage += 1
return
fin = offset + len(payload)
if (fin - 1) // self.bloc + 1 > self.max_blocs:
self.debordements += 1
return
ech = array("h")
ech.frombytes(b"".join(map(table.__getitem__, payload)))
self._etendre((fin - 1) // self.bloc + 1)
self.trames += 1
self.echantillons += len(ech)
if self.pcm is not None:
self._placer_pcm(offset, ech)
i, n = 0, len(ech)
while i < n:
b = (offset + i) // self.bloc
j = min(n, (b + 1) * self.bloc - offset)
seg = ech[i:j]
mn, mx = min(seg), max(seg)
if self.bn[b]:
if mn < self.bmin[b]:
self.bmin[b] = mn
if mx > self.bmax[b]:
self.bmax[b] = mx
else:
self.bmin[b] = mn
self.bmax[b] = mx
self.bsq[b] += sum(v * v for v in seg)
self.bn[b] += j - i
i = j
# -- restitution -------------------------------------------------------
@property
def vide(self):
return self.echantillons == 0
@property
def duree(self):
return self.n_blocs * self.bloc / float(self.clock)
@property
def t_fin(self):
return self.t_origine + self.duree
def enveloppe(self, n_cols, t_debut, t_fin):
"""Reduit la piste a `n_cols` colonnes sur la fenetre [t_debut, t_fin].
Chaque colonne vaut None (hors du flux), False (dans le flux mais sans
echantillon : trou) ou un triplet (mini, maxi, rms) normalise sur
[-1, +1].
"""
if t_fin <= t_debut or self.n_blocs == 0:
return [None] * n_cols
par_sec = float(self.clock) / self.bloc
etendue = t_fin - t_debut
cols = []
for c in range(n_cols):
ta = t_debut + etendue * c / n_cols
tb = t_debut + etendue * (c + 1) / n_cols
b0 = int(math.floor((ta - self.t_origine) * par_sec))
b1 = int(math.ceil((tb - self.t_origine) * par_sec))
if b1 <= b0:
b1 = b0 + 1
if b1 <= 0 or b0 >= self.n_blocs:
cols.append(None)
continue
b0 = max(0, b0)
b1 = min(self.n_blocs, b1)
mn, mx, sq, n = 0, 0, 0.0, 0
for b in range(b0, b1):
if not self.bn[b]:
continue
if n == 0:
mn, mx = self.bmin[b], self.bmax[b]
else:
if self.bmin[b] < mn:
mn = self.bmin[b]
if self.bmax[b] > mx:
mx = self.bmax[b]
sq += self.bsq[b]
n += self.bn[b]
if n == 0:
cols.append(False) # trou a l'interieur du flux
else:
cols.append((mn / PLEINE_ECHELLE, mx / PLEINE_ECHELLE,
math.sqrt(sq / n) / PLEINE_ECHELLE))
return cols
def niveaux_blocs(self):
"""Niveau RMS de chaque bloc porteur d'echantillons, en dBov."""
return [db_ov(math.sqrt(self.bsq[b] / self.bn[b]))
for b in range(self.n_blocs) if self.bn[b]]
def stats(self, seuil_silence_db=-50.0):
"""Indicateurs de niveau sonore du flux."""
if self.vide:
return None
sq = sum(self.bsq[b] for b in range(self.n_blocs) if self.bn[b])
crete = max(max(abs(self.bmin[b]), abs(self.bmax[b]))
for b in range(self.n_blocs) if self.bn[b])
niveaux = sorted(self.niveaux_blocs())
actifs = [d for d in niveaux if d >= seuil_silence_db]
rms_actif = None
if actifs:
energie = sum(10 ** (d / 10.0) for d in actifs) / len(actifs)
rms_actif = 10.0 * math.log10(energie)
def p(q):
return niveaux[min(len(niveaux) - 1, int(q * (len(niveaux) - 1)))]
return {
"rms_db": db_ov(math.sqrt(sq / self.echantillons)),
"rms_actif_db": rms_actif,
"crete_db": db_ov(crete),
"crete_lin": crete / PLEINE_ECHELLE,
"bruit_db": p(0.10),
"median_db": p(0.50),
"p95_db": p(0.95),
"plage_db": p(0.95) - p(0.10),
"blocs": len(niveaux),
"blocs_actifs": len(actifs),
"silence_pct": 100.0 * (len(niveaux) - len(actifs)) / len(niveaux),
"trous": self.n_blocs - len(niveaux),
"trous_ms": (self.n_blocs - len(niveaux)) * self.bloc_ms,
"duree": self.duree,
"seuil_silence_db": seuil_silence_db,
"sature": crete >= 32700,
}
# ---------------------------------------------------------------------------
# Graphes en mode texte (facon Audacity : forme d'onde + niveaux)
# ---------------------------------------------------------------------------
JEU_ASCII = {"crete": "|", "rms": "#", "zero": "-", "trou": "x", "vide": " "}
JEU_UNICODE = {"crete": "▒", "rms": "█", "zero": "─",
"trou": "×", "vide": " "}
def _pas_joli(etendue, cibles=8):
"""Pas d'axe 1/2/5 x 10^n couvrant `etendue` en environ `cibles` graduations."""
if etendue <= 0:
return 1.0
brut = etendue / float(cibles)
exp = math.floor(math.log10(brut))
base = brut / (10.0 ** exp)
for m in (1.0, 2.0, 5.0):
if base <= m:
return m * (10.0 ** exp)
return 10.0 ** (exp + 1)
def _fmt_t(t, pas):
if pas >= 1:
return "%.0f" % t
if pas >= 0.1:
return "%.1f" % t
return "%.2f" % t
def _axe_temps(largeur, t_debut, t_fin, marge, base=0.0):
"""Deux lignes de texte : graduations et etiquettes de l'axe des temps.
Les dates sont affichees relativement a `base` (debut de capture).
"""
etendue = t_fin - t_debut
pas = _pas_joli(etendue, max(4, largeur // 14))
ticks = [" "] * largeur
lbl = [" "] * (largeur + 16)
k = math.ceil((t_debut - base) / pas)
while base + k * pas <= t_fin + 1e-9:
t = base + k * pas
x = int(round((t - t_debut) / etendue * (largeur - 1))) if etendue else 0
k += 1
if not 0 <= x < largeur:
continue
ticks[x] = "+"
texte = _fmt_t(t - base, pas)
deb = max(0, min(len(lbl) - len(texte) - 1, x - len(texte) // 2))
if all(lbl[deb + i] == " " for i in range(len(texte) + 1)):
for i, ch in enumerate(texte):
lbl[deb + i] = ch
return (marge + "".join(ticks).replace(" ", "-"),
marge + "".join(lbl).rstrip() + " s")
def graphe_forme_onde(piste, largeur, hauteur, t_debut, t_fin,
indent="", unicode_=False, gain=None, base=0.0):
"""Forme d'onde facon Audacity : enveloppe crete + coeur RMS, 0 au centre."""
jeu = JEU_UNICODE if unicode_ else JEU_ASCII
if hauteur % 2 == 0:
hauteur += 1
demi = hauteur // 2
cols = piste.enveloppe(largeur, t_debut, t_fin)
pics = [max(abs(c[0]), abs(c[1])) for c in cols if c]
if not pics:
print(indent + "aucun echantillon audio dans cette fenetre.")
return
if gain is None:
gain = 1.0 / max(pics) if max(pics) > 0 else 1.0
gain = min(gain, 512.0)
grille = [[jeu["vide"]] * largeur for _ in range(hauteur)]
for x, col in enumerate(cols):
if col is None:
continue
if col is False:
grille[demi][x] = jeu["trou"]
continue
mn, mx, rms = col[0] * gain, col[1] * gain, col[2] * gain
r_haut = max(0, demi - int(round(min(1.0, max(0.0, mx)) * demi)))
r_bas = min(hauteur - 1, demi + int(round(min(1.0, max(0.0, -mn)) * demi)))
for r in range(r_haut, r_bas + 1):
grille[r][x] = jeu["crete"]
r_haut = max(0, demi - int(round(min(1.0, rms) * demi)))
r_bas = min(hauteur - 1, demi + int(round(min(1.0, rms) * demi)))
for r in range(r_haut, r_bas + 1):
grille[r][x] = jeu["rms"]
marge = indent + " " * 8
for r, ligne in enumerate(grille):
if r == demi:
etiq = "%7s " % "0"
ligne = [jeu["zero"] if ch == jeu["vide"] else ch for ch in ligne]
elif r == 0:
etiq = "%7s " % ("%+.2f" % (1.0 / gain))
elif r == hauteur - 1:
etiq = "%7s " % ("%+.2f" % (-1.0 / gain))
else:
etiq = " " * 8
print(indent + etiq + "".join(ligne))
for ligne in _axe_temps(largeur, t_debut, t_fin, marge, base):
print(ligne)
print(marge + "echelle verticale : +/- %.3f de la pleine echelle (crete a %+.1f dBov)"
% (1.0 / gain, db_ov(max(pics) * PLEINE_ECHELLE)))
return gain
def graphe_niveaux(piste, largeur, hauteur, t_debut, t_fin, plancher=-60.0,
indent="", unicode_=False, base=0.0):
"""Niveau sonore en dBov au fil du temps : barres crete + barres RMS."""
jeu = JEU_UNICODE if unicode_ else JEU_ASCII
cols = piste.enveloppe(largeur, t_debut, t_fin)
if not any(c for c in cols):
print(indent + "aucun echantillon audio dans cette fenetre.")
return
grille = [[jeu["vide"]] * largeur for _ in range(hauteur)]
def ligne_de(db):
"""0 dBov -> rangee 0 (haut) ; plancher -> rangee hauteur-1 (bas)."""
f = (0.0 - db) / (0.0 - plancher)
return max(0, min(hauteur - 1, int(round(f * (hauteur - 1)))))
for x, col in enumerate(cols):
if col is None:
continue
if col is False:
grille[hauteur - 1][x] = jeu["trou"]
continue
crete = db_ov(max(abs(col[0]), abs(col[1])) * PLEINE_ECHELLE)
rms = db_ov(col[2] * PLEINE_ECHELLE)
if crete > plancher:
for r in range(ligne_de(crete), hauteur):
grille[r][x] = jeu["crete"]
if rms > plancher:
for r in range(ligne_de(rms), hauteur):
grille[r][x] = jeu["rms"]
pas_db = _pas_joli(-plancher, min(6, hauteur - 1))
def fmt_db(d):
return "0 dB" if not d else "%+.0f dB" % d
etiquettes = {}
db = 0.0
while db >= plancher:
etiquettes[ligne_de(db)] = fmt_db(db)
db -= pas_db
etiquettes[hauteur - 1] = fmt_db(plancher)
marge = indent + " " * 8
for r, ligne in enumerate(grille):
etiq = "%7s " % etiquettes.get(r, "")
print(indent + etiq + "".join(ligne))
for ligne in _axe_temps(largeur, t_debut, t_fin, marge, base):
print(ligne)
# ---------------------------------------------------------------------------
# Graphe SVG (aucune dependance : le SVG est du texte)
# ---------------------------------------------------------------------------
# Palette : rampe bleue ordinale (crete clair / RMS fonce), rouge d'etat pour
# les trous. Les deux modes clair et sombre sont choisis pour leur surface.
SVG_STYLE = """
.fond{fill:#fcfcfb;}
.piste{fill:#f4f4f1;}
text{font-family:system-ui,-apple-system,"Segoe UI",sans-serif;fill:#52514e;}
.titre{fill:#0b0b0b;font-size:17px;font-weight:600;}
.sstitre{font-size:12px;}
.flux{fill:#0b0b0b;font-size:13px;font-weight:600;}
.note{fill:#898781;font-size:11px;}
.tick{fill:#898781;font-size:10.5px;font-variant-numeric:tabular-nums;}
.grille{stroke:#e1e0d9;stroke-width:1;fill:none;}
.axe,.zero{stroke:#c3c2b7;stroke-width:1;fill:none;}
.c-crete{fill:#86b6ef;}
.c-rms{fill:#1c5cab;}
.c-trou{fill:#d03b3b;opacity:.18;}
.b-trou{fill:#d03b3b;}
.survol{fill:#0b0b0b;fill-opacity:0;}
.survol:hover{fill-opacity:.07;}
@media (prefers-color-scheme: dark){
.fond{fill:#1a1a19;}
.piste{fill:#0d0d0d;}
text{fill:#c3c2b7;}
.titre,.flux{fill:#ffffff;}
.note,.tick{fill:#898781;}
.grille{stroke:#2c2c2a;}
.axe,.zero{stroke:#383835;}
.c-crete{fill:#256abf;}
.c-rms{fill:#86b6ef;}
.c-trou{fill:#e66767;opacity:.22;}
.b-trou{fill:#e66767;}
.survol{fill:#ffffff;}
}
"""
def _esc(s):
return (str(s).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
def _sous_chemins(points, ferme_a):
"""Chemin SVG : une sous-courbe par plage continue de colonnes renseignees.
`points` : liste de (x, y) ou None. `ferme_a` : ordonnee de la ligne de base.
"""
morceaux, courant = [], []
for pt in list(points) + [None]:
if pt is None:
if len(courant) >= 1:
d = ["M %.1f %.1f" % (courant[0][0], ferme_a)]
d += ["L %.1f %.2f" % (x, y) for x, y in courant]
d.append("L %.1f %.1f Z" % (courant[-1][0], ferme_a))
morceaux.append(" ".join(d))
courant = []
else:
courant.append(pt)
return " ".join(morceaux)
def _enveloppe_chemin(cols, x0, dx, y_zero, demi, gain):
"""Polygone mini/maxi de la forme d'onde, une sous-courbe par plage continue."""
morceaux, courant = [], []
for i, col in enumerate(list(cols) + [None]):
if not col:
if len(courant) >= 1:
haut = ["%.1f %.2f" % (x, yh) for x, yh, _ in courant]
bas = ["%.1f %.2f" % (x, yb) for x, _, yb in reversed(courant)]
morceaux.append("M " + " L ".join(haut) + " L " + " L ".join(bas) + " Z")
courant = []
continue
x = x0 + (i + 0.5) * dx
mx = max(-1.0, min(1.0, col[1] * gain))
mn = max(-1.0, min(1.0, col[0] * gain))
yh = y_zero - mx * demi
yb = y_zero - mn * demi
if yb - yh < 0.7: # toujours visible
yh, yb = (yh + yb) / 2 - 0.35, (yh + yb) / 2 + 0.35
courant.append((x, yh, yb))
return " ".join(morceaux)
def _plages_trous(cols):
"""Regroupe les colonnes sans echantillon en plages (debut, fin_exclue)."""
plages, debut = [], None
for i, col in enumerate(list(cols) + [None]):
manque = col is False
if manque and debut is None:
debut = i
elif not manque and debut is not None:
plages.append((debut, i))
debut = None
return plages
def svg_niveaux(r, chemin, largeur_px=1180, plancher=-60.0, max_colonnes=1200):
"""Ecrit un SVG : forme d'onde + niveau sonore (dBov) pour chaque flux."""
pistes = [(f, f.audio) for f in sorted(r["flux"].values(), key=lambda x: -x.n)
if f.audio is not None and not f.audio.vide]
if not pistes:
return None
ML, MR = 78, 26
H_ONDE, H_NIV, H_ETIQ, H_AXE, ECART = 118, 104, 24, 30, 26
largeur = max(320, largeur_px - ML - MR)
n_cols = min(max_colonnes, largeur)
dx = largeur / float(n_cols)
t0 = min(p.t_origine for _, p in pistes)
t1 = max(p.t_fin for _, p in pistes)
if t1 <= t0:
t1 = t0 + 1.0
base = r["premiere_cap"] or t0
etendue = t1 - t0
y = 34
corps = []
corps.append('<text class="titre" x="%d" y="%d">Niveaux sonores RTP / G.711</text>'
% (ML, y))
y += 19
corps.append('<text class="sstitre" x="%d" y="%d">%s &#183; %d flux &#183; '
'resolution %.0f ms &#183; 0 dBov = pleine echelle PCM 16 bits</text>'
% (ML, y, _esc(os.path.basename(r["fichier"])), len(pistes),
pistes[0][1].bloc_ms))
y += 24
# Legende (2 series + l'etat "trou") : l'identite ne passe jamais par la
# seule couleur, chaque pastille porte son libelle.
lx = ML
for classe, libelle in (("c-crete", "crete"), ("c-rms", "RMS"),
("b-trou", "trame manquante")):
corps.append('<rect class="%s" x="%d" y="%d" width="11" height="11" rx="2"/>'
% (classe, lx, y - 9))
corps.append('<text class="note" x="%d" y="%d">%s</text>'
% (lx + 16, y, libelle))
lx += 22 + 7 * len(libelle)
y += 18
pas_t = _pas_joli(etendue, max(4, largeur // 110))
for flux, piste in pistes:
cols = piste.enveloppe(n_cols, t0, t1)
pics = [max(abs(c[0]), abs(c[1])) for c in cols if c]
gain = min(512.0, 1.0 / max(pics)) if pics and max(pics) > 0 else 1.0
trous = _plages_trous(cols)
st = piste.stats()
y += H_ETIQ
corps.append('<text class="flux" x="%d" y="%d">%s</text>'
% (ML, y - 8, _esc(flux.label)))
corps.append('<text class="note" x="%d" y="%d" text-anchor="end">'
'RMS %+.1f dBov &#183; crete %+.1f dBov &#183; silence %.0f %% '
'&#183; manquant %.0f ms</text>'
% (ML + largeur, y - 8, st["rms_db"], st["crete_db"],
st["silence_pct"], st["trous_ms"]))
# --- piste 1 : forme d'onde (amplitude lineaire) ---
yo = y
y_zero = yo + H_ONDE / 2.0
demi = H_ONDE / 2.0 - 6
corps.append('<rect class="piste" x="%d" y="%.1f" width="%.1f" height="%d" rx="3"/>'
% (ML, yo, largeur, H_ONDE))
for a, b in trous:
corps.append('<rect class="c-trou" x="%.1f" y="%.1f" width="%.1f" height="%d"/>'
% (ML + a * dx, yo, max(1.0, (b - a) * dx), H_ONDE))
corps.append('<path class="c-crete" d="%s"/>'
% _enveloppe_chemin(cols, ML, dx, y_zero, demi, gain))
corps.append('<path class="c-rms" d="%s"/>'
% _enveloppe_chemin([(-c[2], c[2], c[2]) if c else c for c in cols],
ML, dx, y_zero, demi, gain))
corps.append('<line class="zero" x1="%d" y1="%.1f" x2="%.1f" y2="%.1f"/>'
% (ML, y_zero, ML + largeur, y_zero))
amp = 1.0 / gain
for val, yy in ((amp, y_zero - demi), (0.0, y_zero), (-amp, y_zero + demi)):
corps.append('<text class="tick" x="%d" y="%.1f" text-anchor="end">%s</text>'
% (ML - 8, yy + 3.5, "%+.2f" % val if val else "0"))
y += H_ONDE + 20
# --- piste 2 : niveau en dBov ---
yn = y
corps.append('<rect class="piste" x="%d" y="%.1f" width="%.1f" height="%d" rx="3"/>'
% (ML, yn, largeur, H_NIV))
def ydb(d, _yn=yn):
return _yn + H_NIV * (0.0 - max(plancher, min(0.0, d))) / (0.0 - plancher)
pas_db = _pas_joli(-plancher, 4)
d = 0.0
while d >= plancher - 1e-9:
corps.append('<line class="grille" x1="%d" y1="%.1f" x2="%.1f" y2="%.1f"/>'
% (ML, ydb(d), ML + largeur, ydb(d)))
corps.append('<text class="tick" x="%d" y="%.1f" text-anchor="end">'
'%s dB</text>'
% (ML - 8, ydb(d) + 3.5, "0" if not d else "%+.0f" % d))
d -= pas_db
for a, b in trous:
corps.append('<rect class="c-trou" x="%.1f" y="%.1f" width="%.1f" height="%d"/>'
% (ML + a * dx, yn, max(1.0, (b - a) * dx), H_NIV))
pc = [(ML + (i + 0.5) * dx,
ydb(db_ov(max(abs(c[0]), abs(c[1])) * PLEINE_ECHELLE))) if c else None
for i, c in enumerate(cols)]
pr = [(ML + (i + 0.5) * dx, ydb(db_ov(c[2] * PLEINE_ECHELLE))) if c else None
for i, c in enumerate(cols)]
corps.append('<path class="c-crete" d="%s"/>' % _sous_chemins(pc, yn + H_NIV))
corps.append('<path class="c-rms" d="%s"/>' % _sous_chemins(pr, yn + H_NIV))
y += H_NIV
# --- axe des temps, commun aux deux pistes ---
corps.append('<line class="axe" x1="%d" y1="%.1f" x2="%.1f" y2="%.1f"/>'
% (ML, y, ML + largeur, y))
k = math.ceil((t0 - base) / pas_t)
while base + k * pas_t <= t1 + 1e-9:
t = base + k * pas_t
k += 1
x = ML + (t - t0) / etendue * largeur
corps.append('<line class="axe" x1="%.1f" y1="%.1f" x2="%.1f" y2="%.1f"/>'
% (x, y, x, y + 4))
corps.append('<text class="tick" x="%.1f" y="%.1f" text-anchor="middle">%s</text>'
% (x, y + 16, _fmt_t(t - base, pas_t)))
corps.append('<text class="note" x="%.1f" y="%.1f">s</text>'
% (ML + largeur + 6, y + 16))
corps.append('<text class="note" x="%d" y="%.1f">echelle verticale : '
'+/- %.3f de la pleine echelle</text>' % (ML, y + 27, amp))
# --- couche de survol : infobulles natives, par tranche de temps ---
n_tr = min(300, n_cols)
pas_tr = n_cols / float(n_tr)
for k in range(n_tr):
a, b = int(k * pas_tr), max(int(k * pas_tr) + 1, int((k + 1) * pas_tr))
lot = [c for c in cols[a:b] if c]
if not lot:
continue
crete = max(max(abs(c[0]), abs(c[1])) for c in lot)
rms = math.sqrt(sum(c[2] ** 2 for c in lot) / len(lot))
corps.append('<rect class="survol" x="%.1f" y="%.1f" width="%.2f" '
'height="%.1f"><title>t = %.2f a %.2f s &#183; '
'crete %+.1f dBov &#183; RMS %+.1f dBov</title></rect>'
% (ML + a * dx, yo, (b - a) * dx, y - yo,
t0 + etendue * a / n_cols - base,
t0 + etendue * b / n_cols - base,
db_ov(crete * PLEINE_ECHELLE),
db_ov(rms * PLEINE_ECHELLE)))
y += H_AXE + ECART
hauteur = int(y + 8)
out = ['<?xml version="1.0" encoding="UTF-8"?>',
'<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" '
'viewBox="0 0 %d %d" role="img" aria-label="Niveaux sonores des flux RTP">'
% (largeur_px, hauteur, largeur_px, hauteur),
"<style>%s</style>" % SVG_STYLE,
'<rect class="fond" width="100%" height="100%"/>']
out += corps
out.append("</svg>")
with open(chemin, "w") as fh:
fh.write("\n".join(out) + "\n")
return chemin
# ---------------------------------------------------------------------------
# Export WAV (pour ouvrir la piste reconstruite dans Audacity)
# ---------------------------------------------------------------------------
def exporter_wav(r, prefixe):
"""Ecrit un WAV PCM 16 bits mono par flux.
La piste est celle de l'echelle de temps RTP : les trames manquantes y
sont comblees par du silence, pour que la duree du WAV corresponde a la
duree du flux. Renvoie [(nom, duree, part de silence ajoute)].
"""
ecrits = []
for f in sorted(r["flux"].values(), key=lambda x: -x.n):
piste = f.audio
if piste is None or piste.pcm is None or not len(piste.pcm):
continue
ech = piste.pcm
if sys.byteorder == "big":
ech = array("h", ech)
ech.byteswap()
nom = "%s-%08X-%s_%d-%s_%d.wav" % (
prefixe, f.ssrc, f.src.replace(":", "_"), f.sport,
f.dst.replace(":", "_"), f.dport)
with contextlib.closing(wave.open(nom, "wb")) as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(piste.clock)
w.writeframes(ech.tobytes())
comble = max(0, len(ech) - piste.echantillons) / float(piste.clock)
ecrits.append((nom, len(ech) / float(piste.clock), comble))
return ecrits
# ---------------------------------------------------------------------------
# Statistiques
# ---------------------------------------------------------------------------
def mean(values):
return sum(values) / len(values) if values else 0.0
def median(values):
if not values:
return 0.0
s = sorted(values)
n = len(s)
return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2
def stdev(values):
if len(values) < 2:
return 0.0
m = mean(values)
return (sum((v - m) ** 2 for v in values) / (len(values) - 1)) ** 0.5
def group_bursts(deltas_ms, seuil_ms):
"""Regroupe des trames consecutives separees de moins de `seuil_ms`.
`deltas_ms` contient les n-1 ecarts entre n trames consecutives.
Renvoie la liste des tailles de groupes (en nombre de trames).
"""
if not deltas_ms:
return []
groupes, courant = [], 1
for d in deltas_ms:
if d < seuil_ms:
courant += 1
else:
groupes.append(courant)
courant = 1
groupes.append(courant)
return groupes
class Flux:
"""Un flux RTP = un SSRC vu sur un 5-uplet IP/UDP donne."""
def __init__(self, key, opts_audio=None):
self.src, self.sport, self.dst, self.dport, self.ssrc = key
self.opts_audio = opts_audio
self.audio = None # Piste, creee a la premiere trame
self.n = 0
self.first_cap = None
self.last_cap = None
self.deltas = [] # ecarts d'arrivee (ms)
self.pts = Counter()
self.plens = Counter()
self.markers = 0
# sequence
self.prev_seq = None
self.seq_cycles = 0
self.first_seq_ext = None
self.last_seq_ext = None
self.seen_seq = set()
self.seen16 = set()
self.dupes = 0
self.reorders = 0
self.gaps = [] # (seq_precedent, seq_recu, nb_perdus, ts_capture)
self.wraps_seq = 0
# horodatage RTP
self.prev_ts = None
self.ts_cycles = 0
self.first_ts_ext = None
self.last_ts_ext = None
self.ts_decroissants = [] # (ts_precedent, ts_recu, ts_capture)
self.ts_egaux = 0
self.ts_deltas = Counter()
self.wraps_ts = 0
@property
def label(self):
return "SSRC 0x%08X %s:%d -> %s:%d" % (
self.ssrc, self.src, self.sport, self.dst, self.dport)
def add(self, t, rtp):
self.n += 1
if self.first_cap is None:
self.first_cap = t
else:
self.deltas.append((t - self.last_cap) * 1000.0)
self.last_cap = t
self.pts[rtp["pt"]] += 1
self.plens[rtp["plen"]] += 1
self.markers += rtp["marker"]
# --- numero de sequence (16 bits, avec rebouclage) ---
seq = rtp["seq"]
if self.prev_seq is None:
self.first_seq_ext = self.last_seq_ext = seq
else:
diff = (seq - self.prev_seq) & 0xFFFF
if diff == 0:
self.dupes += 1
elif diff < 0x8000: # en avant
if seq < self.prev_seq:
self.seq_cycles += 1
self.wraps_seq += 1
if diff > 1:
self.gaps.append((self.prev_seq, seq, diff - 1, self.last_cap))
else: # en arriere : desequencement
self.reorders += 1
seq_ext = (self.seq_cycles << 16) | seq
if self.last_seq_ext is None or seq_ext > self.last_seq_ext:
self.last_seq_ext = seq_ext
if seq_ext in self.seen_seq:
pass
self.seen_seq.add(seq_ext)
self.seen16.add(seq)
self.prev_seq = seq
# --- horodatage RTP (32 bits, avec rebouclage) ---
ts = rtp["ts"]
if self.prev_ts is None:
self.first_ts_ext = self.last_ts_ext = ts
else:
diff = (ts - self.prev_ts) & 0xFFFFFFFF
if diff == 0:
self.ts_egaux += 1
elif diff < 0x80000000: # croissant
if ts < self.prev_ts:
self.ts_cycles += 1
self.wraps_ts += 1
self.ts_deltas[diff] += 1
else:
self.ts_decroissants.append((self.prev_ts, ts, self.last_cap))
ts_ext = (self.ts_cycles << 32) | ts
if self.last_ts_ext is None or ts_ext > self.last_ts_ext:
self.last_ts_ext = ts_ext
self.prev_ts = ts
# --- signal audio, replace sur l'echelle de temps RTP ---
o = self.opts_audio
if o is not None and o.actif:
if self.audio is None:
self.audio = Piste(t, G711_CLOCK, o.bloc_ms, o.max_secondes, o.pcm)
self.audio.ajouter(ts_ext - self.first_ts_ext,
rtp.get("payload", b""), rtp["pt"])
# --- indicateurs derives ---
@property
def duree_capture(self):
if self.first_cap is None or self.n < 2:
return 0.0
return self.last_cap - self.first_cap
@property
def duree_rtp(self):
if self.first_ts_ext is None or self.n < 2:
return 0.0
return (self.last_ts_ext - self.first_ts_ext) / float(G711_CLOCK)
def trous_reels(self):
"""Pour chaque discontinuite : (prev, seq, manquants, recus_plus_tard, t)."""
out = []
for prev, seq, nb, t in self.gaps:
tardifs = sum(1 for k in range(1, nb + 1) if ((prev + k) & 0xFFFF) in self.seen16)
out.append((prev, seq, nb, tardifs, t))
return out
@property
def attendus(self):
"""Nombre de trames attendues d'apres l'etendue des seqnum."""
if self.first_seq_ext is None:
return 0
return self.last_seq_ext - self.first_seq_ext + 1
@property
def perdus(self):
return max(0, self.attendus - len(self.seen_seq))
# ---------------------------------------------------------------------------
# Analyse
# ---------------------------------------------------------------------------
DEFAUTS_GRAPHE = {
"actif": True, "largeur": 100, "hauteur": 17, "hauteur_niveaux": 12,
"plancher": DEFAULT_PLANCHER_DB, "seuil_silence": DEFAULT_SILENCE_DB,
"unicode": False,
}
def analyse(path, seuil_ms, limite, opts_audio=None, graphe=None):
it, fmt = iter_capture(path)
if opts_audio is None:
opts_audio = OptionsAudio()
total = 0
linktypes = Counter()
non_ip = Counter()
non_udp = Counter()
udp_total = 0
rtp_ok = 0
rejets = Counter()
exemples_rejets = []
trames = [] # (ts_capture, cle_flux, rtp) pour l'analyse globale
flux = {}
premiere_cap = derniere_cap = None
for ts, linktype, data, _origlen in it:
total += 1
linktypes[linktype] += 1
if premiere_cap is None:
premiere_cap = ts
derniere_cap = ts
ip_payload, info = decode_link(linktype, data)
if ip_payload is None:
non_ip[info] += 1
continue
res, raison = decode_ip_udp(ip_payload, info)
if res is None:
non_udp[raison] += 1
continue
src, dst, sport, dport, udp_data = res
udp_total += 1
rtp, raison = parse_rtp(udp_data)
if rtp is None:
rejets[raison] += 1
if len(exemples_rejets) < limite:
exemples_rejets.append((total, ts, "%s:%d -> %s:%d" % (src, sport, dst, dport),
len(udp_data), raison))
continue
rtp_ok += 1
key = (src, sport, dst, dport, rtp["ssrc"])
f = flux.get(key)
if f is None:
f = flux[key] = Flux(key, opts_audio)
f.add(ts, rtp)
trames.append((ts, key, rtp))
return {
"fichier": path, "format": fmt, "total": total, "linktypes": linktypes,
"non_ip": non_ip, "non_udp": non_udp, "udp_total": udp_total,
"rtp_ok": rtp_ok, "rejets": rejets, "exemples_rejets": exemples_rejets,
"flux": flux, "trames": trames,
"premiere_cap": premiere_cap, "derniere_cap": derniere_cap,
"seuil_ms": seuil_ms, "limite": limite,
"audio": opts_audio, "graphe": dict(DEFAUTS_GRAPHE, **(graphe or {})),
}
# ---------------------------------------------------------------------------
# Restitution
# ---------------------------------------------------------------------------
def titre(s):
print()
print("=" * 78)
print(s)
print("=" * 78)
def fmt_duree(s):
if s < 1:
return "%.3f s (%.1f ms)" % (s, s * 1000)
m, sec = divmod(s, 60)
if m < 1:
return "%.3f s" % s
return "%.3f s (%dm %.3fs)" % (s, int(m), sec)
def stats_ecarts(deltas, seuil_ms, indent=""):
if not deltas:
print(indent + "moins de 2 trames : pas d'ecart mesurable.")
return
print(indent + "ecart minimal : %9.3f ms" % min(deltas))
print(indent + "ecart maximal : %9.3f ms" % max(deltas))
print(indent + "ecart moyen : %9.3f ms" % mean(deltas))
print(indent + "ecart median : %9.3f ms" % median(deltas))
print(indent + "ecart type : %9.3f ms" % stdev(deltas))
sous = [d for d in deltas if d < seuil_ms]
print(indent + "ecarts < %g ms : %d / %d (%.1f %%)" % (
seuil_ms, len(sous), len(deltas), 100.0 * len(sous) / len(deltas)))
def rapport_bursts(deltas, seuil_ms, indent=""):
groupes = group_bursts(deltas, seuil_ms)
if not groupes:
print(indent + "pas assez de trames.")
return
paquets = [g for g in groupes if g >= 2]
isolees = sum(1 for g in groupes if g == 1)
print(indent + "groupes au total : %d (dont %d trames isolees)" % (len(groupes), isolees))
print(indent + "\"paquets\" (>= 2 trames) : %d" % len(paquets))
if paquets:
print(indent + " trames par paquet mini : %d" % min(paquets))
print(indent + " trames par paquet maxi : %d" % max(paquets))
print(indent + " trames par paquet moyen : %.2f" % mean(paquets))
print(indent + " trames par paquet median: %.1f" % median(paquets))
print(indent + " trames regroupees : %d / %d (%.1f %%)" % (
sum(paquets), sum(groupes), 100.0 * sum(paquets) / sum(groupes)))
else:
print(indent + " aucun groupe de 2 trames ou plus sous le seuil.")
def afficher(r):
seuil = r["seuil_ms"]
limite = r["limite"]
flux = r["flux"]
titre("FICHIER")
print("chemin : %s" % r["fichier"])
print("format : %s" % r["format"])
print("taille : %d octets" % os.path.getsize(r["fichier"]))
couches = ", ".join("%s (%d paquets)" % (LINKTYPE_NAMES.get(lt, "linktype %d" % lt), n)
for lt, n in r["linktypes"].most_common())
print("couches liaison : %s" % (couches or "-"))
titre("PAQUETS")
print("paquets captures : %d" % r["total"])
print("paquets UDP : %d" % r["udp_total"])
print("trames RTP valides : %d" % r["rtp_ok"])
print("paquets UDP ignores : %d" % sum(r["rejets"].values()))
print("paquets non UDP/non IP : %d" % (sum(r["non_ip"].values()) + sum(r["non_udp"].values())))
if r["non_ip"] or r["non_udp"]:
print("\n Paquets ecartes avant UDP :")
for raison, n in (r["non_ip"] + r["non_udp"]).most_common():
print(" %-55s %6d" % (raison, n))
if r["rejets"]:
print("\n /!\\ Paquets UDP qui ne sont PAS des trames RTP valides :")
for raison, n in r["rejets"].most_common():
print(" %-55s %6d" % (raison, n))
print("\n Details (%d premiers) :" % min(limite, len(r["exemples_rejets"])))
print(" %-8s %-14s %-42s %7s %s" % ("paquet", "t (s)", "flux", "udp", "raison"))
t0 = r["premiere_cap"] or 0
for idx, ts, flow, ln, raison in r["exemples_rejets"]:
print(" #%-7d %-14.6f %-42s %7d %s" % (idx, ts - t0, flow, ln, raison))
else:
print("\n Tous les paquets UDP portent une entete RTP valide.")
if not flux:
titre("RESULTAT")
print("Aucune trame RTP exploitable dans cette capture.")
return
# ---------------- Flux ----------------
titre("FLUX RTP DETECTES (%d)" % len(flux))
print("%-12s %-21s %-21s %8s %10s %9s" % ("SSRC", "source", "destination", "trames", "PT", "duree(s)"))
for f in sorted(flux.values(), key=lambda x: -x.n):
pt, _ = f.pts.most_common(1)[0]
pt_lbl = "%d%s" % (pt, "" if pt in G711_PT else "*")
print("0x%08X %-21s %-21s %8d %10s %9.3f" % (
f.ssrc, "%s:%d" % (f.src, f.sport), "%s:%d" % (f.dst, f.dport),
f.n, pt_lbl, f.duree_capture))
hors_g711 = sorted({pt for f in flux.values() for pt in f.pts if pt not in G711_PT})
if hors_g711:
print("\n /!\\ (*) types de charge utile non G.711 rencontres : %s" %
", ".join(str(p) for p in hors_g711))
print(" l'horloge RTP de %d Hz (G.711) est tout de meme appliquee." % G711_CLOCK)
# ---------------- Nombre de trames ----------------
titre("NOMBRE DE TRAMES")
print("trames RTP totales : %d" % r["rtp_ok"])
for f in sorted(flux.values(), key=lambda x: -x.n):
pt, _ = f.pts.most_common(1)[0]
taille, n_taille = f.plens.most_common(1)[0]
print(" %-46s %7d trames | PT %-3d %-22s | charge utile dominante %d o (%.1f ms) x%d" % (
f.label, f.n, pt, G711_PT.get(pt, "non G.711"), taille,
1000.0 * taille / G711_CLOCK, n_taille))
# ---------------- Espacement temporel ----------------
titre("ESPACEMENT TEMPOREL (horodatage de capture)")
globaux = []
trames = r["trames"]
for i in range(1, len(trames)):
globaux.append((trames[i][0] - trames[i - 1][0]) * 1000.0)
if len(flux) > 1:
print("Toutes trames RTP confondues, flux entrelaces (%d ecarts) :" % len(globaux))
stats_ecarts(globaux, seuil, " ")
for f in sorted(flux.values(), key=lambda x: -x.n):
print("\n%s (%d ecarts) :" % (f.label, len(f.deltas)))
stats_ecarts(f.deltas, seuil, " ")
else:
stats_ecarts(globaux, seuil, " ")
# ---------------- Bursts ----------------
titre("REGROUPEMENT EN \"PAQUETS\" DE TRAMES (ecart < %g ms)" % seuil)
print("Un \"paquet\" est une suite maximale de trames consecutives dont chaque")
print("ecart avec la precedente est strictement inferieur a %g ms.\n" % seuil)
print("Toutes trames RTP confondues :")
rapport_bursts(globaux, seuil, " ")
if len(flux) > 1:
for f in sorted(flux.values(), key=lambda x: -x.n):
print("\n%s :" % f.label)
rapport_bursts(f.deltas, seuil, " ")
# ---------------- Sequence ----------------
titre("CONTINUITE DES NUMEROS DE SEQUENCE (par SSRC)")
for f in sorted(flux.values(), key=lambda x: -x.n):
t0 = r["premiere_cap"] or 0
ok = not f.gaps and not f.dupes and not f.reorders and f.perdus == 0
print("\n%s" % f.label)
print(" seq premiere/derniere : %d -> %d (etendue %d, rebouclages 16 bits : %d)" % (
f.first_seq_ext & 0xFFFF, f.last_seq_ext & 0xFFFF, f.attendus, f.wraps_seq))
print(" trames recues : %d attendues : %d manquantes : %d (%.2f %%)" % (
f.n, f.attendus, f.perdus,
100.0 * f.perdus / f.attendus if f.attendus else 0.0))
print(" doublons : %d" % f.dupes)
print(" desequencements : %d" % f.reorders)
print(" discontinuites : %d" % len(f.gaps))
if ok:
print(" => OK : sequence continue, sans trou ni doublon ni desequencement.")
else:
print(" => ANOMALIE(S) detectee(s).")
for prev, seq, nb, tardifs, ts in f.trous_reels()[:limite]:
suite = ""
if tardifs:
suite = " dont %d recue(s) plus tard (desequencement)" % tardifs
print(" t=%10.6f s : saut %d -> %d (%d trame(s) manquante(s)%s)" % (
ts - t0, prev, seq, nb, suite))
if len(f.gaps) > limite:
print(" ... %d discontinuite(s) supplementaire(s)" % (len(f.gaps) - limite))
# ---------------- Timestamps ----------------
titre("CROISSANCE DES HORODATAGES RTP (par SSRC)")
for f in sorted(flux.values(), key=lambda x: -x.n):
t0 = r["premiere_cap"] or 0
print("\n%s" % f.label)
print(" timestamp premier/dernier : %u -> %u (rebouclages 32 bits : %d)" % (
f.first_ts_ext & 0xFFFFFFFF, f.last_ts_ext & 0xFFFFFFFF, f.wraps_ts))
print(" progressions decroissantes: %d" % len(f.ts_decroissants))
print(" timestamps identiques : %d" % f.ts_egaux)
if f.ts_deltas:
communs = ", ".join("%d (x%d, %.1f ms)" % (d, n, 1000.0 * d / G711_CLOCK)
for d, n in f.ts_deltas.most_common(5))
print(" increments les plus vus : %s" % communs)
if not f.ts_decroissants and not f.ts_egaux:
print(" => OK : horodatage RTP strictement croissant.")
elif not f.ts_decroissants:
print(" => OK : horodatage croissant au sens large (%d repetitions)." % f.ts_egaux)
else:
print(" => ANOMALIE : horodatage non croissant.")
for prev, cur, ts in f.ts_decroissants[:limite]:
print(" t=%10.6f s : %u -> %u (recul de %u)" % (
ts - t0, prev, cur, (prev - cur) & 0xFFFFFFFF))
if len(f.ts_decroissants) > limite:
print(" ... %d cas supplementaire(s)" % (len(f.ts_decroissants) - limite))
# ---------------- Durees ----------------
titre("DUREES")
d_cap_tot = (r["derniere_cap"] - r["premiere_cap"]) if r["total"] > 1 else 0.0
print("Capture complete (tous paquets) : %s" % fmt_duree(d_cap_tot))
if trames:
d_rtp_cap = trames[-1][0] - trames[0][0]
print("Premiere -> derniere trame RTP : %s" % fmt_duree(d_rtp_cap))
print()
print("%-46s %14s %14s %10s" % ("flux", "duree capture", "duree RTP", "ecart"))
for f in sorted(flux.values(), key=lambda x: -x.n):
dc, dr = f.duree_capture, f.duree_rtp
print("%-46s %11.3f s %11.3f s %8.1f ms" % (f.label, dc, dr, (dc - dr) * 1000.0))
print()
print("Rappel : la duree RTP est calculee sur l'horloge G.711 de %d Hz," % G711_CLOCK)
print("soit (dernier timestamp - premier timestamp) / %d, hors duree de la" % G711_CLOCK)
print("derniere trame. L'ecart capture/RTP traduit la gigue et les pertes.")
for f in sorted(flux.values(), key=lambda x: -x.n):
taille, _ = f.plens.most_common(1)[0]
if taille:
audio = f.n * taille / float(G711_CLOCK)
print(" %s : %d trames x %d o = %.3f s d'audio effectivement capture." % (
f.label, f.n, taille, audio))
# ---------------- Niveaux sonores ----------------
if r["audio"].actif:
rapport_niveaux(r)
def rapport_niveaux(r):
"""Rubrique "niveaux sonores" : chiffres, puis graphes facon Audacity."""
g = r["graphe"]
flux = sorted(r["flux"].values(), key=lambda x: -x.n)
pistes = [(f, f.audio) for f in flux if f.audio is not None and not f.audio.vide]
titre("NIVEAUX SONORES (charge utile G.711 decodee)")
if not pistes:
print("Aucune charge utile G.711 decodable (PT 0 / PT 8) : pas de graphe.")
pts = sorted({pt for f in flux if f.audio for pt in f.audio.non_decodees})
if pts:
print("Types de charge utile rencontres et non decodes : %s"
% ", ".join(str(p) for p in pts))
return
print("Le signal est reconstitue sur l'echelle de temps de l'horodatage RTP :")
print("chaque trame est replacee a sa position theorique, un trou de sequence")
print("laisse donc un trou dans la piste (il n'est pas comble par du silence).")
print("Reference : 0 dBov = pleine echelle PCM 16 bits. Resolution : %.0f ms."
% pistes[0][1].bloc_ms)
print("Seuil de silence retenu : %g dBov.\n" % g["seuil_silence"])
print("%-46s %9s %9s %9s %9s %8s %9s" % (
"flux", "RMS", "RMS actif", "crete", "bruit p10", "silence", "manquant"))
for f, piste in pistes:
st = piste.stats(g["seuil_silence"])
print("%-46s %8.1f%s %8s%s %8.1f%s %8.1f%s %7.1f%% %7.0f ms" % (
f.label, st["rms_db"], "dB",
("%.1f" % st["rms_actif_db"]) if st["rms_actif_db"] is not None else "-",
"dB", st["crete_db"], "dB", st["bruit_db"], "dB",
st["silence_pct"], st["trous_ms"]))
print()
for f, piste in pistes:
st = piste.stats(g["seuil_silence"])
print("%s" % f.label)
print(" duree de piste : %.3f s (%d blocs de %.0f ms, %d sans echantillon)"
% (st["duree"], piste.n_blocs, piste.bloc_ms, st["trous"]))
print(" trames decodees : %d (%d echantillons, %.3f s d'audio)"
% (piste.trames, piste.echantillons,
piste.echantillons / float(piste.clock)))
print(" niveau median / p95 : %+.1f dBov / %+.1f dBov (plage utile %.1f dB)"
% (st["median_db"], st["p95_db"], st["plage_db"]))
print(" crete : %+.1f dBov (%.3f de la pleine echelle)%s"
% (st["crete_db"], st["crete_lin"],
" /!\\ SATURATION" if st["sature"] else ""))
if piste.non_decodees:
print(" trames non decodees : %s" % ", ".join(
"PT %d x%d" % (pt, n) for pt, n in piste.non_decodees.most_common()))
if piste.hors_plage:
print(" trames hors piste : %d (anterieures au premier horodatage)"
% piste.hors_plage)
if piste.debordements:
print(" trames ignorees : %d (au-dela de --graphe-max-duree)"
% piste.debordements)
if st["crete_db"] <= g["seuil_silence"]:
print(" => flux silencieux sur toute sa duree.")
elif st["silence_pct"] >= 90:
print(" => quasi silencieux : %.1f %% des blocs sous %g dBov."
% (st["silence_pct"], g["seuil_silence"]))
elif st["sature"]:
print(" => ANOMALIE : le signal atteint la pleine echelle (ecretage).")
else:
print(" => niveau nominal.")
print()
if g["actif"]:
for f, piste in pistes:
t0, t1 = piste.t_origine, piste.t_fin
base = r["premiere_cap"] or t0
titre("FORME D'ONDE ET NIVEAUX -- %s" % f.label)
print("fenetre : t = %.3f s -> %.3f s (origine = debut de capture)\n"
% (t0 - base, t1 - base))
print("Forme d'onde (amplitude, '%s' = crete, '%s' = RMS, '%s' = trame manquante) :"
% (("▒", "█", "×") if g["unicode"] else ("|", "#", "x")))
graphe_forme_onde(piste, g["largeur"], g["hauteur"], t0, t1,
unicode_=g["unicode"], base=base)
print()
print("Niveau sonore (dBov, plancher %g dB) :" % g["plancher"])
graphe_niveaux(piste, g["largeur"], g["hauteur_niveaux"], t0, t1,
plancher=g["plancher"], unicode_=g["unicode"], base=base)
print()
# ---------------------------------------------------------------------------
def main(argv=None):
ap = argparse.ArgumentParser(
description="Analyse des trames RTP (G.711) d'une capture pcap/pcapng.")
ap.add_argument("capture", help="fichier .pcap ou .pcapng")
ap.add_argument("--seuil", type=float, default=DEFAULT_BURST_MS, metavar="MS",
help="seuil de regroupement en paquets de trames (defaut : %g ms)"
% DEFAULT_BURST_MS)
ap.add_argument("--limite", type=int, default=20, metavar="N",
help="nombre maximal d'anomalies detaillees par rubrique (defaut : 20)")
gr = ap.add_argument_group("graphe des niveaux sonores (G.711 decode)")
gr.add_argument("--sans-graphe", action="store_true",
help="ne pas decoder l'audio ni tracer les niveaux")
gr.add_argument("--sans-graphe-texte", action="store_true",
help="garder les statistiques de niveau, sans les graphes ASCII")
gr.add_argument("--graphe-largeur", type=int, default=0, metavar="N",
help="largeur des graphes en caracteres (defaut : largeur du terminal)")
gr.add_argument("--graphe-hauteur", type=int, default=17, metavar="N",
help="hauteur de la forme d'onde en lignes (defaut : 17)")
gr.add_argument("--graphe-plancher", type=float, default=DEFAULT_PLANCHER_DB,
metavar="DB",
help="bas de l'echelle du graphe des niveaux (defaut : %g dBov)"
% DEFAULT_PLANCHER_DB)
gr.add_argument("--graphe-silence", type=float, default=DEFAULT_SILENCE_DB,
metavar="DB",
help="seuil de detection du silence (defaut : %g dBov)"
% DEFAULT_SILENCE_DB)
gr.add_argument("--graphe-resolution", type=float, default=DEFAULT_BLOC_MS,
metavar="MS",
help="resolution temporelle des niveaux (defaut : %g ms)"
% DEFAULT_BLOC_MS)
gr.add_argument("--graphe-max-duree", type=float, default=DEFAULT_MAX_DUREE,
metavar="S",
help="duree maximale reconstruite par flux (defaut : %g s)"
% DEFAULT_MAX_DUREE)
gr.add_argument("--graphe-unicode", action="store_true",
help="tracer les graphes ASCII avec des pleins Unicode")
gr.add_argument("--graphe-svg", metavar="FICHIER",
help="ecrire aussi le graphe au format SVG")
gr.add_argument("--graphe-svg-largeur", type=int, default=1180, metavar="PX",
help="largeur du SVG en pixels (defaut : 1180)")
gr.add_argument("--wav", metavar="PREFIXE",
help="exporter la piste reconstruite de chaque flux en WAV")
args = ap.parse_args(argv)
audio_actif = not args.sans_graphe
opts_audio = OptionsAudio(actif=audio_actif,
bloc_ms=max(1.0, args.graphe_resolution),
max_secondes=args.graphe_max_duree,
pcm=bool(args.wav) and audio_actif)
largeur = args.graphe_largeur or max(
40, min(160, shutil.get_terminal_size((100, 24)).columns - 10))
graphe = {
"actif": audio_actif and not args.sans_graphe_texte,
"largeur": largeur,
"hauteur": max(5, args.graphe_hauteur),
"hauteur_niveaux": max(5, args.graphe_hauteur - 5),
"plancher": min(-6.0, args.graphe_plancher),
"seuil_silence": args.graphe_silence,
"unicode": args.graphe_unicode,
}
try:
r = analyse(args.capture, args.seuil, args.limite, opts_audio, graphe)
except FileNotFoundError:
print("erreur : fichier introuvable : %s" % args.capture, file=sys.stderr)
return 2
except CaptureError as e:
print("erreur : capture illisible : %s" % e, file=sys.stderr)
return 2
afficher(r)
if audio_actif and args.graphe_svg:
titre("EXPORT SVG")
chemin = svg_niveaux(r, args.graphe_svg, args.graphe_svg_largeur,
graphe["plancher"])
if chemin:
print("graphe ecrit : %s (%d octets)" % (chemin, os.path.getsize(chemin)))
else:
print("rien a tracer : aucune charge utile G.711 decodee.")
if audio_actif and args.wav:
titre("EXPORT WAV")
fichiers = exporter_wav(r, args.wav)
if fichiers:
print("Pistes reconstruites (PCM 16 bits, %d Hz, mono) ;" % G711_CLOCK)
print("les trames manquantes sont comblees par du silence :")
print(" %-60s %10s %12s" % ("fichier", "duree", "dont comble"))
for nom, duree, comble in fichiers:
print(" %-60s %8.3f s %10.3f s" % (nom, duree, comble))
if comble > 0.25 * duree:
print(" /!\\ %.0f %% de silence ajoute : l'horodatage RTP de ce flux"
% (100.0 * comble / duree))
print(" comporte un bond important (voir --graphe-max-duree).")
else:
print("rien a exporter : aucune charge utile G.711 decodee.")
print()
return 0
if __name__ == "__main__":
sys.exit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment