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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270

"""
NAV State Machine: Complete Specification
Based on IEEE 802.11-2020 standard behavior
"""
 
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Optional
import time
 
class NAVUpdateSource(Enum):
    """Source that triggered the NAV update."""
    NONE = auto()
    RTS = auto()
    CTS = auto()
    DATA = auto()
    ACK = auto()
    CF_END = auto()
    BEACON = auto()
    OTHER_CONTROL = auto()
    OTHER_DATA = auto()
    OTHER_MANAGEMENT = auto()
 
@dataclass
class NAVState:
    """
    Complete NAV state for an 802.11 station.
    
    IEEE 802.11 Spec References:
    - 10.3.2.4 NAV setting rules
    - 10.3.2.5 NAV reset rules
    - 10.3.2.6 NAV assertion during PCF
    """
    
    # Current NAV value in microseconds
    nav_value_us: int = 0
    
    # Timestamp when NAV was last updated (for countdown calculation)
    nav_set_time_us: int = 0
    
    # Source of the current NAV value
    update_source: NAVUpdateSource = NAVUpdateSource.NONE
    
    # Address of station that triggered NAV (TA from received frame)
    nav_source_address: Optional[str] = None
    
    # Flag for CF (Contention Free) period NAV
    cf_nav_active: bool = False
    
    def get_current_tsf(self) -> int:
        """Get current TSF timestamp in microseconds."""
        # In real implementation, this would read the hardware TSF counter
        return int(time.time() * 1_000_000)
    
    def get_remaining_nav(self) -> int:
        """Calculate remaining NAV time in microseconds."""
        if self.nav_value_us == 0:
            return 0
        
        elapsed = self.get_current_tsf() - self.nav_set_time_us
        remaining = self.nav_value_us - elapsed
        
        return max(0, remaining)
    
    def is_nav_zero(self) -> bool:
        """Check if NAV has expired (virtual channel idle)."""
        return self.get_remaining_nav() == 0
    
    def update_nav(self, duration_us: int, source: NAVUpdateSource,
                   source_address: Optional[str] = None) -> bool:
        """
        Update NAV based on received Duration field.
        
        IEEE 802.11 Rule: NAV is only updated if the new value would
        result in a NAV expiration time later than the current value.
        
        Returns: True if NAV was updated, False if ignored
        """
        if duration_us <= 0:
            return False
        
        current_remaining = self.get_remaining_nav()
        
        if duration_us > current_remaining:
            self.nav_value_us = duration_us
            self.nav_set_time_us = self.get_current_tsf()
            self.update_source = source
            self.nav_source_address = source_address
            return True
        
        return False
    
    def reset_nav(self, source: NAVUpdateSource = NAVUpdateSource.CF_END) -> None:
        """
        Reset NAV to zero.
        
        Triggered by:
        - CF-End frame reception
        - NAV timeout (automatic)
        - Certain error recovery procedures
        """
        self.nav_value_us = 0
        self.nav_set_time_us = 0
        self.update_source = source
        self.nav_source_address = None
        self.cf_nav_active = False
    
    def set_cf_nav(self, duration_us: int) -> None:
        """Set Contention-Free period NAV (from beacon DTIM)."""
        self.nav_value_us = duration_us
        self.nav_set_time_us = self.get_current_tsf()
        self.cf_nav_active = True
        self.update_source = NAVUpdateSource.BEACON
    
    def get_status(self) -> dict:
        """Get current NAV status for monitoring."""
        return {
            'remaining_us': self.get_remaining_nav(),
            'remaining_ms': self.get_remaining_nav() / 1000,
            'is_zero': self.is_nav_zero(),
            'source': self.update_source.name,
            'source_address': self.nav_source_address,
            'cf_active': self.cf_nav_active
        }
 
 
