diff --git a/.vscode/launch.json b/.vscode/launch.json index 57f25a0..f981039 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -68,6 +68,14 @@ "justMyCode": true, "args": ["wan", "--port", "2"] }, + { + "name": "CLI: Set Port Port PoE", + "type": "debugpy", + "request": "launch", + "module": "tplink_omada_client.cli", + "justMyCode": true, + "args": ["poe", "Main Switch", "--port", "2", "--on", "-d"] + }, { "name": "CLI: Get Clients", "type": "debugpy", diff --git a/pyproject.toml b/pyproject.toml index 1497c65..7054808 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "tplink_omada_client" -version = "1.3.11" +version = "1.3.12" authors = [ { name="Mark Godwin", email="author@example.com" }, ] diff --git a/src/tplink_omada_client/cli/__init__.py b/src/tplink_omada_client/cli/__init__.py index 23af2cd..f98d1ea 100644 --- a/src/tplink_omada_client/cli/__init__.py +++ b/src/tplink_omada_client/cli/__init__.py @@ -24,8 +24,8 @@ command_unblock_client, command_set_device_led, command_set_client_name, - command_wan - + command_wan, + command_poe ) def main(argv: Union[Sequence[str], None] = None) -> int: @@ -62,6 +62,7 @@ def main(argv: Union[Sequence[str], None] = None) -> int: command_set_device_led.arg_parser(subparsers) command_set_client_name.arg_parser(subparsers) command_wan.arg_parser(subparsers) + command_poe.arg_parser(subparsers) try: args = parser.parse_args(args=argv) diff --git a/src/tplink_omada_client/cli/command_poe.py b/src/tplink_omada_client/cli/command_poe.py new file mode 100644 index 0000000..9c3b70b --- /dev/null +++ b/src/tplink_omada_client/cli/command_poe.py @@ -0,0 +1,89 @@ +"""Implementation for 'poe' command""" + +from argparse import ArgumentError, ArgumentParser + +from tplink_omada_client.definitions import OmadaApiData, PoEMode +from tplink_omada_client.devices import OmadaDevice +from tplink_omada_client.omadasiteclient import AccessPointPortSettings, GatewayPortSettings, OmadaSiteClient, SwitchPortOverrides + +from .config import get_target_config, to_omada_connection +from .util import dump_raw_data, get_device_by_mac_or_name, get_target_argument + +async def set_gateway_poe(site_client: OmadaSiteClient, device: OmadaDevice, port: int, change: bool, on: bool) -> OmadaApiData: + if change: + result = await site_client.set_gateway_port_settings(port, GatewayPortSettings(enable_poe=on), device) + else: + result = await site_client.get_gateway_port(port, device) + print(f"Gateway {device.name} Port {port} PoE is {result.poe_mode.name}") + return result + +async def set_switch_poe(site_client: OmadaSiteClient, device: OmadaDevice, port: int, change: bool, on: bool) -> OmadaApiData: + if change: + result = await site_client.update_switch_port(device, port, overrides=SwitchPortOverrides(enable_poe=on)) + else: + result = await site_client.get_switch_port(device, port) + print(f"Switch {device.name} Port {port} PoE now is {result.poe_mode.name}") + return result + +async def set_access_point_poe(site_client: OmadaSiteClient, device: OmadaDevice, port: int, change: bool, on: bool) -> OmadaApiData: + if change: + result = await site_client.update_access_point_port(device, f"ETH{port}", AccessPointPortSettings(enable_poe=on)) + else: + result = await site_client.get_access_point_port(device, f"ETH{port}") + print(f"Access point {device.name} Port {result.port_name} PoE is {(PoEMode.ENABLED if result.poe_enable else (PoEMode.DISABLED if result.supports_poe else PoEMode.NONE)).name}") + return result + +async def command_poe(args) -> int: + """Executes 'poe' command""" + controller = get_target_argument(args) + config = get_target_config(controller) + + async with to_omada_connection(config) as client: + site_client = await client.get_site_client(config.site) + device = await get_device_by_mac_or_name(site_client, args['mac']) + + port = int(args['port']) + change = args['on'] or args['off'] + on = bool(args['on']) + + handlers = {"gateway": set_gateway_poe, "switch": set_switch_poe, "ap": set_access_point_poe} + handler = handlers.get(device.type) + if not handler: + raise ArgumentError(args["mac"], "Device type not supported") + + result = await handler(site_client, device, port, change, on) + dump_raw_data(args, result) + + return 0 + +def arg_parser(subparsers) -> None: + """Configures arguments parser for 'poe' command""" + switch_parser: ArgumentParser = subparsers.add_parser( + "poe", + help="Controls a device's PoE ports" + ) + switch_parser.set_defaults(func=command_poe) + + switch_parser.add_argument( + "mac", + help="The MAC address or name of the gateway, switch or access point with PoE ports" + ) + switch_parser.add_argument( + "-p", "--port", + help="The port number on the device to set the PoE state.", + required=True + ) + + con_discon_grp = switch_parser.add_mutually_exclusive_group() + con_discon_grp.add_argument( + "--on", + help="Turn PoE On", + action="store_true" + ) + con_discon_grp.add_argument( + "--off", + help="Turn PoE Off", + action="store_true" + ) + switch_parser.add_argument('-d', '--dump', help="Output raw port information", action='store_true') + diff --git a/src/tplink_omada_client/devices.py b/src/tplink_omada_client/devices.py index f30bc9e..fb69c37 100644 --- a/src/tplink_omada_client/devices.py +++ b/src/tplink_omada_client/devices.py @@ -795,6 +795,15 @@ def port_configs(self) -> List[OmadaGatewayPortConfig]: return [ OmadaGatewayPortConfig(p, poeData.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) -> Union[str, None]: + return self._data.get("echoServer") @property def is_combined_gateway(self) -> bool: diff --git a/src/tplink_omada_client/omadasiteclient.py b/src/tplink_omada_client/omadasiteclient.py index 9dbac4e..0d3bfbd 100644 --- a/src/tplink_omada_client/omadasiteclient.py +++ b/src/tplink_omada_client/omadasiteclient.py @@ -17,6 +17,7 @@ OmadaDevice, OmadaFirmwareUpdate, OmadaGateway, + OmadaGatewayPortConfig, OmadaGatewayPortStatus, OmadaListDevice, OmadaPortProfile, @@ -79,6 +80,18 @@ def __init__( self.vlan_enable = vlan_enable self.vlan_id = vlan_id +class GatewayPortSettings: + """ + Settings that can be applied to network ports on gateways + + Specify the values you want to modify. The remaining values will be unaffected + """ + def __init__( + self, + enable_poe: Optional[bool] = None + ): + self.enable_poe = enable_poe + class OmadaSiteClient: """Client for querying an Omada site's devices.""" @@ -167,7 +180,7 @@ async def get_access_points(self) -> List[OmadaAccessPoint]: for d in await self.get_devices() if d.type == "ap" ] - + async def get_access_point( self, mac_or_device: Union[str, OmadaDevice] ) -> OmadaAccessPoint: @@ -185,6 +198,18 @@ async def get_access_point( ) return OmadaAccessPoint(result) + + async def get_access_point_port( + self, mac_or_device: Union[str, OmadaDevice], + port_name: str) -> OmadaAccesPointLanPortSettings: + """Get the config of a single network port on an access point.""" + ap = await self.get_access_point(mac_or_device) + + port = next(p for p in ap.lan_port_settings if p.port_name == port_name) + if(port is None): + raise InvalidDevice(f"Port {port_name} not found") + return port + async def get_switch(self, mac_or_device: Union[str, OmadaDevice]) -> OmadaSwitch: """Get a switch by Mac address or Omada device.""" @@ -463,6 +488,14 @@ async def get_gateway(self, mac_or_device: Union[str, OmadaDevice, None] = None) return OmadaGateway(result) + async def get_gateway_port(self, port_id: int, mac_or_deviec: Union[str, OmadaDevice, None] = None) -> OmadaGatewayPortConfig: + """Get the port config for a specified port on the gateway""" + gw = await self.get_gateway(mac_or_deviec) + port_config = next(p for p in gw.port_configs if p.port_number == port_id) + if port_config is None: + raise InvalidDevice(f"Port {port_id} not found") + return port_config + async def set_gateway_wan_port_connect_state(self, port_id: int, connect: bool, mac_or_device: Union[str, OmadaDevice, None] = None, ipv6:bool = False) -> OmadaGatewayPortStatus: """Connects or disconnects the specified WAN port of the gateway to the internet.""" mac = await self._get_gateway_mac(mac_or_device) @@ -471,6 +504,39 @@ async def set_gateway_wan_port_connect_state(self, port_id: int, connect: bool, result = await self._api.request( "post", self._api.format_url(f"cmd/gateways/{mac}/{'ipv6State' if ipv6 else 'internetState'}", self._site_id), payload=payload) return OmadaGatewayPortStatus(result) + + async def set_gateway_port_settings(self, port_id: int, settings: GatewayPortSettings, mac_or_device: Union[str, OmadaDevice, None] = None) -> OmadaGatewayPortConfig: + """Sets the settings for the specified port of the gateway.""" + mac = await self._get_gateway_mac(mac_or_device) + + # Currently, we (and the Omada API) only supports PoE, so if the caller isn't asking for a PoE change, it's a no-op, but we should still return the current settings + if settings.enable_poe is not None: + + # Reject requests that ask to set PoE on gateways that don't support it + gw = await self.get_gateway(mac) + if not gw.supports_poe and settings.enable_poe is not None: + raise InvalidDevice("This gateway does not support PoE") + + # Thanks to dkriegner, we know the request format is: + # {"lldpEnable":false,"echoServer":"0.0.0.0","poeSettings":[{"enable":true,"portId":5},{"enable":true,"portId":6},{"enable":true,"portId":7},{"enable":true,"portId":8},{"enable":true,"portId":9},{"enable":true,"portId":10},{"enable":true,"portId":11},{"enable":true,"portId":12}]} + # We probably don't need to specify all of these for PATCH, but it's what the UI does, and I have no way of testing + payload = { + "lldpEnable": gw.lldp_enabled, + "echoServer": gw.echo_server, + "poeSettings": [ + # Output an entry for every port that supports PoE, setting the appropriate port as requested + { + "enable": settings.enable_poe if settings.enable_poe is not None and port_id == p.port_number else p.poe_mode == PoEMode.ENABLED, + "portId": p.port_number + } for p in gw.port_configs if p.poe_mode != PoEMode.NONE + ] + } + + await self._api.request( + "patch", self._api.format_url(gw.resource_path, self._site_id), payload=payload) + + # The result data includes an incomplete representation of the gateway port state, so we just request a new update + return await self.get_gateway_port(port_id, mac) async def set_led_setting(self, mac_or_device: Union[str, OmadaDevice], setting: LedSetting) -> bool: """Sets the onboard LED setting for the device"""