Add basic controller firmware update information

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "tplink_omada_client"
-version = "1.5.4"
+version = "1.5.5"
 authors = [
   { name="Mark Godwin", email="10632972+MarkGodwin@users.noreply.github.com" },
 ]

src/tplink_omada_client/__init__.py

@@ -11,6 +11,12 @@
     PortProfileOverrides,
     SwitchPortSettings,
 )
+from .definitions import (
+    OmadaControllerUpdateInfo,
+    OmadaHardwareUpgradeStatus,
+    OmadaHardwareUpdateInfo,
+)
+
 from . import definitions
 from . import exceptions
 from . import clients
@@ -19,6 +25,9 @@
     "OmadaClient",
     "OmadaSite",
     "OmadaSiteClient",
+    "OmadaControllerUpdateInfo",
+    "OmadaHardwareUpgradeStatus",
+    "OmadaHardwareUpdateInfo",
     "AccessPointPortSettings",
     "GatewayPortSettings",
     "OmadaClientSettings",

src/tplink_omada_client/cli/__init__.py

@@ -16,6 +16,7 @@
     command_clients,
     command_default,
     command_devices,
+    command_firmware,
     command_gateway,
     command_known_clients,
     command_poe,
@@ -53,6 +54,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     command_clients.arg_parser(subparsers)
     command_default.arg_parser(subparsers)
     command_devices.arg_parser(subparsers)
+    command_firmware.arg_parser(subparsers)
     command_gateway.arg_parser(subparsers)
     command_known_clients.arg_parser(subparsers)
     command_poe.arg_parser(subparsers)

src/tplink_omada_client/cli/command_firmware.py

@@ -0,0 +1,52 @@
+"""Implementation for 'firmware' command"""
+
+from argparse import ArgumentParser
+from .config import get_target_config, to_omada_connection
+from .util import dump_raw_data, get_target_argument
+
+
+async def command_firmware(args) -> int:
+    """Executes 'firmware' command"""
+    controller = get_target_argument(args)
+    config = get_target_config(controller)
+
+    async with to_omada_connection(config) as client:
+        controller_updates = await client.check_firmware_updates()
+        if controller_updates.hardware:
+            hardware_update = controller_updates.hardware
+            status = "\u2757 UPDATE" if hardware_update.upgrade else "\u2713 UP-TO-DATE"
+            print(f"{'Controller':<30} {hardware_update.current_version:<36} {hardware_update.latest_version:<36} {status}")
+            if hardware_update.upgrade and hardware_update.release_notes and args["release_notes"]:
+                print(f"  Release Notes: {hardware_update.release_notes}")
+        else:
+            print("No controller firmware updates available.")
+
+        dump_raw_data(args, controller_updates)
+
+        site_client = await client.get_site_client(config.site)
+        devices = await site_client.get_devices()
+        if not devices:
+            print("No devices found in the specified site.")
+            return 0
+
+        for device in devices:
+            firmware = await site_client.get_firmware_details(device)
+            status = "\u2757 UPDATE" if firmware.current_version != firmware.latest_version else "\u2713 UP-TO-DATE"
+            print(f"{device.name:<30} {firmware.current_version:<36} {firmware.latest_version:<36} {status}")
+            if firmware.current_version != firmware.latest_version and firmware.release_notes and args["release_notes"]:
+                print(f"  Release Notes: {firmware.release_notes}")
+
+            dump_raw_data(args, firmware)
+
+    return 0
+
+
+def arg_parser(subparsers) -> None:
+    """Configures arguments parser for 'firmware' command"""
+    firmware_parser: ArgumentParser = subparsers.add_parser(
+        "firmware", help="Shows firmware information and update availability for the specified device"
+    )
+    firmware_parser.set_defaults(func=command_firmware)
+
+    firmware_parser.add_argument("-d", "--dump", help="Output raw firmware information", action="store_true")
+    firmware_parser.add_argument("-rn", "--release-notes", help="Show release notes for firmware updates", action="store_true")

src/tplink_omada_client/definitions.py

@@ -277,3 +277,65 @@ class LedSetting(IntEnum):
     @classmethod
     def _missing_(cls, _):
         return LedSetting.UNKNOWN
