Merge pull request #36 from MarkGodwin/feature/wan_control

Added WAN port connect/disconnect

.vscode/launch.json

@@ -6,55 +6,63 @@
     "configurations": [
         {
             "name": "CLI: List Devices",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["devices"]
         },
         {
             "name": "CLI: List Targets",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["targets"]
         },
         {
             "name": "CLI: List Switches",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["switches"]
         },
         {
             "name": "CLI: Get Switch",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["switch", "Main Switch", "--dump"]
         },
         {
             "name": "CLI: Get Switch Ports",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["switch_ports", "Main Switch", "-t", "-p", "5"]
         },
         {
             "name": "CLI: Get Gateway",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,
             "args": ["gateway", "--dump"]
         },
+        {
+            "name": "CLI: Get Gateway Port Details",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "tplink_omada_client.cli",
+            "justMyCode": true,
+            "args": ["wan", "--port", "2"]
+        },
         {
             "name": "CLI: Get Clients",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "module": "tplink_omada_client.cli",
             "justMyCode": true,

README.md

@@ -20,8 +20,9 @@ Only a subset of the controller's features are supported:
     * Get firmware information and initiating automatic updates
     * Port status and configuraton for Switches
     * Lan port configuration for Access Points
+    * Gateway port status and WAN port Connect/Disconnect control
 
-Tested with OC200 on Omada Controller Version 5.5.7 - 5.7.6. Other versions may not be fully compatible.
+Tested with OC200 on Omada Controller Version 5.5.7 - 5.12.x. Other versions may not be fully compatible.
 Version 5.0.x is definitely not compatible.
 
 ## CLI
@@ -63,6 +64,13 @@ There is an undocumented Websocket API which could potentially be used to get a
 I'm not sure how fully featured this subscription channel is on the controller. It seems to be rarely used,
 so probably doesn't include client connect/disconnect notifications.
 
+The Omada platform is transitioning to a new OpenAPI API which this library will need to switch over to using
+eventually. We will try to avoid breaking changes when this happens, but some will be unavoidable - particularly
+authentication.
+
+At the moment, the new API imposes severe daily call limits, even though it is a local device API.
+Hopefully this will change, because it is unusable as it stands.
+
 ## Contributing
 
 There is a VS Code development container, which sets up all of the requirements for running the package.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "tplink_omada_client"
-version = "1.3.9"
+version = "1.3.10"
 authors = [
   { name="Mark Godwin", email="[email protected]" },
 ]

src/tplink_omada_client/cli/__init__.py

@@ -23,7 +23,8 @@
     command_switch_ports,
     command_unblock_client,
     command_set_device_led,
-    command_set_client_name
+    command_set_client_name,
+    command_wan
 
 )
 
@@ -60,6 +61,7 @@ def main(argv: Union[Sequence[str], None] = None) -> int:
     command_unblock_client.arg_parser(subparsers)
     command_set_device_led.arg_parser(subparsers)
     command_set_client_name.arg_parser(subparsers)
+    command_wan.arg_parser(subparsers)
 
     try:
         args = parser.parse_args(args=argv)

src/tplink_omada_client/cli/command_gateway.py

@@ -2,10 +2,10 @@
 
 from argparse import ArgumentParser
 
-from tplink_omada_client.definitions import GatewayPortMode
+from tplink_omada_client.definitions import GatewayPortMode, PoEMode
 
 from .config import get_target_config, to_omada_connection
-from .util import dump_raw_data, get_link_status_char, get_device_mac, get_target_argument
+from .util import dump_raw_data, get_checkbox_char, get_display_bytes, get_link_status_char, get_device_mac, get_power_char, get_target_argument
 
 async def command_gateway(args) -> int:
     """Executes 'gateway' command"""
@@ -27,14 +27,21 @@ async def command_gateway(args) -> int:
         gateway.status_category
         print(f"Uptime: {gateway.display_uptime}")
         wan_ports = (p for p in gateway.port_status if p.mode == GatewayPortMode.WAN)
-        lan_ports = (p for p in gateway.port_status if p.mode == GatewayPortMode.LAN)
+        lan_ports = (p for p in gateway.port_configs if p.port_status.mode == GatewayPortMode.LAN)
         print("WAN Ports:")
+        print("         No. Name        IP Addresss    Proto Link Online Received  Transmitted")
         for p in wan_ports:
             print(f"    Port: {p.port_number:>2} {p.type.name:7} {p.ip:>15} {p.wan_protocol:>8} ", end="")
-            print('\u2611' if p.wan_connected else '\u2610')
+            print('\u2611    ' if p.wan_connected else '\u2610    ', end="")
+            print('\u2611' if p.online_detection else '\u2610', end="")
+            print(f"  {get_display_bytes(p.bytes_rx):>12} {get_display_bytes(p.bytes_tx):>12}")
+
         print("LAN Ports:")
+        print("         No. Name    Link PoE      Received  Transmitted")
         for p in lan_ports:
-            print(f"    Port: {p.port_number:>2} {p.type.name:7} {get_link_status_char(p.link_status)}")
+            ps = p.port_status
+            print(f"    Port: {ps.port_number:>2} {ps.type.name:7} {get_link_status_char(ps.link_status)}    {get_checkbox_char(p.poe_mode == PoEMode.ENABLED)} {get_power_char(ps.poe_active)} ", end="")
+            print(f"{get_display_bytes(ps.bytes_rx):>12} {get_display_bytes(ps.bytes_tx):>12}")
         print(f"LED Setting: {gateway.led_setting.name}")
 
         dump_raw_data(args, gateway)

src/tplink_omada_client/cli/command_switch_ports.py

@@ -7,7 +7,7 @@
 from tplink_omada_client.devices import OmadaSwitch, OmadaSwitchPortDetails
 
 from .config import get_target_config, to_omada_connection
-from .util import dump_raw_data, get_checkbox_char, get_link_status_char, get_device_mac, get_power_char, get_target_argument
+from .util import dump_raw_data, get_checkbox_char, get_display_bytes, get_link_status_char, get_device_mac, get_power_char, get_target_argument
 
 async def command_switch_ports(args) -> int:
     """Executes 'switch_ports' command"""
@@ -63,13 +63,6 @@ def print_port_table(switch: OmadaSwitch, ports: List[OmadaSwitchPortDetails]):
             print(" x      -W ", end="")
         print(f" {get_display_bytes(p.port_status.bytes_rx):>12} {get_display_bytes(p.port_status.bytes_tx):>12}")
 
-def get_display_bytes(bytes: int, short: bool = True) -> str:
-    if bytes / (1 if short else 1024) > 1048576 * 1024 * 512:
-        return f"{bytes / (1048576.0 * 1048576.0):,.1f}TB"
-    if bytes / (1 if short else 1024) > 1048576 * 512:
-        return f"{bytes / (1048576.0 * 1024):,.1f}GB"
-    return f"{bytes / (1048576.0):,.1f}MB"
-
 def arg_parser(subparsers) -> None:
     """Configures arguments parser for 'switch_ports' command"""
     switch_ports_parser: ArgumentParser = subparsers.add_parser(

src/tplink_omada_client/cli/command_wan.py

@@ -0,0 +1,95 @@
+"""Implementation for 'wan' command"""
+
+from argparse import ArgumentError, ArgumentParser
+
+from tplink_omada_client.definitions import GatewayPortMode
+
+from .config import get_target_config, to_omada_connection
+from .util import dump_raw_data, get_checkbox_char, get_link_status_char, get_device_mac, get_target_argument
+
+async def command_wan(args) -> int:
+    """Executes 'wan' 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)
+        mac = args['mac']
+        if mac:
+            mac = await get_device_mac(site_client, mac)
+        gateway = await site_client.get_gateway(mac)
+
+        port = int(args['port'])
+        port_status = next((p for p in gateway.port_status if p.port_number == port), None) 
+        if not port_status:
+            print(f"Port {port} not found")
+            return -1
+
+        if(port_status.mode != GatewayPortMode.WAN):
+            print(f"Port {port} is not in WAN mode")
+            return -1
+
+        if(args['connect'] or args['disconnect']):
+            if(args['ipv6'] and not port_status.wan_ipv6_enabled):
+                print(f"Port {port} is not configured for IPv6")
+                return -1
+            port_status = await site_client.set_gateway_wan_port_connect_state(port, args['connect'], mac, args['ipv6'])
+
+            print("Ok! Note that the gateway may take a few seconds or more to apply the change.")
+
+        print(f"Port:      {port_status.port_number}")
+        print(f"Name:      {port_status.display_name} ({port_status.name})")
+        print(f"Link:      {get_link_status_char(port_status.link_status)}")
+        print(f"Mode:      {port_status.mode.name}")
+        print(f"Ipv4:      {get_checkbox_char(port_status.wan_connected)}", end="")
+        if(port_status.wan_connected):
+            print(f"    {port_status.wan_ip_address}")
+        else:
+            print()
+        print(f"Ipv4Proto: {port_status.wan_protocol}")
+        if(port_status.wan_ipv6_enabled):
+            print(f"Ipv6:      {get_checkbox_char(port_status.ipv6_wan_connected)}", end="")
+            if(port_status.ipv6_wan_connected):
+                print(f"    {port_status.wan_ipv6_address}")
+            else:
+                print()
+        print(f"Online:    {get_checkbox_char(port_status.online_detection)}")
+        print(f"Speed:     {port_status.link_speed.name}")
+        print(f"Duplex:    {port_status.link_duplex.name}")
+
+        dump_raw_data(args, port_status)
+
+    return 0
+
+def arg_parser(subparsers) -> None:
+    """Configures arguments parser for 'wan' command"""
+    switch_parser: ArgumentParser = subparsers.add_parser(
+        "wan",
+        help="Controls the gateway's wan ports"
+    )
+    switch_parser.set_defaults(func=command_wan)
+
+    switch_parser.add_argument(
+        "--mac",
+        help="The MAC address of the gateway (optional)",
+        required=False
+    )
+    switch_parser.add_argument(
+        "-p", "--port",
+        help="The port number of the gateway.",
+        required=True
+    )
+    con_discon_grp = switch_parser.add_mutually_exclusive_group()
+    con_discon_grp.add_argument(
+        "--connect",
+        help="Connect the port to the internet",
+        action="store_true"
+    )
+    con_discon_grp.add_argument(
+        "--disconnect",
+        help="Connect the port from the internet",
+        action="store_true"        
+    )
+    switch_parser.add_argument("--ipv6", help="Connect/Disconnect IPv6 Wan (defaults to IPv4)", action="store_true")
+    switch_parser.add_argument('-d', '--dump', help="Output raw port information",  action='store_true')
+

src/tplink_omada_client/cli/util.py

@@ -71,3 +71,10 @@ def get_power_char(power: bool) -> str:
         return "\u26a1"
     else:
         return "  "
+
+def get_display_bytes(bytes: int, short: bool = True) -> str:
+    if bytes / (1 if short else 1024) > 1048576 * 1024 * 512:
+        return f"{bytes / (1048576.0 * 1048576.0):,.1f}TB"
+    if bytes / (1 if short else 1024) > 1048576 * 512:
+        return f"{bytes / (1048576.0 * 1024):,.1f}GB"
+    return f"{bytes / (1048576.0):,.1f}MB"