Coverage for src / bluetooth_sig / gatt / characteristics / location_and_speed.py: 92%

1"""Location and Speed characteristic implementation.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime 

6from enum import IntEnum, IntFlag 

7 

8import msgspec 

9 

10from ...types.gatt_enums import ValueType 

11from ...types.location import PositionStatus 

12from ..context import CharacteristicContext 

13from .base import BaseCharacteristic 

14from .utils import DataParser, IEEE11073Parser 

15 

16 

17class LocationAndSpeedFlags(IntFlag): 

18 """Location and Speed flags as per Bluetooth SIG specification.""" 

19 

20 INSTANTANEOUS_SPEED_PRESENT = 0x0001 

21 TOTAL_DISTANCE_PRESENT = 0x0002 

22 LOCATION_PRESENT = 0x0004 

23 ELEVATION_PRESENT = 0x0008 

24 HEADING_PRESENT = 0x0010 

25 ROLLING_TIME_PRESENT = 0x0020 

26 UTC_TIME_PRESENT = 0x0040 

27 

28 

29class SpeedAndDistanceFormat(IntEnum): 

30 """Speed and distance format enumeration.""" 

31 

32 FORMAT_2D = 0 

33 FORMAT_3D = 1 

34 

35 

36class ElevationSource(IntEnum): 

37 """Elevation source enumeration.""" 

38 

39 POSITIONING_SYSTEM = 0 

40 BAROMETRIC_AIR_PRESSURE = 1 

41 DATABASE_SERVICE = 2 

42 OTHER = 3 

43 

44 

45class HeadingSource(IntEnum): 

46 """Heading source enumeration.""" 

47 

48 HEADING_BASED_ON_MOVEMENT = 0 

49 HEADING_BASED_ON_MAGNETIC_COMPASS = 1 

50 

51 

52class LocationAndSpeedData(msgspec.Struct, frozen=True, kw_only=True): # pylint: disable=too-few-public-methods,too-many-instance-attributes 

53 """Parsed data from Location and Speed characteristic.""" 

54 

55 flags: LocationAndSpeedFlags 

56 instantaneous_speed: float | None = None 

57 total_distance: float | None = None 

58 latitude: float | None = None 

59 longitude: float | None = None 

60 elevation: float | None = None 

61 heading: float | None = None 

62 rolling_time: int | None = None 

63 utc_time: datetime | None = None 

64 position_status: PositionStatus | None = None 

65 speed_and_distance_format: SpeedAndDistanceFormat | None = None 

66 elevation_source: ElevationSource | None = None 

67 heading_source: HeadingSource | None = None 

68 

69 

70class LocationAndSpeedCharacteristic(BaseCharacteristic[LocationAndSpeedData]): 

71 """Location and Speed characteristic. 

72 

73 Used to represent data related to a location and speed sensor. 

74 Note that it is possible for this characteristic to exceed the default LE ATT_MTU size. 

75 """ 

76 

77 _manual_value_type: ValueType | str | None = ValueType.DICT # Override since decode_value returns dataclass 

78 

79 min_length = 2 # Flags(2) minimum 

80 max_length = 28 # Flags(2) + InstantaneousSpeed(2) + TotalDistance(3) + Location(8) + 

81 # Elevation(3) + Heading(2) + RollingTime(1) + UTCTime(7) maximum 

82 allow_variable_length: bool = True # Variable optional fields 

83 

84 # Bit masks and shifts for status information in flags 

85 POSITION_STATUS_MASK = 0x0300 

86 POSITION_STATUS_SHIFT = 8 

87 SPEED_DISTANCE_FORMAT_MASK = 0x0400 

88 SPEED_DISTANCE_FORMAT_SHIFT = 10 

89 ELEVATION_SOURCE_MASK = 0x1800 

90 ELEVATION_SOURCE_SHIFT = 11 

91 HEADING_SOURCE_MASK = 0x2000 

92 HEADING_SOURCE_SHIFT = 13 

93 

94 def _decode_value(self, data: bytearray, ctx: CharacteristicContext | None = None) -> LocationAndSpeedData: # pylint: disable=too-many-locals 

95 """Parse location and speed data according to Bluetooth specification. 

96 

97 Format: Flags(2) + [Instantaneous Speed(2)] + [Total Distance(3)] + [Location - Latitude(4)] + 

98 [Location - Longitude(4)] + [Elevation(3)] + [Heading(2)] + [Rolling Time(1)] + [UTC Time(7)]. 

99 

100 Args: 

101 data: Raw bytearray from BLE characteristic 

102 ctx: Optional context providing surrounding context (may be None) 

103 

104 Returns: 

105 LocationAndSpeedData containing parsed location and speed data 

106 

