Merge "Add EATT support for dump_le_audio.py tool"

system/tools/scripts/dump_le_audio.py

@@ -63,7 +63,10 @@ ISO_PACKET = 5
 SENT = 0
 RECEIVED = 1
 
-L2CAP_ATT_CID = 4
+L2CAP_ATT_CID = 0x0004
+L2CAP_CID = 0x0005
+
+PSM_EATT = 0x0027
 
 # opcode for att protocol
 OPCODE_ATT_READ_BY_TYPE_RSP = 0x09
@@ -83,6 +86,10 @@ OPCODE_LE_SET_PERIODIC_ADVERTISING_DATA = 0x203F
 OPCODE_LE_CREATE_BIG = 0x2068
 OPCODE_LE_SETUP_ISO_DATA_PATH = 0x206E
 
+# opcode for L2CAP channel
+OPCODE_L2CAP_CREDIT_BASED_CONNECTION_REQ = 0x17
+OPCODE_L2CAP_CREDIT_BASED_CONNECTION_RSP = 0x18
+
 # HCI event
 EVENT_CODE_LE_META_EVENT = 0x3E
 SUBEVENT_CODE_LE_CREATE_BIG_COMPLETE = 0x1B
@@ -129,6 +136,10 @@ debug_enable = False
 add_header = False
 ase_handle = 0xFFFF
 
+l2cap_identifier_set = set()
+source_cid = set()
+destinate_cid = set()
+
 
 class Connection:
 
@@ -676,9 +687,44 @@ def parse_acl_packet(packet, flags, timestamp):
         return
 
     if debug_enable:
-        debug_print("ACL connection_handle - " + str(connection_handle))
+        debug_print("ACL connection_handle - " + str(connection_handle) + " channel id - " + (str(channel_id)))
+
+    # Gather EATT CID
+    if channel_id == L2CAP_CID:
+        global l2cap_identifier_set
+        global source_cid
+        global destinate_cid
+        opcode, packet = unpack_data(packet, 1, False)
+        identifier, packet = unpack_data(packet, 1, False)
+        l2cap_length, packet = unpack_data(packet, 2, False)
+        if opcode == OPCODE_L2CAP_CREDIT_BASED_CONNECTION_REQ:
+            spsm, packet = unpack_data(packet, 2, False)
+            if spsm == PSM_EATT:
+                if opcode == OPCODE_L2CAP_CREDIT_BASED_CONNECTION_REQ:
+                    l2cap_identifier_set.add(identifier)
+                    packet = unpack_data(packet, 6, True)
+                    for i in range(0, l2cap_length - 8, 2):
+                        cid, packet = unpack_data(packet, 2, False)
+                        source_cid.add(cid)
+
+        if opcode == OPCODE_L2CAP_CREDIT_BASED_CONNECTION_RSP:
+            if identifier in l2cap_identifier_set:
+                l2cap_identifier_set.remove(identifier)
+                packet = unpack_data(packet, 8, True)
+                for i in range(0, l2cap_length - 8, 2):
+                    cid, packet = unpack_data(packet, 2, False)
+                    destinate_cid.add(cid)
+
     # Parse ATT protocol
     if channel_id == L2CAP_ATT_CID:
+        if debug_enable:
+            debug_print("parse_att_packet")
+        parse_att_packet(packet, connection_handle, flags, timestamp)
+
+    if channel_id in source_cid or channel_id in destinate_cid:
+        if debug_enable:
+            debug_print("parse_eatt_packet")
+        packet = unpack_data(packet, 2, True)
         parse_att_packet(packet, connection_handle, flags, timestamp)
 
 
@@ -720,8 +766,8 @@ def parse_next_packet(btsnoop_file):
     if len(packet_header) != 25:
         return False
 
-    (length_original, length_captured, flags, dropped_packets, timestamp, type) = struct.unpack(
-        ">IIIIqB", packet_header)
+    (length_original, length_captured, flags, dropped_packets, timestamp,
+     type) = struct.unpack(">IIIIqB", packet_header)
 
     if length_original != length_captured:
         debug_print("Filtered btnsoop, can not be parsed")
@@ -751,10 +797,9 @@ def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("btsnoop_file", help="btsnoop file contains LE audio start procedure")
     parser.add_argument("-v", "--verbose", help="Enable verbose log.", action="store_true")
-    parser.add_argument(
-        "--header",
-        help="Add the header for LC3 Conformance Interoperability Test Software V.1.0.3.",
-        action="store_true")
+    parser.add_argument("--header",
+                        help="Add the header for LC3 Conformance Interoperability Test Software V.1.0.3.",
+                        action="store_true")
     parser.add_argument("--ase_handle", help="Set the ASE handle manually.", type=int)
 
     argv = parser.parse_args()