class ChannelAccessController:
    """
    Combines physical and virtual carrier sensing for channel access decisions.
    """
    
    def __init__(self, station_address: str):
        self.address = station_address
        self.nav = NAVState()
        self.phy_busy = False
        self.cca_threshold_dbm = -82  # Typical CCA threshold
    
    def receive_frame(self, frame_type: str, duration_us: int, 
                      ta: str, ra: str, fcs_valid: bool) -> None:
        """
        Process a received frame for NAV update.
        
        NAV is only updated from frames with valid FCS.
        Special handling for frames addressed to this station.
        """
        
        if not fcs_valid:
            # Invalid frames do not update NAV
            return
        
        # Determine NAV update source type
        source_map = {
            'RTS': NAVUpdateSource.RTS,
            'CTS': NAVUpdateSource.CTS,
            'ACK': NAVUpdateSource.ACK,
            'DATA': NAVUpdateSource.DATA,
            'CF-END': NAVUpdateSource.CF_END,
            'BEACON': NAVUpdateSource.BEACON,
        }
        source = source_map.get(frame_type, NAVUpdateSource.OTHER_DATA)
        
        # Special case: CF-End resets NAV
        if frame_type == 'CF-END':
            self.nav.reset_nav(NAVUpdateSource.CF_END)
            return
        
        # Special case: Frames addressed to us
        if ra == self.address:
            # We are the intended recipient
            # For CTS/ACK to us: we should not update NAV from our own exchange
            if frame_type in ['CTS', 'ACK']:
                return
        
        # Normal case: Update NAV if new duration is longer
        self.nav.update_nav(duration_us, source, ta)
    
    def update_phy_carrier_sense(self, energy_dbm: float) -> None:
        """Update physical carrier sense state based on RF energy."""
        self.phy_busy = energy_dbm > self.cca_threshold_dbm
    
    def is_channel_idle(self) -> bool:
        """
        Determine if channel is idle for transmission.
        
        Channel is idle only if BOTH:
        1. Physical carrier sense indicates idle (no RF energy above threshold)
        2. NAV is zero (no virtual reservation)
        """
        phy_idle = not self.phy_busy
        nav_idle = self.nav.is_nav_zero()
        
        return phy_idle and nav_idle
    
    def can_start_backoff(self) -> bool:
        """
        Determine if station can enter/continue backoff procedure.
        
        Backoff counter only decrements when channel is idle.
        """
        return self.is_channel_idle()
    
    def get_channel_status(self) -> dict:
        """Get complete channel status for monitoring."""
        return {
            'phy_busy': self.phy_busy,
            'nav_status': self.nav.get_status(),
            'channel_idle': self.is_channel_idle(),
            'can_transmit': self.is_channel_idle()
        }
 
 
# Demonstration
def demonstrate_nav_behavior():
    """Show NAV behavior through a complete RTS/CTS/Data/ACK exchange."""
    
    print("=== NAV Behavior Demonstration ===")
    print()
    
    # Create three stations
    controller_sender = ChannelAccessController("AA:AA:AA:AA:AA:AA")
    controller_hidden = ChannelAccessController("BB:BB:BB:BB:BB:BB")
    controller_visible = ChannelAccessController("CC:CC:CC:CC:CC:CC")
    
    # Initial state
    print("--- Initial State ---")
    print(f"Hidden station NAV: {controller_hidden.nav.get_remaining_nav()} μs")
    print(f"Visible station NAV: {controller_visible.nav.get_remaining_nav()} μs")
    print()
    
    # Simulate RTS reception at visible station (hidden doesn't hear it)
    print("--- RTS Transmitted (only visible station hears) ---")
    controller_visible.receive_frame(
        frame_type='RTS',
        duration_us=2000,  # Duration until end of ACK
        ta="AA:AA:AA:AA:AA:AA",
        ra="DD:DD:DD:DD:DD:DD",  # AP
        fcs_valid=True
    )
    print(f"Hidden station NAV: {controller_hidden.nav.get_remaining_nav()} μs (did not hear RTS)")
    print(f"Visible station NAV: {controller_visible.nav.get_remaining_nav()} μs")
    print()
    
    # Simulate CTS reception (both stations hear from AP)
    print("--- CTS from AP (both stations hear) ---")
    cts_duration = 2000 - 44 - 16  # RTS duration minus CTS time minus SIFS
    
    controller_hidden.receive_frame(
        frame_type='CTS',
        duration_us=cts_duration,
        ta="DD:DD:DD:DD:DD:DD",  # AP
        ra="AA:AA:AA:AA:AA:AA",  # Original sender
        fcs_valid=True
    )
    controller_visible.receive_frame(
        frame_type='CTS',
        duration_us=cts_duration,
        ta="DD:DD:DD:DD:DD:DD",
        ra="AA:AA:AA:AA:AA:AA",
        fcs_valid=True
    )
    
    print(f"Hidden station NAV: {controller_hidden.nav.get_remaining_nav()} μs (updated from CTS)")
    print(f"Visible station NAV: {controller_visible.nav.get_remaining_nav()} μs (kept from RTS, higher)")
    print()
    
    print("--- Can Stations Transmit? ---")
    print(f"Hidden station can transmit: {controller_hidden.is_channel_idle()}")
    print(f"Visible station can transmit: {controller_visible.is_channel_idle()}")
 
demonstrate_nav_behavior()

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

