Add iw link endpoint

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)

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"))