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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

"""
Jitter measurement and analysis utilities.
Implements multiple jitter metrics for comprehensive analysis.
"""
 
import statistics
from typing import List, Tuple
import subprocess
import re
 
def calculate_jitter_metrics(delays_ms: List[float]) -> dict:
    """
    Calculate comprehensive jitter metrics from a series of delay measurements.
    
    Args:
        delays_ms: List of delay measurements in milliseconds
    
    Returns:
        Dictionary with various jitter metrics
    """
    if len(delays_ms) < 2:
        return {"error": "Need at least 2 samples"}
    
    # Calculate packet delay variations (PDV)
    pdv = []
    for i in range(1, len(delays_ms)):
        pdv.append(abs(delays_ms[i] - delays_ms[i-1]))
    
    # RFC 3550 style smoothed jitter
    rfc_jitter = 0.0
    for i in range(1, len(delays_ms)):
        diff = abs(delays_ms[i] - delays_ms[i-1])
        rfc_jitter = rfc_jitter + (diff - rfc_jitter) / 16.0
    
    # Statistical analysis
    mean_delay = statistics.mean(delays_ms)
    
    return {
        # Basic delay statistics
        "delay_mean_ms": round(mean_delay, 3),
        "delay_min_ms": round(min(delays_ms), 3),
        "delay_max_ms": round(max(delays_ms), 3),
        "delay_range_ms": round(max(delays_ms) - min(delays_ms), 3),
        
        # Jitter metrics
        "pdv_mean_ms": round(statistics.mean(pdv), 3),
        "pdv_max_ms": round(max(pdv), 3),
        "jitter_stddev_ms": round(statistics.stdev(delays_ms), 3),
        "jitter_rfc3550_ms": round(rfc_jitter, 3),
        
        # Percentiles (for tail jitter)
        "delay_p95_ms": round(sorted(delays_ms)[int(len(delays_ms) * 0.95)], 3),
        "delay_p99_ms": round(sorted(delays_ms)[int(len(delays_ms) * 0.99)], 3),
        
        # Coefficient of variation (normalized jitter)
        "cv_percent": round((statistics.stdev(delays_ms) / mean_delay) * 100, 1),
        
        "sample_count": len(delays_ms),
    }
 
def analyze_jitter_pattern(delays_ms: List[float], window_size: int = 10) -> dict:
    """
    Analyze jitter patterns over time to identify trends.
    
    Args:
        delays_ms: Delay measurements
        window_size: Rolling window size for analysis
    """
    if len(delays_ms) < window_size:
        return {"error": f"Need at least {window_size} samples"}
    
    # Calculate rolling jitter
    rolling_jitter = []
    for i in range(window_size, len(delays_ms)):
        window = delays_ms[i-window_size:i]
        rolling_jitter.append(statistics.stdev(window))
    
    # Detect jitter spikes (> 2x average)
    avg_jitter = statistics.mean(rolling_jitter)
    spikes = [i + window_size for i, j in enumerate(rolling_jitter) if j > 2 * avg_jitter]
    
    return {
        "rolling_jitter_mean": round(avg_jitter, 3),
        "rolling_jitter_max": round(max(rolling_jitter), 3),
        "spike_count": len(spikes),
        "spike_indices": spikes[:10],  # First 10 spikes
        "stability_score": round(100 * (1 - statistics.stdev(rolling_jitter) / avg_jitter), 1),
    }
 
# Example: Parse ping output and calculate jitter
def parse_ping_output(ping_output: str) -> List[float]:
    """Extract RTT values from ping command output."""
    pattern = r'time[=<](d+.?d*)s*ms'
    matches = re.findall(pattern, ping_output)
    return [float(m) for m in matches]
 
# Demonstration with synthetic data
import random
random.seed(42)
 
# Scenario 1: Low jitter (good network)
low_jitter = [50 + random.gauss(0, 2) for _ in range(100)]
 
# Scenario 2: High jitter (congested network)
high_jitter = [50 + random.gauss(0, 15) for _ in range(100)]
 
# Scenario 3: Bufferbloat (occasional massive spikes)
bufferbloat = [50 + random.gauss(0, 3) if random.random() > 0.1 
               else 50 + random.uniform(100, 300) for _ in range(100)]
 
print("=== Jitter Analysis Examples ===\n")
 
for name, data in [("Low Jitter", low_jitter), 
                    ("High Jitter", high_jitter),
                    ("Bufferbloat", bufferbloat)]:
    metrics = calculate_jitter_metrics(data)
    pattern = analyze_jitter_pattern(data)
    
    print(f"--- {name} ---")
    print(f"  Delay: {metrics['delay_mean_ms']}ms mean, "
          f"{metrics['delay_min_ms']}-{metrics['delay_max_ms']}ms range")
    print(f"  Jitter (StdDev): {metrics['jitter_stddev_ms']}ms")
    print(f"  Jitter (RFC3550): {metrics['jitter_rfc3550_ms']}ms")
    print(f"  Coefficient of Variation: {metrics['cv_percent']}%")
    print(f"  Stability Score: {pattern['stability_score']}")
    print(f"  Spikes Detected: {pattern['spike_count']}")
    print()

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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

"""
Adaptive jitter buffer implementation demonstrating key concepts.
Used in VoIP, streaming, and other real-time applications.
"""
 
from collections import deque
from dataclasses import dataclass
from typing import Optional, List
import time
 
