Coverage for src / bluetooth_sig / gatt / characteristics / activity_goal.py: 69%

1"""Activity Goal characteristic (0x2B4E).""" 

2 

3from __future__ import annotations 

4 

5from enum import IntFlag 

6 

7import msgspec 

8 

9from ..constants import UINT16_MAX 

10from ..context import CharacteristicContext 

11from ..exceptions import InsufficientDataError, ValueRangeError 

12from .base import BaseCharacteristic 

13from .utils import DataParser 

14 

15 

16# Hard type for presence flags bit field 

17class ActivityGoalPresenceFlags(IntFlag): 

18 """Presence flags for Activity Goal characteristic.""" 

19 

20 TOTAL_ENERGY_EXPENDITURE = 1 << 0 

21 NORMAL_WALKING_STEPS = 1 << 1 

22 INTENSITY_STEPS = 1 << 2 

23 FLOOR_STEPS = 1 << 3 

24 DISTANCE = 1 << 4 

25 DURATION_NORMAL_WALKING = 1 << 5 

26 DURATION_INTENSITY_WALKING = 1 << 6 

27 

28 

29class ActivityGoalData(msgspec.Struct, frozen=True, kw_only=True): 

30 """Activity Goal data structure.""" 

31 

32 presence_flags: ActivityGoalPresenceFlags 

33 total_energy_expenditure: int | None = None 

34 normal_walking_steps: int | None = None 

35 intensity_steps: int | None = None 

36 floor_steps: int | None = None 

37 distance: int | None = None 

38 duration_normal_walking: int | None = None 

39 duration_intensity_walking: int | None = None 

40 

41 

42class ActivityGoalCharacteristic(BaseCharacteristic[ActivityGoalData]): 

43 """Activity Goal characteristic (0x2B4E). 

44 

45 org.bluetooth.characteristic.activity_goal 

46 

47 The Activity Goal characteristic is used to represent the goal or target of a user, 

48 such as number of steps or total energy expenditure, related to a physical activity session. 

49 """ 

50 

51 min_length: int = 1 # At least presence flags byte 

52 max_length: int = 22 # Max length with all optional fields present 

53 allow_variable_length: bool = True # Variable based on presence flags 

54 

55 def _decode_value(self, data: bytearray, ctx: CharacteristicContext | None = None) -> ActivityGoalData: 

56 """Decode Activity Goal from raw bytes. 

57 

58 Args: 

59 data: Raw bytes from BLE characteristic 

60 ctx: Optional context for parsing 

61 

62 Returns: 

63 ActivityGoalData: Parsed activity goal 

64 

65 Raises: 

66 InsufficientDataError: If data is insufficient for parsing 

67 """ # pylint: disable=too-many-branches 

68 # NOTE: Required by Bluetooth SIG Activity Goal characteristic specification 

69 # Each branch corresponds to a mandatory presence flag check per spec 

70 # Refactoring would violate spec compliance and reduce readability 

71 

72 presence_flags = data[0] 

73 

74 # Start after presence flags 

75 pos = 1 

76 

77 # Parse conditional fields based on presence flags 

78 total_energy_expenditure = None 

79 if presence_flags & ActivityGoalPresenceFlags.TOTAL_ENERGY_EXPENDITURE: 

80 if len(data) < pos + 2: 

81 raise InsufficientDataError("Activity Goal", data, pos + 2) 

82 total_energy_expenditure = DataParser.parse_int16(data, offset=pos, signed=False) 

83 pos += 2 

84 

85 normal_walking_steps = None 

86 if presence_flags & ActivityGoalPresenceFlags.NORMAL_WALKING_STEPS: 

87 if len(data) < pos + 3: 

88 raise InsufficientDataError("Activity Goal", data, pos + 3) 

89 normal_walking_steps = DataParser.parse_int24(data, offset=pos, signed=False) 

90 pos += 3 

91 

92 intensity_steps = None 

93 if presence_flags & ActivityGoalPresenceFlags.INTENSITY_STEPS: 

94 if len(data) < pos + 3: 

95 raise InsufficientDataError("Activity Goal", data, pos + 3) 

96 intensity_steps = DataParser.parse_int24(data, offset=pos, signed=False) 

97 pos += 3 

98 

99 floor_steps = None 

100 if presence_flags & ActivityGoalPresenceFlags.FLOOR_STEPS: 

101 if len(data) < pos + 3: 

102 raise InsufficientDataError("Activity Goal", data, pos + 3) 

103 floor_steps = DataParser.parse_int24(data, offset=pos, signed=False) 

104 pos += 3 

105 

106 distance = None 

107 if presence_flags & ActivityGoalPresenceFlags.DISTANCE: 

108 if len(data) < pos + 3: 

109 raise InsufficientDataError("Activity Goal", data, pos + 3) 

110 distance = DataParser.parse_int24(data, offset=pos, signed=False) 

111 pos += 3 

112 

113 duration_normal_walking = None 

114 if presence_flags & ActivityGoalPresenceFlags.DURATION_NORMAL_WALKING: 

115 if len(data) < pos + 3: 

116 raise InsufficientDataError("Activity Goal", data, pos + 3) 

117 duration_normal_walking = DataParser.parse_int24(data, offset=pos, signed=False) 

118 pos += 3 

119 

120 duration_intensity_walking = None 