"""
Duration Field Calculation for Different Frame Types
802.11 MAC layer timing analysis
"""
 
# Timing constants (802.11g/n with OFDM)
SIFS_US = 16          # Short Interframe Space
SLOT_TIME_US = 9      # Slot time
CTS_TIME_US = 44      # CTS transmission time at 6 Mbps
ACK_TIME_US = 44      # ACK transmission time at 6 Mbps
RTS_TIME_US = 52      # RTS transmission time at 6 Mbps
 
def calculate_ofdm_txtime(bytes: int, rate_mbps: float) -> float:
    """Calculate OFDM transmission time in microseconds."""
    import math
    # PLCP preamble and header: 20 μs
    # Symbol time: 4 μs
    # Bits per symbol = rate_mbps × 4
    preamble = 20
    bits = 16 + (bytes * 8) + 6  # Service + MPDU + Tail
    bits_per_symbol = rate_mbps * 4
    n_symbols = math.ceil(bits / bits_per_symbol)
    return preamble + (n_symbols * 4)
 
def calculate_rts_duration(data_bytes: int, data_rate_mbps: float) -> int:
    """
    Calculate Duration field value for RTS frame.
    
    Duration = SIFS + CTS + SIFS + DATA + SIFS + ACK
    """
    data_time = calculate_ofdm_txtime(data_bytes + 38, data_rate_mbps)  # +38 for MAC header/FCS
    
    duration = (SIFS_US + CTS_TIME_US + 
                SIFS_US + data_time + 
                SIFS_US + ACK_TIME_US)
    
    return int(duration)
 
def calculate_cts_duration(rts_duration: int) -> int:
    """
    Calculate Duration field value for CTS frame.
    
    Duration = RTS_Duration - SIFS - CTS_time
    """
    return max(0, rts_duration - SIFS_US - CTS_TIME_US)
 
def calculate_data_duration() -> int:
    """
    Calculate Duration field value for unicast Data frame.
    
    Duration = SIFS + ACK
    """
    return SIFS_US + ACK_TIME_US
 
def display_duration_chain(data_bytes: int, data_rate_mbps: float):
    """Display the Duration field values across an RTS/CTS/Data/ACK exchange."""
    
    rts_duration = calculate_rts_duration(data_bytes, data_rate_mbps)
    cts_duration = calculate_cts_duration(rts_duration)
    data_duration = calculate_data_duration()
    ack_duration = 0  # ACK has Duration = 0
    
    print(f"=== Duration Field Chain ({data_bytes} bytes @ {data_rate_mbps} Mbps) ===")
    print()
    print("Frame        | Duration (μs) | Protects")
    print("-" * 50)
    print(f"RTS          | {rts_duration:>13} | CTS + Data + ACK")
    print(f"CTS          | {cts_duration:>13} | Data + ACK")
    print(f"Data         | {data_duration:>13} | ACK only")
    print(f"ACK          | {ack_duration:>13} | Nothing (end)")
    print()
    print("NAV decreases through the exchange:")
    print(f"  After RTS heard: NAV = {rts_duration} μs")
    print(f"  After CTS heard: NAV = max(current, {cts_duration}) μs")
    print(f"  After Data heard: NAV = max(current, {data_duration}) μs")
    print(f"  After ACK heard: NAV → 0 (expired or stays current)")
 
# Examples
print("Example 1: Small Frame (256 bytes @ 54 Mbps)")
display_duration_chain(256, 54)
print()
 
print("Example 2: Large Frame (1500 bytes @ 54 Mbps)")
display_duration_chain(1500, 54)
print()
 
print("Example 3: Maximum Frame (2304 bytes @ 6 Mbps)")
display_duration_chain(2304, 6)

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

"""
NAV Reset Scenarios and Edge Cases
"""
 
from enum import Enum, auto
from dataclasses import dataclass
from typing import Optional
 
class NAVResetScenario(Enum):
    """Different scenarios that affect NAV state."""
    NORMAL_EXPIRATION = auto()      # NAV counts down to zero
    CF_END_RECEIVED = auto()        # Explicit reset from CF-End
    EXCHANGE_FAILURE = auto()       # RTS sent, no CTS received
    MANAGEMENT_RESET = auto()       # Disassociation/deauth
    CHANNEL_SWITCH = auto()         # Moving to different channel
 
@dataclass
class NAVEvent:
    """Represents a NAV-related event."""
    timestamp_us: int
    event_type: str
    nav_before: int
    nav_after: int
    reason: str
 
