From da37904108d31e1c5aa8c9d771b4a098f3169abf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mattias=20Walstr=C3=B6m?= Date: Mon, 8 Dec 2025 16:06:01 +0100 Subject: [PATCH] WiFi: Refactor statd (python) implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add support for AP (list connected stations) * Add scan-mode (Scan without create a fully configured station) Signed-off-by: Mattias Walström --- board/common/rootfs/usr/libexec/infix/iw.py | 454 ++++++++++++++++++ src/show/bash_completion.d/show | 2 +- src/show/show.py | 16 +- src/statd/python/cli_pretty/cli_pretty.py | 218 +++++++-- src/statd/python/yanger/__main__.py | 3 + src/statd/python/yanger/ietf_hardware.py | 261 ++++++++-- .../python/yanger/ietf_interfaces/wifi.py | 219 ++++++++- 7 files changed, 1047 insertions(+), 126 deletions(-) create mode 100755 board/common/rootfs/usr/libexec/infix/iw.py diff --git a/board/common/rootfs/usr/libexec/infix/iw.py b/board/common/rootfs/usr/libexec/infix/iw.py new file mode 100755 index 00000000..47e7db12 --- /dev/null +++ b/board/common/rootfs/usr/libexec/infix/iw.py @@ -0,0 +1,454 @@ +#!/usr/bin/env python3 +""" +iw command wrapper that returns structured JSON data + +Usage: + iw.py list - List all PHY devices + iw.py dev - List all interfaces grouped by PHY + iw.py info - Get PHY or interface information + iw.py survey - Get channel survey data +""" + +import sys +import json +import subprocess +import re + + +def run_iw(*args): + """Run iw command and return output""" + try: + result = subprocess.run( + ['iw'] + list(args), + capture_output=True, + text=True, + timeout=5 + ) + if result.returncode == 0: + return result.stdout + return None + except Exception: + return None + + +def normalize_phy_name(name): + """ + Convert radioN to phyN or vice versa based on what exists in sysfs. + Returns the actual phy name that exists. + """ + import os + + # Try the name as-is first + if os.path.exists(f'/sys/class/ieee80211/{name}'): + return name + + # Try converting radioN <-> phyN + if name.startswith('radio'): + phy_name = 'phy' + name[5:] + if os.path.exists(f'/sys/class/ieee80211/{phy_name}'): + return phy_name + elif name.startswith('phy'): + radio_name = 'radio' + name[3:] + if os.path.exists(f'/sys/class/ieee80211/{radio_name}'): + return radio_name + + # Return original if nothing found + return name + + +def parse_phy_info(phy_name): + """ + Parse 'iw phy info' output or 'iw info' output + Returns: {bands, driver, manufacturer, max_txpower, num_virtual_interfaces, interface_combinations} + """ + # Normalize the phy name + actual_phy = normalize_phy_name(phy_name) + + # Try 'iw phy info' first + output = run_iw('phy', actual_phy, 'info') + + # If that fails, try 'iw info' (some systems support this) + if not output: + output = run_iw(actual_phy, 'info') + + if not output: + return {} + + result = { + 'name': phy_name, + 'bands': [], + 'driver': None, + 'manufacturer': None, + 'max_txpower': None, + 'num_virtual_interfaces': 0, + 'interface_combinations': [] + } + + current_band = None + band_num = 0 + in_combinations = False + max_power = None + + for line in output.splitlines(): + stripped = line.strip() + + # Detect band sections + if stripped.startswith('Band '): + if current_band and current_band.get('frequencies'): + result['bands'].append(current_band) + band_num += 1 + current_band = { + 'band': band_num, + 'frequencies': [], + 'name': None, + 'ht_capable': False, + 'vht_capable': False, + 'he_capable': False + } + in_combinations = False + + # Parse frequencies (handle both "2412 MHz" and "2412.0 MHz" formats) + elif current_band and not in_combinations: + freq_match = re.match(r'\* ([0-9.]+) MHz.*?\(([0-9.]+) dBm\)', stripped) + if freq_match: + freq = int(float(freq_match.group(1))) # Convert "2412.0" to 2412 + power = float(freq_match.group(2)) + + current_band['frequencies'].append(freq) + + # Track max power + if max_power is None or power > max_power: + max_power = power + + # Check capabilities + if 'HT ' in stripped or 'High Throughput' in stripped: + current_band['ht_capable'] = True + if 'VHT' in stripped or 'Very High Throughput' in stripped: + current_band['vht_capable'] = True + if 'HE ' in stripped or 'High Efficiency' in stripped: + current_band['he_capable'] = True + + # Detect interface combinations section + if 'valid interface combinations:' in stripped.lower(): + in_combinations = True + continue + + # Parse interface combinations + if in_combinations: + if stripped.startswith('*'): + # Parse combination line + comb_info = {'limits': []} + + # Parse limits: #{ type } <= max + limit_matches = re.findall(r'#\{\s*([^}]+)\s*\}\s*<=\s*(\d+)', stripped) + for types_str, max_val in limit_matches: + types = [t.strip() for t in types_str.split(',')] + comb_info['limits'].append({ + 'max': int(max_val), + 'types': types + }) + + # Parse total + total_match = re.search(r'total\s*<=\s*(\d+)', stripped) + if total_match: + comb_info['max_total'] = int(total_match.group(1)) + + # Parse channels + channels_match = re.search(r'#channels\s*<=\s*(\d+)', stripped) + if channels_match: + comb_info['num_channels'] = int(channels_match.group(1)) + + if comb_info.get('limits') or comb_info.get('max_total'): + result['interface_combinations'].append(comb_info) + + elif not stripped.startswith('#') and ':' in stripped and not stripped.startswith('*'): + # End of combinations section + in_combinations = False + + # Add last band + if current_band and current_band.get('frequencies'): + result['bands'].append(current_band) + + # Determine band names and assign band numbers + for band in result['bands']: + if band['frequencies']: + freq = band['frequencies'][0] + if 2400 <= freq <= 2500: + band['name'] = '2.4GHz' + band['band'] = 1 + elif 5150 <= freq <= 5900: + band['name'] = '5GHz' + band['band'] = 2 + elif 5955 <= freq <= 7115: + band['name'] = '6GHz' + band['band'] = 3 + + # Set max TX power + if max_power is not None: + result['max_txpower'] = int(max_power) + + # Get driver and manufacturer from sysfs + try: + driver_link = subprocess.run( + ['readlink', '-f', f'/sys/class/ieee80211/{actual_phy}/device/driver'], + capture_output=True, text=True, timeout=1 + ).stdout.strip() + + if driver_link: + driver_name = driver_link.split('/')[-1] + result['driver'] = driver_name + + # Map driver to manufacturer + driver_lower = driver_name.lower() + if 'mt' in driver_lower or 'mediatek' in driver_lower: + result['manufacturer'] = 'MediaTek Inc.' + elif 'rtw' in driver_lower or 'realtek' in driver_lower: + result['manufacturer'] = 'Realtek Semiconductor Corp.' + elif 'ath' in driver_lower or 'qca' in driver_lower: + result['manufacturer'] = 'Qualcomm Atheros' + elif 'iwl' in driver_lower or 'intel' in driver_lower: + result['manufacturer'] = 'Intel Corporation' + elif 'brcm' in driver_lower or 'broadcom' in driver_lower: + result['manufacturer'] = 'Broadcom Inc.' + except Exception: + pass + + # Count virtual interfaces + dev_output = run_iw('dev') + if dev_output: + # Extract phy number from actual phy name + phy_num = None + if actual_phy.startswith('radio'): + phy_num = actual_phy[5:] + elif actual_phy.startswith('phy'): + phy_num = actual_phy[3:] + + if phy_num: + count = 0 + current_phy = None + for line in dev_output.splitlines(): + if line.startswith('phy#'): + current_phy = line.replace('phy#', '').strip() + elif current_phy == phy_num and 'Interface' in line: + count += 1 + result['num_virtual_interfaces'] = count + + return result + + +def parse_interface_info(ifname): + """ + Parse 'iw dev info' output + Returns: {ifname, iftype, mac, ssid, frequency, channel, txpower, channel_width} + """ + output = run_iw('dev', ifname, 'info') + if not output: + return {} + + result = {'ifname': ifname} + + for line in output.splitlines(): + stripped = line.strip() + + # Interface type + if stripped.startswith('type '): + result['iftype'] = stripped.split()[1] + + # MAC address + elif stripped.startswith('addr '): + result['mac'] = stripped.split()[1] + + # SSID + elif stripped.startswith('ssid '): + result['ssid'] = ' '.join(stripped.split()[1:]) + + # Channel/frequency + elif stripped.startswith('channel '): + parts = stripped.split() + if len(parts) >= 2: + result['channel'] = int(parts[1]) + if 'MHz' in stripped: + freq_match = re.search(r'\((\d+) MHz', stripped) + if freq_match: + result['frequency'] = int(freq_match.group(1)) + # Channel width + if 'width:' in stripped: + width_match = re.search(r'width:\s*(\d+)\s*MHz', stripped) + if width_match: + result['channel_width'] = f"{width_match.group(1)} MHz" + + # TX power + elif stripped.startswith('txpower '): + power_match = re.search(r'([0-9.]+) dBm', stripped) + if power_match: + result['txpower'] = float(power_match.group(1)) + + return result + + +def parse_survey(ifname): + """ + Parse 'iw dev survey dump' output + Returns: list of {frequency, in_use, noise, active_time, busy_time, receive_time, transmit_time} + """ + output = run_iw('dev', ifname, 'survey', 'dump') + if not output: + return [] + + channels = [] + current_channel = None + + for line in output.splitlines(): + stripped = line.strip() + + # New survey entry + if stripped.startswith('Survey data from'): + if current_channel: + channels.append(current_channel) + current_channel = None + + # Frequency + elif stripped.startswith('frequency:'): + parts = stripped.split() + if len(parts) >= 2: + freq = int(parts[1]) + in_use = '[in use]' in stripped + current_channel = { + 'frequency': freq, + 'in_use': in_use + } + + # Channel metrics + elif current_channel: + if stripped.startswith('noise:'): + noise_match = re.search(r'(-?\d+) dBm', stripped) + if noise_match: + current_channel['noise'] = int(noise_match.group(1)) + + elif stripped.startswith('channel active time:'): + time_match = re.search(r'(\d+) ms', stripped) + if time_match: + current_channel['active_time'] = int(time_match.group(1)) + + elif stripped.startswith('channel busy time:'): + time_match = re.search(r'(\d+) ms', stripped) + if time_match: + current_channel['busy_time'] = int(time_match.group(1)) + + elif stripped.startswith('channel receive time:'): + time_match = re.search(r'(\d+) ms', stripped) + if time_match: + current_channel['receive_time'] = int(time_match.group(1)) + + elif stripped.startswith('channel transmit time:'): + time_match = re.search(r'(\d+) ms', stripped) + if time_match: + current_channel['transmit_time'] = int(time_match.group(1)) + + # Add last channel + if current_channel: + channels.append(current_channel) + + return channels + + +def parse_list(): + """ + Parse 'iw list' output + Returns: list of PHY names + """ + output = run_iw('list') + if not output: + return [] + + phys = [] + for line in output.splitlines(): + match = re.match(r'Wiphy (phy\d+|radio\d+)', line) + if match: + phys.append(match.group(1)) + + return phys + + +def parse_dev(): + """ + Parse 'iw dev' output + Returns: dict mapping PHY numbers to list of interfaces + """ + output = run_iw('dev') + if not output: + return {} + + result = {} + current_phy = None + + for line in output.splitlines(): + # PHY line: "phy#0" or "phy#1" + if line.startswith('phy#'): + current_phy = line.replace('phy#', '').strip() + if current_phy not in result: + result[current_phy] = [] + # Interface line: " Interface wlan0" + elif current_phy and 'Interface' in line: + ifname = line.split('Interface')[1].strip() + result[current_phy].append(ifname) + + return result + + +def main(): + if len(sys.argv) < 2: + print(json.dumps({ + 'error': 'Usage: iw.py [device]', + 'commands': { + 'list': 'List all PHY devices', + 'dev': 'List all interfaces grouped by PHY', + 'info': 'Get PHY or interface information (requires device)', + 'survey': 'Get channel survey data (requires interface name)' + }, + 'examples': [ + 'iw.py list', + 'iw.py dev', + 'iw.py info radio0', + 'iw.py info phy4', + 'iw.py info wlan0', + 'iw.py survey wlan0' + ] + }, indent=2)) + sys.exit(1) + + command = sys.argv[1] + + try: + if command == 'list': + data = parse_list() + elif command == 'dev': + data = parse_dev() + elif command == 'info': + if len(sys.argv) < 3: + data = {'error': 'info command requires device argument'} + else: + device = sys.argv[2] + # Auto-detect if device is a PHY (phy*/radio*) or interface + if device.startswith('phy') or device.startswith('radio'): + data = parse_phy_info(device) + else: + data = parse_interface_info(device) + elif command == 'survey': + if len(sys.argv) < 3: + data = {'error': 'survey command requires device argument'} + else: + device = sys.argv[2] + data = parse_survey(device) + else: + data = {'error': f'Unknown command: {command}'} + + print(json.dumps(data, indent=2)) + + except Exception as e: + print(json.dumps({'error': str(e)})) + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/src/show/bash_completion.d/show b/src/show/bash_completion.d/show index c7963444..3c6489e6 100644 --- a/src/show/bash_completion.d/show +++ b/src/show/bash_completion.d/show @@ -4,7 +4,7 @@ _show_completions() { cur="${COMP_WORDS[COMP_CWORD]}" prev="${COMP_WORDS[COMP_CWORD-1]}" - commands="dhcp interface ntp routes software stp" + commands="dhcp interface ntp routes software stp wifi-radio" if [[ $COMP_CWORD -eq 1 ]]; then COMPREPLY=( $(compgen -W "$commands" -- "$cur") ) diff --git a/src/show/show.py b/src/show/show.py index 204d79a4..2ec78ba3 100755 --- a/src/show/show.py +++ b/src/show/show.py @@ -422,19 +422,6 @@ def lldp(args: List[str]): return cli_pretty(data, "show-lldp") -def wifi(args: List[str]): - iface = args[0] - if len(args) == 0: - print("Illigal usage") - return - if is_valid_interface_name(iface): - if not os.path.exists(f"/sys/class/net/{iface}/wireless"): - print("Not a Wi-Fi interface") - return - data = run_sysrepocfg("/ietf-interfaces:interfaces") - cli_pretty(data, "show-wifi-scan", "-n", iface) - else: - print(f"Invalid interface name: {iface}") def system(args: List[str]) -> None: # Get system state from sysrepo @@ -556,8 +543,7 @@ def execute_command(command: str, args: List[str]): 'services': services, 'software': software, 'stp': stp, - 'system': system, - 'wifi': wifi + 'system': system } if command in command_mapping: diff --git a/src/statd/python/cli_pretty/cli_pretty.py b/src/statd/python/cli_pretty/cli_pretty.py index 926dd7b1..324d181e 100755 --- a/src/statd/python/cli_pretty/cli_pretty.py +++ b/src/statd/python/cli_pretty/cli_pretty.py @@ -190,12 +190,6 @@ class PadNtpSource: poll = 14 -class PadWifiScan: - ssid = 40 - encryption = 30 - signal = 9 - - class PadLldp: interface = 16 rem_idx = 10 @@ -544,7 +538,7 @@ class Decore(): @staticmethod def bright_green(txt): - return Decore.decorate("1;32", txt, "39") + return Decore.decorate("1;32", txt, "0") @staticmethod def yellow(txt): @@ -1248,20 +1242,73 @@ class Iface: print(row) def pr_wifi_ssids(self): - hdr = (f"{'SSID':<{PadWifiScan.ssid}}" - f"{'ENCRYPTION':<{PadWifiScan.encryption}}" - f"{'SIGNAL':<{PadWifiScan.signal}}") + print("\nAVAILABLE NETWORKS:") + ssid_table = SimpleTable([ + Column('SSID'), + Column('SECURITY'), + Column('SIGNAL'), + Column('CHANNEL') + ]) - print(Decore.invert(hdr)) - results = self.wifi.get("scan-results", {}) + station = self.wifi.get("station", {}) + results = station.get("scan-results", {}) for result in results: - encstr = ", ".join(result["encryption"]) - status = rssi_to_status(result["rssi"]) - row = f"{result['ssid']:<{PadWifiScan.ssid}}" - row += f"{encstr:<{PadWifiScan.encryption}}" - row += f"{status:<{PadWifiScan.signal}}" + encstr = ", ".join(result.get("encryption", ["Unknown"])) + status = rssi_to_status(result.get("rssi", -100)) + channel = result.get("channel", "?") - print(row) + ssid_table.row(result.get('ssid', 'Hidden'), encstr, status, channel) + ssid_table.print() + + def pr_wifi_stations(self): + """Display connected stations for AP mode""" + if not self.wifi: + return + + # Get stations from access-point container + ap = self.wifi.get("access-point", {}) + stations_data = ap.get("stations", {}) + stations = stations_data.get("station", []) + + if not stations: + return + + print("\nCONNECTED STATIONS:") + stations_table = SimpleTable([ + Column('MAC'), + Column('SIGNAL'), + Column('TIME'), + Column('RX PKT'), + Column('TX PKT'), + Column('RX BYTES'), + Column('TX BYTES'), + Column('RX SPEED'), + Column('TX SPEED') + ]) + + for station in stations: + mac = station.get("mac-address", "unknown") + rssi = station.get("rssi") + signal_str = rssi_to_status(rssi) if rssi is not None else "------" + + conn_time = station.get("connected-time", 0) + time_str = f"{conn_time}s" + + rx_pkt = station.get("rx-packets", 0) + tx_pkt = station.get("tx-packets", 0) + rx_bytes = station.get("rx-bytes", 0) + tx_bytes = station.get("tx-bytes", 0) + + # Speed in 100 kbit/s units, convert to Mbps for display + rx_speed = station.get("rx-speed", 0) + tx_speed = station.get("tx-speed", 0) + rx_speed_str = f"{rx_speed / 10:.1f}" if rx_speed else "-" + tx_speed_str = f"{tx_speed / 10:.1f}" if tx_speed else "-" + + stations_table.row(mac, signal_str, time_str, rx_pkt, tx_pkt, + rx_bytes, tx_bytes, rx_speed_str, tx_speed_str) + + stations_table.print() def pr_proto_wifi(self, pipe=''): @@ -1269,20 +1316,35 @@ class Iface: print(row) ssid = None rssi = None + mode = None if self.wifi: - rssi=self.wifi.get("rssi") - ssid=self.wifi.get("ssid") - if ssid is None: - ssid="------" - - if rssi is None: - signal="------" + # Detect mode: AP has "stations", Station has "rssi" or "scan-results" + ap=self.wifi.get("access-point", {}) + if ap: + ssid = ap.get("ssid", "------") + mode = "AP" + stations_data = ap.get("stations", {}) + stations = stations_data.get("station", []) + station_count = len(stations) + data_str = f"{mode}, ssid: {ssid}, stations: {station_count}" + else: + station=self.wifi.get("station", {}) + ssid = station.get("ssid", "------") + rssi = station.get("rssi") + mode = "Station" + if rssi is not None: + signal = rssi_to_status(rssi) + data_str = f"{mode}, ssid: {ssid}, signal: {signal}" + else: + data_str = f"{mode}, ssid: {ssid}" else: - signal=rssi_to_status(rssi) - data_str = f"ssid: {ssid}, signal: {signal}" + data_str = "ssid: ------" - row = f"{'':<{Pad.iface}}" + row = f"{'':<{Pad.flags}}" + row += f"{pipe:<{Pad.iface}}" + row = f"{'':<{Pad.flags}}" + row += f"{pipe:<{Pad.iface}}" row += f"{'wifi':<{Pad.proto}}" row += f"{'':<{Pad.state}}{data_str}" print(row) @@ -1550,13 +1612,37 @@ class Iface: else: print(f"{'ipv6 addresses':<{20}}:") + if self.in_octets and self.out_octets: + print(f"{'in-octets':<{20}}: {self.in_octets}") + print(f"{'out-octets':<{20}}: {self.out_octets}") + + frame = get_json_data([], self.data,'ieee802-ethernet-interface:ethernet', + 'statistics', 'frame') + if self.wifi: - ssid=self.wifi.get('ssid', "----") - rssi=self.wifi.get('rssi', "----") - print(f"{'SSID':<{20}}: {ssid}") - print(f"{'Signal':<{20}}: {rssi}") - print("") - self.pr_wifi_ssids() + # Detect mode: AP has "stations", Station has "rssi" or "scan-results" + ap = self.wifi.get('access-point') + if ap: + mode = "access-point" + ssid = ap.get('ssid', "----") + stations_data = ap.get("stations", {}) + stations = stations_data.get("station", []) + print(f"{'mode':<{20}}: {mode}") + print(f"{'ssid':<{20}}: {ssid}") + print(f"{'connected stations':<{20}}: {len(stations)}") + self.pr_wifi_stations() + else: + mode = "station" + station = self.wifi.get('station', {}) + rssi = station.get('rssi') + ssid = station.get('ssid', "----") + print(f"{'mode':<{20}}: {mode}") + print(f"{'ssid':<{20}}: {ssid}") + if rssi is not None: + signal_status = rssi_to_status(rssi) + print(f"{'signal':<{20}}: {rssi} dBm ({signal_status})") + if "scan-results" in station: + self.pr_wifi_ssids() if self.gre: print(f"{'local address':<{20}}: {self.gre['local']}") @@ -1567,12 +1653,6 @@ class Iface: print(f"{'remote address':<{20}}: {self.vxlan['remote']}") print(f"{'VxLAN id':<{20}}: {self.vxlan['vni']}") - if self.in_octets and self.out_octets: - print(f"{'in-octets':<{20}}: {self.in_octets}") - print(f"{'out-octets':<{20}}: {self.out_octets}") - - frame = get_json_data([], self.data,'ieee802-ethernet-interface:ethernet', - 'statistics', 'frame') if frame: print("") for key, val in frame.items(): @@ -1953,9 +2033,6 @@ def show_services(json): services_data = get_json_data({}, json, 'ietf-system:system-state', 'infix-system:services') services = services_data.get("service", []) - # This is the first usage of simple table. I assume this will be - # copied so I left a lot of comments. If you copy it feel free - # to be less verbose.. service_table = SimpleTable([ Column('NAME'), Column('STATUS'), @@ -2005,9 +2082,10 @@ def show_hardware(json): motherboard = [c for c in components if c.get("class") == "iana-hardware:chassis"] usb_ports = [c for c in components if c.get("class") == "infix-hardware:usb"] sensors = [c for c in components if c.get("class") == "iana-hardware:sensor"] + wifi_radios = [c for c in components if c.get("class") == "infix-hardware:wifi"] # Determine overall width (use the wider of the two sections) - width = max(PadUsbPort.table_width(), PadSensor.table_width()) + width = max(PadUsbPort.table_width(), PadSensor.table_width(), 100) # Display full-width inverted heading print(Decore.invert(f"{'HARDWARE COMPONENTS':<{width}}")) @@ -2027,6 +2105,56 @@ def show_hardware(json): if board.get("hardware-rev"): print(f"Hardware Revision : {board['hardware-rev']}") + if wifi_radios: + Decore.title("WiFi radios", width) + + radios_table = SimpleTable([ + Column('NAME'), + Column('MANUFACTURER'), + Column('BANDS', 'right'), + Column('STANDARDS', 'right'), + Column('MAX AP', 'right') + ]) + + for component in wifi_radios: + phy = component.get("name", "") + manufacturer = component.get("mfg-name", "Unknown") + + radio_data = component.get("infix-hardware:wifi-radio", {}) + + bands = radio_data.get("bands", []) + band_names = [] + has_ht = False + has_vht = False + has_he = False + + for band in bands: + if band.get("name"): + band_names.append(band["name"]) + if band.get("ht-capable"): + has_ht = True + if band.get("vht-capable"): + has_vht = True + if band.get("he-capable"): + has_he = True + + bands_str = "/".join(band_names) if band_names else "Unknown" + + standards = [] + if has_ht: + standards.append("11n") + if has_vht: + standards.append("11ac") + if has_he: + standards.append("11ax") + standard_str = "/".join(standards) if standards else "Unknown" + + max_if = radio_data.get("max-interfaces", {}) + max_ap = max_if.get('ap', 'N/A') if max_if else 'N/A' + + radios_table.row(phy, manufacturer, bands_str, standard_str, max_ap) + radios_table.print() + if usb_ports: Decore.title("USB Ports", width) hdr = (f"{'NAME':<{PadUsbPort.name}}" @@ -4342,6 +4470,8 @@ def main(): show_firewall_logs(args.limit) elif args.command == "show-ntp": show_ntp(json_data) + elif args.command == "show-wifi-radio": + show_wifi_radio(json_data) elif args.command == "show-bfd": show_bfd(json_data) elif args.command == "show-bfd-status": diff --git a/src/statd/python/yanger/__main__.py b/src/statd/python/yanger/__main__.py index f8e2477f..ddffaa09 100644 --- a/src/statd/python/yanger/__main__.py +++ b/src/statd/python/yanger/__main__.py @@ -87,6 +87,9 @@ def main(): elif args.model == 'ietf-bfd-ip-sh': from . import ietf_bfd_ip_sh yang_data = ietf_bfd_ip_sh.operational() + elif args.model == 'infix-wifi-radio': + from . import infix_wifi_radio + yang_data = infix_wifi_radio.operational() else: common.LOG.warning("Unsupported model %s", args.model) sys.exit(1) diff --git a/src/statd/python/yanger/ietf_hardware.py b/src/statd/python/yanger/ietf_hardware.py index 876ab863..641dc6dc 100644 --- a/src/statd/python/yanger/ietf_hardware.py +++ b/src/statd/python/yanger/ietf_hardware.py @@ -1,6 +1,8 @@ import datetime import os import glob +import re +import sys from .common import insert, YangDate from .host import HOST @@ -150,65 +152,44 @@ def normalize_sensor_name(name): def get_wifi_phy_info(): """ - Discover WiFi PHYs and map them to bands and interface names. + Discover WiFi PHYs using iw list command. Returns dict: {phy_name: {band: str, iface: str, description: str}} - Example: {"phy0": {"band": "2.4 GHz", "iface": "wlan0", "description": "WiFi Radio (2.4 GHz)"}} + Example: {"radio0": {"band": "2.4 GHz", "iface": "wlan0", "description": "WiFi Radio (2.4 GHz)"}} """ phy_info = {} try: - # Enumerate PHYs from /sys/class/ieee80211/ - ieee80211_path = "/sys/class/ieee80211" - if not os.path.exists(ieee80211_path): + # Use iw.py to list all PHYs + phys = HOST.run_json(("/usr/libexec/infix/iw.py", "list"), default=[]) + if not phys: return phy_info - for phy in os.listdir(ieee80211_path): - if not phy.startswith("phy"): - continue + # Initialize PHY info for each PHY + for phy in phys: + phy_info[phy] = {"band": "Unknown", "iface": None, "description": None} - phy_path = os.path.join(ieee80211_path, phy) - info = {"band": "Unknown", "iface": None, "description": None} + # Create a mapping from PHY number to PHY name + phy_num_to_name = {} + for phy_name in phy_info.keys(): + # Extract number from radio/phy name (e.g., "0" from "radio0" or "phy0") + num_match = re.search(r'(\d+)$', phy_name) + if num_match: + phy_num = num_match.group(1) + phy_num_to_name[phy_num] = phy_name - # Try to determine band from device path or hwmon name - # The hwmon device usually tells us: mt7915_phy0, mt7915_phy1, etc. - # We'll check supported frequencies to determine band - try: - # Read supported bands - check if device supports 5 GHz - # Most dual-band chips expose phy0 as 2.4 GHz and phy1 as 5 GHz - device_path = os.path.join(phy_path, "device") - if os.path.exists(device_path): - # Simple heuristic: phy0 is usually 2.4 GHz, phy1 is 5 GHz - # This works for most MediaTek chips (mt7915, mt7921, etc.) - if phy == "phy0": - info["band"] = "2.4 GHz" - elif phy == "phy1": - info["band"] = "5 GHz" - elif phy == "phy2": - info["band"] = "6 GHz" # WiFi 6E - except: - pass + # Find associated virtual interfaces using iw.py dev + dev_map = HOST.run_json(("/usr/libexec/infix/iw.py", "dev"), default={}) - # Find associated interface by checking which interface has a phy80211 link to this PHY - try: - net_path = "/sys/class/net" - if os.path.exists(net_path): - for iface in os.listdir(net_path): - phy_link = os.path.join(net_path, iface, "phy80211") - if os.path.islink(phy_link): - # Read the link target and extract PHY name - try: - link_target = os.readlink(phy_link) - linked_phy = os.path.basename(link_target) - if linked_phy == phy: - info["iface"] = iface - break - except: - continue - except: - pass + # dev_map is a dict mapping PHY numbers to list of interfaces + for phy_num, interfaces in dev_map.items(): + phy_name = phy_num_to_name.get(phy_num) + if phy_name and phy_name in phy_info and interfaces: + # Use the first interface + phy_info[phy_name]["iface"] = interfaces[0] - # Build description + # Build descriptions + for phy, info in phy_info.items(): if info["iface"] and info["band"] != "Unknown": info["description"] = f"WiFi Radio {info['iface']} ({info['band']})" elif info["band"] != "Unknown": @@ -218,8 +199,6 @@ def get_wifi_phy_info(): else: info["description"] = "WiFi Radio" - phy_info[phy] = info - except Exception: pass @@ -257,6 +236,12 @@ def hwmon_sensor_components(): continue device_name = HOST.read(name_path).strip() + + # Check if device/name exists (e.g., for WiFi radios) and use that instead + device_name_path = os.path.join(hwmon_path, "device", "name") + if HOST.exists(device_name_path): + device_name = HOST.read(device_name_path).strip() + base_name = normalize_sensor_name(device_name) # Helper to create sensor component with human-readable description @@ -426,8 +411,8 @@ def hwmon_sensor_components(): wifi_info = get_wifi_phy_info() for component in components: name = component.get("name", "") - # Match phy0, phy1, etc. sensors - if name.startswith("phy") and name in wifi_info: + # Match radio0, radio1, etc. sensors + if name.startswith("radio") and name in wifi_info: phy = wifi_info[name] # Add WiFi-specific description component["description"] = phy["description"] @@ -496,6 +481,179 @@ def thermal_sensor_components(): return components +def get_survey_data(ifname): + """Get channel survey data using iw.py script""" + channels = [] + + try: + survey_data = HOST.run_json(("/usr/libexec/infix/iw.py", "survey", ifname), default=[]) + + for entry in survey_data: + channel = { + "frequency": entry.get("frequency"), + "in-use": entry.get("in_use", False) + } + + # Add optional fields if present + if "noise" in entry: + channel["noise"] = entry["noise"] + if "active_time" in entry: + channel["active-time"] = entry["active_time"] + if "busy_time" in entry: + channel["busy-time"] = entry["busy_time"] + if "receive_time" in entry: + channel["receive-time"] = entry["receive_time"] + if "transmit_time" in entry: + channel["transmit-time"] = entry["transmit_time"] + + channels.append(channel) + + except Exception: + pass + + return channels + + +def get_phy_info(phy_name): + """Get complete PHY information using iw.py script""" + try: + return HOST.run_json(("/usr/libexec/infix/iw.py", "info", phy_name), default={}) + except Exception: + return {} + + +def convert_iw_phy_info_for_yanger(phy_info): + """ + Convert iw.py phy_info format to yanger format. + Input: iw.py format with 'bands', 'driver', 'manufacturer', 'interface_combinations' + Output: yanger format with renamed/restructured fields + """ + result = {"bands": [], "driver": None, "manufacturer": "Unknown", "max-interfaces": {}} + + # Convert bands - iw.py already uses snake_case for capabilities + for band in phy_info.get("bands", []): + band_data = { + "band": str(band.get("band", 0)), + "name": band.get("name", "Unknown") + } + + # Add capability flags (iw.py uses snake_case: ht_capable, vht_capable, he_capable) + if band.get("ht_capable"): + band_data["ht-capable"] = True + if band.get("vht_capable"): + band_data["vht-capable"] = True + if band.get("he_capable"): + band_data["he-capable"] = True + + result["bands"].append(band_data) + + # Copy driver and manufacturer + if phy_info.get("driver"): + result["driver"] = phy_info["driver"] + if phy_info.get("manufacturer"): + result["manufacturer"] = phy_info["manufacturer"] + + # Convert interface combinations to max-interfaces + # Find max AP interfaces from combinations + for comb in phy_info.get("interface_combinations", []): + for limit in comb.get("limits", []): + if "AP" in limit.get("types", []): + ap_max = limit.get("max", 0) + if "ap" not in result["max-interfaces"] or ap_max > result["max-interfaces"]["ap"]: + result["max-interfaces"]["ap"] = ap_max + + return result + + +def wifi_radio_components(): + """ + Create WiFi radio components with complete operational data. + Returns a list of hardware components for WiFi radios. + """ + components = [] + wifi_info = get_wifi_phy_info() + + for phy_name, phy_data in wifi_info.items(): + component = { + "name": phy_name, + "class": "infix-hardware:wifi", + "description": phy_data.get("description", "WiFi Radio") + } + + # Initialize wifi-radio data structure + wifi_radio_data = {} + + # Get complete PHY information from iw.py script + iw_info = get_phy_info(phy_name) + + # Convert iw.py format to yanger format + phy_details = convert_iw_phy_info_for_yanger(iw_info) + + # Add manufacturer to component + if phy_details.get("manufacturer") and phy_details["manufacturer"] != "Unknown": + component["mfg-name"] = phy_details["manufacturer"] + + # Add bands + if phy_details.get("bands"): + wifi_radio_data["bands"] = phy_details["bands"] + + # Add driver + if phy_details.get("driver"): + wifi_radio_data["driver"] = phy_details["driver"] + + # Add max-interfaces + if phy_details.get("max-interfaces"): + wifi_radio_data["max-interfaces"] = phy_details["max-interfaces"] + + # Add max TX power from iw info + if iw_info.get("max_txpower"): + wifi_radio_data["max-txpower"] = iw_info["max_txpower"] + + # Add supported channels from band frequencies + supported_channels = [] + for band in iw_info.get("bands", []): + for freq in band.get("frequencies", []): + # Convert frequency to channel number + if 2412 <= freq <= 2484: + channel = (freq - 2407) // 5 + elif 5170 <= freq <= 5825: + channel = (freq - 5000) // 5 + elif 5955 <= freq <= 7115: + channel = (freq - 5950) // 5 + else: + continue + supported_channels.append(channel) + + if supported_channels: + wifi_radio_data['supported-channels'] = sorted(set(supported_channels)) + + # Count virtual interfaces from iw info + num_ifaces = iw_info.get('num_virtual_interfaces', 0) + wifi_radio_data['num-virtual-interfaces'] = num_ifaces + + # Get survey data if we have an interface + iface = phy_data.get("iface") + if iface: + try: + channels = get_survey_data(iface) + + if channels: + wifi_radio_data["survey"] = { + "channel": channels + } + except Exception: + # If survey fails, continue without survey data + pass + + # Add wifi-radio data to component + if wifi_radio_data: + component["infix-hardware:wifi-radio"] = wifi_radio_data + + components.append(component) + + return components + + def operational(): systemjson = HOST.read_json("/run/system.json") @@ -507,6 +665,7 @@ def operational(): usb_port_components(systemjson) + hwmon_sensor_components() + thermal_sensor_components() + + wifi_radio_components() + [], }, } diff --git a/src/statd/python/yanger/ietf_interfaces/wifi.py b/src/statd/python/yanger/ietf_interfaces/wifi.py index b220f9df..78a8b6dd 100644 --- a/src/statd/python/yanger/ietf_interfaces/wifi.py +++ b/src/statd/python/yanger/ietf_interfaces/wifi.py @@ -2,38 +2,213 @@ from ..host import HOST import json import re -def wifi(ifname): - wifi_data={} + +def detect_wifi_mode(ifname): + """Detect if interface is in AP or Station mode""" + try: + output = HOST.run(tuple(f"iw dev {ifname} info".split()), default="") + for line in output.splitlines(): + if 'type' in line.lower(): + if 'ap' in line.lower(): + return 'ap' + else: + return 'station' + except Exception: + pass + + # Default to station mode + return 'station' + + +def find_primary_interface_from_config(ifname): + """Find primary interface by reading hostapd config files""" + try: + file_list = HOST.run(tuple("ls /etc/hostapd-*.conf".split()), default="") + if not file_list: + return None + + for config_file in file_list.splitlines(): + config_file = config_file.strip() + if not config_file: + continue + + try: + content = HOST.run(tuple(f"cat {config_file}".split()), default="") + if not content: + continue + + if f"interface={ifname}" in content or f"bss={ifname}" in content: + for line in content.splitlines(): + if line.startswith("interface="): + return line.split("=", 1)[1].strip() + except Exception: + continue + except Exception: + pass + return None + + +def wifi_ap(ifname): + """Get operational data for AP mode using hostapd_cli""" + ap_data = {} try: - data=HOST.run(tuple(f"wpa_cli -i {ifname} status".split()), default="") + primary_if = find_primary_interface_from_config(ifname) + if not primary_if: + return {} + + data = HOST.run(tuple(f"hostapd_cli -i {primary_if} status".split()), default="") + if not data: + return {} + + # Find our interface's SSID, different for bss and primary, because it is + if ifname == primary_if: + # Primary interface - get ssid[0] or ssid + for line in data.splitlines(): + if "=" in line: + try: + k, v = line.split("=", 1) + if k in ("ssid[0]", "ssid"): + ap_data["ssid"] = v + break + except ValueError: + continue + else: + # Secondary BSS - find in BSS array + bss_idx = None + for line in data.splitlines(): + if "=" in line: + try: + k, v = line.split("=", 1) + if v == ifname and k.startswith("bss["): + bss_idx = k[4:-1] # Extract index from bss[N] + break + except ValueError: + continue + + if bss_idx: + for line in data.splitlines(): + if "=" in line: + try: + k, v = line.split("=", 1) + if k == f"ssid[{bss_idx}]": + ap_data["ssid"] = v + break + except ValueError: + continue + + stations_data = HOST.run(tuple(f"iw dev {ifname} station dump".split()), default="") + stations = parse_iw_stations(stations_data) + + if stations: + ap_data["stations"] = { + "station": stations + } + + except Exception: + pass + + # Nest data inside access-point container to match YANG schema + return { + "access-point": ap_data + } if ap_data else {} + + +def parse_iw_stations(output): + """Parse iw station dump output to get connected stations""" + stations = [] + current_station = None + + for line in output.splitlines(): + line = line.strip() + + # Station line: "Station aa:bb:cc:dd:ee:ff (on wifiX)" + if line.startswith("Station "): + if current_station: + stations.append(current_station) + # Extract MAC address + parts = line.split() + if len(parts) >= 2: + current_station = { + "mac-address": parts[1].lower() + } + elif current_station: + # Parse station attributes + try: + # Lines are in format "key: value" with tabs + if ":" not in line: + continue + + parts = line.split(":", 1) + key = parts[0].strip() + value = parts[1].strip() + + if key == "signal": + # Format: "-42 dBm" or "-42 [-44] dBm" + rssi = int(value.split()[0]) + current_station["rssi"] = rssi + elif key == "connected time": + # Format: "123 seconds" + seconds = int(value.split()[0]) + current_station["connected-time"] = seconds + elif key == "rx packets": + current_station["rx-packets"] = int(value) + elif key == "tx packets": + current_station["tx-packets"] = int(value) + elif key == "rx bytes": + current_station["rx-bytes"] = int(value) + elif key == "tx bytes": + current_station["tx-bytes"] = int(value) + elif key == "tx bitrate": + # Format: "866.7 MBit/s ..." - extract speed and convert to 100kbit/s units + speed_mbps = float(value.split()[0]) + current_station["tx-speed"] = int(speed_mbps * 10) + elif key == "rx bitrate": + # Format: "780.0 MBit/s ..." - extract speed and convert to 100kbit/s units + speed_mbps = float(value.split()[0]) + current_station["rx-speed"] = int(speed_mbps * 10) + except (ValueError, KeyError, IndexError): + # Skip invalid values + continue + + # Add last station + if current_station: + stations.append(current_station) + + return stations + + +def wifi_station(ifname): + """Get operational data for Station mode using wpa_cli""" + station_data = {} + + try: + data = HOST.run(tuple(f"wpa_cli -i {ifname} status".split()), default="") if data != "": for line in data.splitlines(): try: if "=" not in line: continue - k,v = line.split("=", 1) + k, v = line.split("=", 1) if k == "ssid": - wifi_data["ssid"] = v - if k == "wpa_state" and v == "DISCONNECTED": # wpa_suppicant has most likely restarted, restart scanning - HOST.run(tuple(f"wpa_cli -i {ifname} scan".split()), default="") + station_data["ssid"] = v except ValueError: # Skip malformed lines continue try: - data=HOST.run(tuple(f"wpa_cli -i {ifname} signal_poll".split()), default="FAIL") + data = HOST.run(tuple(f"wpa_cli -i {ifname} signal_poll".split()), default="FAIL") - # signal_poll return FAIL not connected + # signal_poll return FAIL if not connected if data.strip() != "FAIL": for line in data.splitlines(): try: if "=" not in line: continue - k,v = line.strip().split("=", 1) + k, v = line.strip().split("=", 1) if k == "RSSI": - wifi_data["rssi"]=int(v) + station_data["rssi"] = int(v) except (ValueError, KeyError): # Skip malformed lines or invalid integers continue @@ -45,14 +220,28 @@ def wifi(ifname): pass try: - data=HOST.run(tuple(f"wpa_cli -i {ifname} scan_result".split()), default="FAIL") + data = HOST.run(tuple(f"wpa_cli -i {ifname} scan_result".split()), default="FAIL") if data != "FAIL": - wifi_data["scan-results"] = parse_wpa_scan_result(data) + scan_results = parse_wpa_scan_result(data) + if scan_results: + station_data["scan-results"] = scan_results except Exception: # If scan results fail, just omit them pass - return wifi_data + # Always nest data inside station container to match YANG schema + # In scan-only mode, this will be just scan-results with no ssid/rssi + return {"station": station_data} if station_data else {} + + +def wifi(ifname): + """Main entry point - detect mode and return appropriate data""" + mode = detect_wifi_mode(ifname) + + if mode == 'ap': + return wifi_ap(ifname) + else: + return wifi_station(ifname) def parse_wpa_scan_result(scan_output): @@ -106,7 +295,7 @@ def parse_wpa_scan_result(scan_output): # Convert to list and sort by RSSI (best first) result = list(networks.values()) - result.sort(key=lambda x: x['rssi'], reverse=False) + result.sort(key=lambda x: x['rssi'], reverse=True) return result