From 958b19bb073e6b497fc68606eaee448e20557f95 Mon Sep 17 00:00:00 2001 From: Alex Erohin Date: Wed, 10 Apr 2024 10:37:20 +0300 Subject: [PATCH] - Added wired clients to Status - Fixed mem_usage and cpu_usage for Archer C7 v5.0 - Added IoT Wi-Fi network clients number --- README.md | 66 ++++--- setup.py | 2 +- test.py | 4 +- test/test_client.py | 374 ++++++++++++++++++++++++++++++++--- test/test_client_deco.py | 67 +++---- test/test_client_mr.py | 40 ++-- tplinkrouterc6u/__init__.py | 2 +- tplinkrouterc6u/client.py | 207 ++++++++++++------- tplinkrouterc6u/dataclass.py | 4 +- tplinkrouterc6u/enum.py | 54 +++-- 10 files changed, 633 insertions(+), 187 deletions(-) diff --git a/README.md b/README.md index 1a23ac7..7e8e36f 100644 --- a/README.md +++ b/README.md @@ -17,12 +17,12 @@ Enter the host & credentials used to log in to your router management page. User ```python from tplinkrouterc6u import ( - TplinkRouterProvider, + TplinkRouterProvider, TplinkRouter, TplinkC1200Router, TPLinkMRClient, TPLinkDecoClient, - Wifi + Connection ) from logging import Logger @@ -44,7 +44,7 @@ try: # Get status info - returns Status status = router.get_status() if not status.guest_2g_enable: # check if guest 2.4G wifi is disable - router.set_wifi(Wifi.WIFI_GUEST_2G, True) # turn on guest 2.4G wifi + router.set_wifi(Connection.GUEST_2G, True) # turn on guest 2.4G wifi # Get Address reservations, sort by ipaddr reservations = router.get_ipv4_reservations() @@ -76,17 +76,17 @@ or you have TP-link C1200 V2 or similar router you need to get web encrypted pas ## Functions | Function | Args | Description | Return | -|--|--|--|--| -| get_firmware | | Gets firmware info about the router | [Firmware](#firmware) | -| get_status | | Gets status about the router info including wifi statuses and wifi clients info | [Status](#status) | -| get_ipv4_status | | Gets WAN and LAN IPv4 status info, gateway, DNS, netmask | [IPv4Status](#IPv4Status) | -| get_ipv4_reservations | | Gets IPv4 reserved addresses (static) | [[IPv4Reservation]](#IPv4Reservation) | -| get_ipv4_dhcp_leases | | Gets IPv4 addresses assigned via DHCP | [[IPv4DHCPLease]](#IPv4DHCPLease) | -| set_wifi | wifi: [Wifi](#wifi), enable: bool | Allow to turn on/of 4 wifi networks | | -| send_sms | phone_number: str, message: str | Send sms for LTE routers | | -| reboot | | reboot router | -| authorize | | authorize for actions | -| logout | | logout after all is done | +|---|---|---|---| +| get_firmware | | Gets firmware info about the router | [Firmware](#firmware) | +| get_status | | Gets status about the router info including wifi statuses and connected devices info | [Status](#status) | +| get_ipv4_status | | Gets WAN and LAN IPv4 status info, gateway, DNS, netmask | [IPv4Status](#IPv4Status) | +| get_ipv4_reservations | | Gets IPv4 reserved addresses (static) | [[IPv4Reservation]](#IPv4Reservation) | +| get_ipv4_dhcp_leases | | Gets IPv4 addresses assigned via DHCP | [[IPv4DHCPLease]](#IPv4DHCPLease) | +| set_wifi | wifi: [Connection](#connection), enable: bool | Allow to turn on/of 4 wifi networks | | +| send_sms | phone_number: str, message: str | Send sms for LTE routers | | +| reboot | | reboot router | +| authorize | | authorize for actions | +| logout | | logout after all is done | ## Dataclass ### Firmware @@ -110,7 +110,7 @@ or you have TP-link C1200 V2 or similar router you need to get web encrypted pas | wan_ipv4_gateway | router wan ipv4 gateway | str, None | | wan_ipv4_gateway_address | router wan ipv4 gateway address | ipaddress.IPv4Address, None | | wired_total | Total amount of wired clients | int | -| wifi_clients_total | Total amount of main wifi clients | int | +| wifi_clients_total | Total amount of host wifi clients | int | | guest_clients_total | Total amount of guest wifi clients | int | | clients_total | Total amount of all connected clients | int | | iot_clients_total | Total amount of all iot connected clients | int, None | @@ -120,18 +120,18 @@ or you have TP-link C1200 V2 or similar router you need to get web encrypted pas | iot_2g_enable | Is IoT wifi 2.4G enabled | bool, None | | iot_5g_enable | Is IoT wifi 5G enabled | bool, None | | iot_6g_enable | Is IoT wifi 6G enabled | bool, None | -| wifi_2g_enable | Is main wifi 2.4G enabled | bool | -| wifi_5g_enable | Is main wifi 5G enabled | bool, None | -| wifi_6g_enable | Is main wifi 6G enabled | bool, None | +| wifi_2g_enable | Is host wifi 2.4G enabled | bool | +| wifi_5g_enable | Is host wifi 5G enabled | bool, None | +| wifi_6g_enable | Is host wifi 6G enabled | bool, None | | wan_ipv4_uptime | Internet Uptime | int, None | -| mem_usage | Memory usage | float, None | -| cpu_usage | CPU usage | float, None | -| devices | List of all wifi clients | list[[Device](#device)] | +| mem_usage | Memory usage in percentage between 0 and 1 | float, None | +| cpu_usage | CPU usage in percentage between 0 and 1 | float, None | +| devices | List of all connectedd devices | list[[Device](#device)] | ### Device | Field | Description | Type | | --- |---|---| -| type | client connection type (2.4G or 5G, guest wifi or main wifi) | [Wifi](#wifi) | +| type | client connection type (2.4G or 5G, guest wifi or host wifi, wired) | [Connection](#connection) | | macaddr | client mac address | str | | macaddress | client mac address | macaddress | | ipaddr | client ip address | str | @@ -186,16 +186,17 @@ or you have TP-link C1200 V2 or similar router you need to get web encrypted pas | remote | router remote | bool, None | ## Enum -### Wifi -- Wifi.WIFI_2G - main wifi 2.4G -- Wifi.WIFI_5G - main wifi 5G -- Wifi.WIFI_6G - main wifi 5G -- Wifi.WIFI_GUEST_2G - guest wifi 2.4G -- Wifi.WIFI_GUEST_5G - guest wifi 5G -- Wifi.WIFI_GUEST_6G - guest wifi 5G -- Wifi.WIFI_IOT_2G - IoT wifi 2.4G -- Wifi.WIFI_IOT_5G - IoT wifi 5G -- Wifi.WIFI_IOT_6G - IoT wifi 6G +### Connection +- Connection.HOST_2G - host wifi 2.4G +- Connection.HOST_5G - host wifi 5G +- Connection.HOST_6G - host wifi 5G +- Connection.GUEST_2G - guest wifi 2.4G +- Connection.GUEST_5G - guest wifi 5G +- Connection.GUEST_6G - guest wifi 5G +- Connection.IOT_2G - IoT wifi 2.4G +- Connection.IOT_5G - IoT wifi 5G +- Connection.IOT_6G - IoT wifi 6G +- Connection.WIRED - Wired ## Supported routers ### Fully tested Hardware Versions @@ -204,6 +205,7 @@ or you have TP-link C1200 V2 or similar router you need to get web encrypted pas - Archer AX12 v1.0 - Archer AX20 v1.0 - Archer AX21 v1.20 +- Archer AX23 v1.0 - Archer AX50 v1.0 - Archer AX55 v1.0 - Archer AX55 V1.60 diff --git a/setup.py b/setup.py index e6cb1d1..79a93d3 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setuptools.setup( name="tplinkrouterc6u", - version="3.5.0", + version="4.0.0", author="Alex Erohin", author_email="alexanderErohin@yandex.ru", description="TP-Link Router API", diff --git a/test.py b/test.py index d45efe3..f974533 100644 --- a/test.py +++ b/test.py @@ -4,7 +4,7 @@ import macaddress import ipaddress from typing import TypeAlias -from tplinkrouterc6u import Wifi, TplinkRouterProvider +from tplinkrouterc6u import Connection, TplinkRouterProvider from tplinkrouterc6u.dataclass import Status, Device from mac_vendor_lookup import MacLookup, BaseMacLookup import pprint @@ -36,7 +36,7 @@ def lookup(mac): def get_device() -> Device: - d = Device(Wifi.WIFI_2G, macaddress.EUI48("11-22-33-44-55-66"), ipaddress.IPv4Address("192.168.0.1"), "router") + d = Device(Connection.HOST_2G, macaddress.EUI48("11-22-33-44-55-66"), ipaddress.IPv4Address("192.168.0.1"), "router") return d diff --git a/test/test_client.py b/test/test_client.py index e741647..e112764 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -4,9 +4,10 @@ import json from tplinkrouterc6u import ( TplinkRouter, - Wifi, + Connection, Status, Device, + ClientException, ) @@ -206,8 +207,9 @@ def request(self, path: str, data: str, ignore_response: bool = False, ignore_errors: bool = False) -> dict | None: if path == 'admin/status?form=all&operation=read': return json.loads(response_status)['data'] - else: + elif path == 'admin/wireless?form=statistics': return json.loads(response_stats)['data'] + raise ClientException() client = TPLinkRouterTest('', '') status = client.get_status() @@ -226,6 +228,7 @@ def request(self, path: str, data: str, self.assertEqual(status.wifi_clients_total, 2) self.assertEqual(status.guest_clients_total, 0) self.assertEqual(status.clients_total, 4) + self.assertEqual(status.iot_clients_total, None) self.assertEqual(status.guest_2g_enable, False) self.assertEqual(status.guest_5g_enable, False) self.assertEqual(status.iot_2g_enable, None) @@ -234,26 +237,40 @@ def request(self, path: str, data: str, self.assertEqual(status.wifi_5g_enable, True) self.assertEqual(status.wan_ipv4_uptime, None) self.assertEqual(status.mem_usage, 0.43) - self.assertEqual(status.cpu_usage, 0.07) - self.assertEqual(len(status.devices), 2) + self.assertEqual(status.cpu_usage, 0.28) + self.assertEqual(len(status.devices), 4) self.assertIsInstance(status.devices[0], Device) - self.assertEqual(status.devices[0].type, Wifi.WIFI_2G) - self.assertEqual(status.devices[0].macaddr, '06-82-9D-2B-8F-C6') + self.assertEqual(status.devices[0].type, Connection.WIRED) + self.assertEqual(status.devices[0].macaddr, '3D-24-25-24-30-79') self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[0].ipaddr, '192.168.1.186') + self.assertEqual(status.devices[0].ipaddr, '192.168.1.228') self.assertIsInstance(status.devices[0].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[0].hostname, 'UNKNOWN') - self.assertEqual(status.devices[0].packets_sent, 450333) - self.assertEqual(status.devices[0].packets_received, 4867482) - self.assertIsInstance(status.devices[1], Device) - self.assertEqual(status.devices[1].type, Wifi.WIFI_5G) - self.assertEqual(status.devices[1].macaddr, '1F-7A-BD-F7-20-0D') + self.assertEqual(status.devices[0].hostname, 'SERVER') + self.assertEqual(status.devices[0].packets_sent, None) + self.assertEqual(status.devices[0].packets_received, None) + self.assertIsInstance(status.devices[0], Device) + self.assertEqual(status.devices[1].type, Connection.WIRED) + self.assertEqual(status.devices[1].macaddr, 'AC-04-D6-25-2A-96') self.assertIsInstance(status.devices[1].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[1].ipaddr, '0.0.0.0') + self.assertEqual(status.devices[1].ipaddr, '192.168.1.254') self.assertIsInstance(status.devices[1].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[1].hostname, '') - self.assertEqual(status.devices[1].packets_sent, 134815) - self.assertEqual(status.devices[1].packets_received, 2953078) + self.assertEqual(status.devices[1].hostname, 'UNKNOWN') + self.assertEqual(status.devices[1].packets_sent, None) + self.assertEqual(status.devices[1].packets_received, None) + self.assertIsInstance(status.devices[2], Device) + self.assertEqual(status.devices[2].type, Connection.HOST_2G) + self.assertEqual(status.devices[2].macaddr, '06-82-9D-2B-8F-C6') + self.assertEqual(status.devices[2].ipaddr, '192.168.1.186') + self.assertEqual(status.devices[2].hostname, 'UNKNOWN') + self.assertEqual(status.devices[2].packets_sent, 450333) + self.assertEqual(status.devices[2].packets_received, 4867482) + self.assertIsInstance(status.devices[3], Device) + self.assertEqual(status.devices[3].type, Connection.HOST_5G) + self.assertEqual(status.devices[3].macaddr, '1F-7A-BD-F7-20-0D') + self.assertEqual(status.devices[3].ipaddr, '0.0.0.0') + self.assertEqual(status.devices[3].hostname, '') + self.assertEqual(status.devices[3].packets_sent, 134815) + self.assertEqual(status.devices[3].packets_received, 2953078) def test_get_status_ax_55(self) -> None: response_status = ''' @@ -305,8 +322,9 @@ def request(self, path: str, data: str, ignore_response: bool = False, ignore_errors: bool = False) -> dict | None: if path == 'admin/status?form=all&operation=read': return json.loads(response_status)['data'] - else: + elif path == 'admin/wireless?form=statistics': return json.loads(response_stats)['data'] + raise ClientException() client = TPLinkRouterTest('', '') status = client.get_status() @@ -322,6 +340,7 @@ def request(self, path: str, data: str, self.assertEqual(status.wifi_clients_total, 2) self.assertEqual(status.guest_clients_total, 0) self.assertEqual(status.clients_total, 4) + self.assertEqual(status.iot_clients_total, None) self.assertEqual(status.guest_2g_enable, True) self.assertEqual(status.guest_5g_enable, None) self.assertEqual(status.guest_6g_enable, None) @@ -334,25 +353,326 @@ def request(self, path: str, data: str, self.assertEqual(status.wan_ipv4_uptime, None) self.assertEqual(status.mem_usage, None) self.assertEqual(status.cpu_usage, None) - self.assertEqual(len(status.devices), 2) + self.assertEqual(len(status.devices), 4) self.assertIsInstance(status.devices[0], Device) - self.assertEqual(status.devices[0].type, Wifi.WIFI_2G) - self.assertEqual(status.devices[0].macaddr, '06-82-9D-2B-8F-C6') + self.assertEqual(status.devices[0].type, Connection.WIRED) + self.assertEqual(status.devices[0].macaddr, '3D-24-25-24-30-79') self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[0].ipaddr, '192.168.1.186') + self.assertEqual(status.devices[0].ipaddr, '192.168.1.228') self.assertIsInstance(status.devices[0].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[0].hostname, 'UNKNOWN') + self.assertEqual(status.devices[0].hostname, 'SERVER') self.assertEqual(status.devices[0].packets_sent, None) self.assertEqual(status.devices[0].packets_received, None) self.assertIsInstance(status.devices[1], Device) - self.assertEqual(status.devices[1].type, Wifi.WIFI_UNKNOWN) - self.assertEqual(status.devices[1].macaddr, '1F-7A-BD-F7-20-0D') + self.assertEqual(status.devices[1].type, Connection.WIRED) + self.assertEqual(status.devices[1].macaddr, 'AC-04-D6-25-2A-96') self.assertIsInstance(status.devices[1].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[1].ipaddr, '0.0.0.0') + self.assertEqual(status.devices[1].ipaddr, '192.168.1.254') self.assertIsInstance(status.devices[1].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[1].hostname, '') + self.assertEqual(status.devices[1].hostname, 'UNKNOWN') self.assertEqual(status.devices[1].packets_sent, None) self.assertEqual(status.devices[1].packets_received, None) + self.assertIsInstance(status.devices[2], Device) + self.assertEqual(status.devices[2].type, Connection.HOST_2G) + self.assertEqual(status.devices[2].macaddr, '06-82-9D-2B-8F-C6') + self.assertIsInstance(status.devices[2].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[2].ipaddr, '192.168.1.186') + self.assertIsInstance(status.devices[2].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[2].hostname, 'UNKNOWN') + self.assertEqual(status.devices[2].packets_sent, None) + self.assertEqual(status.devices[2].packets_received, None) + self.assertIsInstance(status.devices[3], Device) + self.assertEqual(status.devices[3].type, Connection.UNKNOWN) + self.assertEqual(status.devices[3].macaddr, '1F-7A-BD-F7-20-0D') + self.assertEqual(status.devices[3].ipaddr, '0.0.0.0') + self.assertEqual(status.devices[3].hostname, '') + self.assertEqual(status.devices[3].packets_sent, None) + self.assertEqual(status.devices[3].packets_received, None) + + def test_get_status_with_game_accelerator(self) -> None: + response_status = ''' +{ + "success": true, + "data": { + "lan_macaddr": "06:e6:97:9e:23:f5", + "access_devices_wired": [ + { + "wire_type": "wired", + "macaddr": "3d:24:25:24:30:79", + "ipaddr": "192.168.1.228", + "hostname": "SERVER" + }, + { + "wire_type": "wired", + "macaddr": "ac:04:d6:25:2a:96", + "ipaddr": "192.168.1.254", + "hostname": "UNKNOWN" + } + ], + "access_devices_wireless_host": [ + { + "wire_type": "2.4G", + "macaddr": "06:82:9d:2b:8f:c6", + "ipaddr": "192.168.1.186", + "hostname": "UNKNOWN" + } + ], + "guest_2g_enable": "on", + "wireless_2g_enable": "on" + } +} +''' + response_game_accelerator = ''' + { + "data": [ + {"mac": "06:82:9d:2b:8f:c6", "deviceTag":"2.4G", "isGuest":false, "ip":"192.168.1.186", "deviceName":"name1"}, + {"mac": "fb:90:b8:2a:8a:b1", "deviceTag":"iot_2.4G", "isGuest":false, "ip":"192.168.1.187", "deviceName":"name2"}, + {"mac": "54:b3:a2:f7:be:ea", "deviceTag":"iot_5G", "isGuest":false, "ip":"192.168.1.188", "deviceName":"name3"}, + {"mac": "3c:ae:e1:83:94:9d", "deviceTag":"iot_6G", "isGuest":false, "ip":"192.168.1.189", "deviceName":"name4"} + ], + "timeout": false, + "success": true + } +''' + response_stats = ''' + { + "data": [ + { + "mac": "06:82:9d:2b:8f:c6", + "type": "2.4GHz", + "encryption": "wpa/wpa2-psk", + "rxpkts": 4867482, + "txpkts": 450333 + }, + { + "mac": "1f:7a:bd:f7:20:0d", + "type": "5GHz", + "encryption": "wpa/wpa2-psk", + "rxpkts": 2953078, + "txpkts": 134815 + } + ], + "timeout": false, + "success": true, + "operator": "load" + } +''' + + class TPLinkRouterTest(TplinkRouter): + def request(self, path: str, data: str, + ignore_response: bool = False, ignore_errors: bool = False) -> dict | None: + if path == 'admin/status?form=all&operation=read': + return json.loads(response_status)['data'] + elif path == 'admin/smart_network?form=game_accelerator': + return json.loads(response_game_accelerator)['data'] + elif path == 'admin/wireless?form=statistics': + return json.loads(response_stats)['data'] + raise ClientException() + + client = TPLinkRouterTest('', '') + status = client.get_status() + + self.assertIsInstance(status, Status) + self.assertEqual(status.wan_macaddr, None) + self.assertEqual(status.lan_macaddr, '06-E6-97-9E-23-F5') + self.assertIsInstance(status.lan_macaddress, macaddress.EUI48) + self.assertEqual(status.wan_ipv4_addr, None) + self.assertEqual(status.lan_ipv4_addr, None) + self.assertEqual(status.wan_ipv4_gateway, None) + self.assertEqual(status.wired_total, 2) + self.assertEqual(status.wifi_clients_total, 2) + self.assertEqual(status.guest_clients_total, 0) + self.assertEqual(status.clients_total, 4) + self.assertEqual(status.iot_clients_total, 3) + self.assertEqual(status.guest_2g_enable, True) + self.assertEqual(status.guest_5g_enable, None) + self.assertEqual(status.guest_6g_enable, None) + self.assertEqual(status.iot_2g_enable, None) + self.assertEqual(status.iot_5g_enable, None) + self.assertEqual(status.iot_6g_enable, None) + self.assertEqual(status.wifi_2g_enable, True) + self.assertEqual(status.wifi_5g_enable, None) + self.assertEqual(status.wifi_6g_enable, None) + self.assertEqual(status.wan_ipv4_uptime, None) + self.assertEqual(status.mem_usage, None) + self.assertEqual(status.cpu_usage, None) + self.assertEqual(len(status.devices), 7) + self.assertIsInstance(status.devices[0], Device) + self.assertEqual(status.devices[0].type, Connection.WIRED) + self.assertEqual(status.devices[0].macaddr, '3D-24-25-24-30-79') + self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[0].ipaddr, '192.168.1.228') + self.assertIsInstance(status.devices[0].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[0].hostname, 'SERVER') + self.assertEqual(status.devices[0].packets_sent, None) + self.assertEqual(status.devices[0].packets_received, None) + self.assertIsInstance(status.devices[1], Device) + self.assertEqual(status.devices[1].type, Connection.WIRED) + self.assertEqual(status.devices[1].macaddr, 'AC-04-D6-25-2A-96') + self.assertIsInstance(status.devices[1].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[1].ipaddr, '192.168.1.254') + self.assertIsInstance(status.devices[1].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[1].hostname, 'UNKNOWN') + self.assertEqual(status.devices[1].packets_sent, None) + self.assertEqual(status.devices[1].packets_received, None) + self.assertIsInstance(status.devices[2], Device) + self.assertEqual(status.devices[2].type, Connection.HOST_2G) + self.assertEqual(status.devices[2].macaddr, '06-82-9D-2B-8F-C6') + self.assertIsInstance(status.devices[2].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[2].ipaddr, '192.168.1.186') + self.assertIsInstance(status.devices[2].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[2].hostname, 'UNKNOWN') + self.assertEqual(status.devices[2].packets_sent, 450333) + self.assertEqual(status.devices[2].packets_received, 4867482) + self.assertIsInstance(status.devices[3], Device) + self.assertEqual(status.devices[3].type, Connection.IOT_2G) + self.assertEqual(status.devices[3].macaddr, 'FB-90-B8-2A-8A-B1') + self.assertIsInstance(status.devices[3].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[3].ipaddr, '192.168.1.187') + self.assertIsInstance(status.devices[3].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[3].hostname, 'name2') + self.assertEqual(status.devices[3].packets_sent, None) + self.assertEqual(status.devices[3].packets_received, None) + self.assertIsInstance(status.devices[4], Device) + self.assertEqual(status.devices[4].type, Connection.IOT_5G) + self.assertEqual(status.devices[4].macaddr, '54-B3-A2-F7-BE-EA') + self.assertIsInstance(status.devices[4].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[4].ipaddr, '192.168.1.188') + self.assertIsInstance(status.devices[4].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[4].hostname, 'name3') + self.assertEqual(status.devices[4].packets_sent, None) + self.assertEqual(status.devices[4].packets_received, None) + self.assertIsInstance(status.devices[5], Device) + self.assertEqual(status.devices[5].type, Connection.IOT_6G) + self.assertEqual(status.devices[5].macaddr, '3C-AE-E1-83-94-9D') + self.assertIsInstance(status.devices[5].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[5].ipaddr, '192.168.1.189') + self.assertIsInstance(status.devices[5].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[5].hostname, 'name4') + self.assertEqual(status.devices[5].packets_sent, None) + self.assertEqual(status.devices[5].packets_received, None) + self.assertIsInstance(status.devices[6], Device) + self.assertEqual(status.devices[6].type, Connection.HOST_5G) + self.assertEqual(status.devices[6].macaddr, '1F-7A-BD-F7-20-0D') + self.assertIsInstance(status.devices[6].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[6].ipaddr, '0.0.0.0') + self.assertIsInstance(status.devices[6].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[6].hostname, '') + self.assertEqual(status.devices[6].packets_sent, 134815) + self.assertEqual(status.devices[6].packets_received, 2953078) + + def test_get_status_with_perf_request(self) -> None: + response_status = ''' + { + "success": true, + "data": { + "lan_macaddr": "06:e6:97:9e:23:f5", + "guest_2g_enable": "on", + "wireless_2g_enable": "on" + } + } + ''' + perf_stats = ''' + { + "data": {"mem_usage":0.47, "cpu_usage":0.25}, + "timeout": false, + "success": true, + "operator": "load" + } + ''' + response_stats = ''' + { + "data": [], + "timeout": false, + "success": true, + "operator": "load" + } + ''' + + class TPLinkRouterTest(TplinkRouter): + def request(self, path: str, data: str, + ignore_response: bool = False, ignore_errors: bool = False) -> dict | None: + if path == 'admin/status?form=all&operation=read': + return json.loads(response_status)['data'] + elif path == 'admin/status?form=perf&operation=read': + return json.loads(perf_stats)['data'] + elif path == 'admin/wireless?form=statistics': + return json.loads(response_stats)['data'] + raise ClientException() + + client = TPLinkRouterTest('', '') + status = client.get_status() + + self.assertIsInstance(status, Status) + self.assertEqual(status.wan_macaddr, None) + self.assertEqual(status.lan_macaddr, '06-E6-97-9E-23-F5') + self.assertIsInstance(status.lan_macaddress, macaddress.EUI48) + self.assertEqual(status.wan_ipv4_addr, None) + self.assertEqual(status.lan_ipv4_addr, None) + self.assertEqual(status.wan_ipv4_gateway, None) + self.assertEqual(status.wired_total, 0) + self.assertEqual(status.wifi_clients_total, 0) + self.assertEqual(status.guest_clients_total, 0) + self.assertEqual(status.clients_total, 0) + self.assertEqual(status.iot_clients_total, None) + self.assertEqual(status.guest_2g_enable, True) + self.assertEqual(status.guest_5g_enable, None) + self.assertEqual(status.guest_6g_enable, None) + self.assertEqual(status.iot_2g_enable, None) + self.assertEqual(status.iot_5g_enable, None) + self.assertEqual(status.iot_6g_enable, None) + self.assertEqual(status.wifi_2g_enable, True) + self.assertEqual(status.wifi_5g_enable, None) + self.assertEqual(status.wifi_6g_enable, None) + self.assertEqual(status.wan_ipv4_uptime, None) + self.assertEqual(status.mem_usage, 0.47) + self.assertEqual(status.cpu_usage, 0.25) + self.assertEqual(len(status.devices), 0) + + def test_set_wifi(self) -> None: + check_url = '' + check_data = '' + + class TPLinkRouterTest(TplinkRouter): + def request(self, path: str, data: str, + ignore_response: bool = False, ignore_errors: bool = False) -> dict | None: + nonlocal check_url, check_data + check_url = path + check_data = data + return None + + client = TPLinkRouterTest('', '') + result = client.set_wifi(Connection.HOST_2G, False) + self.assertIsNone(result) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=wireless_2g') + self.assertEqual(check_data, 'operation=write&wireless_2g_enable=off') + client.set_wifi(Connection.HOST_2G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=wireless_2g') + self.assertEqual(check_data, 'operation=write&wireless_2g_enable=on') + client.set_wifi(Connection.HOST_5G, False) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=wireless_5g') + self.assertEqual(check_data, 'operation=write&wireless_5g_enable=off') + client.set_wifi(Connection.HOST_6G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=wireless_6g') + self.assertEqual(check_data, 'operation=write&wireless_6g_enable=on') + client.set_wifi(Connection.GUEST_2G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=guest_2g') + self.assertEqual(check_data, 'operation=write&guest_2g_enable=on') + client.set_wifi(Connection.GUEST_5G, False) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=guest_5g') + self.assertEqual(check_data, 'operation=write&guest_5g_enable=off') + client.set_wifi(Connection.GUEST_6G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=guest_6g') + self.assertEqual(check_data, 'operation=write&guest_6g_enable=on') + client.set_wifi(Connection.IOT_2G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=iot_2g') + self.assertEqual(check_data, 'operation=write&iot_2g_enable=on') + client.set_wifi(Connection.IOT_5G, False) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=iot_5g') + self.assertEqual(check_data, 'operation=write&iot_5g_enable=off') + client.set_wifi(Connection.IOT_6G, True) + self.assertEqual(check_url, 'admin/wireless?&form=guest&form=iot_6g') + self.assertEqual(check_data, 'operation=write&iot_6g_enable=on') if __name__ == '__main__': diff --git a/test/test_client_deco.py b/test/test_client_deco.py index d4f3b79..5e55221 100644 --- a/test/test_client_deco.py +++ b/test/test_client_deco.py @@ -4,7 +4,7 @@ import ipaddress from tplinkrouterc6u import ( TPLinkDecoClient, - Wifi, + Connection, Firmware, Status, Device, @@ -116,9 +116,9 @@ def request(self, path: str, data: str, self.assertEqual(status.wan_ipv4_uptime, None) self.assertEqual(status.mem_usage, 0.43) self.assertEqual(status.cpu_usage, 0.1) - self.assertEqual(len(status.devices), 4) + self.assertEqual(len(status.devices), 5) self.assertIsInstance(status.devices[0], Device) - self.assertEqual(status.devices[0].type, Wifi.WIFI_5G) + self.assertEqual(status.devices[0].type, Connection.HOST_5G) self.assertEqual(status.devices[0].macaddr, 'CF-51-C9-04-E1-02') self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) self.assertEqual(status.devices[0].ipaddr, '192.168.68.101') @@ -127,32 +127,33 @@ def request(self, path: str, data: str, self.assertEqual(status.devices[0].packets_sent, None) self.assertEqual(status.devices[0].packets_received, None) self.assertIsInstance(status.devices[1], Device) - self.assertEqual(status.devices[1].type, Wifi.WIFI_2G) - self.assertEqual(status.devices[1].macaddr, '6B-35-FE-21-A7-73') - self.assertIsInstance(status.devices[1].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[1].ipaddr, '192.168.68.103') - self.assertIsInstance(status.devices[1].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[1].hostname, 'wireless3') + self.assertEqual(status.devices[1].type, Connection.WIRED) + self.assertEqual(status.devices[1].macaddr, '5F-F8-08-28-AF-54') + self.assertEqual(status.devices[1].ipaddr, '192.168.68.100') + self.assertEqual(status.devices[1].hostname, 'wired1') self.assertEqual(status.devices[1].packets_sent, None) self.assertEqual(status.devices[1].packets_received, None) self.assertIsInstance(status.devices[2], Device) - self.assertEqual(status.devices[2].type, Wifi.WIFI_GUEST_5G) - self.assertEqual(status.devices[2].macaddr, '19-90-F7-61-66-B2') - self.assertIsInstance(status.devices[2].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[2].ipaddr, '192.168.68.104') - self.assertIsInstance(status.devices[2].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[2].hostname, 'wireless4') + self.assertEqual(status.devices[2].type, Connection.HOST_2G) + self.assertEqual(status.devices[2].macaddr, '6B-35-FE-21-A7-73') + self.assertEqual(status.devices[2].ipaddr, '192.168.68.103') + self.assertEqual(status.devices[2].hostname, 'wireless3') self.assertEqual(status.devices[2].packets_sent, None) self.assertEqual(status.devices[2].packets_received, None) self.assertIsInstance(status.devices[3], Device) - self.assertEqual(status.devices[3].type, Wifi.WIFI_GUEST_2G) - self.assertEqual(status.devices[3].macaddr, '56-32-C3-DE-CE-F0') - self.assertIsInstance(status.devices[3].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[3].ipaddr, '192.168.68.105') - self.assertIsInstance(status.devices[3].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[3].hostname, 'wireless5') + self.assertEqual(status.devices[3].type, Connection.GUEST_5G) + self.assertEqual(status.devices[3].macaddr, '19-90-F7-61-66-B2') + self.assertEqual(status.devices[3].ipaddr, '192.168.68.104') + self.assertEqual(status.devices[3].hostname, 'wireless4') self.assertEqual(status.devices[3].packets_sent, None) self.assertEqual(status.devices[3].packets_received, None) + self.assertIsInstance(status.devices[4], Device) + self.assertEqual(status.devices[4].type, Connection.GUEST_2G) + self.assertEqual(status.devices[4].macaddr, '56-32-C3-DE-CE-F0') + self.assertEqual(status.devices[4].ipaddr, '192.168.68.105') + self.assertEqual(status.devices[4].hostname, 'wireless5') + self.assertEqual(status.devices[4].packets_sent, None) + self.assertEqual(status.devices[4].packets_received, None) def test_get_status_iot(self) -> None: response_network = ''' @@ -243,20 +244,20 @@ def request(self, path: str, data: str, self.assertEqual(len(status.devices), 5) self.assertIsInstance(status.devices[0], Device) - self.assertEqual(status.devices[0].type, Wifi.WIFI_5G) + self.assertEqual(status.devices[0].type, Connection.HOST_5G) self.assertEqual(status.devices[0].macaddr, 'CF-51-C9-04-E1-02') self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) self.assertIsInstance(status.devices[1], Device) - self.assertEqual(status.devices[1].type, Wifi.WIFI_IOT_2G) + self.assertEqual(status.devices[1].type, Connection.IOT_2G) self.assertEqual(status.devices[1].macaddr, '5F-F8-08-28-AF-54') self.assertIsInstance(status.devices[2], Device) - self.assertEqual(status.devices[2].type, Wifi.WIFI_IOT_5G) + self.assertEqual(status.devices[2].type, Connection.IOT_5G) self.assertEqual(status.devices[2].macaddr, '5F-F8-08-28-AF-55') self.assertIsInstance(status.devices[3], Device) - self.assertEqual(status.devices[3].type, Wifi.WIFI_IOT_6G) + self.assertEqual(status.devices[3].type, Connection.IOT_6G) self.assertEqual(status.devices[3].macaddr, '5F-F8-08-28-AF-56') self.assertIsInstance(status.devices[4], Device) - self.assertEqual(status.devices[4].type, Wifi.WIFI_UNKNOWN) + self.assertEqual(status.devices[4].type, Connection.UNKNOWN) self.assertEqual(status.devices[4].macaddr, '5F-F8-08-28-AF-57') def test_get_status_no_internet(self) -> None: @@ -478,21 +479,21 @@ def request(self, path: str, data: str, check_data = data client = TPLinkRouterTest('', '') - result = client.set_wifi(Wifi.WIFI_2G, False) + result = client.set_wifi(Connection.HOST_2G, False) self.assertIsNone(result) self.assertEqual(check_url, 'admin/wireless?form=wlan') self.assertEqual(check_data, '{"operation": "write", "params": {"band2_4": {"host": {"enable": false}}}}') - client.set_wifi(Wifi.WIFI_2G, True) + client.set_wifi(Connection.HOST_2G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band2_4": {"host": {"enable": true}}}}') - client.set_wifi(Wifi.WIFI_5G, True) + client.set_wifi(Connection.HOST_5G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band5_1": {"host": {"enable": true}}}}') - client.set_wifi(Wifi.WIFI_GUEST_2G, True) + client.set_wifi(Connection.GUEST_2G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band2_4": {"guest": {"enable": true}}}}') - client.set_wifi(Wifi.WIFI_GUEST_5G, True) + client.set_wifi(Connection.GUEST_5G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band5_1": {"guest": {"enable": true}}}}') - client.set_wifi(Wifi.WIFI_6G, True) + client.set_wifi(Connection.HOST_6G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band6": {"host": {"enable": true}}}}') - client.set_wifi(Wifi.WIFI_GUEST_6G, True) + client.set_wifi(Connection.GUEST_6G, True) self.assertEqual(check_data, '{"operation": "write", "params": {"band6": {"guest": {"enable": true}}}}') def test_reboot_with_firmware(self) -> None: diff --git a/test/test_client_mr.py b/test/test_client_mr.py index 420ced1..dc30d82 100644 --- a/test/test_client_mr.py +++ b/test/test_client_mr.py @@ -3,7 +3,7 @@ import ipaddress from tplinkrouterc6u import ( TPLinkMRClient, - Wifi, + Connection, Firmware, Status, Device, @@ -181,16 +181,25 @@ def _request(self, url, method='POST', data_str=None, encrypt=False): self.assertEqual(status.wan_ipv4_uptime, None) self.assertEqual(status.mem_usage, None) self.assertEqual(status.cpu_usage, None) - self.assertEqual(len(status.devices), 1) + self.assertEqual(len(status.devices), 2) self.assertIsInstance(status.devices[0], Device) - self.assertEqual(status.devices[0].type, Wifi.WIFI_5G) - self.assertEqual(status.devices[0].macaddr, 'F4-A3-86-2D-41-B5') + self.assertEqual(status.devices[0].type, Connection.WIRED) + self.assertEqual(status.devices[0].macaddr, '66-E2-02-BD-B5-1B') self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) - self.assertEqual(status.devices[0].ipaddr, '192.168.30.11') + self.assertEqual(status.devices[0].ipaddr, '192.168.30.10') self.assertIsInstance(status.devices[0].ipaddress, ipaddress.IPv4Address) - self.assertEqual(status.devices[0].hostname, 'host2') - self.assertEqual(status.devices[0].packets_sent, 176) - self.assertEqual(status.devices[0].packets_received, 467) + self.assertEqual(status.devices[0].hostname, 'host1') + self.assertEqual(status.devices[0].packets_sent, None) + self.assertEqual(status.devices[0].packets_received, None) + self.assertIsInstance(status.devices[1], Device) + self.assertEqual(status.devices[1].type, Connection.HOST_5G) + self.assertEqual(status.devices[1].macaddr, 'F4-A3-86-2D-41-B5') + self.assertIsInstance(status.devices[1].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[1].ipaddr, '192.168.30.11') + self.assertIsInstance(status.devices[1].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[1].hostname, 'host2') + self.assertEqual(status.devices[1].packets_sent, 176) + self.assertEqual(status.devices[1].packets_received, 467) def test_get_status_without_5G(self) -> None: response = '''[1,1,0,0,0,0]0 @@ -245,7 +254,16 @@ def _request(self, url, method='POST', data_str=None, encrypt=False): self.assertEqual(status.wan_ipv4_uptime, None) self.assertEqual(status.mem_usage, None) self.assertEqual(status.cpu_usage, None) - self.assertEqual(len(status.devices), 0) + self.assertEqual(len(status.devices), 1) + self.assertIsInstance(status.devices[0], Device) + self.assertEqual(status.devices[0].type, Connection.WIRED) + self.assertEqual(status.devices[0].macaddr, '66-E2-02-BD-B5-1B') + self.assertIsInstance(status.devices[0].macaddress, macaddress.EUI48) + self.assertEqual(status.devices[0].ipaddr, '192.168.30.10') + self.assertIsInstance(status.devices[0].ipaddress, ipaddress.IPv4Address) + self.assertEqual(status.devices[0].hostname, 'host1') + self.assertEqual(status.devices[0].packets_sent, None) + self.assertEqual(status.devices[0].packets_received, None) def test_get_status_with_wlan_dev(self) -> None: response = ''' @@ -303,7 +321,7 @@ def _request(self, url, method='POST', data_str=None, encrypt=False): self.assertEqual(status.mem_usage, None) self.assertEqual(status.cpu_usage, None) self.assertEqual(len(status.devices), 1) - self.assertEqual(status.devices[0].type, Wifi.WIFI_2G) + self.assertEqual(status.devices[0].type, Connection.HOST_2G) self.assertEqual(status.devices[0].macaddr, 'F4-A3-86-2D-41-B8') self.assertEqual(status.devices[0].ipaddr, '0.0.0.0') self.assertEqual(status.devices[0].hostname, '') @@ -586,7 +604,7 @@ def _request(self, url, method='POST', data_str=None, encrypt=False): return 200, response client = TPLinkMRClientTest('', '') - client.set_wifi(Wifi.WIFI_2G, True) + client.set_wifi(Connection.HOST_2G, True) self.assertIn('http:///cgi_gdpr?_=', check_url) self.assertEqual(check_data, '2\r\n[LAN_WLAN#1,1,0,0,0,0#0,0,0,0,0,0]0,1\r\nenable=1\r\n') diff --git a/tplinkrouterc6u/__init__.py b/tplinkrouterc6u/__init__.py index 7ba7d94..ae6720b 100644 --- a/tplinkrouterc6u/__init__.py +++ b/tplinkrouterc6u/__init__.py @@ -6,7 +6,7 @@ AbstractRouter, TPLinkDecoClient, ) -from tplinkrouterc6u.enum import Wifi +from tplinkrouterc6u.enum import Connection from tplinkrouterc6u.dataclass import ( Firmware, Status, diff --git a/tplinkrouterc6u/client.py b/tplinkrouterc6u/client.py index 94b1f21..fb0ce2a 100644 --- a/tplinkrouterc6u/client.py +++ b/tplinkrouterc6u/client.py @@ -10,7 +10,7 @@ import ipaddress from logging import Logger from tplinkrouterc6u.encryption import EncryptionWrapper, EncryptionWrapperMR -from tplinkrouterc6u.enum import Wifi +from tplinkrouterc6u.enum import Connection from tplinkrouterc6u.dataclass import Firmware, Status, Device, IPv4Reservation, IPv4DHCPLease, IPv4Status from tplinkrouterc6u.exception import ClientException, ClientError from abc import ABC, abstractmethod @@ -55,7 +55,7 @@ def reboot(self) -> None: pass @abstractmethod - def set_wifi(self, wifi: Wifi, enable: bool) -> None: + def set_wifi(self, wifi: Connection, enable: bool) -> None: pass @@ -265,12 +265,14 @@ def _decrypt_response(self, data: dict) -> dict: class TplinkBaseRouter(AbstractRouter, TplinkRequest): + _smart_network = True + _perf_status = True + def __init__(self, host: str, password: str, username: str = 'admin', logger: Logger = None, verify_ssl: bool = True, timeout: int = 10) -> None: super().__init__(host, password, username, logger, verify_ssl, timeout) self._url_firmware = 'admin/firmware?form=upgrade&operation=read' - self._url_wireless_stats = 'admin/wireless?form=statistics' self._url_ipv4_reservations = 'admin/dhcps?form=reservation&operation=load' self._url_ipv4_dhcp_leases = 'admin/dhcps?form=client&operation=load' referer = '{}/webpages/index.html'.format(self.host) @@ -281,9 +283,21 @@ def __init__(self, host: str, password: str, username: str = 'admin', logger: Lo def authorize(self) -> bool: pass - def set_wifi(self, wifi: Wifi, enable: bool) -> None: - path = f"admin/wireless?&form=guest&form={wifi.value}" - data = f"operation=write&{wifi.value}_enable={'on' if enable else 'off'}" + def set_wifi(self, wifi: Connection, enable: bool) -> None: + values = { + Connection.HOST_2G: 'wireless_2g', + Connection.HOST_5G: 'wireless_5g', + Connection.HOST_6G: 'wireless_6g', + Connection.GUEST_2G: 'guest_2g', + Connection.GUEST_5G: 'guest_5g', + Connection.GUEST_6G: 'guest_6g', + Connection.IOT_2G: 'iot_2g', + Connection.IOT_5G: 'iot_5g', + Connection.IOT_6G: 'iot_6g', + } + value = values.get(wifi) + path = f"admin/wireless?&form=guest&form={value}" + data = f"operation=write&{value}_enable={'on' if enable else 'off'}" self.request(path, data) def reboot(self) -> None: @@ -302,12 +316,6 @@ def get_firmware(self) -> Firmware: return firmware def get_status(self) -> Status: - - def _calc_cpu_usage(data: dict) -> float | None: - cpu_usage = (data.get('cpu_usage', 0) + data.get('cpu1_usage', 0) - + data.get('cpu2_usage', 0) + data.get('cpu3_usage', 0)) - return cpu_usage / 4 if cpu_usage != 0 else None - data = self.request('admin/status?form=all&operation=read', 'operation=read') status = Status() status._wan_macaddr = macaddress.EUI48(data['wan_macaddr']) if 'wan_macaddr' in data else None @@ -318,7 +326,7 @@ def _calc_cpu_usage(data: dict) -> float | None: data['wan_ipv4_gateway']) if 'wan_ipv4_gateway' in data else None status.wan_ipv4_uptime = data.get('wan_ipv4_uptime') status.mem_usage = data.get('mem_usage') - status.cpu_usage = _calc_cpu_usage(data) + status.cpu_usage = data.get('cpu_usage') status.wired_total = len(data.get('access_devices_wired', [])) status.wifi_clients_total = len(data.get('access_devices_wireless_host', [])) status.guest_clients_total = len(data.get('access_devices_wireless_guest', [])) @@ -332,13 +340,25 @@ def _calc_cpu_usage(data: dict) -> float | None: status.wifi_5g_enable = self._str2bool(data.get('wireless_5g_enable')) status.wifi_6g_enable = self._str2bool(data.get('wireless_6g_enable')) + if (status.mem_usage is None or status.mem_usage is None) and self._perf_status: + try: + performance = self.request('admin/status?form=perf&operation=read', 'operation=read') + status.mem_usage = performance.get('mem_usage') + status.cpu_usage = performance.get('cpu_usage') + except: + self._perf_status = False + devices = {} - def _add_device(type: Wifi, item: dict) -> None: - devices[item['macaddr']] = Device(type, macaddress.EUI48(item['macaddr']), + def _add_device(conn: Connection, item: dict) -> None: + devices[item['macaddr']] = Device(conn, macaddress.EUI48(item['macaddr']), ipaddress.IPv4Address(item['ipaddr']), item['hostname']) + for item in data.get('access_devices_wired', []): + type = self._map_wire_type(item.get('wire_type')) + _add_device(type, item) + for item in data.get('access_devices_wireless_host', []): type = self._map_wire_type(item.get('wire_type')) _add_device(type, item) @@ -347,11 +367,31 @@ def _add_device(type: Wifi, item: dict) -> None: type = self._map_wire_type(item.get('wire_type'), False) _add_device(type, item) - for item in self.request(self._url_wireless_stats, 'operation=load'): + smart_network = None + if self._smart_network: + try: + smart_network = self.request('admin/smart_network?form=game_accelerator', 'operation=loadDevice') + except Exception: + self._smart_network = False + + if smart_network: + for item in smart_network: + if item['mac'] not in devices: + conn = self._map_wire_type(item.get('deviceTag'), not item.get('isGuest')) + devices[item['mac']] = Device(conn, macaddress.EUI48(item['mac']), + ipaddress.IPv4Address(item['ip']), + item['deviceName']) + if conn.is_iot(): + if status.iot_clients_total is None: + status.iot_clients_total = 0 + status.iot_clients_total += 1 + + for item in self.request('admin/wireless?form=statistics', 'operation=load'): if item['mac'] not in devices: status.wifi_clients_total += 1 type = self._map_wire_type(item.get('type')) - devices[item['mac']] = Device(type, macaddress.EUI48(item['mac']), ipaddress.IPv4Address('0.0.0.0'), '') + devices[item['mac']] = Device(type, macaddress.EUI48(item['mac']), ipaddress.IPv4Address('0.0.0.0'), + '') devices[item['mac']].packets_sent = item.get('txpkts') devices[item['mac']].packets_received = item.get('rxpkts') @@ -405,16 +445,24 @@ def _str2bool(v) -> bool | None: return str(v).lower() in ("yes", "true", "on") if v is not None else None @staticmethod - def _map_wire_type(data: str | None, host: bool = True) -> Wifi: - result = Wifi.WIFI_UNKNOWN + def _map_wire_type(data: str | None, host: bool = True) -> Connection: + result = Connection.UNKNOWN if data is None: return result + if data == 'wired': + result = Connection.WIRED if data.startswith('2.4'): - result = Wifi.WIFI_2G if host else Wifi.WIFI_GUEST_2G + result = Connection.HOST_2G if host else Connection.GUEST_2G elif data.startswith('5'): - result = Wifi.WIFI_5G if host else Wifi.WIFI_GUEST_5G + result = Connection.HOST_5G if host else Connection.GUEST_5G elif data.startswith('6'): - result = Wifi.WIFI_6G if host else Wifi.WIFI_GUEST_6G + result = Connection.HOST_6G if host else Connection.GUEST_6G + elif data.startswith('iot_2'): + result = Connection.IOT_2G + elif data.startswith('iot_5'): + result = Connection.IOT_5G + elif data.startswith('iot_6'): + result = Connection.IOT_6G return result @@ -424,7 +472,6 @@ def __init__(self, host: str, password: str, username: str = 'admin', logger: Lo super().__init__(host, password, username, logger, verify_ssl, timeout) self._url_firmware = 'admin/firmware?form=upgrade' - self._url_wireless_stats = 'admin/wireless?form=statistics' self._url_ipv4_reservations = 'admin/dhcps?form=reservation' self._url_ipv4_dhcp_leases = 'admin/dhcps?form=client' @@ -445,17 +492,17 @@ def logout(self) -> None: self._sysauth = '' self._logged = False - def set_wifi(self, wifi: Wifi, enable: bool) -> None: + def set_wifi(self, wifi: Connection, enable: bool) -> None: en = {'enable': enable} - if Wifi.WIFI_2G == wifi: + if Connection.HOST_2G == wifi: params = {'band2_4': {'host': en}} - elif Wifi.WIFI_5G == wifi: + elif Connection.HOST_5G == wifi: params = {'band5_1': {'host': en}} - elif Wifi.WIFI_GUEST_5G == wifi: + elif Connection.GUEST_5G == wifi: params = {'band5_1': {'guest': en}} - elif Wifi.WIFI_6G == wifi: + elif Connection.HOST_6G == wifi: params = {'band6': {'host': en}} - elif Wifi.WIFI_GUEST_6G == wifi: + elif Connection.GUEST_6G == wifi: params = {'band6': {'guest': en}} else: params = {'band2_4': {'guest': en}} @@ -515,21 +562,20 @@ def get_status(self) -> Status: for item in data: if not item.get('online'): continue - if item.get('wire_type') == 'wired': + conn = self._map_wire_type(item) + if conn == Connection.WIRED: status.wired_total += 1 - continue - wifi = self._map_wire_type(item) - if wifi in [Wifi.WIFI_2G, Wifi.WIFI_5G, Wifi.WIFI_6G]: + elif conn.is_host_wifi(): status.wifi_clients_total += 1 - elif wifi in [Wifi.WIFI_GUEST_2G, Wifi.WIFI_GUEST_5G, Wifi.WIFI_GUEST_5G]: + elif conn.is_guest_wifi(): status.guest_clients_total += 1 - elif wifi in [Wifi.WIFI_IOT_2G, Wifi.WIFI_IOT_5G, Wifi.WIFI_IOT_6G]: + elif conn.is_iot(): if status.iot_clients_total is None: status.iot_clients_total = 0 status.iot_clients_total += 1 ip = item['ip'] if item.get('ip') else '0.0.0.0' - devices.append(Device(wifi, + devices.append(Device(conn, macaddress.EUI48(item['mac']), ipaddress.IPv4Address(ip), base64.b64decode(item['name']).decode())) @@ -571,14 +617,16 @@ def _get_value(dictionary: dict, keys: list): return None return nested_dict - def _map_wire_type(self, data: dict) -> Wifi: - mapping = {'band2_4': {'main': Wifi.WIFI_2G, 'guest': Wifi.WIFI_GUEST_2G, 'iot': Wifi.WIFI_IOT_2G}, - 'band5': {'main': Wifi.WIFI_5G, 'guest': Wifi.WIFI_GUEST_5G, 'iot': Wifi.WIFI_IOT_5G}, - 'band6': {'main': Wifi.WIFI_6G, 'guest': Wifi.WIFI_GUEST_6G, 'iot': Wifi.WIFI_IOT_6G} + def _map_wire_type(self, data: dict) -> Connection: + if data.get('wire_type') == 'wired': + return Connection.WIRED + mapping = {'band2_4': {'main': Connection.HOST_2G, 'guest': Connection.GUEST_2G, 'iot': Connection.IOT_2G}, + 'band5': {'main': Connection.HOST_5G, 'guest': Connection.GUEST_5G, 'iot': Connection.IOT_5G}, + 'band6': {'main': Connection.HOST_6G, 'guest': Connection.GUEST_6G, 'iot': Connection.IOT_6G} } result = self._get_value(mapping, [data.get('connection_type'), data.get('interface')]) - return result if result else Wifi.WIFI_UNKNOWN + return result if result else Connection.UNKNOWN @staticmethod def _get_login_data(crypted_pwd: str) -> str: @@ -593,6 +641,37 @@ def _is_valid_response(self, data: dict) -> bool: return 'error_code' in data and data['error_code'] == 0 +class TplinkC6V4Router(AbstractRouter): + def supports(self) -> bool: + url = '{}/?code=2&asyn=1'.format(self.host) + try: + response = requests.post(url, timeout=self.timeout, verify=self._verify_ssl) + except: + return False + if response.status_code == 401 and response.text.startswith('00'): + raise ClientException( + 'Your router is not supported. Please add your router support to https://github.com/AlexandrErohin/TP-Link-Archer-C6U by implementing methods for TplinkC6V4Router class') + return False + + def authorize(self) -> None: + raise ClientException('Not Implemented') + + def logout(self) -> None: + raise ClientException('Not Implemented') + + def get_firmware(self) -> Firmware: + raise ClientException('Not Implemented') + + def get_status(self) -> Status: + raise ClientException('Not Implemented') + + def reboot(self) -> None: + raise ClientException('Not Implemented') + + def set_wifi(self, wifi: Connection, enable: bool) -> None: + raise ClientException('Not Implemented') + + class TplinkC1200Router(TplinkBaseRouter): def supports(self) -> bool: return True @@ -638,24 +717,19 @@ class TPLinkMRClient(AbstractRouter): HTTP_ERR_USER_PWD_NOT_CORRECT = 71233 HTTP_ERR_USER_BAD_REQUEST = 71234 - LAN = 0 - WIFI_2G = 1 - WIFI_5G = 3 - WIFI_GUEST_2G = 2 - WIFI_GUEST_5G = 4 - CLIENT_TYPES = { - WIFI_2G: Wifi.WIFI_2G, - WIFI_5G: Wifi.WIFI_5G, - WIFI_GUEST_2G: Wifi.WIFI_GUEST_2G, - WIFI_GUEST_5G: Wifi.WIFI_GUEST_5G, + 0: Connection.WIRED, + 1: Connection.HOST_2G, + 3: Connection.HOST_5G, + 2: Connection.GUEST_2G, + 4: Connection.GUEST_5G, } WIFI_SET = { - Wifi.WIFI_2G: '1,1,0,0,0,0', - Wifi.WIFI_5G: '1,2,0,0,0,0', - Wifi.WIFI_GUEST_2G: '1,1,1,0,0,0', - Wifi.WIFI_GUEST_5G: '1,2,1,0,0,0', + Connection.HOST_2G: '1,1,0,0,0,0', + Connection.HOST_5G: '1,2,0,0,0,0', + Connection.GUEST_2G: '1,1,1,0,0,0', + Connection.GUEST_5G: '1,2,1,0,0,0', } class ActItem: @@ -796,17 +870,16 @@ def get_status(self) -> Status: for val in self._to_list(values.get('4')): if int(val['active']) == 0: continue - type = int(val['X_TP_ConnType']) - if type == self.LAN: - status.wired_total += 1 + conn = self.CLIENT_TYPES.get(int(val['X_TP_ConnType'])) + if conn is None: continue - if type in [self.WIFI_GUEST_2G, self.WIFI_GUEST_5G]: + elif conn == Connection.WIRED: + status.wired_total += 1 + elif conn.is_guest_wifi(): status.guest_clients_total += 1 - elif type in [self.WIFI_2G, self.WIFI_5G]: + elif conn.is_host_wifi(): status.wifi_clients_total += 1 - else: - continue - devices[val['MACAddress']] = Device(self.CLIENT_TYPES[type], + devices[val['MACAddress']] = Device(conn, macaddress.EUI48(val['MACAddress']), ipaddress.IPv4Address(val['IPAddress']), val['hostName']) @@ -815,7 +888,7 @@ def get_status(self) -> Status: if val['associatedDeviceMACAddress'] not in devices: status.wifi_clients_total += 1 devices[val['associatedDeviceMACAddress']] = Device( - Wifi.WIFI_2G, + Connection.HOST_2G, macaddress.EUI48(val['associatedDeviceMACAddress']), ipaddress.IPv4Address('0.0.0.0'), '') @@ -895,11 +968,11 @@ def get_ipv4_status(self) -> IPv4Status: return ipv4_status - def set_wifi(self, wifi: Wifi, enable: bool) -> None: + def set_wifi(self, wifi: Connection, enable: bool) -> None: acts = [ self.ActItem( self.ActItem.SET, - 'LAN_WLAN' if wifi in [Wifi.WIFI_2G, Wifi.WIFI_5G] else 'LAN_WLAN_MSSIDENTRY', + 'LAN_WLAN' if wifi in [Connection.HOST_2G, Connection.HOST_5G] else 'LAN_WLAN_MSSIDENTRY', self.WIFI_SET[wifi], attrs=['enable={}'.format(int(enable))]), ] @@ -1179,7 +1252,7 @@ class TplinkRouterProvider: @staticmethod def get_client(host: str, password: str, username: str = 'admin', logger: Logger = None, verify_ssl: bool = True, timeout: int = 10) -> AbstractRouter | None: - for client in [TPLinkMRClient, TPLinkDecoClient, TplinkRouter, TplinkC1200Router]: + for client in [TPLinkMRClient, TplinkC6V4Router, TPLinkDecoClient, TplinkRouter, TplinkC1200Router]: router = client(host, password, username, logger, verify_ssl, timeout) if router.supports(): return router diff --git a/tplinkrouterc6u/dataclass.py b/tplinkrouterc6u/dataclass.py index b656ade..9eab6a1 100644 --- a/tplinkrouterc6u/dataclass.py +++ b/tplinkrouterc6u/dataclass.py @@ -1,7 +1,7 @@ import macaddress import ipaddress from dataclasses import dataclass, field -from tplinkrouterc6u.enum import Wifi +from tplinkrouterc6u.enum import Connection @dataclass @@ -14,7 +14,7 @@ def __init__(self, hardware: str, model: str, firmware: str) -> None: @dataclass class Device: - def __init__(self, type: Wifi, macaddr: macaddress, ipaddr: ipaddress, hostname: str) -> None: + def __init__(self, type: Connection, macaddr: macaddress, ipaddr: ipaddress, hostname: str) -> None: self.type = type self._macaddr = macaddr self._ipaddr = ipaddr diff --git a/tplinkrouterc6u/enum.py b/tplinkrouterc6u/enum.py index 1300bda..a43a27a 100644 --- a/tplinkrouterc6u/enum.py +++ b/tplinkrouterc6u/enum.py @@ -1,14 +1,46 @@ from enum import Enum -class Wifi(Enum): - WIFI_2G = 'wireless_2g' - WIFI_5G = 'wireless_5g' - WIFI_6G = 'wireless_6g' - WIFI_GUEST_2G = 'guest_2g' - WIFI_GUEST_5G = 'guest_5g' - WIFI_GUEST_6G = 'guest_6g' - WIFI_IOT_2G = 'iot_2g' - WIFI_IOT_5G = 'iot_5g' - WIFI_IOT_6G = 'iot_6g' - WIFI_UNKNOWN = 'unknown' +class Connection(Enum): + HOST_2G = 'host_2g' + HOST_5G = 'host_5g' + HOST_6G = 'host_6g' + GUEST_2G = 'guest_2g' + GUEST_5G = 'guest_5g' + GUEST_6G = 'guest_6g' + IOT_2G = 'iot_2g' + IOT_5G = 'iot_5g' + IOT_6G = 'iot_6g' + WIRED = 'wired' + UNKNOWN = 'unknown' + + def is_host_wifi(self) -> bool: + return self in [Connection.HOST_2G, Connection.HOST_5G, Connection.HOST_6G] + + def is_guest_wifi(self) -> bool: + return self in [Connection.GUEST_2G, Connection.GUEST_5G, Connection.GUEST_5G] + + def is_iot(self) -> bool: + return self in [Connection.IOT_2G, Connection.IOT_5G, Connection.IOT_6G] + + def get_band(self) -> str | None: + band = None + if self in [Connection.HOST_2G, Connection.GUEST_2G, Connection.IOT_2G]: + band = '2G' + elif self in [Connection.HOST_5G, Connection.GUEST_5G, Connection.IOT_5G]: + band = '5G' + elif self in [Connection.HOST_6G, Connection.GUEST_6G, Connection.IOT_6G]: + band = '6G' + return band + + def get_type(self) -> str | None: + band = None + if self.is_host_wifi(): + band = 'host' + elif self.is_guest_wifi(): + band = 'guest' + elif self.is_iot(): + band = 'IoT' + elif self == Connection.WIRED: + band = 'wired' + return band