feat: get_channel_scan() + push_radio_channel() in OpenWrtWirelessMixin
get_channel_scan(mode='active'|'passive'): - Discovers AP-mode interfaces via 'iw dev' - Runs 'iw dev <iface> scan [passive]' on each AP interface - Parses BSS blocks: BSSID, SSID, frequency, channel, signal, channel width (HT/VHT/HE), band (2.4/5/6 GHz) - Returns dict[iface → list[ChannelScanEntryDict]] push_radio_channel(radio, channel): - Sets channel via 'uci set wireless.<radio>.channel=<ch|auto>' - Commits and reloads: 'uci commit wireless && wifi reload' - channel=0 writes 'auto' Tests: 19 new unit tests covering parse logic, active/passive flag, AP-only interface filtering, 6 GHz band detection, push command order. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
e3a0a9e4b3
commit
512871b1bb
@@ -380,6 +380,164 @@ class OpenWrtWirelessMixin:
|
||||
_flush()
|
||||
return results
|
||||
|
||||
def get_channel_scan(self, mode: str = "active") -> dict[str, Any]:
|
||||
"""Return channel scan results for all AP-mode interfaces.
|
||||
|
||||
Uses ``iw dev <iface> scan`` (active) or
|
||||
``iw dev <iface> scan passive`` (passive) on each AP interface.
|
||||
|
||||
Keys are interface names (e.g. ``"wlan0"``). Values are lists of
|
||||
:class:`~napalm_device_types.models.ChannelScanEntryDict`.
|
||||
"""
|
||||
from napalm_device_types.models import ChannelScanEntryDict
|
||||
|
||||
# Step 1: discover interfaces and filter to AP mode only
|
||||
iw_out = self._send_command("iw dev 2>/dev/null || true")
|
||||
ap_ifaces: list[str] = []
|
||||
current_iface: str = ""
|
||||
current_type: str = ""
|
||||
iface_order: list[tuple[str, str]] = [] # (iface, type)
|
||||
|
||||
for line in iw_out.splitlines():
|
||||
stripped = line.strip()
|
||||
im = re.match(r"^Interface\s+(\S+)$", stripped)
|
||||
if im:
|
||||
if current_iface:
|
||||
iface_order.append((current_iface, current_type))
|
||||
current_iface = im.group(1)
|
||||
current_type = ""
|
||||
continue
|
||||
tm = re.match(r"^type\s+(\S+)$", stripped)
|
||||
if tm and current_iface:
|
||||
current_type = tm.group(1)
|
||||
|
||||
if current_iface:
|
||||
iface_order.append((current_iface, current_type))
|
||||
|
||||
ap_ifaces = [
|
||||
name for name, itype in iface_order
|
||||
if itype.upper() in ("AP", "AP/VLAN")
|
||||
]
|
||||
|
||||
if not ap_ifaces:
|
||||
return {}
|
||||
|
||||
# Step 2: run scan on each AP interface
|
||||
passive_flag = " passive" if mode == "passive" else ""
|
||||
result: dict[str, list[Any]] = {}
|
||||
|
||||
for iface in ap_ifaces:
|
||||
scan_cmd = f"iw dev {iface} scan{passive_flag} 2>/dev/null || true"
|
||||
scan_out = self._send_command(scan_cmd)
|
||||
entries = self._parse_iw_scan(scan_out, iface)
|
||||
result[iface] = entries
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _parse_iw_scan(scan_out: str, iface: str) -> list[Any]:
|
||||
"""Parse ``iw dev <iface> scan`` output into ChannelScanEntryDict list."""
|
||||
from napalm_device_types.models import ChannelScanEntryDict
|
||||
|
||||
entries: list[Any] = []
|
||||
current: dict[str, Any] = {}
|
||||
|
||||
def _freq_to_band(freq: int) -> str:
|
||||
if freq < 3000:
|
||||
return "2.4GHz"
|
||||
if freq < 5950:
|
||||
return "5GHz"
|
||||
return "6GHz"
|
||||
|
||||
def _freq_to_channel(freq: int) -> int:
|
||||
if 2412 <= freq <= 2484:
|
||||
if freq == 2484:
|
||||
return 14
|
||||
return (freq - 2407) // 5
|
||||
if 5160 <= freq <= 5885:
|
||||
return (freq - 5000) // 5
|
||||
if 5955 <= freq <= 7115:
|
||||
return (freq - 5950) // 5 + 1
|
||||
return 0
|
||||
|
||||
def _flush() -> None:
|
||||
if current.get("bssid"):
|
||||
freq = current.get("frequency", 0)
|
||||
ch = current.get("channel") or _freq_to_channel(freq)
|
||||
width = current.get("channel_width", 0)
|
||||
# If width is still 0 try to derive from HT/VHT/HE info
|
||||
if width == 0 and current.get("_ht_width"):
|
||||
width = current["_ht_width"]
|
||||
entries.append(ChannelScanEntryDict(
|
||||
bssid=current["bssid"],
|
||||
ssid=current.get("ssid", ""),
|
||||
frequency=freq,
|
||||
channel=ch,
|
||||
signal_dbm=current.get("signal_dbm", 0),
|
||||
channel_width=width,
|
||||
band=_freq_to_band(freq),
|
||||
))
|
||||
|
||||
for line in scan_out.splitlines():
|
||||
stripped = line.strip()
|
||||
|
||||
# New BSS block: "BSS aa:bb:cc:dd:ee:ff(on wlan0)"
|
||||
bss_m = re.match(r"^BSS\s+([\da-fA-F:]{17})\(", stripped)
|
||||
if bss_m:
|
||||
_flush()
|
||||
current = {"bssid": bss_m.group(1).lower()}
|
||||
continue
|
||||
|
||||
# freq: 2437
|
||||
freq_m = re.match(r"^freq:\s+(\d+)$", stripped)
|
||||
if freq_m:
|
||||
current["frequency"] = int(freq_m.group(1))
|
||||
continue
|
||||
|
||||
# signal: -72.00 dBm
|
||||
sig_m = re.match(r"^signal:\s+([-\d.]+)\s+dBm", stripped)
|
||||
if sig_m:
|
||||
current["signal_dbm"] = int(float(sig_m.group(1)))
|
||||
continue
|
||||
|
||||
# SSID: NeighborNet (may be empty for hidden)
|
||||
ssid_m = re.match(r"^SSID:\s*(.*)?$", stripped)
|
||||
if ssid_m:
|
||||
current["ssid"] = ssid_m.group(1).strip()
|
||||
continue
|
||||
|
||||
# HT operation primary channel: 6
|
||||
ht_ch_m = re.match(r"^\*\s+primary channel:\s+(\d+)", stripped)
|
||||
if ht_ch_m:
|
||||
current["channel"] = int(ht_ch_m.group(1))
|
||||
continue
|
||||
|
||||
# HT STA channel width: 20 MHz
|
||||
ht_w_m = re.match(r"^\*\s+STA channel width:\s+(\d+)\s+MHz", stripped)
|
||||
if ht_w_m:
|
||||
current["_ht_width"] = int(ht_w_m.group(1))
|
||||
continue
|
||||
|
||||
# VHT channel width: 80 MHz
|
||||
vht_w_m = re.match(r"^\*\s+channel width:\s+(\d+)\s+MHz", stripped)
|
||||
if vht_w_m:
|
||||
current["channel_width"] = int(vht_w_m.group(1))
|
||||
continue
|
||||
|
||||
_flush()
|
||||
return entries
|
||||
|
||||
def push_radio_channel(self, radio: str, channel: int) -> None:
|
||||
"""Set a new channel on *radio* via UCI and reload the wireless stack.
|
||||
|
||||
:param radio: UCI radio name, e.g. ``"radio0"``.
|
||||
:param channel: Channel number to set; ``0`` sets UCI ``auto``.
|
||||
"""
|
||||
ch_val = "auto" if channel == 0 else str(channel)
|
||||
self._send_command(f"uci set wireless.{radio}.channel={ch_val}")
|
||||
self._send_command("uci commit wireless")
|
||||
self._send_command("wifi reload")
|
||||
|
||||
def get_radio_status(self) -> dict[str, Any]:
|
||||
"""Return radio status from UCI and iwinfo.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user