Merge pull request #893 from zxzxwu/le-emu

Emulation: Support LE Read features

bumble/controller.py

@@ -544,6 +544,20 @@ class Controller:
                 self.on_le_cis_disconnected(packet.cig_id, packet.cis_id)
             case ll.EncReq():
                 self.on_le_encrypted(connection)
+            case ll.FeatureReq() | ll.PeripheralFeatureReq():
+                connection.send_ll_control_pdu(
+                    ll.FeatureRsp(
+                        feature_set=self.le_features.value.to_bytes(8, 'little')
+                    )
+                )
+            case ll.FeatureRsp(feature_set):
+                self.send_hci_packet(
+                    hci.HCI_LE_Read_Remote_Features_Complete_Event(
+                        status=hci.HCI_ErrorCode.SUCCESS,
+                        connection_handle=connection.handle,
+                        le_features=feature_set,
+                    )
+                )
 
     def on_ll_advertising_pdu(self, packet: ll.AdvertisingPdu) -> None:
         logger.debug("[%s] <<< Advertising PDU: %s", self.name, packet)
@@ -2084,7 +2098,7 @@ class Controller:
 
         handle = command.connection_handle
 
-        if not self.find_connection_by_handle(handle):
+        if not (connection := self.find_le_connection_by_handle(handle)):
             self._send_hci_command_status(
                 hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code
             )
@@ -2093,14 +2107,16 @@ class Controller:
         # First, say that the command is pending
         self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code)
 
-        # Then send the remote features
+        if connection.role == hci.Role.CENTRAL:
-        self.send_hci_packet(
+            connection.send_ll_control_pdu(
-            hci.HCI_LE_Read_Remote_Features_Complete_Event(
+                ll.FeatureReq(feature_set=self.le_features.value.to_bytes(8, 'little'))
-                status=hci.HCI_ErrorCode.SUCCESS,
+            )
-                connection_handle=handle,
+        else:
-                le_features=bytes.fromhex('dd40000000000000'),
+            connection.send_ll_control_pdu(
+                ll.PeripheralFeatureReq(
+                    feature_set=self.le_features.value.to_bytes(8, 'little')
+                )
             )
-        )
         return None
 
     def on_hci_le_rand_command(

bumble/ll.py

@@ -198,3 +198,24 @@ class CisTerminateInd(ControlPdu):
     cig_id: int
     cis_id: int
     error_code: int
+
+
+@dataclasses.dataclass
+class FeatureReq(ControlPdu):
+    opcode = ControlPdu.Opcode.LL_FEATURE_REQ
+
+    feature_set: bytes
+
+
+@dataclasses.dataclass
+class FeatureRsp(ControlPdu):
+    opcode = ControlPdu.Opcode.LL_FEATURE_RSP
+
+    feature_set: bytes
+
+
+@dataclasses.dataclass
+class PeripheralFeatureReq(ControlPdu):
+    opcode = ControlPdu.Opcode.LL_PERIPHERAL_FEATURE_REQ
+
+    feature_set: bytes

tests/device_test.py

@@ -466,7 +466,9 @@ async def test_get_remote_le_features():
     devices = TwoDevices()
     await devices.setup_connection()
 
-    assert (await devices.connections[0].get_remote_le_features()) is not None
+    assert (
+        await devices.connections[0].get_remote_le_features()
+    ) == devices.controllers[1].le_features
 
 
 # -----------------------------------------------------------------------------