mirror of
https://github.com/kernelkit/infix.git
synced 2026-08-03 14:23:02 +02:00
WiFi: Refactor statd (python) implementation
* Add support for AP (list connected stations) * Add scan-mode (Scan without create a fully configured station) Signed-off-by: Mattias Walström <lazzer@gmail.com>
This commit is contained in:
Mattias Walström
@@ -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":
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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() +
|
||||
[],
|
||||
},
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user