sipmon 0.1.2

Passive SIP/RTP signaling & media quality monitoring (pcap/live, TUI + JSONL export)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

//! Simplified E-model (ITU-T G.107) MOS estimation.
//!
//! `R = 93.2 - Id(delay) - Ie-eff(codec, loss)`
//! `MOS = 1 + 0.035*R + 7e-6 * R*(R-60)*(100-R)` for R<100; R>=100 -> 4.5.

/// Codec impairment parameters `(Ie, Bpl)` with packet-loss concealment
/// (ITU-T G.107 Annex B typical values).
fn codec_params(codec: Option<&str>, pt: Option<u8>) -> (f64, f64) {
    let s = codec.map(|c| c.to_ascii_uppercase());
    match (s.as_deref(), pt) {
        (Some(c), _) if c.contains("PCMU") || c.contains("PCMA") || c.contains("G.711") => {
            (0.0, 25.1)
        }
        (Some(c), _) if c.contains("G.729") => (10.0, 19.0),
        (Some(c), _) if c.contains("G.723") => (15.0, 13.0),
        (Some(c), _) if c.contains("G.722") => (4.0, 15.7),
        (Some(c), _) if c.contains("OPUS") => (0.0, 25.0),
        (Some(c), _) if c.contains("TELEPHONE") => (0.0, 25.0),
        (_, Some(0)) | (_, Some(8)) => (0.0, 25.1),
        _ => (10.0, 10.0),
    }
}

/// Delay impairment Id (simplified G.107 default, no echo term).
fn delay_impairment(oneway_ms: f64) -> f64 {
    let d = oneway_ms.max(0.0);
    0.024 * d + 0.11 * (d - 177.3).max(0.0)
}

/// Effective equipment impairment Ie-eff (G.107):
/// `Ie + (95 - Ie) * Ppl / (Ppl / BurstR + Bpl)`, BurstR = 1 for random loss.
fn equipment_impairment(codec: Option<&str>, pt: Option<u8>, loss_pct: f64) -> f64 {
    let (ie, bpl) = codec_params(codec, pt);
    let p = loss_pct.max(0.0);
    ie + (95.0 - ie) * p / (p / 1.0 + bpl)
}

/// Compute MOS.
///
/// `oneway_ms` is the one-way delay (use RTT/2 if only round-trip is known).
/// `jitter_ms` adds a de-jitter buffer contribution to the effective delay.
pub fn estimate_mos(
    codec: Option<&str>,
    pt: Option<u8>,
    loss_pct: f64,
    oneway_ms: Option<f64>,
    jitter_ms: Option<f64>,
) -> Option<f64> {
    let oneway = oneway_ms.unwrap_or(0.0);
    let jitter = jitter_ms.unwrap_or(0.0);
    // Approximate mouth-to-ear delay: network one-way + de-jitter buffer.
    let dejitter = (40.0_f64).max(jitter * 3.0);
    let d = oneway + dejitter;

    let id = delay_impairment(d);
    let ie = equipment_impairment(codec, pt, loss_pct);
    let r = (93.2 - id - ie).clamp(0.0, 100.0);
    let mos = if r >= 100.0 {
        4.5
    } else {
        1.0 + 0.035 * r + 7e-6 * r * (r - 60.0) * (100.0 - r)
    };
    Some(mos.clamp(1.0, 4.5))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn g711_lossless_low_delay_is_high_mos() {
        let m = estimate_mos(Some("PCMU"), Some(0), 0.0, Some(10.0), Some(2.0)).unwrap();
        assert!(m > 4.0, "expected high MOS, got {m}");
    }

    #[test]
    fn high_loss_lowers_mos() {
        let good = estimate_mos(Some("PCMU"), Some(0), 0.0, Some(20.0), None).unwrap();
        let bad = estimate_mos(Some("PCMU"), Some(0), 20.0, Some(20.0), None).unwrap();
        assert!(bad < good, "{bad} should be < {good}");
    }

    #[test]
    fn high_delay_lowers_mos() {
        let low = estimate_mos(Some("PCMU"), Some(0), 0.0, Some(50.0), None).unwrap();
        let high = estimate_mos(Some("PCMU"), Some(0), 0.0, Some(300.0), None).unwrap();
        assert!(high < low, "{high} should be < {low}");
    }
}