yaro/wifi_test
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Enhance SSID tracking in query_metrics by introducing window-based data structures for improved signal statistics and debugging output

Browse source
This commit is contained in:
Yaro Kasear 2025-04-29 13:34:03 -05:00
parent 21fcc807ad
commit 6c231a0830
1 changed files with 25 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

29
enrichment/indexed_capture.py
View file

@ -146,6 +146,9 @@ class IndexedCapture:
window_clients = defaultdict(set)
window_aps = defaultdict(set)
window_signals = defaultdict(list)
window_ssid_to_bssids = defaultdict(set)
window_ssid_signals = defaultdict(list)
for packet in packets:
try:
@ -177,6 +180,17 @@ class IndexedCapture:
if mac and mac != 'ff:ff:ff:ff:ff:ff':
window_clients[channel].add(mac)
# Only beacon/probe have SSID
ssid = self.bssid_to_ssid.get(bssid)
if ssid:
window_ssid_to_bssids[ssid].add(bssid)
signal = getattr(radio, 'dbm_antsignal', None)
if signal:
window_ssid_signals[ssid].append(int(signal))
if channel == ap_channel:
print(f"[DEBUG] [AP] Adding bssid={bssid} to window_aps[{channel}]")
# Track signals
signal = getattr(radio, 'dbm_antsignal', None)
if signal:
@ -195,10 +209,10 @@ class IndexedCapture:
unlinked_devices = self._count_unlinked_devices(packets, ap_channel)
our_ssid = self.bssid_to_ssid.get(ap_bssid)
num_bssids = len(self.ssid_to_bssids.get(our_ssid, [])) if our_ssid else 0
avg_ssid_signal = mean(self.ssid_signals.get(our_ssid, [])) if our_ssid else 0
max_ssid_signal = max(self.ssid_signals.get(our_ssid, [])) if our_ssid else 0
num_channels_ssid = len(self.ssid_to_bssids.get(our_ssid, [])) if our_ssid else 0
num_bssids = len(window_ssid_to_bssids.get(our_ssid, [])) if our_ssid else 0
avg_ssid_signal = mean(window_ssid_signals.get(our_ssid, [])) if our_ssid else 0
max_ssid_signal = max(window_ssid_signals.get(our_ssid, [])) if our_ssid else 0
num_channels_ssid = len({self._channel_of_bssid(b) for b in window_ssid_to_bssids.get(our_ssid, [])}) if our_ssid else 0
packet_count = len(packets)
return (
@ -285,3 +299,10 @@ class IndexedCapture:
if ssid in self.cisco_ssid_clients:
return max(self.cisco_ssid_clients[ssid])
return 0
def _channel_of_bssid(self, bssid):
for channel, bssids in self.channel_to_aps.items():
if bssid in bssids:
return channel
return None