class NAVResilienceAnalyzer:
    """Analyze NAV behavior in various scenarios."""
    
    def __init__(self):
        self.events: list[NAVEvent] = []
        self.current_nav = 0
        self.nav_set_time = 0
    
    def simulate_normal_exchange(self, base_time: int):
        """Simulate successful RTS/CTS/Data/ACK exchange."""
        
        # RTS sets NAV to 2000 μs
        self.events.append(NAVEvent(
            timestamp_us=base_time,
            event_type="RTS_RECEIVED",
            nav_before=self.current_nav,
            nav_after=2000,
            reason="Duration field = 2000"
        ))
        self.current_nav = 2000
        self.nav_set_time = base_time
        
        # CTS at +60 μs (would set NAV = 1900, but current is higher)
        at_cts = base_time + 60
        remaining = self.current_nav - 60  # 1940
        cts_duration = 1900  # From CTS
        
        self.events.append(NAVEvent(
            timestamp_us=at_cts,
            event_type="CTS_RECEIVED",
            nav_before=remaining,
            nav_after=remaining,  # Not updated (1940 > 1900)
            reason=f"CTS Duration {cts_duration} < remaining {remaining}; no update"
        ))
        
        # NAV expires naturally
        expiration_time = base_time + 2000
        self.events.append(NAVEvent(
            timestamp_us=expiration_time,
            event_type="NAV_EXPIRED",
            nav_before=0,
            nav_after=0,
            reason="Natural countdown to zero"
        ))
        
        return self.events
    
    def simulate_rts_failure(self, base_time: int):
        """Simulate RTS without CTS response (failed exchange)."""
        
        # RTS sets NAV to 2000 μs
        self.events.append(NAVEvent(
            timestamp_us=base_time,
            event_type="RTS_RECEIVED",
            nav_before=self.current_nav,
            nav_after=2000,
            reason="Duration field = 2000"
        ))
        self.current_nav = 2000
        self.nav_set_time = base_time
        
        # CTS never arrives
        self.events.append(NAVEvent(
            timestamp_us=base_time + 200,
            event_type="CTS_TIMEOUT",
            nav_before=1800,  # 2000 - 200
            nav_after=1800,  # No change from timeout
            reason="Sender notices timeout; NAV in other stations unchanged"
        ))
        
        # NAV still expires at original time
        # This is 'wasted' but ensures protection even for partial exchanges
        expiration_time = base_time + 2000
        self.events.append(NAVEvent(
            timestamp_us=expiration_time,
            event_type="NAV_EXPIRED",
            nav_before=0,
            nav_after=0,
            reason="Natural expiration despite no data transfer"
        ))
        
        return self.events
    
    def simulate_cf_end_reset(self, base_time: int):
        """Simulate CF period with CF-End reset."""
        
        # Beacon sets NAV for CFP (5000 μs)
        self.events.append(NAVEvent(
            timestamp_us=base_time,
            event_type="BEACON_CF_PARAMS",
            nav_before=self.current_nav,
            nav_after=5000,
            reason="CFP Duration from beacon = 5000"
        ))
        self.current_nav = 5000
        
        # CFP ends early, CF-End sent at +3000 μs
        cf_end_time = base_time + 3000
        self.events.append(NAVEvent(
            timestamp_us=cf_end_time,
            event_type="CF_END_RECEIVED",
            nav_before=2000,  # Would be 5000 - 3000
            nav_after=0,  # Explicit reset!
            reason="CF-End explicitly resets NAV to 0"
        ))
        self.current_nav = 0
        
        return self.events
    
    def display_events(self):
        """Display all recorded events."""
        print("=== NAV Event Log ===")
        print()
        for event in self.events:
            print(f"T={event.timestamp_us:>8} μs | {event.event_type:20} | "
                  f"NAV: {event.nav_before:>5} → {event.nav_after:>5} | {event.reason}")
 
# Demonstrate scenarios
analyzer1 = NAVResilienceAnalyzer()
analyzer1.simulate_normal_exchange(0)
analyzer1.display_events()
 
print()
 
analyzer2 = NAVResilienceAnalyzer()
analyzer2.simulate_cf_end_reset(10000)
analyzer2.display_events()

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

"""
NAV Analysis Script for Wireshark/tshark Captures
Analyzes Duration fields and NAV behavior
"""
 
import subprocess
import json
from collections import defaultdict
from datetime import datetime
 
