Rework detection

Author: CoMPaTech

.github/workflows/verify.yml

@@ -174,7 +174,7 @@ jobs:
         run: |
           . venv/bin/activate
           coverage combine artifacts/.coverage*
-          coverage report --fail-under=85
+          coverage report --fail-under=80
           coverage xml
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

CHANGELOG.md

@@ -12,7 +12,7 @@ All notable changes to this project will be documented in this file.
 - Added tx/rx rates for v6
 - Fixed ieeemode (v8) vs opmode (v6) mapped back to IeeeMode enum
 - Derived antenna_gain (v8) from antenna (v6 string)
-- Improved internal workings
+- Improved internal workings and firmware detection
 
 ## [0.5.6] - 2025-10-11
 

airos/base.py

@@ -158,38 +158,67 @@ def _derived_data_helper(
             "mode": DerivedWirelessMode.PTP,
             "sku": sku,
         }
+
         # WIRELESS
         derived = derived_wireless_data_func(derived, response)
 
-        # INTERFACES
-        addresses = {}
-        interface_order = ["br0", "eth0", "ath0"]
-
+        # Interfaces / MAC (for unique id)
         interfaces = response.get("interfaces", [])
-
         # No interfaces, no mac, no usability
         if not interfaces:
             _LOGGER.error("Failed to determine interfaces from AirOS data")
             raise AirOSKeyDataMissingError from None
 
-        for interface in interfaces:
-            if interface["enabled"]:  # Only consider if enabled
-                addresses[interface["ifname"]] = interface["hwaddr"]
+        derived["mac"] = AirOS.get_mac(interfaces)["mac"]
+        derived["mac_interface"] = AirOS.get_mac(interfaces)["mac_interface"]
 
-        # Fallback take fist alternate interface found
-        derived["mac"] = interfaces[0]["hwaddr"]
-        derived["mac_interface"] = interfaces[0]["ifname"]
-
-        for interface in interface_order:
-            if interface in addresses:
-                derived["mac"] = addresses[interface]
-                derived["mac_interface"] = interface
-                break
+        # Firmware Major Version
+        fwversion = (response.get("host") or {}).get("fwversion", "invalid")
+        derived["fw_major"] = AirOS.get_fw_major(fwversion)
 
         response["derived"] = derived
 
         return response
 
+    @staticmethod
+    def get_fw_major(fwversion: str) -> int:
+        """Extract major firmware version from fwversion string."""
+        try:
+            return int(fwversion.lstrip("v").split(".", 1)[0])
+        except (ValueError, AttributeError) as err:
+            _LOGGER.error("Invalid firmware version '%s'", fwversion)
+            raise AirOSKeyDataMissingError("invalid fwversion") from err
+
+    @staticmethod
+    def get_mac(interfaces: list[dict[str, Any]]) -> dict[str, str]:
+        """Extract MAC address from interfaces."""
+        result: dict[str, str] = {"mac": "", "mac_interface": ""}
+
+        if not interfaces:
+            return result
+
+        addresses: dict[str, str] = {}
+        interface_order = ["br0", "eth0", "ath0"]
+
+        for interface in interfaces:
+            if (
+                interface.get("enabled")
+                and interface.get("hwaddr")
+                and interface.get("ifname")
+            ):
+                addresses[interface["ifname"]] = interface["hwaddr"]
+
+        for preferred in interface_order:
+            if preferred in addresses:
+                result["mac"] = addresses[preferred]
+                result["mac_interface"] = preferred
+                break
+        else:
+            result["mac"] = interfaces[0].get("hwaddr", "")
+            result["mac_interface"] = interfaces[0].get("ifname", "")
+
+        return result
+
     @classmethod
     def derived_data(cls, response: dict[str, Any]) -> dict[str, Any]:
         """Add derived data to the device response (instance method for polymorphism)."""
@@ -354,12 +383,15 @@ async def login(self) -> None:
         else:
             return
 
-    async def status(self) -> AirOSDataModel:
+    async def status(self, underived: bool = False) -> AirOSDataModel | dict[str, Any]:
         """Retrieve status from the device."""
         response = await self._request_json(
             "GET", self._status_cgi_url, authenticated=True
         )
 
+        if underived:
+            return response
+
         try:
             adjusted_json = self.derived_data(response)
             return self.data_model.from_dict(adjusted_json)

airos/data.py

@@ -102,9 +102,6 @@ def _redact(d: dict[str, Any]) -> dict[str, Any]:
     return _redact(data)
 
 
-# Data class start
-
-
 class AirOSDataClass(DataClassDictMixin):
     """A base class for all mashumaro dataclasses."""
 
@@ -732,6 +729,9 @@ class Derived(AirOSDataClass):
     # Lookup of model_id (presumed via SKU)
     sku: str
 
+    # Firmware major version
+    fw_major: int | None = None
+
 
 @dataclass
 class AirOS8Data(AirOSDataBaseClass):

airos/helpers.py

