Alternative v6 XM approach

.gitignore

@@ -12,3 +12,4 @@ tests/__pycache__
 tmp
 todo
 .DS_Store
+test.py

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.5.7] - 2025-10-20
+
+### Added
+
+- Support for v6 firmware XM models using a different login path (XW already was successful)
+- Calculated cpuload on v6 if values available (to prevent reporting close to 100%)
+- Fix frequency on v6 firmware (if value is cast as a string ending in MHz)
+- Added tx/rx rates for v6
+
 ## [0.5.6] - 2025-10-11
 
 ### Added

airos/base.py

@@ -55,6 +55,8 @@ def __init__(
         self.username = username
         self.password = password
 
+        self.api_version: int = 8
+
         parsed_host = urlparse(host)
         scheme = (
             parsed_host.scheme
@@ -74,11 +76,13 @@ def __init__(
         self.current_csrf_token: str | None = None
 
         # Mostly 8.x API endpoints, login/status are the same in 6.x
-        self._login_urls = {
-            "default": f"{self.base_url}/api/auth",
-            "v6_alternative": f"{self.base_url}/login.cgi",
-        }
+        self._login_url = f"{self.base_url}/api/auth"
         self._status_cgi_url = f"{self.base_url}/status.cgi"
+
+        # Presumed 6.x XM only endpoint
+        self._v6_xm_login_url = f"{self.base_url}/login.cgi"
+        self._v6_form_url = "/index.cgi"
+
         # Presumed 8.x only endpoints
         self._stakick_cgi_url = f"{self.base_url}/stakick.cgi"
         self._provmode_url = f"{self.base_url}/api/provmode"
@@ -88,6 +92,8 @@ def __init__(
         self._download_progress_url = f"{self.base_url}/api/fw/download-progress"
         self._install_url = f"{self.base_url}/fwflash.cgi"
 
+        self._login_urls = [self._login_url, self._v6_xm_login_url]
+
     @staticmethod
     def derived_wireless_data(
         derived: dict[str, Any], response: dict[str, Any]
@@ -129,7 +135,7 @@ def _derived_data_helper(
             sku = UispAirOSProductMapper().get_sku_by_devmodel(devmodel)
         except KeyError:
             _LOGGER.warning(
-                "Unknown SKU/Model ID for %s. Please report at "
+                "Unknown SKU/Model ID for '%s'. Please report at "
                 "https://github.com/CoMPaTech/python-airos/issues so we can add support.",
                 devmodel,
             )
@@ -204,7 +210,8 @@ def _get_authenticated_headers(
             headers["X-CSRF-ID"] = self._csrf_id
 
         if self._auth_cookie:  # pragma: no cover
-            headers["Cookie"] = f"AIROS_{self._auth_cookie}"
+            # headers["Cookie"] = f"AIROS_{self._auth_cookie}"
+            headers["Cookie"] = self._auth_cookie
 
         return headers
 
@@ -218,7 +225,8 @@ def _store_auth_data(self, response: aiohttp.ClientResponse) -> None:
             cookie.load(set_cookie)
         for key, morsel in cookie.items():
             if key.startswith("AIROS_"):
-                self._auth_cookie = morsel.key[6:] + "=" + morsel.value
+                # self._auth_cookie = morsel.key[6:] + "=" + morsel.value
+                self._auth_cookie = f"{morsel.key}={morsel.value}"
                 break
 
     async def _request_json(
@@ -243,7 +251,7 @@ async def _request_json(
             request_headers.update(headers)
 
         try:
-            if url not in self._login_urls.values() and not self.connected:
+            if url not in self._login_urls and not self.connected:
                 _LOGGER.error("Not connected, login first")
                 raise AirOSDeviceConnectionError from None
 
@@ -259,7 +267,7 @@ async def _request_json(
                 _LOGGER.debug("Successfully fetched JSON from %s", url)
 
                 # If this is the login request, we need to store the new auth data
-                if url in self._login_urls.values():
+                if url in self._login_urls:
                     self._store_auth_data(response)
                     self.connected = True
 
@@ -283,32 +291,68 @@ async def _request_json(
             _LOGGER.warning("Request to %s was cancelled", url)
             raise
 
+    async def _login_v6(self) -> None:
+        """Login to airOS v6 (XM) devices."""
+        # Handle session cookie from login url
+        async with self.session.request(
+            "GET",
+            self._v6_xm_login_url,
+            allow_redirects=False,
+        ) as response:
+            session_cookie = next(
+                (c for n, c in response.cookies.items() if n.startswith("AIROS")), None
+            )
+            if not session_cookie:
+                raise AirOSDeviceConnectionError("No session cookie received.")
+            self._auth_cookie = f"{session_cookie.key}={session_cookie.value}"
+
+        # Handle login expecting 302 redirect
+        payload = {
+            "username": self.username,
+            "password": self.password,
+            "uri": self._v6_form_url,
+        }
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Origin": self.base_url,
+            "Referer": self._v6_xm_login_url,
+            "Cookie": self._auth_cookie,
+        }
+        async with self.session.request(
+            "POST",
+            self._v6_xm_login_url,
+            data=payload,
+            headers=headers,
+            allow_redirects=False,
+        ) as response:
+            if response.status != 302:
+                raise AirOSConnectionAuthenticationError("Login failed.")
+
+        # Activate session by accessing the form URL
+        headers = {"Referer": self._v6_xm_login_url, "Cookie": self._auth_cookie}
+        async with self.session.request(
+            "GET",
+            f"{self.base_url}{self._v6_form_url}",
+            headers=headers,
+            allow_redirects=True,
+        ) as response:
+            if "login.cgi" in str(response.url):
+                raise AirOSConnectionAuthenticationError("Session activation failed.")
+        self.connected = True
+        self.api_version = 6
+
     async def login(self) -> None:
         """Login to AirOS device."""
         payload = {"username": self.username, "password": self.password}
         try:
-            await self._request_json(
-                "POST", self._login_urls["default"], json_data=payload
-            )
+            await self._request_json("POST", self._login_url, json_data=payload)
         except AirOSUrlNotFoundError:
-            pass  # Try next URL
+            await self._login_v6()
         except AirOSConnectionSetupError as err:
             raise AirOSConnectionSetupError("Failed to login to AirOS device") from err
         else:
             return
 
-        try:  # Alternative URL
-            await self._request_json(
-                "POST",
-                self._login_urls["v6_alternative"],
-                form_data=payload,
-                ct_form=True,
-            )
-        except AirOSConnectionSetupError as err:
-            raise AirOSConnectionSetupError(
-                "Failed to login to default and alternate AirOS device urls"
-            ) from err
-
     async def status(self) -> AirOSDataModel:
         """Retrieve status from the device."""
         response = await self._request_json(

airos/data.py

@@ -246,11 +246,20 @@ class Host6(AirOSDataClass):
     totalram: int
     freeram: int
     cpuload: float | int | None
+    cputotal: float | int | None  # Reported on XM firmware
+    cpubusy: float | int | None  # Reported on XM firmware
 
     @classmethod
     def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
         """Pre-deserialize hook for Host."""
         _check_and_log_unknown_enum_value(d, "netrole", NetRole, "Host", "netrole")
+
+        # Calculate cpufloat from actuals instead on relying on near 100% value
+        if (
+            all(isinstance(d.get(k), (int, float)) for k in ("cpubusy", "cputotal"))
+            and d["cputotal"] > 0
+        ):
+            d["cpuload"] = round((d["cpubusy"] / d["cputotal"]) * 100, 2)
         return d
 
 
@@ -562,7 +571,7 @@ class Wireless6(AirOSDataClass):
     apmac: str
     countrycode: int
     channel: int
-    frequency: str
+    frequency: int
     dfs: int
     opmode: str
     antenna: str
@@ -574,8 +583,8 @@ class Wireless6(AirOSDataClass):
     ack: int
     distance: int  # In meters
     ccq: int
-    txrate: str
-    rxrate: str
+    txrate: int
+    rxrate: int
     security: Security
     qos: str
     rstatus: int
@@ -584,6 +593,7 @@ class Wireless6(AirOSDataClass):
     wds: int
     aprepeater: int  # Not bool as v8
     chanbw: int
+    throughput: Throughput
     mode: Wireless6Mode | None = None
 
     @classmethod
@@ -593,6 +603,18 @@ def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
         _check_and_log_unknown_enum_value(
             d, "security", Security, "Wireless", "security"
         )
+
+        freq = d.get("frequency")
+        if isinstance(freq, str) and "MHz" in freq:
+            d["frequency"] = int(freq.split()[0])
+
+        rxrate = d.get("rxrate")
+        txrate = d.get("txrate")
+        d["throughput"] = {
+            "rx": int(float(rxrate)) if rxrate else 0,
+            "tx": int(float(txrate)) if txrate else 0,
+        }
+
         return d
 
 

airos/model_map.py

@@ -2,8 +2,8 @@
 
 from .exceptions import AirOSMultipleMatchesFoundException
 
-MODELS: dict[str, str] = {
-    # Generated list from https://store.ui.com/us/en/category/wireless
+# Generated list from https://store.ui.com/us/en/category/wireless
+SITE_MODELS: dict[str, str] = {
     "Wave MLO5": "Wave-MLO5",
     "airMAX Rocket Prism 5AC": "RP-5AC-Gen2",
     "airFiber 5XHD": "AF-5XHD",
@@ -76,10 +76,16 @@
     "airMAX 5 GHz, 19/20 dBi Sector": "AM-5G2",
     "airMAX 2.4 GHz, 10 dBi Omni": "AMO-2G10",
     "airMAX 2.4 GHz, 15 dBi, 120º Sector": "AM-2G15-120",
-    # Manually added entries for common unofficial names
+}
+
+# Manually added entries for common unofficial names
+MANUAL_MODELS: dict[str, str] = {
     "LiteAP GPS": "LAP-GPS",  # Shortened name for airMAX Lite Access Point GPS
+    "NanoStation loco M5": "LocoM5",  # XM firmware version 6 - note the reversed names
 }
 
+MODELS: dict[str, str] = {**SITE_MODELS, **MANUAL_MODELS}
+
 
 class UispAirOSProductMapper:
     """Utility class to map product model names to SKUs and vice versa."""
@@ -90,6 +96,7 @@ def __init__(self) -> None:
 
     def get_sku_by_devmodel(self, devmodel: str) -> str:
         """Retrieves the SKU for a given device model name."""
+        devmodel = devmodel.strip()
         if devmodel in MODELS:
             return MODELS[devmodel]
 

fixtures/airos_NanoStation_M5_sta_v6.3.16.json

@@ -10,7 +10,7 @@
     "ptmp": false,
     "ptp": true,
     "role": "station",
-    "sku": "LocoM5",
+    "sku": "NSM5",
     "station": true
   },
   "firewall": {
@@ -21,7 +21,9 @@
   },
   "genuine": "/images/genuine.png",
   "host": {
-    "cpuload": 24.0,
+    "cpubusy": 3786414,
+    "cpuload": 25.51,
+    "cputotal": 14845531,
     "devmodel": "NanoStation M5 ",
     "freeram": 42516480,
     "fwprefix": "XW",
@@ -140,7 +142,7 @@
     "dfs": 0,
     "distance": 750,
     "essid": "Nano",
-    "frequency": "5180 MHz",
+    "frequency": 5180,
     "hide_essid": 0,
     "mode": "sta",
     "noisef": -99,
@@ -149,11 +151,15 @@
     "qos": "No QoS",
     "rssi": 32,
     "rstatus": 5,
-    "rxrate": "216",
+    "rxrate": 216,
     "security": "WPA2",
     "signal": -64,
+    "throughput": {
+      "rx": 216,
+      "tx": 270
+    },
     "txpower": 24,
-    "txrate": "270",
+    "txrate": 270,
     "wds": 1
   }
 }