+
+
+class OmadaHardwareUpdateInfo(OmadaApiData):
+    """Information about available hardware firmware updates."""
+
+    @property
+    def upgrade(self) -> bool:
+        """Whether a firmware upgrade is available."""
+        return self._data["upgrade"]
+
+    @property
+    def latest_version(self) -> str:
+        """The latest available firmware version."""
+        return self._data.get("latestVersion", self.current_version)
+
+    @property
+    def current_version(self) -> str:
+        """The currently installed firmware version."""
+        return self._data["currentVersion"]
+
+    @property
+    def release_notes(self) -> str | None:
+        """Release notes for the latest firmware version."""
+        return self._data.get("fwReleaseLog", None)
+
+
+class OmadaControllerUpdateInfo(OmadaApiData):
+    """Information about available controller and device firmware updates."""
+
+    @property
+    def hardware(self) -> OmadaHardwareUpdateInfo | None:
+        """Information about available hardware controller firmware updates."""
+        return OmadaHardwareUpdateInfo(self._data["hardware"]) if "hardware" in self._data else None
+
+
+class OmadaHardwareUpgradeStatus(OmadaApiData):
+    """Information about the status of a hardware controller firmware upgrade."""
+
+    @property
+    def upgrade_status(self) -> int:
+        """The current status of the firmware upgrade process."""
+        return self._data["upgradeStatus"]
+
+    @property
+    def upgrade_msg(self) -> str:
+        """Any message associated with the current firmware upgrade status."""
+        return self._data.get("upgradeMsg", "")
+
+    @property
+    def upgrade_time(self) -> int:
+        """How often to refresh to watch the download progress?"""
+        return self._data.get("upgradeTime", 0)
+
+    @property
+    def download_progress(self) -> int:
+        """The current progress of the firmware download, as a percentage."""
+        return self._data.get("downloadProgress", 0)
+
+    @property
+    def reboot_time(self) -> int:
+        """How long after the donwload completes before we expect the controller to come back online, in seconds."""
+        return self._data.get("rebootTime", 300)

src/tplink_omada_client/devices.py

@@ -51,7 +51,7 @@ def name(self) -> str:
     @property
     def model(self) -> str:
         """The device model, such as EAP225."""
-        return self._data("model", "Unknown")
+        return self._data.get("model", "Unknown")
 
     @property
     def model_display_name(self) -> str:
@@ -723,7 +723,7 @@ def current_version(self) -> str:
     @property
     def latest_version(self) -> str:
         """Latest firmware version available."""
-        return self._data["lastFwVer"]
+        return self._data.get("lastFwVer", self.current_version)
 
     @property
     def release_notes(self) -> str:

src/tplink_omada_client/omadaclient.py

@@ -7,6 +7,7 @@
 from awesomeversion import AwesomeVersion
 from multidict import CIMultiDict
 
+from .definitions import OmadaControllerUpdateInfo, OmadaHardwareUpgradeStatus
 from .omadasiteclient import OmadaSiteClient
 from .omadaapiconnection import OmadaApiConnection
 
@@ -141,3 +142,25 @@ async def set_certificate(self, file: str, cert_password: str):
         }
         url = self._api.format_url("controller/setting")
         await self._api.request("patch", url, json=payload)
+
+    async def check_firmware_updates(self) -> OmadaControllerUpdateInfo:
+        """Check if firmware updates are available for the Omada hardware controller
+
+        Software Controller updates are probably not available via this API
+        """
+        result = await self._api.request("get", self._api.format_url("controller/notification/updateInfo"))
+
+        return OmadaControllerUpdateInfo(result)
+
+    async def upgrade_controller_firmware(self, target_version: str) -> bool:
+        """Upgrade the Omada hardware controller firmware to the specified version."""
+        url = self._api.format_url("cmd/upgradeFirmware")
+        payload = {"targetVersion": target_version}
+        await self._api.request("post", url, json=payload)
+        return True
+
+    async def get_controller_upgrade_status(self) -> OmadaHardwareUpgradeStatus:
+        """Get the status of an ongoing hardware controller firmware upgrade."""
+        url = self._api.format_url("maintenance/hardware/upgradeStatus")
+        result = await self._api.request("get", url)
+        return OmadaHardwareUpgradeStatus(result)