diff --git a/.gitignore b/.gitignore index b6e4761..0ce37c3 100644 --- a/.gitignore +++ b/.gitignore @@ -127,3 +127,9 @@ dmypy.json # Pyre type checker .pyre/ + +# Local exploration / probe scripts that may contain real credentials +tplink.py + +# Decompiled Tether source — license + size; do not commit +tether-src/ diff --git a/CLAUDE.md b/CLAUDE.md index 24d311c..5050734 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -59,17 +59,29 @@ Set `include_tapo=False` on `TPLinkDeviceManager` to disable Tapo cloud support. - **Power strips**: `HS300`, `KP303` — parent + child devices per outlet - **Outdoor plugs**: `KP200`, `KP400` — parent + child devices per outlet, `EP40` — single outlet - **Light strips**: `KL420L5`, `KL430` — color, brightness, color temp via `smartlife.iot.smartbulb.lightingservice` +- **Routers** (Archer family, both wireless and xDSL modem-router): `Router` — metadata-only class. Cloud control of router *operations* is not possible via this library; the cloud relay rejects every JSON method (`-20103`) for router-typed devices because routers speak the proprietary TMP binary protocol. The class surfaces alias/model/MAC/firmware/hardware/region/online-status/V2-URL and raises `NotImplementedError` on plug-style action methods. See `docs/superpowers/router-api/discovery-findings.md` for the empirical evidence and reasoning. Devices with multiple outlets (`HS300`, `KP303`, `KP200`, `KP400`) have `has_children() -> True`. Child devices are separate class instances (e.g., `HS300Child`) with a `child_id`. +### Device dispatch in `_construct_device` + +Two-tier: + +1. **`device_type`-first** (router shortcut): if the cloud returns `deviceType` `WIRELESSROUTER` or `XDSLMODEMROUTER`, instantiate `Router` directly. Used for devices where model-name disambiguation is unnecessary because every router goes to the same metadata-only class. +2. **Model-prefix dispatch** (everything else): walk `DEVICE_MODEL_MAP` looking for a model-name prefix match (`HS103`, `KL430`, etc.). Fall back to base `TPLinkDevice` if no prefix matches. + ### Adding a new device +For a plug/switch/light/strip: + 1. Create `tplinkcloud/.py` extending `TPLinkDevice` (or `EmeterDevice` for energy monitoring) 2. Define the `DeviceType` enum value in `device_type.py` -3. Add the model mapping in `device_manager.py` `_construct_device()` +3. Add the model mapping in `device_manager.py` `DEVICE_MODEL_MAP` 4. Add wiremock stubs in `tests/wiremock/mappings/` and `tests/wiremock/__files/` 5. Add tests in `tests/test_device_manager.py` +For a new router-family `deviceType` (e.g., a mesh node) that should also use the metadata-only `Router` class: add the new `deviceType` string to the tuple in `_construct_device`'s router check and add a wiremock fixture entry; no new file needed. + ## Controlling devices ```python diff --git a/tests/test_device_manager.py b/tests/test_device_manager.py index 0f1d7cf..6280d72 100644 --- a/tests/test_device_manager.py +++ b/tests/test_device_manager.py @@ -25,12 +25,13 @@ class TestGetDevices(object): async def test_gets_devices(self, client): device_list = await client.get_devices() assert device_list is not None - # Kasa: 9 parents + 10 children = 19 + # Kasa plugs/lights: 9 parents + 10 children = 19 # HS103, HS105, HS110, HS300 (6 children), KL430, # HS200, KP200 (2 children), KP400 (2 children), KL420L5 + # Kasa routers: 2 (Archer AX53, Archer VR400) # Tapo: 3 devices (P100, P110, L530) - # Total: 19 + 3 = 22 - assert len(device_list) == 22 + # Total: 19 + 2 + 3 = 24 + assert len(device_list) == 24 @pytest.mark.usefixtures('client') class TestFindDevice(object): @@ -330,3 +331,29 @@ async def test_auth_no_password(self, client): term_id=os.environ.get('TPLINK_KASA_TERM_ID') ) device_manager.login(os.environ.get('TPLINK_KASA_USERNAME'), None) + + +@pytest.mark.usefixtures('client') +class TestRouterDispatch(object): + + @pytest.mark.asyncio + async def test_finds_archer_ax53_router(self, client): + from tplinkcloud.router import Router + device_name = 'Archer AX53' + device = await client.find_device(device_name) + assert device is not None + assert isinstance(device, Router) + assert device.device_info.device_type == 'WIRELESSROUTER' + assert device.device_info.device_model == 'Archer AX53(EU)' + assert device.app_server_url_v2 == 'http://127.0.0.1:8080' + assert device.is_online() is True + + @pytest.mark.asyncio + async def test_finds_archer_vr400_xdsl_router(self, client): + from tplinkcloud.router import Router + device_name = 'Archer VR400' + device = await client.find_device(device_name) + assert device is not None + assert isinstance(device, Router) + assert device.device_info.device_type == 'XDSLMODEMROUTER' + assert device.is_online() is False # status 0 in fixture diff --git a/tests/test_router.py b/tests/test_router.py new file mode 100644 index 0000000..8ead3cd --- /dev/null +++ b/tests/test_router.py @@ -0,0 +1,141 @@ +"""Tests for the Router device class and supporting metadata.""" +import pytest + +from tplinkcloud.device_info import TPLinkDeviceInfo +from tplinkcloud.device_type import TPLinkDeviceType + + +class TestDeviceInfoAppServerUrlV2: + + def test_reads_app_server_url_v2_when_present(self): + info = TPLinkDeviceInfo({ + "deviceType": "WIRELESSROUTER", + "appServerUrl": "https://primary.example", + "appServerUrlV2": "https://v2.example", + }) + assert info.app_server_url_v2 == "https://v2.example" + + def test_app_server_url_v2_is_none_when_missing(self): + info = TPLinkDeviceInfo({"deviceType": "IOT.SMARTPLUGSWITCH"}) + assert info.app_server_url_v2 is None + + +class TestRouterEnum: + + def test_router_enum_value_exists(self): + assert TPLinkDeviceType.ROUTER.value == 1000 + + def test_router_does_not_collide_with_other_values(self): + values = [m.value for m in TPLinkDeviceType] + assert len(values) == len(set(values)), "duplicate enum value" + + +def _make_router_info(status=1, **overrides): + base = { + "deviceType": "WIRELESSROUTER", + "fwVer": "1.6.2 Build 20260119 rel.87654", + "appServerUrl": "https://v1.example", + "appServerUrlV2": "https://v2.example", + "deviceRegion": "ap-southeast-1", + "deviceId": "ROUTER-DEVICE-ID", + "deviceName": "Archer AX53", + "deviceHwVer": "1.0", + "alias": "My Router", + "deviceMac": "AABBCCDDEEFF", + "deviceModel": "Archer AX53(EU)", + "status": status, + } + base.update(overrides) + return TPLinkDeviceInfo(base) + + +def _make_router(**overrides): + from tplinkcloud.router import Router + info = _make_router_info(**overrides) + # Router doesn't talk to the network; pass None for the client. + return Router(client=None, device_id=info.device_id, device_info=info) + + +class TestRouterMetadata: + + def test_alias(self): + r = _make_router() + assert r.get_alias() == "My Router" + + def test_mac(self): + r = _make_router() + assert r.mac == "AABBCCDDEEFF" + + def test_firmware_version(self): + r = _make_router() + assert r.firmware_version == "1.6.2 Build 20260119 rel.87654" + + def test_hardware_version(self): + r = _make_router() + assert r.hardware_version == "1.0" + + def test_region(self): + r = _make_router() + assert r.region == "ap-southeast-1" + + def test_app_server_url_v2(self): + r = _make_router() + assert r.app_server_url_v2 == "https://v2.example" + + def test_is_online_true_when_status_1(self): + r = _make_router(status=1) + assert r.is_online() is True + + def test_is_online_false_when_status_0(self): + r = _make_router(status=0) + assert r.is_online() is False + + def test_model_type_is_router(self): + r = _make_router() + assert r.model_type == TPLinkDeviceType.ROUTER + + +class TestRouterUnsupportedActions: + + _METHOD_ARGS = { + "set_led_state": (True,), + "get_schedule_rule": ("rule-id",), + "edit_schedule_rule": ({"id": "rule-id"},), + "add_schedule_rule": ({"enable": 1},), + "delete_schedule_rule": ("rule-id",), + "get_runtime_day": (2026, 4), + "get_runtime_month": (2026,), + } + + @pytest.mark.asyncio + @pytest.mark.parametrize("method_name", [ + "power_on", "power_off", "toggle", "is_on", "is_off", + "set_led_state", + "get_sys_info", + "get_schedule_rules", "get_schedule_rule", + "edit_schedule_rule", "add_schedule_rule", + "delete_all_scheduled_rules", "delete_schedule_rule", + "get_runtime_day", "get_runtime_month", + "get_net_info", "get_time", "get_timezone", + ]) + async def test_unsupported_methods_raise(self, method_name): + r = _make_router() + method = getattr(r, method_name) + args = self._METHOD_ARGS.get(method_name, ()) + with pytest.raises(NotImplementedError) as exc_info: + await method(*args) + msg = str(exc_info.value) + assert method_name in msg + assert "discovery-findings" in msg + + +class TestPackageReExport: + + def test_router_importable_from_package_root(self): + from tplinkcloud import Router + from tplinkcloud.router import Router as RouterDirect + assert Router is RouterDirect + + def test_router_in_dunder_all(self): + import tplinkcloud + assert "Router" in tplinkcloud.__all__ diff --git a/tests/wiremock/__files/get_device_list_response.json b/tests/wiremock/__files/get_device_list_response.json index 58f13f6..fc832db 100644 --- a/tests/wiremock/__files/get_device_list_response.json +++ b/tests/wiremock/__files/get_device_list_response.json @@ -163,6 +163,44 @@ "fwId": "00000000000000000000000000000000", "isSameRegion": true, "status": 1 + }, + { + "deviceType": "WIRELESSROUTER", + "role": 0, + "fwVer": "1.6.2 Build 20260119 rel.87654", + "appServerUrl": "http://127.0.0.1:8080", + "appServerUrlV2": "http://127.0.0.1:8080", + "deviceRegion": "us-east-1", + "deviceId": "E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2E5F60005", + "deviceName": "Archer AX53", + "deviceHwVer": "1.0", + "alias": "Archer AX53", + "deviceMac": "AABBCCDDEEE5", + "oemId": "ARCHER_AX53_OEM_ID_PLACEHOLDER0001", + "deviceModel": "Archer AX53(EU)", + "hwId": "ARCHER_AX53_HW_ID_PLACEHOLDER00001", + "fwId": "00000000000000000000000000000000", + "isSameRegion": true, + "status": 1 + }, + { + "deviceType": "XDSLMODEMROUTER", + "role": 0, + "fwVer": "1.5.0 0.9.1 v00a1.0 Build 220321 Rel.60385n", + "appServerUrl": "http://127.0.0.1:8080", + "appServerUrlV2": "http://127.0.0.1:8080", + "deviceRegion": "us-east-1", + "deviceId": "F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3F6A10006", + "deviceName": "Archer VR400", + "deviceHwVer": "3.0", + "alias": "Archer VR400", + "deviceMac": "AABBCCDDEEE6", + "oemId": "ARCHER_VR400_OEM_ID_PLACEHOLDER0002", + "deviceModel": "Archer VR400(EU)", + "hwId": "ARCHER_VR400_HW_ID_PLACEHOLDER0002", + "fwId": "00000000000000000000000000000000", + "isSameRegion": true, + "status": 0 } ] } diff --git a/tplinkcloud/__init__.py b/tplinkcloud/__init__.py index f9b4c74..b339ed7 100644 --- a/tplinkcloud/__init__.py +++ b/tplinkcloud/__init__.py @@ -8,11 +8,13 @@ TPLinkMFARequiredError, TPLinkTokenExpiredError, ) +from .router import Router __all__ = [ 'TPLinkDeviceManager', 'TPLinkDeviceManagerPowerTools', 'TPLinkDeviceScheduleRuleBuilder', + 'Router', 'TPLinkAuthError', 'TPLinkCloudError', 'TPLinkDeviceOfflineError', diff --git a/tplinkcloud/device_info.py b/tplinkcloud/device_info.py index 8c3c751..f6832b0 100644 --- a/tplinkcloud/device_info.py +++ b/tplinkcloud/device_info.py @@ -17,4 +17,5 @@ def __init__(self, device_info, cloud_type="kasa"): self.fw_id = device_info.get('fwId') self.is_same_region = device_info.get('isSameRegion') self.status = device_info.get('status') + self.app_server_url_v2 = device_info.get('appServerUrlV2') self.cloud_type = cloud_type diff --git a/tplinkcloud/device_manager.py b/tplinkcloud/device_manager.py index d527693..fbe54e5 100644 --- a/tplinkcloud/device_manager.py +++ b/tplinkcloud/device_manager.py @@ -19,6 +19,7 @@ from .kp303 import KP303 from .kp400 import KP400 from .ep40 import EP40 +from .router import Router from .device import TPLinkDevice DEVICE_MODEL_MAP: dict[str, type[TPLinkDevice]] = { @@ -195,12 +196,15 @@ def _construct_device(self, device_info, api, token, cloud_type): app_name=api._app_name, cloud_type=cloud_type, ) - model = tplink_device_info.device_model - device_cls = next( - (cls for prefix, cls in DEVICE_MODEL_MAP.items() if model.startswith(prefix)), - TPLinkDevice, - ) - device = device_cls(client, tplink_device_info.device_id, tplink_device_info) + if tplink_device_info.device_type in ('WIRELESSROUTER', 'XDSLMODEMROUTER'): + device = Router(client, tplink_device_info.device_id, tplink_device_info) + else: + model = tplink_device_info.device_model + device_cls = next( + (cls for prefix, cls in DEVICE_MODEL_MAP.items() if model.startswith(prefix)), + TPLinkDevice, + ) + device = device_cls(client, tplink_device_info.device_id, tplink_device_info) device.cloud_type = cloud_type return device diff --git a/tplinkcloud/device_type.py b/tplinkcloud/device_type.py index 2eda0fa..945243c 100644 --- a/tplinkcloud/device_type.py +++ b/tplinkcloud/device_type.py @@ -19,6 +19,7 @@ class TPLinkDeviceType(Enum): KP400CHILD = 4002 KL420L5 = 420 KL430 = 430 - UNKNOWN = 9999 EP40 = 400 EP40CHILD = 4000 + ROUTER = 1000 + UNKNOWN = 9999 diff --git a/tplinkcloud/router.py b/tplinkcloud/router.py new file mode 100644 index 0000000..01a225b --- /dev/null +++ b/tplinkcloud/router.py @@ -0,0 +1,115 @@ +"""TP-Link Router device class. + +NOTE: TP-Link routers cannot be controlled through the TP-Link Cloud API +that powers this library. They use the proprietary TMP binary protocol, +reachable only via local-network TCP or a persistent device-to-cloud +mTLS tunnel. See docs/superpowers/router-api/discovery-findings.md for +the evidence and reasoning. + +This class therefore surfaces router metadata only and raises +NotImplementedError on plug-style action methods. For local control, +use a dedicated local-network library (e.g. python-kasa). +""" +from .device import TPLinkDevice +from .device_type import TPLinkDeviceType + + +_UNSUPPORTED_MSG_FMT = ( + "Router.{method}() is not supported. TP-Link routers cannot be " + "controlled via the cloud API. See " + "docs/superpowers/router-api/discovery-findings.md." +) + + +def _unsupported(method_name: str) -> NotImplementedError: + return NotImplementedError(_UNSUPPORTED_MSG_FMT.format(method=method_name)) + + +class Router(TPLinkDevice): + """A TP-Link router (wireless or xDSL modem-router) recognized via the + cloud device-list API. Surfaces metadata only — see module docstring.""" + + def __init__(self, client, device_id, device_info, child_id=None): + super().__init__(client, device_id, device_info, child_id=child_id) + self.model_type = TPLinkDeviceType.ROUTER + + # ---- metadata accessors ---- + + @property + def mac(self): + return self.device_info.device_mac + + @property + def firmware_version(self): + return self.device_info.fw_ver + + @property + def hardware_version(self): + return self.device_info.device_hw_ver + + @property + def region(self): + return self.device_info.device_region + + @property + def app_server_url_v2(self): + return self.device_info.app_server_url_v2 + + def is_online(self) -> bool: + return self.device_info.status == 1 + + # ---- explicitly unsupported actions ---- + + async def power_on(self): + raise _unsupported("power_on") + + async def power_off(self): + raise _unsupported("power_off") + + async def toggle(self): + raise _unsupported("toggle") + + async def is_on(self): + raise _unsupported("is_on") + + async def is_off(self): + raise _unsupported("is_off") + + async def set_led_state(self, on): + raise _unsupported("set_led_state") + + async def get_sys_info(self): + raise _unsupported("get_sys_info") + + async def get_schedule_rules(self): + raise _unsupported("get_schedule_rules") + + async def get_schedule_rule(self, rule_id): + raise _unsupported("get_schedule_rule") + + async def edit_schedule_rule(self, rule): + raise _unsupported("edit_schedule_rule") + + async def add_schedule_rule(self, rule): + raise _unsupported("add_schedule_rule") + + async def delete_all_scheduled_rules(self): + raise _unsupported("delete_all_scheduled_rules") + + async def delete_schedule_rule(self, rule_id): + raise _unsupported("delete_schedule_rule") + + async def get_runtime_day(self, year, month): + raise _unsupported("get_runtime_day") + + async def get_runtime_month(self, year): + raise _unsupported("get_runtime_month") + + async def get_net_info(self): + raise _unsupported("get_net_info") + + async def get_time(self): + raise _unsupported("get_time") + + async def get_timezone(self): + raise _unsupported("get_timezone") diff --git a/tplinkcloud/router_client.py b/tplinkcloud/router_client.py new file mode 100644 index 0000000..11908a2 --- /dev/null +++ b/tplinkcloud/router_client.py @@ -0,0 +1,354 @@ +""" +Router Web API client for TP-Link routers. + +Based on Tether app decompilation - uses /cgi/ endpoints. +""" + +import asyncio +import binascii +import hashlib +import re +import urllib.parse +from typing import Optional + +import aiohttp +from aiohttp import CookieJar + + +class TPLinkRouterWebClient: + """Client for TP-Link router local web interface.""" + + def __init__( + self, + host: str = "192.168.0.1", + username: str = "admin", + password: str = None, + verbose: bool = False + ): + self.host = host + self.username = username + self.password = password + self.verbose = verbose + + self._session: Optional[aiohttp.ClientSession] = None + self._cookies: Optional[dict] = None + self._stok: Optional[str] = None # Session token + + @property + def base_url(self) -> str: + return f"http://{self.host}" + + async def _get_session(self) -> aiohttp.ClientSession: + if self._session is None: + jar = CookieJar() + self._session = aiohttp.ClientSession( + cookie_jar=jar, + timeout=aiohttp.ClientTimeout(total=30) + ) + return self._session + + async def close(self): + if self._session: + await self._session.close() + self._session = None + + async def login(self, password: str = None) -> bool: + """Login to the router's web interface.""" + password = password or self.password + if not password: + raise ValueError("Password required for router login") + + session = await self._get_session() + + # Try the standard CGI login endpoint + login_url = f"{self.base_url}/cgi/login" + params = { + "UserName": self.username, + "Passwd": password, + "enableSSH": "1" + } + + if self.verbose: + print(f"POST {login_url} with params: {params}") + + try: + async with session.post(login_url, params=params) as resp: + text = await resp.text() + + if self.verbose: + print(f"Login response status: {resp.status}") + print(f"Login response: {text[:500]}") + + # Check for successful login + if resp.status == 200: + # Check cookies for session + cookies = session.cookie_jar.filter_cookies(self.base_url) + + # Look for auth token in response + if "stok" in text.lower(): + # Try to extract the token + match = re.search(r'stok[=/]?([^&"\'\s]+)', text, re.IGNORECASE) + if match: + self._stok = match.group(1) + if self.verbose: + print(f"Found token: {self._stok}") + return True + + # Check cookies + for cookie in cookies: + if 'stok' in cookie.key.lower(): + self._stok = cookie.value + if self.verbose: + print(f"Found token in cookie: {self._stok}") + return True + + # Check if we have any cookies set + if cookies: + if self.verbose: + print(f"Cookies set: {[c.key for c in cookies]}") + return True + + return False + + except aiohttp.ClientError as e: + if self.verbose: + print(f"Login error: {e}") + return False + + async def _request(self, method: str, path: str, **kwargs) -> dict: + """Make authenticated request to router.""" + session = await self._get_session() + + url = f"{self.base_url}{path}" + + # Add token to URL if we have one + if self._stok: + # Some endpoints use token in query string + if '?' in path: + url = f"{url}&stok={self._stok}" + else: + url = f"{url}?stok={self._stok}" + + if self.verbose: + print(f"{method} {url}") + + try: + async with session.request(method, url, **kwargs) as resp: + text = await resp.text() + + if self.verbose: + print(f"Response status: {resp.status}") + print(f"Response: {text[:1000]}") + + # Try to parse as JSON + try: + return {"status": resp.status, "data": await resp.json()} + except: + return {"status": resp.status, "text": text} + + except aiohttp.ClientError as e: + if self.verbose: + print(f"Request error: {e}") + return {"error": str(e)} + + async def get_connected_devices(self) -> list: + """Get list of connected devices.""" + # Try various endpoints to find client list + + endpoints = [ + # Newer API style with token + f"/api/connected-device/list", + f"/api/online-devices", + f"/api/device/online-list", + # Try the /cgi/ endpoints + f"/cgi/get_client_list", + f"/cgi/get_stations", + # Old style + f"/data/conndev.json", + f"/data/devlist.json", + # Generic + "/api/v2/device/list", + "/api/v1/device-list", + ] + + for endpoint in endpoints: + result = await self._request("GET", endpoint) + if self.verbose: + print(f"Try {endpoint}: {result}") + + # Check if we got valid data + if "data" in result: + data = result.get("data", {}) + if isinstance(data, dict): + # Look for device list in common keys + for key in ["device_list", "client_list", "devices", "online_devices", "wlan_stations"]: + if key in data: + return data[key] + return data + elif isinstance(data, list): + return data + + return [] + + async def get_wireless_clients(self) -> list: + """Get wireless clients specifically.""" + endpoints = [ + f"/api/wlan/station-list", + f"/api/wireless/client-list", + f"/api/wlan-stations", + f"/cgi/wlan_stations", + ] + + for endpoint in endpoints: + result = await self._request("GET", endpoint) + if self.verbose: + print(f"Try {endpoint}: {result}") + + if "data" in result: + data = result.get("data", {}) + if isinstance(data, dict): + for key in ["wlan_stations", "stations", "wireless_clients"]: + if key in data: + return data[key] + return data + + return [] + + async def get_dhcp_clients(self) -> list: + """Get DHCP client list.""" + endpoints = [ + "/api/dhcp-server/client-list", + "/api/dhcp/clients", + "/cgi/dhcp_clients", + ] + + for endpoint in endpoints: + result = await self._request("GET", endpoint) + if self.verbose: + print(f"Try {endpoint}: {result}") + + if "data" in result: + data = result.get("data", {}) + if isinstance(data, dict): + for key in ["dhcp_clients", "client_list", "lease_list"]: + if key in data: + return data[key] + return data + + return [] + + async def get_device_info(self) -> dict: + """Get router device info.""" + result = await self._request("GET", "/api/device/info") + + if "data" in result: + return result["data"] + + # Try alternative endpoints + alt = await self._request("GET", "/api/router-status") + if "data" in alt: + return alt["data"] + + return result + + async def get_wan_status(self) -> dict: + """Get WAN/Internet status.""" + result = await self._request("GET", "/api/wan/status") + + if "data" in result: + return result["data"] + return result + + async def get_wireless_status(self) -> dict: + """Get wireless status.""" + result = await self._request("GET", "/api/wireless/status") + + if "data" in result: + return result["data"] + return result + + async def get_clients_mib(self) -> list: + """ + Get connected clients using MIB-style request. + Uses LAN_WLAN_ASSOC_DEV to get wireless clients. + """ + result = await self._mib_request( + "LAN_WLAN_ASSOC_DEV", + ["AssociatedDeviceMACAddress", "AssociatedDeviceIPAddress", + "X_TP_TotalPacketsSent", "X_TP_TotalPacketsReceived", "X_TP_HostName"] + ) + + if self.verbose: + print(f"MIB clients result: {result}") + + clients = [] + if "text" in result and result.get("status") == 200: + text = result["text"] + lines = text.strip().split('\n') + for line in lines[1:]: + if line.strip(): + parts = line.split('|') + if len(parts) >= 2: + client = { + "mac": parts[0] if parts[0] else "", + "ip": parts[1] if len(parts) > 1 else "", + } + if len(parts) > 4: + client["hostname"] = parts[4] + clients.append(client) + return clients + + async def _mib_request(self, mib_obj: str, fields: list) -> dict: + """ + Make MIB-style request to router. + """ + session = await self._get_session() + mib_request = f"[{mib_obj}#0,0,0,0,0,0#1,0,0,0,0,0]0,{len(fields)}" + for field in fields: + mib_request += f"\n{field}" + url = f"{self.base_url}/cgi" + params = {"ctx": "1", "obj": mib_obj, "format": "nodev"} + headers = {"Content-Type": "application/x-www-form-urlencoded"} + if self._stok: + params["stok"] = self._stok + try: + async with session.post(url, params=params, data=mib_request, headers=headers) as resp: + return {"status": resp.status, "text": await resp.text()} + except aiohttp.ClientError as e: + return {"error": str(e)} + + +async def test_router(): + """Test the router client.""" + client = TPLinkRouterWebClient( + host="192.168.0.1", + username="admin", + password="admin", # Replace with your router password + verbose=True + ) + + try: + # Try to login + success = await client.login() + print(f"Login success: {success}") + + if success: + # Try to get device info + info = await client.get_device_info() + print(f"Device info: {info}") + + # Try to get connected devices + devices = await client.get_connected_devices() + print(f"Connected devices: {devices}") + + wireless = await client.get_wireless_clients() + print(f"Wireless clients: {wireless}") + else: + print("Login failed") + + finally: + await client.close() + + +if __name__ == "__main__": + asyncio.run(test_router()) \ No newline at end of file