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

1"""General Activity Instantaneous Data characteristic (0x2B3C). 

2 

3Reports instantaneous general activity data with segmented header. 

4 

5References: 

6 Bluetooth SIG Physical Activity Monitor Service 1.0 

7""" 

8 

9from __future__ import annotations 

10 

11from enum import IntFlag 

12 

13import msgspec 

14 

15from ..context import CharacteristicContext 

16from .base import BaseCharacteristic 

17from .utils import DataParser 

18 

19 

20class GeneralActivityInstFlags(IntFlag): 

21 """General Activity Instantaneous Data flags (24-bit, 3 octets).""" 

22 

23 NORMAL_WALKING_EE_PER_HOUR_PRESENT = 0x000001 

24 INTENSITY_EE_PER_HOUR_PRESENT = 0x000002 

25 TOTAL_EE_PER_HOUR_PRESENT = 0x000004 

26 FAT_BURNED_PER_HOUR_PRESENT = 0x000008 

27 METABOLIC_EQUIVALENT_PRESENT = 0x000010 

28 SPEED_PRESENT = 0x000020 

29 MOTION_CADENCE_PRESENT = 0x000040 

30 ELEVATION_PRESENT = 0x000080 

31 ACTIVITY_COUNT_PER_MINUTE_PRESENT = 0x000100 

32 ACTIVITY_LEVEL_PRESENT = 0x000200 

33 ACTIVITY_TYPE_PRESENT = 0x000400 

34 DEVICE_WORN = 0x800000 

35 

36 

37class GeneralActivityInstantaneousData(msgspec.Struct, frozen=True, kw_only=True): 

38 """Parsed data from General Activity Instantaneous Data characteristic. 

39 

40 Attributes: 

41 header: Segmentation header byte. 

42 flags: Presence flags for optional fields (24-bit). 

43 session_id: Session identifier (uint16). 

44 sub_session_id: Sub-session identifier (uint16). 

45 relative_timestamp: Relative timestamp in seconds (uint32). 

46 sequence_number: Sequence number (uint32). 

47 

48 """ 

49 

50 header: int 

51 flags: GeneralActivityInstFlags 

52 session_id: int 

53 sub_session_id: int 

54 relative_timestamp: int 

55 sequence_number: int 

56 

57 

58class GeneralActivityInstantaneousDataCharacteristic(BaseCharacteristic[GeneralActivityInstantaneousData]): 

59 """General Activity Instantaneous Data characteristic (0x2B3C). 

60 

61 org.bluetooth.characteristic.general_activity_instantaneous_data 

62 

63 Reports instantaneous general activity data. 

64 """ 

65 

66 min_length = 16 # header(1) + flags(3) + session(2) + subsession(2) + timestamp(4) + sequence(4) 

67 allow_variable_length = True 

68 

69 def _decode_value( 

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

71 ) -> GeneralActivityInstantaneousData: 

72 """Parse General Activity Instantaneous Data. 

73 

74 Format: Header (uint8) + Flags (uint24) + SessionID (uint16) 

75 + SubSessionID (uint16) + RelativeTimestamp (uint32) 

76 + SequenceNumber (uint32) + [optional fields]. 

77 """ 

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

79 flags_raw = DataParser.parse_int24(data, 1, signed=False) 

80 flags = GeneralActivityInstFlags(flags_raw) 

81 session_id = DataParser.parse_int16(data, 4, signed=False) 

82 sub_session_id = DataParser.parse_int16(data, 6, signed=False) 

83 relative_timestamp = DataParser.parse_int32(data, 8, signed=False) 

84 sequence_number = DataParser.parse_int32(data, 12, signed=False) 

85 

86 return GeneralActivityInstantaneousData( 

87 header=header, 

88 flags=flags, 

89 session_id=session_id, 

90 sub_session_id=sub_session_id, 

91 relative_timestamp=relative_timestamp, 

92 sequence_number=sequence_number, 

93 ) 

94 

95 def _encode_value(self, data: GeneralActivityInstantaneousData) -> bytearray: 

96 """Encode General Activity Instantaneous data.""" 

97 result = bytearray() 

98 result.extend(DataParser.encode_int8(data.header, signed=False)) 

99 result.extend(DataParser.encode_int24(int(data.flags), signed=False)) 

100 result.extend(DataParser.encode_int16(data.session_id, signed=False)) 

101 result.extend(DataParser.encode_int16(data.sub_session_id, signed=False)) 

102 result.extend(DataParser.encode_int32(data.relative_timestamp, signed=False)) 

103 result.extend(DataParser.encode_int32(data.sequence_number, signed=False)) 

104 return result