added get_sms, set_sms_read, delete_sms and send_ussd to TPLinkMRClient

README.md

@@ -86,6 +86,10 @@ or you have TP-link C5400X or similar router you need to get web encrypted passw
 | 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 |   |
+| send_ussd | command: str | Send USSD command for LTE routers | str |
+| get_sms | | Get sms messages from the first page for LTE routers | [[SMS]](#sms) |
+| set_sms_read | sms: [SMS](#sms) | Set sms message read from the first page for LTE routers |   |
+| delete_sms | sms: [SMS](#sms) | Delete sms message from the first page for LTE routers |   |
 | reboot |   | reboot router |
 | authorize |   | authorize for actions |
 | logout |   | logout after all is done |
@@ -189,6 +193,15 @@ or you have TP-link C5400X or similar router you need to get web encrypted passw
 | lan_ipv4_netmask_address | router LAN gateway IP netmask | ipaddress |
 | remote | router remote | bool, None |
 
+### <a id="sms">SMS</a>
+| Field | Description | Type |
+| --- |---|---|
+| id | message index | int |
+| sender| sender | str |
+| content| sms text | str |
+| received_at| received datetime | datetime |
+| unread| is message unread | bool |
+
 ## Enum
 ### <a id="connection">Connection</a>
 - Connection.HOST_2G - host wifi 2.4G

test/test_client_mr.py

@@ -1,6 +1,7 @@
 from unittest import main, TestCase
 from macaddress import EUI48
 from ipaddress import IPv4Address
+from datetime import datetime
 from tplinkrouterc6u import (
     TPLinkMRClient,
     Connection,
@@ -10,6 +11,8 @@
     IPv4Reservation,
     IPv4DHCPLease,
     IPv4Status,
+    ClientError,
+    SMS,
 )
 
 
@@ -679,6 +682,188 @@ def _request(self, url, method='POST', data_str=None, encrypt=False):
         self.assertEqual(check_data, ('2\r\n[LTE_SMS_SENDNEWMSG#0,0,0,0,0,0#0,0,0,0,0,0]0,3\r\nindex=1\r\n'
                                       'to=534324724234\r\ntextContent=test sms\r\n'))
 
+    def test_send_ussd(self) -> None:
+        responses = ['''[error]0
+
+''', '''[0,0,0,0,0,0]0
+sessionStatus=1
+sendResult=1
+response=
+ussdStatus=0
+[error]0
+
+''', '''[0,0,0,0,0,0]0
+sessionStatus=0
+sendResult=1
+response=some text
+ussdStatus=1
+[error]0
+
+''']
+
+        check_url = []
+        check_data = []
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                nonlocal check_url, check_data
+                check_url.append(url)
+                check_data.append(data_str)
+                return 200, responses.pop(0)
+
+        client = TPLinkMRClientTest('', '')
+
+        self.assertEqual('some text', client.send_ussd('534324724234'))
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url.pop(0))
+        self.assertEqual(check_data.pop(0),
+                         '2\r\n[LTE_USSD#0,0,0,0,0,0#0,0,0,0,0,0]0,2\r\naction=1\r\nreqContent=534324724234\r\n')
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url.pop(0))
+        self.assertEqual(check_data.pop(0),
+                         ('1\r\n[LTE_USSD#0,0,0,0,0,0#0,0,0,0,0,0]0,4\r\nsessionStatus\r\n'
+                          'sendResult\r\nresponse\r\nussdStatus\r\n'))
+
+    def test_send_ussd_error(self) -> None:
+        responses = ['''[error]0
+
+''', '''[0,0,0,0,0,0]0
+sessionStatus=1
+sendResult=1
+response=
+ussdStatus=0
+[error]0
+
+''', '''[0,0,0,0,0,0]0
+sessionStatus=0
+sendResult=1
+response=
+ussdStatus=2
+[error]0
+
+''']
+
+        check_url = []
+        check_data = []
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                nonlocal check_url, check_data
+                check_url.append(url)
+                check_data.append(data_str)
+                return 200, responses.pop(0)
+
+        client = TPLinkMRClientTest('', '')
+
+        with self.assertRaises(ClientError) as context:
+            client.send_ussd('534324724234')
+
+        self.assertTrue('Cannot send USSD!' in str(context.exception))
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url.pop(0))
+        self.assertEqual(check_data.pop(0),
+                         '2\r\n[LTE_USSD#0,0,0,0,0,0#0,0,0,0,0,0]0,2\r\naction=1\r\nreqContent=534324724234\r\n')
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url.pop(0))
+        self.assertEqual(check_data.pop(0),
+                         ('1\r\n[LTE_USSD#0,0,0,0,0,0#0,0,0,0,0,0]0,4\r\nsessionStatus\r\nsendResult\r\n'
+                          'response\r\nussdStatus\r\n'))
+
+    def test_get_sms(self) -> None:
+        response = '''[1,0,0,0,0,0]1
+index=3
+from=sender1
+content=text second
+receivedTime=2024-11-15 22:28:09
+unread=1
+[2,0,0,0,0,0]1
+index=2
+from=sender2
+content=text first
+receivedTime=2024-11-15 22:23:59
+unread=0
+[error]0
+'''
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                return 200, response
+
+        client = TPLinkMRClientTest('', '')
+        messages = client.get_sms()
+
+        self.assertEqual(len(messages), 2)
+        self.assertIsInstance(messages[0], SMS)
+        self.assertEqual(messages[0].id, 1)
+        self.assertEqual(messages[0].sender, 'sender1')
+        self.assertEqual(messages[0].content, 'text second')
+        self.assertEqual(messages[0].received_at, datetime.fromisoformat('2024-11-15 22:28:09'))
+        self.assertEqual(messages[0].unread, True)
+        self.assertIsInstance(messages[1], SMS)
+        self.assertEqual(messages[1].id, 2)
+        self.assertEqual(messages[1].sender, 'sender2')
+        self.assertEqual(messages[1].content, 'text first')
+        self.assertEqual(messages[1].received_at, datetime.fromisoformat('2024-11-15 22:23:59'))
+        self.assertEqual(messages[1].unread, False)
+
+    def test_get_sms_empty(self) -> None:
+        response = '''[error]0
+
+'''
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                return 200, response
+
+        client = TPLinkMRClientTest('', '')
+        messages = client.get_sms()
+
+        self.assertEqual([], messages)
+
+    def test_set_sms_read(self) -> None:
+        response = '''
+[error]0
+
+'''
+
+        check_url = ''
+        check_data = ''
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                nonlocal check_url, check_data
+                check_url = url
+                check_data = data_str
+                return 200, response
+
+        client = TPLinkMRClientTest('', '')
+        client.set_sms_read(SMS(2, '', '', datetime.now(), True))
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url)
+        self.assertEqual(check_data, '2\r\n[LTE_SMS_RECVMSGENTRY#2,0,0,0,0,0#0,0,0,0,0,0]0,1\r\nunread=0\r\n')
+
+    def test_delete_sms(self) -> None:
+        response = '''
+[error]0
+
+'''
+
+        check_url = ''
+        check_data = ''
+
+        class TPLinkMRClientTest(TPLinkMRClient):
+            def _request(self, url, method='POST', data_str=None, encrypt=False):
+                nonlocal check_url, check_data
+                check_url = url
+                check_data = data_str
+                return 200, response
+
+        client = TPLinkMRClientTest('', '')
+        client.delete_sms(SMS(2, '', '', datetime.now(), True))
+
+        self.assertIn('http:///cgi_gdpr?_=', check_url)
+        self.assertEqual(check_data, '4\r\n[LTE_SMS_RECVMSGENTRY#2,0,0,0,0,0#0,0,0,0,0,0]0,0\r\n\r\n')
+
 
 if __name__ == '__main__':
     main()

