Merge pull request #38 from MarkGodwin/feature/no_verify_ssl

Add PoE control on gateways and PoE cli command

.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",

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="[email protected]" },
 ]

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)

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')
+

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:

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"""