diff --git a/bumble/device.py b/bumble/device.py index e90dbf01..f3b431c5 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -1575,6 +1575,18 @@ class CisParameters: rtn_c_to_p: int = DEVICE_DEFAULT_ISO_CIS_RTN # Number of C->P retransmissions rtn_p_to_c: int = DEVICE_DEFAULT_ISO_CIS_RTN # Number of P->C retransmissions + def __post_init__(self) -> None: + # For unidirectional CIS (e.g., Central-to-Peripheral only), the unused direction's + # SDU size is 0. If SDU size is 0, the corresponding retransmission count and PHY + # must also be set to 0. Otherwise, some controllers will reject the parameters + # with error 0x30 (Parameter Out Of Mandatory Range). + if self.max_sdu_c_to_p == 0: + self.rtn_c_to_p = 0 + self.phy_c_to_p = hci.PhyBit(0) + if self.max_sdu_p_to_c == 0: + self.rtn_p_to_c = 0 + self.phy_p_to_c = hci.PhyBit(0) + cig_id: int cis_parameters: list[CisParameters] sdu_interval_c_to_p: int # C->P SDU interval, in microseconds diff --git a/tests/device_test.py b/tests/device_test.py index a45c43dd..3efe4db1 100644 --- a/tests/device_test.py +++ b/tests/device_test.py @@ -595,6 +595,25 @@ def on_hci_le_cis_established_event(host, event): await asyncio.wait_for(cis_create_task, _TIMEOUT) +# ----------------------------------------------------------------------------- +def test_cis_parameters_unidirectional(): + # Test C2P unidirectional (P to C not used) + cis_c2p = CigParameters.CisParameters(cis_id=1, max_sdu_p_to_c=0) + assert cis_c2p.max_sdu_c_to_p != 0 + assert cis_c2p.rtn_c_to_p != 0 + assert cis_c2p.phy_c_to_p != hci.PhyBit(0) + assert cis_c2p.rtn_p_to_c == 0 + assert cis_c2p.phy_p_to_c == hci.PhyBit(0) + + # Test P2C unidirectional (C to P not used) + cis_p2c = CigParameters.CisParameters(cis_id=2, max_sdu_c_to_p=0) + assert cis_p2c.max_sdu_p_to_c != 0 + assert cis_p2c.rtn_p_to_c != 0 + assert cis_p2c.phy_p_to_c != hci.PhyBit(0) + assert cis_p2c.rtn_c_to_p == 0 + assert cis_p2c.phy_c_to_p == hci.PhyBit(0) + + # ----------------------------------------------------------------------------- @pytest.mark.asyncio async def test_enter_and_exit_sniff_mode():