diff --git a/pyproject.toml b/pyproject.toml index 6b62f86..c6c7413 100644 --- a/pyproject.toml +++ b/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" }, ] diff --git a/src/tplink_omada_client/__init__.py b/src/tplink_omada_client/__init__.py index c0940bb..629bc0e 100644 --- a/src/tplink_omada_client/__init__.py +++ b/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", diff --git a/src/tplink_omada_client/cli/__init__.py b/src/tplink_omada_client/cli/__init__.py index 7fa4b4e..47ace33 100644 --- a/src/tplink_omada_client/cli/__init__.py +++ b/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) diff --git a/src/tplink_omada_client/cli/command_firmware.py b/src/tplink_omada_client/cli/command_firmware.py new file mode 100644 index 0000000..4918c81 --- /dev/null +++ b/src/tplink_omada_client/cli/command_firmware.py @@ -0,0 +1,49 @@ +"""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() + + 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 all devices" + ) + 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") diff --git a/src/tplink_omada_client/definitions.py b/src/tplink_omada_client/definitions.py index cacb4f4..ad2a5f7 100644 --- a/src/tplink_omada_client/definitions.py +++ b/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 download completes before we expect the controller to come back online, in seconds.""" + return self._data.get("rebootTime", 300) diff --git a/src/tplink_omada_client/devices.py b/src/tplink_omada_client/devices.py index 3ecfe65..070c2c6 100644 --- a/src/tplink_omada_client/devices.py +++ b/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,12 +723,12 @@ 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: """Release notes for the new firmware.""" - return self._data["fwReleaseLog"] + return self._data.get("fwReleaseLog", "") class OmadaGatewayPortStatus(OmadaApiData, OmadaPortStatus): diff --git a/src/tplink_omada_client/omadaclient.py b/src/tplink_omada_client/omadaclient.py index e65befb..3320953 100644 --- a/src/tplink_omada_client/omadaclient.py +++ b/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) diff --git a/uv.lock b/uv.lock index d69d067..c80b1cb 100644 --- a/uv.lock +++ b/uv.lock @@ -499,7 +499,7 @@ wheels = [ [[package]] name = "tplink-omada-client" -version = "1.5.4" +version = "1.5.5" source = { editable = "." } dependencies = [ { name = "aiohttp" },