@dataclass
class Packet:
    sequence: int
    timestamp_ms: float  # Original send timestamp
    arrival_ms: float    # Actual arrival timestamp
    payload: bytes
 
class AdaptiveJitterBuffer:
    """
    Adaptive jitter buffer that adjusts size based on observed jitter.
    
    The buffer trades latency for consistency:
    - Larger buffer = more latency, smoother playback
    - Smaller buffer = less latency, risk of underruns
    """
    
    def __init__(
        self,
        target_buffer_ms: float = 60.0,
        min_buffer_ms: float = 20.0,
        max_buffer_ms: float = 200.0,
        adaptation_rate: float = 0.1
    ):
        self.target_buffer_ms = target_buffer_ms
        self.min_buffer_ms = min_buffer_ms
        self.max_buffer_ms = max_buffer_ms
        self.adaptation_rate = adaptation_rate
        
        self.buffer: deque[Packet] = deque()
        self.last_played_sequence: int = -1
        self.jitter_estimate: float = 0.0
        self.last_delay: Optional[float] = None
        
        # Statistics
        self.packets_received: int = 0
        self.packets_dropped_late: int = 0
        self.packets_dropped_old: int = 0
        self.underruns: int = 0
        
    def _update_jitter_estimate(self, delay_ms: float) -> None:
        """Update RFC 3550-style smoothed jitter estimate."""
        if self.last_delay is not None:
            diff = abs(delay_ms - self.last_delay)
            # Exponential smoothing: J = J + (|D| - J) / 16
            self.jitter_estimate += (diff - self.jitter_estimate) / 16.0
        self.last_delay = delay_ms
        
    def _adapt_buffer_size(self) -> None:
        """
        Adjust target buffer based on observed jitter.
        
        Strategy:
        - If jitter is low, gradually reduce buffer (lower latency)
        - If jitter is high, increase buffer (better quality)
        """
        # Target buffer = 3x jitter estimate (covers 99.7% of variation)
        ideal_buffer = max(self.jitter_estimate * 3, self.min_buffer_ms)
        
        # Smooth adaptation to avoid oscillation
        diff = ideal_buffer - self.target_buffer_ms
        self.target_buffer_ms += diff * self.adaptation_rate
        
        # Clamp to bounds
        self.target_buffer_ms = max(self.min_buffer_ms, 
                                     min(self.max_buffer_ms, 
                                         self.target_buffer_ms))
        
    def receive_packet(self, packet: Packet) -> None:
        """Add a packet to the buffer."""
        self.packets_received += 1
        
        # Calculate one-way delay (assuming synchronized clocks)
        delay_ms = packet.arrival_ms - packet.timestamp_ms
        self._update_jitter_estimate(delay_ms)
        self._adapt_buffer_size()
        
        # Drop packets that are too old (already missed their playback time)
        if packet.sequence <= self.last_played_sequence:
            self.packets_dropped_old += 1
            return
            
        # Insert packet in sequence order
        inserted = False
        for i, buffered in enumerate(self.buffer):
            if packet.sequence < buffered.sequence:
                self.buffer.insert(i, packet)
                inserted = True
                break
        if not inserted:
            self.buffer.append(packet)
            
    def get_next_packet(self, current_time_ms: float) -> Optional[Packet]:
        """
        Get the next packet for playback if available and on time.
        
        Returns None if buffer is empty (underrun) or no packet is ready.
        """
        if not self.buffer:
            self.underruns += 1
            return None
            
        # Check if first packet is ready for playback
        next_packet = self.buffer[0]
        playout_time = next_packet.timestamp_ms + self.target_buffer_ms
        
        if current_time_ms >= playout_time:
            # Packet is due or overdue
            packet = self.buffer.popleft()
            self.last_played_sequence = packet.sequence
            return packet
            
        return None  # Not yet time to play
        
    def get_stats(self) -> dict:
        """Return buffer statistics."""
        return {
            "target_buffer_ms": round(self.target_buffer_ms, 1),
            "current_jitter_ms": round(self.jitter_estimate, 1),
            "packets_buffered": len(self.buffer),
            "packets_received": self.packets_received,
            "packets_dropped_late": self.packets_dropped_late,
            "packets_dropped_old": self.packets_dropped_old,
            "underruns": self.underruns,
            "quality_score": round(100 * (1 - (self.packets_dropped_late + 
                            self.underruns) / max(self.packets_received, 1)), 1)
        }
 
# Demonstration
print("=== Adaptive Jitter Buffer Demo ===\n")
 
buffer = AdaptiveJitterBuffer(target_buffer_ms=60)
 
# Simulate receiving packets with varying delays
import random
random.seed(42)
 
base_time = 0
for seq in range(100):
    send_time = base_time + seq * 20  # 20ms packet intervals
    # Simulate jitter: normal delay + random variation
    jitter = random.gauss(0, 10)  # 10ms standard deviation
    if random.random() < 0.05:  # 5% chance of spike
        jitter += random.uniform(50, 100)
    arrival_time = send_time + 50 + jitter  # 50ms base delay
    
    packet = Packet(
        sequence=seq,
        timestamp_ms=send_time,
        arrival_ms=arrival_time,
        payload=b'audio_frame'
    )
    buffer.receive_packet(packet)
 
stats = buffer.get_stats()
print("After receiving 100 packets:")
print(f"  Target buffer: {stats['target_buffer_ms']}ms")
print(f"  Measured jitter: {stats['current_jitter_ms']}ms")
print(f"  Quality score: {stats['quality_score']}%")