def analyze_nav_from_capture(pcap_file: str) -> dict:
    """
    Analyze NAV-related fields from a packet capture.
    
    Requires tshark installed and capture taken in monitor mode.
    """
    
    # tshark command to extract Duration and frame info
    cmd = [
        'tshark', '-r', pcap_file,
        '-T', 'json',
        '-e', 'frame.time_relative',
        '-e', 'wlan.fc.type',
        '-e', 'wlan.fc.subtype',
        '-e', 'wlan.duration',
        '-e', 'wlan.ta',
        '-e', 'wlan.ra',
        '-Y', 'wlan'  # Filter to WLAN frames only
    ]
    
    # This would run tshark - for demonstration, we'll simulate data
    # result = subprocess.run(cmd, capture_output=True, text=True)
    # frames = json.loads(result.stdout)
    
    # Simulated frame data for demonstration
    frames = [
        {'time': 0.0, 'type': 1, 'subtype': 11, 'duration': 500, 'ta': 'AA:AA:AA:AA:AA:AA'},
        {'time': 0.00005, 'type': 1, 'subtype': 12, 'duration': 440, 'ta': 'BB:BB:BB:BB:BB:BB'},
        {'time': 0.0001, 'type': 2, 'subtype': 0, 'duration': 60, 'ta': 'AA:AA:AA:AA:AA:AA'},
        {'time': 0.0005, 'type': 1, 'subtype': 13, 'duration': 0, 'ta': 'BB:BB:BB:BB:BB:BB'},
        {'time': 0.001, 'type': 1, 'subtype': 11, 'duration': 1200, 'ta': 'CC:CC:CC:CC:CC:CC'},
        {'time': 0.0015, 'type': 1, 'subtype': 11, 'duration': 32767, 'ta': 'DD:DD:DD:DD:DD:DD'},  # Suspicious!
    ]
    
    analysis = {
        'total_frames': len(frames),
        'nav_updates': 0,
        'max_duration_seen': 0,
        'duration_histogram': defaultdict(int),
        'suspicious_frames': [],
        'frame_type_duration': defaultdict(list)
    }
    
    frame_type_names = {
        (1, 11): 'RTS',
        (1, 12): 'CTS',
        (1, 13): 'ACK',
        (2, 0): 'Data',
        (0, 8): 'Beacon'
    }
    
    for frame in frames:
        duration = frame.get('duration', 0)
        ftype = frame.get('type', 0)
        subtype = frame.get('subtype', 0)
        frame_name = frame_type_names.get((ftype, subtype), f'Type{ftype}/Sub{subtype}')
        
        if duration > 0:
            analysis['nav_updates'] += 1
            analysis['max_duration_seen'] = max(analysis['max_duration_seen'], duration)
            
            # Histogram in 100μs buckets
            bucket = (duration // 100) * 100
            analysis['duration_histogram'][bucket] += 1
            
            analysis['frame_type_duration'][frame_name].append(duration)
            
            # Flag suspicious Duration values (> 10ms unusual for normal operation)
            if duration > 10000:
                analysis['suspicious_frames'].append({
                    'time': frame.get('time'),
                    'type': frame_name,
                    'duration': duration,
                    'ta': frame.get('ta'),
                    'warning': 'Unusually high Duration value'
                })
    
    return analysis
 
def display_nav_analysis(analysis: dict):
    """Display NAV analysis results."""
    
    print("=== NAV Analysis Report ===")
    print()
    print(f"Total frames analyzed: {analysis['total_frames']}")
    print(f"Frames with NAV updates: {analysis['nav_updates']}")
    print(f"Maximum Duration seen: {analysis['max_duration_seen']} μs")
    print()
    
    print("--- Duration by Frame Type ---")
    for frame_type, durations in analysis['frame_type_duration'].items():
        if durations:
            avg = sum(durations) / len(durations)
            print(f"  {frame_type:10}: count={len(durations):4}, "
                  f"avg={avg:7.1f} μs, max={max(durations):6} μs")
    print()
    
    if analysis['suspicious_frames']:
        print("!!! SUSPICIOUS FRAMES DETECTED !!!")
        for sf in analysis['suspicious_frames']:
            print(f"  T={sf['time']:.6f}s: {sf['type']} from {sf['ta']}, "
                  f"Duration={sf['duration']} μs")
            print(f"    Warning: {sf['warning']}")
        print()
    else:
        print("No suspicious frames detected.")
    
    print("--- Duration Histogram (100μs buckets) ---")
    for bucket in sorted(analysis['duration_histogram'].keys()):
        count = analysis['duration_histogram'][bucket]
        bar = '█' * min(count, 50)
        print(f"  {bucket:>6}-{bucket+99:>5} μs: {count:>4} {bar}")
 
# Run analysis
analysis = analyze_nav_from_capture("sample_capture.pcap")
display_nav_analysis(analysis)