107 """ 

108 if len(data) < 2: 

109 raise ValueError("Location and Speed data must be at least 2 bytes") 

110 

111 flags = LocationAndSpeedFlags(DataParser.parse_int16(data, 0, signed=False)) 

112 

113 # Extract status information from flags 

114 position_status_bits = (flags & self.POSITION_STATUS_MASK) >> self.POSITION_STATUS_SHIFT 

115 position_status = PositionStatus(position_status_bits) if position_status_bits <= 3 else None 

116 

117 speed_distance_format_bit = (flags & self.SPEED_DISTANCE_FORMAT_MASK) >> self.SPEED_DISTANCE_FORMAT_SHIFT 

118 speed_and_distance_format = SpeedAndDistanceFormat(speed_distance_format_bit) 

119 

120 elevation_source_bits = (flags & self.ELEVATION_SOURCE_MASK) >> self.ELEVATION_SOURCE_SHIFT 

121 elevation_source = ElevationSource(elevation_source_bits) if elevation_source_bits <= 3 else None 

122 

123 heading_source_bit = (flags & self.HEADING_SOURCE_MASK) >> self.HEADING_SOURCE_SHIFT 

124 heading_source = HeadingSource(heading_source_bit) 

125 

126 # Parse optional fields 

127 instantaneous_speed: float | None = None 

128 total_distance: float | None = None 

129 latitude: float | None = None 

130 longitude: float | None = None 

131 elevation: float | None = None 

132 heading: float | None = None 

133 rolling_time: int | None = None 

134 utc_time: datetime | None = None 

135 offset = 2 

136 

137 if (flags & LocationAndSpeedFlags.INSTANTANEOUS_SPEED_PRESENT) and len(data) >= offset + 2: 

138 # Unit is 1/100 of a m/s 

139 instantaneous_speed = DataParser.parse_int16(data, offset, signed=False) / 100.0 

140 offset += 2 

141 

142 if (flags & LocationAndSpeedFlags.TOTAL_DISTANCE_PRESENT) and len(data) >= offset + 3: 

143 # Unit is 1/10 m 

144 total_distance = DataParser.parse_int24(data, offset, signed=False) / 10.0 

145 offset += 3 

146 

147 if (flags & LocationAndSpeedFlags.LOCATION_PRESENT) and len(data) >= offset + 8: 

148 # Unit is 1*10^-7 degrees 

149 latitude = DataParser.parse_int32(data, offset, signed=True) / 10000000.0 

150 longitude = DataParser.parse_int32(data, offset + 4, signed=True) / 10000000.0 

151 offset += 8 

152 

153 if (flags & LocationAndSpeedFlags.ELEVATION_PRESENT) and len(data) >= offset + 3: 

154 # Unit is 1/100 m 

155 elevation = DataParser.parse_int24(data, offset, signed=True) / 100.0 

156 offset += 3 

157 

158 if (flags & LocationAndSpeedFlags.HEADING_PRESENT) and len(data) >= offset + 2: 

159 # Unit is 1*10^-2 degrees 

160 heading = DataParser.parse_int16(data, offset, signed=False) / 100.0 

161 offset += 2 

162 

163 if (flags & LocationAndSpeedFlags.ROLLING_TIME_PRESENT) and len(data) >= offset + 1: 

164 rolling_time = data[offset] 

165 offset += 1 

166 

167 if (flags & LocationAndSpeedFlags.UTC_TIME_PRESENT) and len(data) >= offset + 7: 

168 utc_time = IEEE11073Parser.parse_timestamp(data, offset) 

169 

170 return LocationAndSpeedData( 

171 flags=flags, 

172 instantaneous_speed=instantaneous_speed, 

173 total_distance=total_distance, 

174 latitude=latitude, 

175 longitude=longitude, 

176 elevation=elevation, 

177 heading=heading, 

178 rolling_time=rolling_time, 

179 utc_time=utc_time, 

180 position_status=position_status, 

181 speed_and_distance_format=speed_and_distance_format, 

182 elevation_source=elevation_source, 

183 heading_source=heading_source, 

184 ) 

185 

186 def _encode_value(self, data: LocationAndSpeedData) -> bytearray: 

187 """Encode LocationAndSpeedData back to bytes. 

188 

189 Args: 

190 data: LocationAndSpeedData instance to encode 

191 

192 Returns: 

193 Encoded bytes representing the location and speed data 

194 

195 """ 

196 result = bytearray() 

197 

198 flags = int(data.flags) 

199 

200 # Set status bits in flags 

201 if data.position_status is not None: 

202 flags |= data.position_status.value << self.POSITION_STATUS_SHIFT 

203 

204 if data.speed_and_distance_format is not None: 

205 flags |= data.speed_and_distance_format.value << self.SPEED_DISTANCE_FORMAT_SHIFT 

206 

207 if data.elevation_source is not None: 

208 flags |= data.elevation_source.value << self.ELEVATION_SOURCE_SHIFT 

209 

210 if data.heading_source is not None: 

211 flags |= data.heading_source.value << self.HEADING_SOURCE_SHIFT 

212 

213 result.extend(DataParser.encode_int16(flags, signed=False)) 

214 

215 if data.instantaneous_speed is not None: 

216 speed_value = int(data.instantaneous_speed * 100) 

217 result.extend(DataParser.encode_int16(speed_value, signed=False)) 

218 

219 if data.total_distance is not None: 

220 distance_value = int(data.total_distance * 10) 

221 result.extend(DataParser.encode_int24(distance_value, signed=False)) 

222 

223 if data.latitude is not None and data.longitude is not None: 

224 lat_value = int(data.latitude * 10000000) 

225 lon_value = int(data.longitude * 10000000) 

226 result.extend(DataParser.encode_int32(lat_value, signed=True)) 

227 result.extend(DataParser.encode_int32(lon_value, signed=True)) 

228 

229 if data.elevation is not None: 

230 elevation_value = int(data.elevation * 100) 

231 result.extend(DataParser.encode_int24(elevation_value, signed=True)) 

232 

233 if data.heading is not None: 

234 heading_value = int(data.heading * 100) 

235 result.extend(DataParser.encode_int16(heading_value, signed=False)) 

236 

237 if data.rolling_time is not None: 

238 result.append(data.rolling_time) 

239 

240 if data.utc_time is not None: 

241 result.extend(IEEE11073Parser.encode_timestamp(data.utc_time)) 

242 

243 return result