Coverage for src/bluetooth_sig/gatt/characteristics/position_quality.py: 94%

1"""Position Quality characteristic implementation.""" 

2 

3from __future__ import annotations 

4 

5from enum import IntEnum, IntFlag 

6 

7import msgspec 

8 

9from ...types.gatt_enums import ValueType 

10from ..context import CharacteristicContext 

11from .base import BaseCharacteristic 

12from .utils import DataParser 

13 

14 

15class PositionQualityFlags(IntFlag): 

16 """Position Quality flags as per Bluetooth SIG specification.""" 

17 

18 NUMBER_OF_BEACONS_IN_SOLUTION_PRESENT = 0x0001 

19 NUMBER_OF_BEACONS_IN_VIEW_PRESENT = 0x0002 

20 TIME_TO_FIRST_FIX_PRESENT = 0x0004 

21 EHPE_PRESENT = 0x0008 

22 EVPE_PRESENT = 0x0010 

23 HDOP_PRESENT = 0x0020 

24 VDOP_PRESENT = 0x0040 

25 

26 

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

28 """Parsed data from Position Quality characteristic.""" 

29 

30 flags: PositionQualityFlags 

31 number_of_beacons_in_solution: int | None = None 

32 number_of_beacons_in_view: int | None = None 

33 time_to_first_fix: float | None = None 

34 ehpe: float | None = None 

35 evpe: float | None = None 

36 hdop: float | None = None 

37 vdop: float | None = None 

38 position_status: PositionStatus | None = None 

39 

40 

41class PositionStatus(IntEnum): 

42 """Position status enumeration.""" 

43 

44 NO_POSITION = 0 

45 POSITION_OK = 1 

46 ESTIMATED_POSITION = 2 

47 LAST_KNOWN_POSITION = 3 

48 

49 

50class PositionQualityCharacteristic(BaseCharacteristic): 

51 """Position Quality characteristic. 

52 

53 Used to represent data related to the quality of a position measurement. 

54 """ 

55 

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

57 

58 min_length = 2 # Flags(2) minimum 

59 max_length = 16 # Flags(2) + NumberOfBeaconsInSolution(1) + NumberOfBeaconsInView(1) + 

60 # TimeToFirstFix(2) + EHPE(4) + EVPE(4) + HDOP(1) + VDOP(1) maximum 

61 allow_variable_length: bool = True # Variable optional fields 

62 

63 # Bit masks and shifts for status information in flags 

64 POSITION_STATUS_MASK = 0x0180 

65 POSITION_STATUS_SHIFT = 7 

66 

67 def decode_value(self, data: bytearray, ctx: CharacteristicContext | None = None) -> PositionQualityData: # pylint: disable=too-many-locals 

68 """Parse position quality data according to Bluetooth specification. 

69 

70 Format: Flags(2) + [Number of Beacons in Solution(1)] + [Number of Beacons in View(1)] + 

71 [Time to First Fix(2)] + [EHPE(4)] + [EVPE(4)] + [HDOP(1)] + [VDOP(1)]. 

72 

73 Args: 

74 data: Raw bytearray from BLE characteristic 

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

76 

77 Returns: 

78 PositionQualityData containing parsed position quality data 

79 

80 """ 

81 if len(data) < 2: 

82 raise ValueError("Position Quality data must be at least 2 bytes") 

83 

84 flags = PositionQualityFlags(DataParser.parse_int16(data, 0, signed=False)) 

85 

86 # Extract status information from flags 

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

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

89 

90 # Parse optional fields 

91 number_of_beacons_in_solution: int | None = None 

92 number_of_beacons_in_view: int | None = None 

93 time_to_first_fix: float | None = None 

94 ehpe: float | None = None 

95 evpe: float | None = None 

96 hdop: float | None = None 

97 vdop: float | None = None 

98 offset = 2 

99 

100 if (flags & PositionQualityFlags.NUMBER_OF_BEACONS_IN_SOLUTION_PRESENT) and len(data) >= offset + 1: 

101 number_of_beacons_in_solution = data[offset] 

102 offset += 1 

103 

104 if (flags & PositionQualityFlags.NUMBER_OF_BEACONS_IN_VIEW_PRESENT) and len(data) >= offset + 1: 

105 number_of_beacons_in_view = data[offset] 

106 offset += 1 

107 

108 if (flags & PositionQualityFlags.TIME_TO_FIRST_FIX_PRESENT) and len(data) >= offset + 2: 

109 # Unit is 1/10 seconds 

110 time_to_first_fix = DataParser.parse_int16(data, offset, signed=False) / 10.0 

111 offset += 2 

112 

113 if (flags & PositionQualityFlags.EHPE_PRESENT) and len(data) >= offset + 4: 

114 # Unit is 1/100 m 

115 ehpe = DataParser.parse_int32(data, offset, signed=False) / 100.0 

116 offset += 4 

117 

118 if (flags & PositionQualityFlags.EVPE_PRESENT) and len(data) >= offset + 4: 

119 # Unit is 1/100 m 

120 evpe = DataParser.parse_int32(data, offset, signed=False) / 100.0 

121 offset += 4 

122 

123 if (flags & PositionQualityFlags.HDOP_PRESENT) and len(data) >= offset + 1: 

124 # Unit is 2*10^-1 

125 hdop = data[offset] / 2.0 

126 offset += 1 

127 

128 if (flags & PositionQualityFlags.VDOP_PRESENT) and len(data) >= offset + 1: 

129 # Unit is 2*10^-1 

130 vdop = data[offset] / 2.0 

131 

132 return PositionQualityData( 

133 flags=flags, 

134 number_of_beacons_in_solution=number_of_beacons_in_solution, 

135 number_of_beacons_in_view=number_of_beacons_in_view, 

136 time_to_first_fix=time_to_first_fix, 

137 ehpe=ehpe, 

138 evpe=evpe, 

139 hdop=hdop, 

140 vdop=vdop, 

141 position_status=position_status, 

142 ) 

143 

144 def encode_value(self, data: PositionQualityData) -> bytearray: 

145 """Encode PositionQualityData back to bytes. 

146 

147 Args: 

148 data: PositionQualityData instance to encode 

149 

150 Returns: 

151 Encoded bytes representing the position quality data 

152 

153 """ 

154 result = bytearray() 

155 

156 flags = int(data.flags) 

157 

158 # Set status bits in flags 

159 if data.position_status is not None: 

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

161 

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

163 

164 if data.number_of_beacons_in_solution is not None: 

165 result.append(data.number_of_beacons_in_solution) 

166 

167 if data.number_of_beacons_in_view is not None: 

168 result.append(data.number_of_beacons_in_view) 

169 

170 if data.time_to_first_fix is not None: 

171 time_value = int(data.time_to_first_fix * 10) 

172 result.extend(DataParser.encode_int16(time_value, signed=False)) 

173 

174 if data.ehpe is not None: 

175 ehpe_value = int(data.ehpe * 100) 

176 result.extend(DataParser.encode_int32(ehpe_value, signed=False)) 

177 

178 if data.evpe is not None: 

179 evpe_value = int(data.evpe * 100) 

180 result.extend(DataParser.encode_int32(evpe_value, signed=False)) 

181 

182 if data.hdop is not None: 

183 hdop_value = int(data.hdop * 2) 

184 result.append(hdop_value) 

185 

186 if data.vdop is not None: 

187 vdop_value = int(data.vdop * 2) 

188 result.append(vdop_value) 

189 

190 return result