121 if presence_flags & ActivityGoalPresenceFlags.DURATION_INTENSITY_WALKING: 

122 if len(data) < pos + 3: 

123 raise InsufficientDataError("Activity Goal", data, pos + 3) 

124 duration_intensity_walking = DataParser.parse_int24(data, offset=pos, signed=False) 

125 pos += 3 

126 

127 return ActivityGoalData( 

128 presence_flags=ActivityGoalPresenceFlags(presence_flags), 

129 total_energy_expenditure=total_energy_expenditure, 

130 normal_walking_steps=normal_walking_steps, 

131 intensity_steps=intensity_steps, 

132 floor_steps=floor_steps, 

133 distance=distance, 

134 duration_normal_walking=duration_normal_walking, 

135 duration_intensity_walking=duration_intensity_walking, 

136 ) 

137 

138 def _encode_value(self, data: ActivityGoalData) -> bytearray: 

139 """Encode Activity Goal to raw bytes. 

140 

141 Args: 

142 data: ActivityGoalData to encode 

143 

144 Returns: 

145 bytearray: Encoded bytes 

146 """ # pylint: disable=too-many-branches 

147 # NOTE: Required by Bluetooth SIG Activity Goal characteristic specification 

148 # Each branch corresponds to a mandatory presence flag encoding per spec 

149 # Refactoring would violate spec compliance and reduce readability 

150 result = bytearray() 

151 

152 result.append(data.presence_flags) 

153 

154 # Encode conditional fields based on presence flags 

155 if data.presence_flags & ActivityGoalPresenceFlags.TOTAL_ENERGY_EXPENDITURE: 

156 if data.total_energy_expenditure is None: 

157 raise ValueRangeError("total_energy_expenditure", data.total_energy_expenditure, 0, UINT16_MAX) 

158 if not 0 <= data.total_energy_expenditure <= UINT16_MAX: 

159 raise ValueRangeError("total_energy_expenditure", data.total_energy_expenditure, 0, UINT16_MAX) 

160 result.extend(data.total_energy_expenditure.to_bytes(2, byteorder="little", signed=False)) 

161 

162 if data.presence_flags & ActivityGoalPresenceFlags.NORMAL_WALKING_STEPS: 

163 if data.normal_walking_steps is None: 

164 raise ValueRangeError("normal_walking_steps", data.normal_walking_steps, 0, UINT16_MAX) 

165 if not 0 <= data.normal_walking_steps <= UINT16_MAX: 

166 raise ValueRangeError("normal_walking_steps", data.normal_walking_steps, 0, UINT16_MAX) 

167 result.extend(data.normal_walking_steps.to_bytes(3, byteorder="little", signed=False)) 

168 

169 if data.presence_flags & ActivityGoalPresenceFlags.INTENSITY_STEPS: 

170 if data.intensity_steps is None: 

171 raise ValueRangeError("intensity_steps", data.intensity_steps, 0, UINT16_MAX) 

172 if not 0 <= data.intensity_steps <= UINT16_MAX: 

173 raise ValueRangeError("intensity_steps", data.intensity_steps, 0, UINT16_MAX) 

174 result.extend(data.intensity_steps.to_bytes(3, byteorder="little", signed=False)) 

175 

176 if data.presence_flags & ActivityGoalPresenceFlags.FLOOR_STEPS: 

177 if data.floor_steps is None: 

178 raise ValueRangeError("floor_steps", data.floor_steps, 0, UINT16_MAX) 

179 if not 0 <= data.floor_steps <= UINT16_MAX: 

180 raise ValueRangeError("floor_steps", data.floor_steps, 0, UINT16_MAX) 

181 result.extend(data.floor_steps.to_bytes(3, byteorder="little", signed=False)) 

182 

183 if data.presence_flags & ActivityGoalPresenceFlags.DISTANCE: 

184 if data.distance is None: 

185 raise ValueRangeError("distance", data.distance, 0, UINT16_MAX) 

186 if not 0 <= data.distance <= UINT16_MAX: 

187 raise ValueRangeError("distance", data.distance, 0, UINT16_MAX) 

188 result.extend(data.distance.to_bytes(3, byteorder="little", signed=False)) 

189 

190 if data.presence_flags & ActivityGoalPresenceFlags.DURATION_NORMAL_WALKING: 

191 if data.duration_normal_walking is None: 

192 raise ValueRangeError("duration_normal_walking", data.duration_normal_walking, 0, UINT16_MAX) 

193 if not 0 <= data.duration_normal_walking <= UINT16_MAX: 

194 raise ValueRangeError("duration_normal_walking", data.duration_normal_walking, 0, UINT16_MAX) 

195 result.extend(data.duration_normal_walking.to_bytes(3, byteorder="little", signed=False)) 

196 

197 if data.presence_flags & ActivityGoalPresenceFlags.DURATION_INTENSITY_WALKING: 

198 if data.duration_intensity_walking is None: 

199 raise ValueRangeError("duration_intensity_walking", data.duration_intensity_walking, 0, UINT16_MAX) 

200 if not 0 <= data.duration_intensity_walking <= UINT16_MAX: 

201 raise ValueRangeError("duration_intensity_walking", data.duration_intensity_walking, 0, UINT16_MAX) 

202 result.extend(data.duration_intensity_walking.to_bytes(3, byteorder="little", signed=False)) 

203 

204 return result