tplinkrouterc6u/__init__.py

@@ -15,5 +15,6 @@
     IPv4Reservation,
     IPv4DHCPLease,
     IPv4Status,
+    SMS,
 )
-from tplinkrouterc6u.common.exception import ClientException
+from tplinkrouterc6u.common.exception import ClientException, ClientError, AuthorizeError

tplinkrouterc6u/client/mr.py

@@ -3,13 +3,21 @@
 from time import time, sleep
 from urllib.parse import quote
 from requests import Session
-from datetime import timedelta
+from datetime import timedelta, datetime
 from macaddress import EUI48
 from ipaddress import IPv4Address
 from logging import Logger
 from tplinkrouterc6u.common.encryption import EncryptionWrapperMR
 from tplinkrouterc6u.common.package_enum import Connection
-from tplinkrouterc6u.common.dataclass import Firmware, Status, Device, IPv4Reservation, IPv4DHCPLease, IPv4Status
+from tplinkrouterc6u.common.dataclass import (
+    Firmware,
+    Status,
+    Device,
+    IPv4Reservation,
+    IPv4DHCPLease,
+    IPv4Status,
+    SMS,
+)
 from tplinkrouterc6u.common.exception import ClientException, ClientError
 from tplinkrouterc6u.client_abstract import AbstractRouter
 
