diff --git a/wlanpi_core/api/api_v1/endpoints/network_api.py b/wlanpi_core/api/api_v1/endpoints/network_api.py index c7919f7..b7efc90 100644 --- a/wlanpi_core/api/api_v1/endpoints/network_api.py +++ b/wlanpi_core/api/api_v1/endpoints/network_api.py @@ -14,7 +14,11 @@ SupplicantNetwork, ) from wlanpi_core.services import network_ethernet_service, network_service -from wlanpi_core.utils.network import list_ethernet_interfaces, list_wlan_interfaces +from wlanpi_core.utils.network import ( + list_ethernet_interfaces, + list_wlan_interfaces, + get_iw_link, +) router = APIRouter() @@ -488,3 +492,23 @@ async def get_interface_details(interface: Optional[str] = None): except Exception as ex: log.error(ex) return Response(content="Internal Server Error", status_code=500) + + +@router.get( + "/wlan/{interface}/link", + response_model=None, + response_model_exclude_none=True, +) +async def get_interface_link_details(interface: str): + """ + Gets interface details via iw. + """ + + try: + validate_wlan_interface(interface, required=True) + return get_iw_link(interface) + except ValidationError as ve: + return Response(content=ve.error_msg, status_code=ve.status_code) + except Exception as ex: + log.error(ex) + return Response(content="Internal Server Error", status_code=500) diff --git a/wlanpi_core/utils/network.py b/wlanpi_core/utils/network.py index 6ae16a2..9f88eab 100644 --- a/wlanpi_core/utils/network.py +++ b/wlanpi_core/utils/network.py @@ -397,3 +397,61 @@ def get_interface_details( def get_interface_mac(interface: str) -> str: return run_command(["jc", "ifconfig", interface]).output_from_json()[0]["mac_addr"] + + +def get_iw_link(interface: str) -> dict[str, Union[bool, str, int, float, None]]: + lines = run_command(["iw", "dev", interface, "link"]).stdout.split("\n") + state = lines[0].strip() + data: dict[str, Union[bool, str, int, float, None]] = { + "connected": state != "Not connected.", + "ssid": None, # SSID # Kronos-5 + "bssid": None, # from state line + "freq": None, # freq # 5220.0 + # "rx": None, # RX # 402298 bytes (2063 packets) + "rx_bytes": None, + "rx_packets": None, + # "tx": None, # TX # 19503 bytes (137 packets) + "tx_bytes": None, + "tx_packets": None, + # "signal": None, # signal # -70 dBm + "signal_dbm": None, + "rx_bitrate": None, # rx bitrate # 54.0 MBit/s VHT-MCS 1 40MHz VHT-NSS 2 + "tx_bitrate": None, # tx bitrate # 351.0 MBit/s VHT-MCS 4 80MHz VHT-NSS 2 + "bss_flags": None, # bss flags # short-slot-time + "dtim_period": None, # dtim period # 3 + "beacon_int": None, # beacon int # 100 + } + + if not data["connected"]: + return data + + # Populate the lines as much as possible + data["bssid"] = lines[0].strip().split(" ")[2] + for line in [x for x in lines[1:] if x != ""]: + key, val = line.split(":", 1) + data[key.strip().lower().replace(" ", "_")] = val.strip() + + # Rebuild him better, faster, stronger + data["freq"] = float(data["freq"]) + data["signal_dbm"] = int(data["signal"].split()[0]) + del data["signal"] + + rx_split = data["rx"].split() + data["rx_bytes"] = int(rx_split[0]) + data["rx_packets"] = int(rx_split[2][1:]) + del data["rx"] + + tx_split = data["tx"].split() + data["tx_bytes"] = int(tx_split[0]) + data["tx_packets"] = int(tx_split[2][1:]) + del data["tx"] + + data["dtim_period"] = int(data["dtim_period"]) + data["beacon_int"] = int(data["beacon_int"]) + + return data + + +if __name__ == "__main__": + print(get_iw_link("wlan0")) + print(get_iw_link("wlan1"))