Coverage for src / bluetooth_sig / gatt / characteristics / acs_data_in.py: 100%

1"""ACS Data In characteristic (0x2B30).""" 

2 

3from __future__ import annotations 

4 

5from ...types import MAX_ROLLING_SEGMENT_COUNTER, ACSDataPacket, ACSSegmentationHeader 

6from ..context import CharacteristicContext 

7from .base import BaseCharacteristic 

8from .utils import DataParser 

9 

10 

11class ACSDataInCharacteristic(BaseCharacteristic[ACSDataPacket]): 

12 """ACS Data In characteristic (0x2B30). 

13 

14 org.bluetooth.characteristic.acs_data_in 

15 

16 Segmented data input for Authorization Control Service. 

17 """ 

18 

19 min_length = 1 

20 allow_variable_length = True 

21 

22 def _decode_value( 

23 self, data: bytearray, ctx: CharacteristicContext | None = None, *, validate: bool = True 

24 ) -> ACSDataPacket: 

25 header = DataParser.parse_int8(data, 0, signed=False) 

26 payload = bytes(data[1:]) 

27 

28 return ACSDataPacket( 

29 header=ACSSegmentationHeader( 

30 first_segment=bool(header & 0x01), 

31 last_segment=bool(header & 0x02), 

32 rolling_segment_counter=(header >> 2) & 0x3F, 

33 ), 

34 payload=payload, 

35 ) 

36 

37 def _encode_value(self, data: ACSDataPacket) -> bytearray: 

38 counter = data.header.rolling_segment_counter 

39 if not 0 <= counter <= MAX_ROLLING_SEGMENT_COUNTER: 

40 raise ValueError(f"rolling_segment_counter must be in range 0-63, got {counter}") 

41 

42 header = 0 

43 if data.header.first_segment: 

44 header |= 0x01 

45 if data.header.last_segment: 

46 header |= 0x02 

47 

48 header |= counter << 2 

49 return bytearray([header, *data.payload])