iutctl: add log parser

Add socat log parser

autopts/ptsprojects/mynewt/iutctl.py

@@ -76,7 +76,7 @@ def start(self, test_case):
 
         self.flush_serial()
 
-        self.socket_srv = BTPSocketSrv()
+        self.socket_srv = BTPSocketSrv(test_case.log_dir)
         self.socket_srv.open(self.btp_address)
         self.btp_socket = BTPWorker(self.socket_srv)
 

autopts/ptsprojects/zephyr/iutctl.py

@@ -112,7 +112,7 @@ def start(self, test_case):
         # partial or whole IUT ready event. Flush serial to ignore it.
         self.flush_serial()
 
-        self.socket_srv = BTPSocketSrv()
+        self.socket_srv = BTPSocketSrv(test_case.log_dir)
         self.socket_srv.open(self.btp_address)
         self.btp_socket = BTPWorker(self.socket_srv)
 

autopts/pybtp/iutctl_common.py

@@ -25,6 +25,8 @@
 from abc import abstractmethod
 
 from autopts.pybtp import defs
+from autopts.pybtp.defs import *
+from datetime import datetime
 from autopts.pybtp.types import BTPError
 from autopts.pybtp.parser import enc_frame, dec_hdr, dec_data, HDR_LEN
 from autopts.utils import get_global_end, raise_on_global_end
@@ -46,9 +48,12 @@ def set_event_handler(event_handler):
 
 class BTPSocket:
 
-    def __init__(self):
+    def __init__(self, log_dir=None):
         self.conn = None
         self.addr = None
+        self.btp_service_id_dict = None
+        self.log_file = open(os.path.join(log_dir, "autopts-iutctl-decoded.log"), "a")
+        self.btp_service_id_dict = self.get_svc_id()
 
     @abstractmethod
     def open(self, address):
@@ -58,6 +63,42 @@ def open(self, address):
     def accept(self, timeout=10.0):
         pass
 
+    @staticmethod
+    def get_svc_id():
+        """Looks for BTP_SVC_ID variables from the defs.py"""
+        btp_service_ids = {}
+
+        for name, value in vars(defs).items():
+            if name.startswith('BTP_SERVICE_ID_') and isinstance(value, int):
+                trimmed_name = name.replace('BTP_SERVICE_ID_', '')
+                btp_service_ids[trimmed_name] = value
+
+        return btp_service_ids
+
+    def write_to_log(self, req, data, hex_data):
+        """Log decoded header"""
+        current_time = datetime.now().strftime('%H:%M:%S:%f')
+        f = self.log_file
+
+        if req:
+            f.write(f'{current_time[:-3]}\t-> Command:  {self.parse_data(data)} {hex_data}\n')
+        else:
+            f.write(f'{current_time[:-3]}\t<- Response: {self.parse_data(data)} {hex_data}\n')
+
+    def write_err_status(self, data, hex_data, status):
+        """Log command status value for error response"""
+        current_time = datetime.now().strftime('%H:%M:%S:%f')
+        f = self.log_file
+        status_values = {
+            1: 'Fail',
+            2: 'Unknown Command',
+            3: 'Not Ready',
+            4: 'Invalid Index'
+        }
+        err_status = status_values[int(status)]
+        f.write(f'{current_time[:-3]}\t<- Response:  {self.parse_data(data)} {hex_data} {err_status}\n')
+
+
     def read(self, timeout=20.0):
         """Read BTP data from socket
 
@@ -77,8 +118,12 @@ def read(self, timeout=20.0):
             hdr_memview = hdr_memview[nbytes:]
             toread_hdr_len -= nbytes
 
+        hex_hdr = hdr.hex()
         tuple_hdr = dec_hdr(hdr)
         toread_data_len = tuple_hdr.data_len
+        if tuple_hdr.op != 0:
+            # 0 for logging response, 1 for command
+            self.write_to_log(0, tuple_hdr, hex_hdr)
 
         logging.debug("Received: hdr: %r %r", tuple_hdr, hdr)
 
@@ -95,6 +140,8 @@ def read(self, timeout=20.0):
             toread_data_len -= nbytes
 
         data_string = binascii.hexlify(data).decode('utf-8')
+        if tuple_hdr.op == 0:
+            self.write_err_status(tuple_hdr, hex_hdr, data_string)
         data_string = ' '.join(f'{data_string[i:i+2]}' for i in range(0, len(data_string), 2))
         log(f"Received data: { {data_string} }, {data}")
 
@@ -109,17 +156,57 @@ def send(self, svc_id, op, ctrl_index, data):
         frame = enc_frame(svc_id, op, ctrl_index, data)
 
         logging.debug("sending frame %r", frame.hex())
+
+        hex_data = frame.hex()
+        tuple_data = (svc_id, op, ctrl_index, len(data) if isinstance(data, (str,  bytearray)) else data)
+        # 0 for logging response, 1 for command
+        self.write_to_log(1, tuple_data, hex_data)
         self.conn.send(frame)
 
+    def parse_data(self, data):
+        def get_btp_cmd_name(prefix, op_code):
+            """Looks for BTP Command variables from the defs.py"""
+            if op_code in ('0x0', '0x00'):
+                return 'BTP_ERROR'
+            for key, value in vars(defs).items():
+                if (key.startswith(f'BTP_CMD_{prefix}') and value == int(op_code, 16)) or\
+                        (key.startswith(f'BTP_EV_{prefix}') and value == int(op_code, 16)):
+                    return key
+
+            return None  # Return None if no matching variable is found
+
+        parsed_data = ''
+        svc_name = ''
+        if isinstance(data, str):
+            svc_id = data[:2]
+            opc = data[2:4]
+            hex_op = int(opc, 16)
+            opc = hex(hex_op)
+        else:
+            svc_id, opc, ctrl_idx, data_len = data[0], hex(data[1]), data[2], data[3]
+        for name, btp_id in self.btp_service_id_dict.items():
+            if btp_id == int(svc_id):
+                svc_name += name
+                break
+
+        indent = "\n\t\t\t"
+        btp_command = get_btp_cmd_name(svc_name, opc)
+        to_hex = lambda x: "0x{:02x}".format(x)
+        parsed_data += f'{btp_command} ({to_hex(svc_id)}|{opc}|{to_hex(ctrl_idx)}){indent} raw data ({data_len}):'
+
+        return parsed_data
+
     @abstractmethod
     def close(self):
+        self.log_file.close()
+        self.log_file = None
         pass
 
 
 class BTPSocketSrv(BTPSocket):
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, log_dir=None):
+        super().__init__(log_dir)
         self.sock = None
 
     def open(self, addres=BTP_ADDRESS):
@@ -148,6 +235,7 @@ def accept(self, timeout=10.0):
         self.sock.settimeout(None)
 
     def close(self):
+        super().close()
         try:
             self.conn.shutdown(socket.SHUT_RDWR)
             self.conn.close()