""" Definitions for Omada device objects APs, Switches and Routers """ from abc import ABC, abstractmethod from typing import Any from .definitions import ( BandwidthControl, DeviceStatus, DeviceStatusCategory, Eth802Dot1X, GatewayPortMode, GatewayPortType, LinkDuplex, LinkSpeed, LinkStatus, NetworkTagsSetting, OmadaApiData, PoEMode, PortType, LedSetting, ) class OmadaDevice(OmadaApiData): """Details of a device connected to the controller""" @property def type(self) -> str: """The type of the device. Its value can be "ap", "gateway", and "switch".""" return self._data["type"] @property def resource_path(self): """The API path for querying detailed data about this device""" types = {"ap": "eaps", "gateway": "gateways", "switch": "switches"} return f"{types[self.type]}/{self.mac}" @property def mac(self) -> str: """The MAC address of the device.""" return self._data["mac"] @property def name(self) -> str: """The device name.""" return self._data.get("name", self.mac) @property def model(self) -> str: """The device model, such as EAP225.""" return self._data.get("model", "Unknown") @property def model_display_name(self) -> str: """Model description for front-end display.""" return self._data.get("showModel", "Unknown Model") @property def status(self) -> DeviceStatus: """The status of the device.""" return DeviceStatus(self._data.get("status", DeviceStatus.UNKNOWN)) @property def status_category(self) -> DeviceStatusCategory: """The high-level status of the device.""" return DeviceStatusCategory(self._data.get("statusCategory", DeviceStatusCategory.UNKNOWN)) @property def ip_address(self) -> str: """IP address of the device.""" return self._data.get("ip", "") @property def display_uptime(self) -> str | None: """Uptime of the device, as a display string""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data["uptime"] else: return None @property def cpu_usage(self) -> int: """Currenct CPU usage of the device.""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data.get("cpuUtil", 0) else: return 0 @property def mem_usage(self) -> int: """Current memory usage of the device.""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data.get("memUtil", 0) else: return 0 @property def uptime(self) -> int: """Uptime of the device, as a display string""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data["uptimeLong"] else: return 0 @property def firmware_version(self) -> str: """Firmware version of the device""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data["firmwareVersion"] else: return "Unknown" class OmadaListDevice(OmadaDevice): """An Omada Device (router, switch, eap) as represented in the device list""" @property def need_upgrade(self) -> bool: """True, if a firmware upgrade is available for the device.""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data.get("needUpgrade", False) else: return False @property def fw_download(self) -> bool: """True, if a firmware upgrade is being downloaded.""" if self._data["statusCategory"] == DeviceStatusCategory.CONNECTED: return self._data.get("fwDownload", False) else: return False class OmadaDetailedDevice(OmadaDevice): """Generic properties for Omada Devices (router, switch, eap) as returned the device-type specific endpoints""" @property def led_setting(self) -> LedSetting: """The onboard LED setting for the device""" return LedSetting(self._data.get("ledSetting", LedSetting.UNKNOWN)) class OmadaLink(OmadaApiData): """Up/Downlink connection from a switch/ap device.""" @property def mac(self) -> str: """The MAC of the linked device.""" return self._data.get("mac", self._data.get("uplinkMac")) @property def name(self) -> str: """The name of the linked device.""" return self._data["name"] @property def type(self) -> str: """The type of device linked to.""" return self._data["type"] @property def port(self) -> int: """The port's number.""" return self._data["port"] class OmadaDownlink(OmadaLink): """Downlink connection from a switch/ap port.""" @property def type(self) -> str: """The type of device downlinked to.""" return "ap" @property def model(self) -> str: """The model name of device linked to.""" return self._data["model"] class OmadaUplink(OmadaLink): """Uplink connection from a switch/ap device.""" class OmadaPortStatus(ABC): """Status information for a port.""" @property @abstractmethod def link_status(self) -> LinkStatus: """Port's link status.""" @property @abstractmethod def link_speed(self) -> LinkSpeed: """Port's link speed.""" @property @abstractmethod def bytes_tx(self) -> int: """Number of bytes transmitted by the port.""" @property @abstractmethod def bytes_rx(self) -> int: """Number of bytes received by the port.""" @property @abstractmethod def poe_active(self) -> bool: """Is the port powering a PoE device?""" class OmadaSwitchPortStatus(OmadaApiData, OmadaPortStatus): """Status information for a switch port.""" @property def link_status(self) -> LinkStatus: """Port's link status.""" return LinkStatus(self._data["linkStatus"]) @property def link_speed(self) -> LinkSpeed: """Port's link speed.""" return LinkSpeed(self._data["linkSpeed"]) @property def poe_active(self) -> bool: """Is the port powering a PoE device?""" return self._data["poe"] @property def poe_power(self) -> float | None: """Power (W) supplied over PoE.""" return self._data.get("poePower") @property def bytes_tx(self) -> int: """Number of bytes transmitted by the port.""" return self._data["tx"] @property def bytes_rx(self) -> int: """Number of bytes received by the port.""" return self._data["rx"] @property def stp_discarding(self) -> bool: """Stp blocking status in spanning tree.""" return self._data["stpDiscarding"] class OmadaSwitchPort(OmadaApiData): """Port on a switch/gateway device.""" @property def port(self) -> int: """The port's number.""" return self._data["port"] @property def name(self) -> str: """The device name.""" return self._data["name"] @property def profile_id(self) -> str: """ID of the port's config profile.""" return self._data["profileId"] @property def type(self) -> PortType: """The type of the port.""" return self._data["type"] @property def operation(self) -> str: """Port config: switching, mirroring or aggregating.""" return self._data["operation"] @property def is_disabled(self) -> bool: """Is the port disabled?""" return self._data["disable"] @property def port_status(self) -> OmadaSwitchPortStatus: """Status of the port.""" return OmadaSwitchPortStatus(self._data["portStatus"]) class OmadaSwitchDeviceCaps(OmadaApiData): """Capabilities of a switch.""" @property def poe_ports(self) -> int: """Number of PoE ports supported.""" return self._data.get("poePortNum", 0) @property def supports_poe(self) -> bool: """Is PoE supported.""" return self._data.get("poeSupport", False) @property def supports_bt(self) -> bool: """Is BT supported.""" return self._data.get("supportBt", False) class OmadaSwitch(OmadaDetailedDevice): """Details of a switch connected to the controller.""" @property def number_of_ports(self) -> int: """The number of ports on the switch.""" if "portNum" in self._data: return self._data["portNum"] # So much for the docs return self._data.get("deviceMisc", {}).get("portNum", 0) @property def ports(self) -> list[OmadaSwitchPort]: """List of ports attached to the switch.""" return [OmadaSwitchPort(p) for p in self._data.get("ports", [])] @property def uplink(self) -> OmadaUplink | None: """Uplink device for this switch.""" if "uplink" not in self._data: return None uplink = self._data["uplink"] if uplink is None: return None return OmadaUplink(uplink) @property def downlink(self) -> list[OmadaDownlink]: """Downlink devices attached to switch.""" if "downlinkList" in self._data: return [OmadaDownlink(d) for d in self._data["downlinkList"]] return [] @property def device_capabilities(self) -> OmadaSwitchDeviceCaps: """Capabilities of the switch.""" return OmadaSwitchDeviceCaps(self._data.get("devCap", {})) class OmadaAccesPointLanPortSettings(OmadaApiData): """A LAN port on an access point.""" @property def port_name(self) -> str: """Name of the port - can't be edited""" return self._data.get("lanPort", "LAN Port") @property def supports_vlan(self) -> bool: """True if the port supports VLAN tagging""" return self._data.get("supportVlan", False) @property def local_vlan_enable(self) -> bool: """True if VLAN tagging is enabled for the port explicitly""" return self._data.get("localVlanEnable", False) @property def local_vlan_id(self) -> int: """VLAN ID for this port""" return self._data.get("localVlanId", -1) @property def supports_poe(self) -> bool: """True if the port supports PoE output""" return self._data.get("supportPoe", False) @property def poe_enable(self) -> bool: """ True to enable PoE. WARNING: Do not enable PoE for EAPs powered from another EAP """ return self._data.get("poeOutEnable", False) class OmadaAccessPoint(OmadaDetailedDevice): """Details of an Access Point connected to the controller.""" @property def wireless_linked(self) -> bool: """True, if the AP is connected wirelessley.""" return self._data.get("wirelessLinked", False) @property def supports_5g(self) -> bool: """True if 5G wifi is supported""" if "deviceMisc" not in self._data: return False return self._data["deviceMisc"].get("support5g", False) @property def supports_5g2(self) -> bool: """True if 5G2 wifi is supported""" if "deviceMisc" not in self._data: return False return self._data["deviceMisc"].get("support5g2", False) @property def supports_6g(self) -> bool: """True if Wifi 6 is supported""" if "deviceMisc" not in self._data: return False return self._data["deviceMisc"].get("support6g", False) @property def supports_11ac(self) -> bool: """True if PoE is supported""" if "deviceMisc" not in self._data: return False return self._data["deviceMisc"].get("support11ac", False) @property def supports_mesh(self) -> bool: """True if mesh networking is supported""" if "deviceMisc" not in self._data: return False return self._data["deviceMisc"].get("supportMesh", False) @property def lan_port_settings(self) -> list[OmadaAccesPointLanPortSettings]: """Settings for the LAN ports on the access point""" return [OmadaAccesPointLanPortSettings(p) for p in self._data.get("lanPortSettings", [])] @property def wired_uplink(self) -> OmadaUplink | None: """Wired Uplink device for this ap.""" uplink = self._data.get("wiredUplink", None) if uplink is None: return None return OmadaUplink(uplink) class OmadaSwitchPortDetails(OmadaSwitchPort): """Full details of a port on a switch.""" @property def port_id(self) -> str: """The ID of the port""" return self._data["id"] @property def max_speed(self) -> LinkSpeed: """The max speed of the port.""" return LinkSpeed(self._data["maxSpeed"]) @property def link_speed(self) -> LinkSpeed: """The link speed of the port.""" return LinkSpeed(self._data["linkSpeed"]) @property def duplex(self) -> LinkDuplex: """The link duplex state of the port.""" return LinkDuplex(self._data["duplex"]) @property def profile_name(self) -> str: """Name of the port's config profile""" return self._data["profileName"] @property def has_profile_override(self) -> bool: """True if the port's config profile has been overridden.""" return self._data["profileOverrideEnable"] @property def supports_poe(self) -> bool: """True if the port supports PoE.""" return self._data.get("supportPoe", True) # default to true, some older versions don't report this @property def poe_mode(self) -> PoEMode: """PoE config for this port.""" # For reasons, Omada may claim Enabled on non-poe ports - probably due to port profiles return PoEMode(self._data.get("poe", PoEMode.NONE)) if self.supports_poe else PoEMode.NONE @property def bandwidth_limit_mode(self) -> BandwidthControl: """Type of bandwidth control applied.""" return BandwidthControl(self._data["bandWidthCtrlType"]) # "bandCtrl": { # "egressEnable": false, # "egressLimit": 0, # "egressUnit": 1, # "ingressEnable": false, # "ingressLimit": 0, # "ingressUnit": 1 # }, # "stormCtrl": { # "unknownUnicastEnable": false, # "unknownUnicast": 0, # "multicastEnable": false, # "multicast": 0, # "broadcastEnable": false, # "broadcast": 0, # "action": 0, # "recoverTime": 3600 # }, @property def eth_802_1x_control(self) -> Eth802Dot1X: """802.1x Auth mode""" return Eth802Dot1X(self._data["dot1x"]) @property def lldp_med_enabled(self) -> bool: """LLDP Mode""" return self._data.get("lldpMedEnable", False) @property def topology_notify_enabled(self) -> bool: """Topology notify mode""" return self._data.get("topoNotifyEnable", False) @property def spanning_tree_enabled(self) -> bool: """Spanning tree loopback control""" return self._data.get("spanningTreeEnable", False) @property def loopback_detect_enabled(self) -> bool: """Loopback detection""" return self._data.get("loopbackDetectEnable", False) @property def port_isolation_enabled(self) -> bool: """Port isolation (Danger!)""" return self._data.get("portIsolationEnable", False) @property def loopback_detect_vlan_based_enabled(self) -> bool: """VLAN based loopback detection""" return self._data.get("loopbackDetectVlanBasedEnable", False) @property def has_loopback_detect_vlan_based(self) -> bool: """True if VLAN based loopback detection settings are available for this port""" return "loopbackDetectVlanBasedEnable" in self._data @property def native_network_id(self) -> str | None: """ID of the port's native network.""" return self._data.get("nativeNetworkId", None) @property def tag_ids(self) -> list[str]: """IDs of the port's tagged networks.""" return self._data.get("tagIds", []) @property def network_tags_setting(self) -> NetworkTagsSetting: """Network tags setting for the port.""" return self._data.get("networkTagsSetting", NetworkTagsSetting.UNKNOWN) @property def tagged_network_ids(self) -> list[str]: """IDs of the port's tagged networks.""" return self._data.get("tagNetworkIds", []) @property def untagged_network_ids(self) -> list[str]: """IDs of the port's untagged networks.""" return self._data.get("untagNetworkIds", []) @property def voice_network_enabled(self) -> bool: """True if Voice Network is enabled for the port""" return self._data.get("voiceNetworkEnable", False) @property def has_voice_network(self) -> bool: """True if Voice Network settings are available for this port""" return "voiceNetworkEnable" in self._data @property def voice_network_id(self) -> str | None: """ID of the port's voice network, if any.""" return self._data.get("voiceNetworkId", None) @property def flow_control_enabled(self) -> bool: """Flow control enabled""" return self._data.get("flowControlEnable", False) @property def has_flow_control(self) -> bool: """True if Flow Control settings are available for this port""" return "flowControlEnable" in self._data @property def eee_enabled(self) -> bool: """Energy Efficient Ethernet enabled""" return self._data.get("eeeEnable", False) @property def has_eee(self) -> bool: """True if EEE settings are available for this port""" return "eeeEnable" in self._data class OmadaPortProfile(OmadaApiData): """Definition of a switch port configuration profile.""" @property def profile_id(self) -> str: """ID of this profile.""" return self._data["id"] @property def site(self) -> str: """Site which this profile is valid for.""" return self._data["site"] @property def name(self) -> str: """Name of the profile.""" return self._data["name"] @property def poe_mode(self) -> PoEMode: """PoE mode.""" return PoEMode(self._data.get("poe", PoEMode.NONE)) @property def bandwidth_limit_mode(self) -> BandwidthControl: """Type of bandwidth control applied.""" return BandwidthControl(self._data["bandWidthCtrlType"]) @property def eth_802_1x_control(self) -> Eth802Dot1X: """802.1x Auth mode""" return Eth802Dot1X(self._data["dot1x"]) @property def lldp_med_enabled(self) -> bool: """LLDP Mode""" return self._data.get("lldpMedEnable", False) @property def topology_notify_enabled(self) -> bool: """Topology notify mode""" return self._data.get("topoNotifyEnable", False) @property def spanning_tree_enabled(self) -> bool: """Spanning tree loopback control""" return self._data.get("spanningTreeEnable", False) @property def loopback_detect_enabled(self) -> bool: """Loopback detection""" return self._data.get("loopbackDetectEnable", False) @property def port_isolation_enabled(self) -> bool: """Port isolation (Danger!)""" return self._data.get("portIsolationEnable", False) @property def loopback_detect_vlan_based_enabled(self) -> bool: """VLAN based loopback detection""" return self._data.get("loopbackDetectVlanBasedEnable", False) @property def has_loopback_detect_vlan_based(self) -> bool: """True if VLAN based loopback detection settings are available for this profile""" return "loopbackDetectVlanBasedEnable" in self._data @property def flow_control_enabled(self) -> bool: """Flow control enabled""" return self._data.get("flowControlEnable", False) @property def has_flow_control(self) -> bool: """True if Flow Control settings are available for this profile""" return "flowControlEnable" in self._data @property def eee_enabled(self) -> bool: """Energy Efficient Ethernet enabled""" return self._data.get("eeeEnable", False) @property def has_eee(self) -> bool: """True if EEE settings are available for this profile""" return "eeeEnable" in self._data class OmadaInterfaceDetails(OmadaApiData): """Basic UI Information about controller.""" @property def controller_name(self) -> str: """Display name of the controller.""" return self._data["controllerName"] class OmadaFirmwareUpdate(OmadaApiData): """Status information for a switch port.""" @property def current_version(self) -> str: """Device's current firmware version.""" return self._data["curFwVer"] @property def latest_version(self) -> str: """Latest firmware version available.""" return self._data.get("lastFwVer", self.current_version) @property def release_notes(self) -> str: """Release notes for the new firmware.""" return self._data.get("fwReleaseLog", "") class OmadaGatewayPortStatus(OmadaApiData, OmadaPortStatus): """Status information for a gateway port.""" @property def port_number(self) -> int: """Port number""" return self._data["port"] @property def name(self) -> str: """Port name""" return self._data["name"] @property def display_name(self) -> str: """Port display name""" return self._data.get("portDesc", self.name) @property def type(self) -> GatewayPortType: """Type of the port - SFP, WAN, WAN/LAN or LAN only.""" return GatewayPortType(self._data["type"]) @property def mode(self) -> GatewayPortMode: """Whether the port is operating in WAN or LAN mode""" return GatewayPortMode(self._data["mode"]) @property def link_status(self) -> LinkStatus: """Low level connectivity status of the link.""" return LinkStatus(self._data["status"]) @property def bytes_tx(self) -> int: """Number of bytes transmitted by the port.""" return self._data["tx"] @property def bytes_rx(self) -> int: """Number of bytes received by the port.""" return self._data["rx"] @property def poe_active(self) -> bool: """True if the port is powering a PoE device.""" return self._data.get("poe", 0) != 0 @property def wan_connected(self) -> bool: """True if the port is connected to the internet/WAN""" return self._data.get("internetState", 0) != 0 @property def ipv6_wan_connected(self) -> bool: """True if the port is connected to the internet/WAN with IPv6""" return dict[str, Any](self._data.get("wanPortIpv6Config", {})).get("internetState", 0) != 0 @property def online_detection(self) -> bool: """True regular internet ping tests are working""" return (self.wan_connected or self.ipv6_wan_connected) and self._data.get("onlineDetection", 0) != 0 @property def ip(self) -> str | None: """DEPRECATED: The WAN IP of the port (for WAN ports only)""" return self._data.get("ip") @property def wan_ip_address(self) -> str | None: """The WAN IPv4 Address of the port (for WAN ports only)""" return self._data.get("ip") @property def wan_ipv6_enabled(self) -> bool: """The WAN IPv6 Address of the port (for WAN ports only)""" return dict[str, Any](self._data.get("wanPortIpv6Config", {})).get("enable", 0) != 0 @property def wan_ipv6_address(self) -> str | None: """The WAN IPv6 Address of the port (for WAN ports only)""" return dict[str, Any](self._data.get("wanPortIpv6Config", {})).get("addr") @property def link_speed(self) -> LinkSpeed: """The established link speed of the port""" return LinkSpeed(self._data.get("speed", LinkSpeed.SPEED_10_MBPS)) @property def link_duplex(self) -> LinkDuplex: """Actual duplex mode of the port""" return LinkDuplex(self._data.get("duplex", LinkDuplex.FULL)) @property def wan_protocol(self) -> str | None: """May be: static, dhcp, pppoe, l2tp, pptp""" return self._data.get("proto") # For connected ports # "rxPkt":217028151,"rxPktRate":35,"rxRate":6,"tx":15157457326,"txPkt":60843861,"txPktRate":24,"txRate":5, # "mirroredPorts":[] # FOR WAN PORTS: # "wanPortIpv6Config":{"enable":0,"addr":"","gateway":"","priDns":"","sndDns":"","internetState":0}, # "wanPortIpv4Config":{ # "ip":"x.x.x.x", # "gateway":"x.x.x.x", # "gateway2":"0.0.0.0", # "priDns":"194.168.4.100", # "sndDns":"194.168.8.100", # "priDns2":"0.0.0.0", # "sndDns2":"0.0.0.0" # } class OmadaGatewayPortConfig(OmadaApiData): """Configuration of a gateway port. Includes status.""" def __init__(self, data: dict, poe_enabled: bool | None): super().__init__(data) self._poe_enabled = poe_enabled @property def port_number(self) -> int: """Port number""" return self._data["port"] @property def duplex(self) -> LinkDuplex: """Configured duplex mode for the port.""" return self._data.get("duplex", LinkDuplex.AUTO) @property def link_speed(self) -> LinkSpeed: """Configured link speed for the port.""" return self._data.get("linkSpeed", LinkSpeed.SPEED_AUTO) @property def mirror_enable(self) -> bool: """True if port mirroring is enabled""" return self._data.get("mirrorEnable", False) @property def port_status(self) -> OmadaGatewayPortStatus: """Full status of the port""" return OmadaGatewayPortStatus(self._data["portStat"]) @property def poe_mode(self) -> PoEMode: """PoE mode for the port""" poe_mode_mapping = { True: PoEMode.ENABLED, False: PoEMode.DISABLED, None: PoEMode.NONE, } return poe_mode_mapping[self._poe_enabled] class OmadaGateway(OmadaDetailedDevice): """Details of an Omada Gateway device.""" @property def number_of_ports(self) -> int: """The number of ports on the switch.""" return self._data.get("portNum", 0) @property def supports_poe(self) -> bool: """True if the device supports PoE.""" return self._data.get("supportPoe", False) @property def ip(self) -> str: """Gateway's LAN IP address.""" return self._data.get("ip", "") @property def port_status(self) -> list[OmadaGatewayPortStatus]: """Status of the gateway's ports.""" return [OmadaGatewayPortStatus(p) for p in self._data.get("portStats", [])] @property def port_configs(self) -> list[OmadaGatewayPortConfig]: """Configuration of the gateway's ports. Also includes status...""" poe_data = {} if self.supports_poe: # Combined Gateway+PoE switch has this extra data poe_data = {int(x["portId"]): bool(x["enable"]) for x in self._data["poeSettings"]} return [OmadaGatewayPortConfig(p, poe_data.get(p["port"])) for p in self._data["portConfigs"]] @property def lldp_enabled(self) -> bool: """LLDP Enabled for the whole gateway""" return self._data.get("lldpEnable", False) @property def echo_server(self) -> str | None: """Address of server to ping for online detection.""" return self._data.get("echoServer") @property def is_combined_gateway(self) -> bool: """True if this is a combined gateway/switch.""" return self._data.get("combinedGateway", False)