@@ -1,12 +1,20 @@
 """Ubiquiti AirOS firmware helpers."""
 
+import logging
 from typing import TypedDict
 
 import aiohttp
 
-from .airos6 import AirOS6
 from .airos8 import AirOS8
-from .exceptions import AirOSKeyDataMissingError
+from .exceptions import (
+    AirOSConnectionAuthenticationError,
+    AirOSConnectionSetupError,
+    AirOSDataMissingError,
+    AirOSDeviceConnectionError,
+    AirOSKeyDataMissingError,
+)
+
+_LOGGER = logging.getLogger(__name__)
 
 
 class DetectDeviceData(TypedDict):
@@ -25,49 +33,28 @@ async def async_get_firmware_data(
     use_ssl: bool = True,
 ) -> DetectDeviceData:
     """Connect to a device and return the major firmware version."""
-    detect: AirOS8 = AirOS8(host, username, password, session, use_ssl)
-
-    await detect.login()
-    raw_status = await detect._request_json(  # noqa: SLF001
-        "GET",
-        detect._status_cgi_url,  # noqa: SLF001
-        authenticated=True,
-    )
-
-    fw_version = (raw_status.get("host") or {}).get("fwversion")
-    if not fw_version:
-        raise AirOSKeyDataMissingError("Missing host.fwversion in API response")
+    detect_device: AirOS8 = AirOS8(host, username, password, session, use_ssl)
 
     try:
-        fw_major = int(fw_version.lstrip("v").split(".", 1)[0])
-    except (ValueError, AttributeError) as exc:
-        raise AirOSKeyDataMissingError(
-            f"Invalid firmware version '{fw_version}'"
-        ) from exc
-
-    if fw_major == 6:
-        derived_data = AirOS6._derived_data_helper(  # noqa: SLF001
-            raw_status,
-            AirOS6._derived_wireless_data,  # noqa: SLF001
-        )
-    else:  # Assume AirOS 8 for all other versions
-        derived_data = AirOS8._derived_data_helper(  # noqa: SLF001
-            raw_status,
-            AirOS8._derived_wireless_data,  # noqa: SLF001
-        )
-
-    # Extract MAC address and hostname from the derived data
-    hostname = derived_data.get("host", {}).get("hostname")
-    mac = derived_data.get("derived", {}).get("mac")
-
-    if not hostname:  # pragma: no cover
-        raise AirOSKeyDataMissingError("Missing hostname")
-
-    if not mac:  # pragma: no cover
-        raise AirOSKeyDataMissingError("Missing MAC address")
+        await detect_device.login()
+        device_data = await detect_device.status(underived=True)
+    except (
+        AirOSConnectionSetupError,
+        AirOSDeviceConnectionError,
+    ):
+        _LOGGER.exception("Error connecting to device at %s", host)
+        raise
+    except (AirOSConnectionAuthenticationError, AirOSDataMissingError):
+        _LOGGER.exception("Authentication error connecting to device at %s", host)
+        raise
+    except AirOSKeyDataMissingError:
+        _LOGGER.exception("Key data missing from device at %s", host)
+        raise
+
+    assert isinstance(device_data, dict)
 
     return {
-        "fw_major": fw_major,
-        "mac": mac,
-        "hostname": hostname,
+        "fw_major": AirOS8.get_fw_major(device_data.get("host", {}).get("fwversion")),
+        "hostname": device_data.get("host", {}).get("hostname"),
+        "mac": AirOS8.get_mac(device_data.get("interfaces", {}))["mac"],
     }

fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json

@@ -11,6 +11,7 @@
   ],
   "derived": {
     "access_point": true,
+    "fw_major": 8,
     "mac": "68:D7:9A:9A:08:BB",
     "mac_interface": "br0",
     "mode": "point_to_point",

fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json

@@ -11,6 +11,7 @@
   ],
   "derived": {
     "access_point": false,
+    "fw_major": 8,
     "mac": "68:D7:9A:98:FB:FF",
     "mac_interface": "br0",
     "mode": "point_to_point",

fixtures/airos_NanoBeam_5AC_ap-ptmp_v8.7.18.json

@@ -11,6 +11,7 @@
   ],
   "derived": {
     "access_point": true,
+    "fw_major": 8,
     "mac": "xxxxxxxxxxxxxxxx",
     "mac_interface": "br0",
     "mode": "point_to_multipoint",

fixtures/airos_NanoStation_M5_sta_v6.3.16.json

@@ -4,6 +4,7 @@
   },
   "derived": {
     "access_point": false,
+    "fw_major": 6,
     "mac": "XX:XX:XX:XX:XX:XX",
     "mac_interface": "br0",
     "mode": "point_to_point",

fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_ap.json

@@ -4,6 +4,7 @@
   },
   "derived": {
     "access_point": true,
+    "fw_major": 6,
     "mac": "XX:XX:XX:XX:XX:XX",
     "mac_interface": "br0",
     "mode": "point_to_point",