[tcat] implementation of tcat Ping command (#10484)

Commit implements Tcat ping command `kTlvPing`.

Signed-off-by: Przemyslaw Bida <[email protected]>

tools/tcat_ble_client/cli/base_commands.py

@@ -37,6 +37,8 @@ from cli.command import Command, CommandResultNone, CommandResultTLV
 from dataset.dataset import ThreadDataset
 from utils import select_device_by_user_input
 from os import path
+from time import time
+from secrets import token_bytes
 
 
 class HelpCommand(Command):
@@ -103,6 +105,37 @@ class DecommissionCommand(Command):
         return CommandResultTLV(tlv_response)
 
 
+class PingCommand(Command):
+
+    def get_help_string(self) -> str:
+        return 'Send echo request to TCAT device.'
+
+    async def execute_default(self, args, context):
+        bless: BleStreamSecure = context['ble_sstream']
+        payload_size = 10
+        max_payload = 512
+        if len(args) > 0:
+            payload_size = int(args[0])
+            if payload_size > max_payload:
+                print(f'Payload size too large. Maximum supported value is {max_payload}')
+                return
+        to_send = token_bytes(payload_size)
+        data = TLV(TcatTLVType.PING.value, to_send).to_bytes()
+        elapsed_time = time()
+        response = await bless.send_with_resp(data)
+        elapsed_time = time() - elapsed_time
+        if not response:
+            return
+
+        tlv_response = TLV.from_bytes(response)
+        if tlv_response.value != to_send:
+            print("Received malformed response.")
+
+        print(f"Roundtrip time {elapsed_time} s.")
+
+        return CommandResultTLV(tlv_response)
+
+
 class ThreadStartCommand(Command):
 
     def get_help_string(self) -> str:

tools/tcat_ble_client/cli/cli.py

@@ -29,8 +29,8 @@
 import readline
 import shlex
 from ble.ble_stream_secure import BleStreamSecure
-from cli.base_commands import (HelpCommand, HelloCommand, CommissionCommand, DecommissionCommand, ThreadStateCommand,
-                               ScanCommand)
+from cli.base_commands import (HelpCommand, HelloCommand, CommissionCommand, DecommissionCommand, PingCommand,
+                               ThreadStateCommand, ScanCommand)
 from cli.dataset_commands import (DatasetCommand)
 from dataset.dataset import ThreadDataset
 from typing import Optional
@@ -44,6 +44,7 @@ class CLI:
             'hello': HelloCommand(),
             'commission': CommissionCommand(),
             'decommission': DecommissionCommand(),
+            'ping': PingCommand(),
             'dataset': DatasetCommand(),
             'thread': ThreadStateCommand(),
             'scan': ScanCommand(),

tools/tcat_ble_client/tlv/tcat_tlv.py

@@ -32,6 +32,7 @@ class TcatTLVType(Enum):
     RESPONSE_W_STATUS = 0x01
     RESPONSE_W_PAYLOAD = 0x02
     DISCONNECT = 0x09
+    PING = 0x0A
     ACTIVE_DATASET = 0x20
     DECOMMISSION = 0x60
     APPLICATION = 0x82

tools/tcat_ble_client/tlv/tlv.py

@@ -58,14 +58,19 @@ class TLV():
     def set_from_bytes(self, data: bytes):
         self.type = data[0]
         header_len = 2
+        size_offset = 1
         if data[1] == 0xFF:
             header_len = 4
-        length = int.from_bytes(data[1:header_len], byteorder='big')
+            size_offset = 2
+        length = int.from_bytes(data[size_offset:header_len], byteorder='big')
         self.value = data[header_len:header_len + length]
 
     def to_bytes(self) -> bytes:
-        has_long_header = len(self.value) >= 255
+        has_long_header = len(self.value) >= 254
         header_len = 4 if has_long_header else 2
-        len_bytes = len(self.value).to_bytes(header_len - 1, byteorder='big')
-        header = bytes([self.type]) + len_bytes
+        len_bytes = len(self.value).to_bytes(header_len // 2, byteorder='big')
+        type = self.type
+        if has_long_header:
+            type = type << 8 | 255
+        header = type.to_bytes(header_len // 2, byteorder='big') + len_bytes
         return header + self.value