berkus
/
wifi-densepose
mirror of https://github.com/ruvnet/wifi-densepose.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: cross-platform WiFi collector factory (ADR-049) (#173) 

feat: cross-platform WiFi collector factory (ADR-049)
This commit is contained in:
rUv 2026-03-06 15:10:26 -05:00 committed by GitHub
parent 45c15b77a5 75d4685d25
commit 7659b0bbe2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 366 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
docs/adr/ADR-049-cross-platform-wifi-interface-detection.md Normal file
View File

@ -0,0 +1,122 @@
# ADR-049: Cross-Platform WiFi Interface Detection and Graceful Degradation
| Field | Value |
|-------|-------|
| Status | Proposed |
| Date | 2026-03-06 |
| Deciders | ruv |
| Depends on | ADR-013 (Feature-Level Sensing), ADR-025 (macOS CoreWLAN) |
| Issue | [#148](https://github.com/ruvnet/wifi-densepose/issues/148) |
## Context
Users report `RuntimeError: Cannot read /proc/net/wireless` when running WiFi DensePose in environments where the Linux wireless proc filesystem is unavailable:
- **Docker containers** on macOS/Windows (Linux kernel detected, but no wireless subsystem)
- **WSL2** without USB WiFi passthrough
- **Headless Linux servers** without WiFi hardware
- **Embedded Linux** boards without wireless-extensions support
The current architecture has two layers of defense:
1. **`ws_server.py`** (line 345-355) checks `os.path.exists("/proc/net/wireless")` before instantiating `LinuxWifiCollector` and falls back to `SimulatedCollector` if missing.
2. **`rssi_collector.py`** `LinuxWifiCollector._validate_interface()` (line 178-196) raises a hard `RuntimeError` if `/proc/net/wireless` is missing or the interface isn't listed.
However, there are gaps:
- **Direct usage**: Any code that instantiates `LinuxWifiCollector` directly (outside `ws_server.py`) hits the unguarded `RuntimeError` with no fallback.
- **Error message**: The RuntimeError message tells users to "use SimulatedCollector instead" but doesn't explain how.
- **No auto-detection**: The collector selection logic is duplicated between `ws_server.py` and `install.sh` with no shared platform-detection utility.
- **Partial `/proc/net/wireless`**: The file may exist (e.g., kernel module loaded) but contain no interfaces, producing a confusing "interface not found" error instead of a clean fallback.
## Decision
### 1. Platform-Aware Collector Factory
Introduce a `create_collector()` factory function in `rssi_collector.py` that encapsulates the platform detection and fallback chain:
```python
def create_collector(
preferred: str = "auto",
interface: str = "wlan0",
sample_rate_hz: float = 10.0,
) -> BaseCollector:
"""
Create the best available WiFi collector for the current platform.
Resolution order (when preferred="auto"):
1. ESP32 CSI (if UDP port 5005 is receiving frames)
2. Platform-native WiFi:
- Linux: LinuxWifiCollector (requires /proc/net/wireless + active interface)
- Windows: WindowsWifiCollector (netsh wlan)
- macOS: MacosWifiCollector (CoreWLAN)
3. SimulatedCollector (always available)
Raises nothing — always returns a usable collector.
"""
```
### 2. Soft Validation in LinuxWifiCollector
Replace the hard `RuntimeError` in `_validate_interface()` with a class method that returns availability status without raising:
```python
@classmethod
def is_available(cls, interface: str = "wlan0") -> tuple[bool, str]:
"""Check if Linux WiFi collection is possible. Returns (available, reason)."""
if not os.path.exists("/proc/net/wireless"):
return False, "/proc/net/wireless not found (Docker, WSL, or no wireless subsystem)"
with open("/proc/net/wireless") as f:
content = f.read()
if interface not in content:
names = cls._parse_interface_names(content)
return False, f"Interface '{interface}' not in /proc/net/wireless. Available: {names}"
return True, "ok"
```
The existing `_validate_interface()` continues to raise `RuntimeError` for direct callers who need fail-fast behavior, but `create_collector()` uses `is_available()` to probe without exceptions.
### 3. Structured Fallback Logging
When auto-detection skips a collector, log at `WARNING` level with actionable context:
```
WiFi collector: LinuxWifiCollector unavailable (/proc/net/wireless not found — likely Docker/WSL).
WiFi collector: Falling back to SimulatedCollector. For real sensing, connect ESP32 nodes via UDP:5005.
```
### 4. Consolidate Platform Detection
Remove duplicated platform-detection logic from `ws_server.py` and `install.sh`. Both should use `create_collector()` (Python) or a shared `detect_wifi_platform()` shell function.
## Consequences
### Positive
- **Zero-crash startup**: `create_collector("auto")` never raises — Docker, WSL, and headless users get `SimulatedCollector` automatically with a clear log message.
- **Single detection path**: Platform logic lives in one place (`rssi_collector.py`), reducing drift between `ws_server.py`, `install.sh`, and future entry points.
- **Better DX**: Error messages explain *why* a collector is unavailable and *what to do* (connect ESP32, install WiFi driver, etc.).
### Negative
- **SimulatedCollector may mask hardware issues**: Users with real WiFi hardware that fails detection might unknowingly run on simulated data. Mitigated by the `WARNING`-level log.
- **Breaking change for direct `LinuxWifiCollector` callers**: Code that catches `RuntimeError` from `_validate_interface()` as a signal needs to migrate to `is_available()` or `create_collector()`. This is a minor change — there are no known external consumers.
### Neutral
- `_validate_interface()` behavior is unchanged for existing direct callers — this is additive.
## Implementation Notes
1. Add `create_collector()` and `BaseCollector.is_available()` to `v1/src/sensing/rssi_collector.py`
2. Refactor `ws_server.py` `_init_collector()` to call `create_collector()`
3. Update `install.sh` `detect_wifi_hardware()` to use shared detection logic
4. Add unit tests for each platform path (mock `/proc/net/wireless` presence/absence)
5. Comment on issue #148 with the fix
## References
- Issue #148: RuntimeError: Cannot read /proc/net/wireless
- ADR-013: Feature-Level Sensing on Commodity Gear
- ADR-025: macOS CoreWLAN WiFi Sensing
- [Linux /proc/net/wireless documentation](https://www.kernel.org/doc/html/latest/networking/statistics.html)

141
v1/src/sensing/rssi_collector.py
View File

@ -12,13 +12,15 @@ from __future__ import annotations
import logging
import math
import os
import platform
import re
import subprocess
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, List, Optional, Protocol
from typing import Deque, List, Optional, Protocol, Union
import numpy as np
@ -173,27 +175,47 @@ class LinuxWifiCollector:
"""Collect a single sample right now (blocking)."""
return self._read_sample()
# -- availability check --------------------------------------------------
@classmethod
def is_available(cls, interface: str = "wlan0") -> tuple[bool, str]:
"""Check if Linux WiFi collection is possible without raising.
Returns
-------
(available, reason) : tuple[bool, str]
``available`` is True when /proc/net/wireless exists and lists
the requested interface. ``reason`` is a human-readable
explanation when unavailable.
"""
if not os.path.exists("/proc/net/wireless"):
return False, (
"/proc/net/wireless not found. "
"This environment has no Linux wireless subsystem "
"(common in Docker, WSL, or headless servers)."
)
try:
with open("/proc/net/wireless", "r") as f:
content = f.read()
except OSError as exc:
return False, f"Cannot read /proc/net/wireless: {exc}"
if interface not in content:
names = cls._parse_interface_names(content)
return False, (
f"Interface '{interface}' not listed in /proc/net/wireless. "
f"Available: {names or '(none)'}. "
f"Ensure the interface is up and associated with an AP."
)
return True, "ok"
# -- internals -----------------------------------------------------------
def _validate_interface(self) -> None:
"""Check that the interface exists on this machine."""
try:
with open("/proc/net/wireless", "r") as f:
content = f.read()
if self._interface not in content:
raise RuntimeError(
f"WiFi interface '{self._interface}' not found in "
f"/proc/net/wireless. Available interfaces may include: "
f"{self._parse_interface_names(content)}. "
f"Ensure the interface is up and associated with an AP."
)
except FileNotFoundError:
raise RuntimeError(
"Cannot read /proc/net/wireless. "
"This collector requires a Linux system with wireless-extensions support. "
"If running in a container or VM without WiFi hardware, use "
"SimulatedCollector instead."
)
available, reason = self.is_available(self._interface)
if not available:
raise RuntimeError(reason)
@staticmethod
def _parse_interface_names(proc_content: str) -> List[str]:
@ -736,3 +758,86 @@ class MacosWifiCollector:
if self._running:
logger.error("macOS WiFi utility exited unexpectedly. Collector stopped.")
self._running = False
# ---------------------------------------------------------------------------
# Collector factory (ADR-049)
# ---------------------------------------------------------------------------
CollectorType = Union[LinuxWifiCollector, WindowsWifiCollector, MacosWifiCollector, SimulatedCollector]
def create_collector(
preferred: str = "auto",
interface: str = "wlan0",
sample_rate_hz: float = 10.0,
) -> CollectorType:
"""Create the best available WiFi collector for the current platform.
Resolution order (when ``preferred="auto"``):
1. Platform-native WiFi:
- Linux: LinuxWifiCollector (requires /proc/net/wireless + active interface)
- Windows: WindowsWifiCollector (netsh wlan)
- macOS: MacosWifiCollector (CoreWLAN)
2. SimulatedCollector (always available)
This function never raises -- it always returns a usable collector.
Parameters
----------
preferred : str
``"auto"`` for platform detection, or one of ``"linux"``,
``"windows"``, ``"macos"``, ``"simulated"`` to force a specific
collector.
interface : str
WiFi interface name (Linux/Windows only).
sample_rate_hz : float
Target sampling rate.
"""
_VALID_PREFERRED = {"auto", "linux", "windows", "macos", "simulated"}
if preferred not in _VALID_PREFERRED:
logger.warning(
"WiFi collector: unknown preferred=%r (valid: %s). Falling back to auto.",
preferred, ", ".join(sorted(_VALID_PREFERRED)),
)
preferred = "auto"
system = platform.system()
if preferred == "auto":
if system == "Linux":
available, reason = LinuxWifiCollector.is_available(interface)
if available:
logger.info("WiFi collector: using LinuxWifiCollector on %s", interface)
return LinuxWifiCollector(interface=interface, sample_rate_hz=sample_rate_hz)
logger.warning("WiFi collector: LinuxWifiCollector unavailable (%s).", reason)
elif system == "Windows":
try:
win_iface = interface if interface != "wlan0" else "Wi-Fi"
collector = WindowsWifiCollector(interface=win_iface, sample_rate_hz=min(sample_rate_hz, 2.0))
collector.collect_once()
logger.info("WiFi collector: using WindowsWifiCollector on '%s'", interface)
return collector
except Exception as exc:
logger.warning("WiFi collector: WindowsWifiCollector unavailable (%s).", exc)
elif system == "Darwin":
try:
collector = MacosWifiCollector(sample_rate_hz=sample_rate_hz)
logger.info("WiFi collector: using MacosWifiCollector")
return collector
except Exception as exc:
logger.warning("WiFi collector: MacosWifiCollector unavailable (%s).", exc)
elif preferred == "linux":
return LinuxWifiCollector(interface=interface, sample_rate_hz=sample_rate_hz)
elif preferred == "windows":
return WindowsWifiCollector(interface=interface, sample_rate_hz=min(sample_rate_hz, 2.0))
elif preferred == "macos":
return MacosWifiCollector(sample_rate_hz=sample_rate_hz)
elif preferred == "simulated":
return SimulatedCollector(seed=42, sample_rate_hz=sample_rate_hz)
logger.info(
"WiFi collector: falling back to SimulatedCollector. "
"For real sensing, connect ESP32 nodes via UDP:5005 or install platform WiFi drivers."
)
return SimulatedCollector(seed=42, sample_rate_hz=sample_rate_hz)

60
v1/src/sensing/ws_server.py
View File

@ -24,7 +24,6 @@ import asyncio
import json
import logging
import math
import platform
import signal
import socket
import struct
@ -38,10 +37,6 @@ import numpy as np
# Sensing pipeline imports
from v1.src.sensing.rssi_collector import (
LinuxWifiCollector,
SimulatedCollector,
WindowsWifiCollector,
MacosWifiCollector,
WifiSample,
RingBuffer,
)
@ -321,7 +316,13 @@ class SensingWebSocketServer:
self._running = False
def _create_collector(self):
"""Auto-detect data source: ESP32 UDP > Windows WiFi > Linux WiFi > simulated."""
"""Auto-detect data source: ESP32 UDP > platform WiFi > simulated.
Uses the ``create_collector`` factory (ADR-049) for platform WiFi
detection, which never raises and logs actionable fallback messages.
"""
from .rssi_collector import create_collector
# 1. Try ESP32 UDP first
print(" Probing for ESP32 on UDP :5005 ...")
if probe_esp32_udp(ESP32_UDP_PORT, timeout=2.0):
@ -329,43 +330,18 @@ class SensingWebSocketServer:
self.source = "esp32"
return Esp32UdpCollector(port=ESP32_UDP_PORT, sample_rate_hz=10.0)
# 2. Platform-specific WiFi
system = platform.system()
if system == "Windows":
try:
collector = WindowsWifiCollector(sample_rate_hz=2.0)
collector.collect_once() # test that it works
logger.info("Using WindowsWifiCollector")
self.source = "windows_wifi"
return collector
except Exception as e:
logger.warning("Windows WiFi unavailable (%s), falling back", e)
elif system == "Linux":
# In Docker on Mac, Linux is detected but no wireless extensions exist.
# Force SimulatedCollector if /proc/net/wireless doesn't exist.
import os
if os.path.exists("/proc/net/wireless"):
try:
collector = LinuxWifiCollector(sample_rate_hz=10.0)
self.source = "linux_wifi"
return collector
except RuntimeError:
logger.warning("Linux WiFi unavailable, falling back")
else:
logger.warning("Linux detected but /proc/net/wireless missing (likely Docker). Falling back.")
elif system == "Darwin":
try:
collector = MacosWifiCollector(sample_rate_hz=10.0)
logger.info("Using MacosWifiCollector")
self.source = "macos_wifi"
return collector
except Exception as e:
logger.warning("macOS WiFi unavailable (%s), falling back", e)
# 2. Platform-specific WiFi (auto-detect with graceful fallback)
collector = create_collector(preferred="auto", sample_rate_hz=10.0)
# 3. Simulated
logger.info("Using SimulatedCollector")
self.source = "simulated"
return SimulatedCollector(seed=42, sample_rate_hz=10.0)
# Map collector class to source label
source_map = {
"LinuxWifiCollector": "linux_wifi",
"WindowsWifiCollector": "windows_wifi",
"MacosWifiCollector": "macos_wifi",
"SimulatedCollector": "simulated",
}
self.source = source_map.get(type(collector).__name__, "unknown")
return collector
def _build_message(self, features: RssiFeatures, result: SensingResult) -> str:
"""Build the JSON message to broadcast."""

103
v1/tests/unit/test_sensing.py
View File

@ -702,3 +702,106 @@ class TestBandPower:
# Band 0.21-0.39 has no power
p = _band_power(freqs, psd, 0.21, 0.39)
assert p == 0.0
# ===========================================================================
# LinuxWifiCollector.is_available() tests (ADR-049)
# ===========================================================================
from unittest.mock import patch, mock_open
from v1.src.sensing.rssi_collector import LinuxWifiCollector, create_collector
class TestLinuxWifiCollectorAvailability:
def test_unavailable_when_proc_missing(self):
"""is_available returns False when /proc/net/wireless doesn't exist."""
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "/proc/net/wireless not found" in reason
def test_unavailable_when_interface_not_listed(self):
"""is_available returns False when the interface isn't in proc."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan1: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "wlan0" in reason
assert "wlan1" in reason
def test_available_when_interface_listed(self):
"""is_available returns True when the interface is present."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan0: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is True
assert reason == "ok"
def test_unavailable_when_file_unreadable(self):
"""is_available returns False when /proc/net/wireless exists but can't be read."""
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", side_effect=PermissionError("Permission denied")):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "Cannot read" in reason
# ===========================================================================
# create_collector() factory tests (ADR-049)
# ===========================================================================
class TestCreateCollector:
def test_returns_simulated_when_no_wifi(self):
"""On Linux without /proc/net/wireless, should return SimulatedCollector."""
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Linux"):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
collector = create_collector(preferred="auto")
assert isinstance(collector, SimulatedCollector)
def test_returns_simulated_for_explicit_preference(self):
"""preferred='simulated' always returns SimulatedCollector."""
collector = create_collector(preferred="simulated")
assert isinstance(collector, SimulatedCollector)
def test_returns_linux_collector_when_available(self):
"""On Linux with /proc/net/wireless, should return LinuxWifiCollector."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan0: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Linux"):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
collector = create_collector(preferred="auto", interface="wlan0")
assert isinstance(collector, LinuxWifiCollector)
def test_never_raises(self):
"""create_collector should never raise, regardless of platform."""
for plat in ["Linux", "Windows", "Darwin", "FreeBSD", "SunOS"]:
with patch("v1.src.sensing.rssi_collector.platform.system", return_value=plat):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
with patch("subprocess.run", side_effect=FileNotFoundError("not found")):
try:
collector = create_collector(preferred="auto")
assert collector is not None
except Exception as exc:
pytest.fail(f"create_collector raised on {plat}: {exc}")
def test_windows_default_interface_mapping(self):
"""On Windows with default interface='wlan0', should map to 'Wi-Fi'."""
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Windows"):
with patch("subprocess.run", side_effect=FileNotFoundError("netsh not found")):
collector = create_collector(preferred="auto", interface="wlan0")
# Should fall back to SimulatedCollector since netsh isn't available
assert isinstance(collector, SimulatedCollector)