@@ -484,7 +492,7 @@ def get_status(self) -> Status:
 
     def get_ipv4_reservations(self) -> [IPv4Reservation]:
         acts = [
-            self.ActItem(5, 'LAN_DHCP_STATIC_ADDR', attrs=['enable', 'chaddr', 'yiaddr']),
+            self.ActItem(self.ActItem.GL, 'LAN_DHCP_STATIC_ADDR', attrs=['enable', 'chaddr', 'yiaddr']),
         ]
         _, values = self.req_act(acts)
 
@@ -502,7 +510,8 @@ def get_ipv4_reservations(self) -> [IPv4Reservation]:
 
     def get_ipv4_dhcp_leases(self) -> [IPv4DHCPLease]:
         acts = [
-            self.ActItem(5, 'LAN_HOST_ENTRY', attrs=['IPAddress', 'MACAddress', 'hostName', 'leaseTimeRemaining']),
+            self.ActItem(self.ActItem.GL, 'LAN_HOST_ENTRY', attrs=['IPAddress', 'MACAddress', 'hostName',
+                                                                   'leaseTimeRemaining']),
         ]
         _, values = self.req_act(acts)
 
@@ -570,3 +579,67 @@ def send_sms(self, phone_number: str, message: str) -> None:
                 ]),
         ]
         self.req_act(acts)
+
+    def get_sms(self) -> [SMS]:
+        acts = [
+            self.ActItem(
+                self.ActItem.SET, 'LTE_SMS_RECVMSGBOX', attrs=['PageNumber=1']),
+            self.ActItem(
+                self.ActItem.GL, 'LTE_SMS_RECVMSGENTRY', attrs=['index', 'from', 'content', 'receivedTime',
+                                                                'unread']),
+        ]
+        _, values = self.req_act(acts)
+
+        messages = []
+        if values:
+            i = 1
+            for item in self._to_list(values.get('1')):
+                messages.append(
+                    SMS(
+                        i, item['from'], item['content'], datetime.fromisoformat(item['receivedTime']),
+                        True if item['unread'] == '1' else False
+                    )
+                )
+                i += 1
+
+        return messages
+
+    def set_sms_read(self, sms: SMS) -> None:
+        acts = [
+            self.ActItem(
+                self.ActItem.SET, 'LTE_SMS_RECVMSGENTRY', f'{sms.id},0,0,0,0,0', attrs=['unread=0']),
+        ]
+        self.req_act(acts)
+
+    def delete_sms(self, sms: SMS) -> None:
+        acts = [
+            self.ActItem(
+                self.ActItem.DEL, 'LTE_SMS_RECVMSGENTRY', f'{sms.id},0,0,0,0,0'),
+        ]
+        self.req_act(acts)
+
+    def send_ussd(self, command: str) -> str:
+        acts = [
+            self.ActItem(
+                self.ActItem.SET, 'LTE_USSD', attrs=[
+                    'action=1',
+                    f"reqContent={command}",
+                ]),
+        ]
+        self.req_act(acts)
+
+        status = '0'
+        while status == '0':
+            sleep(1)
+            acts = [
+                self.ActItem(
+                    self.ActItem.GET, 'LTE_USSD', attrs=['sessionStatus', 'sendResult', 'response', 'ussdStatus']),
+            ]
+            _, values = self.req_act(acts)
+
+            status = values.get('ussdStatus', '2')
+
+            if status == '1':
+                return values.get('response')
+            elif status == '2':
+                raise ClientError('Cannot send USSD!')

tplinkrouterc6u/common/dataclass.py

@@ -1,6 +1,7 @@
 from macaddress import EUI48
 from ipaddress import IPv4Address
 from dataclasses import dataclass
+from datetime import datetime
 from tplinkrouterc6u.common.package_enum import Connection
 
 
@@ -246,3 +247,13 @@ def lan_ipv4_netmask(self):
     @property
     def lan_ipv4_netmask_address(self):
         return self._lan_ipv4_netmask
+
+
+@dataclass
+class SMS:
+    def __init__(self, index: int, sender: str, content: str, received_at: datetime, unread: bool) -> None:
+        self.id = index
+        self.sender = sender
+        self.content = content
+        self.received_at = received_at
+        self.unread = unread