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

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

2 

3from __future__ import annotations 

4 

5from enum import IntFlag 

6 

7import msgspec 

8 

9from ...types.gatt_enums import ValueType 

10from ...types.location import PositionStatus 

11from ..context import CharacteristicContext 

12from .base import BaseCharacteristic 

13from .utils import DataParser 

14 

15 

16class PositionQualityFlags(IntFlag): 

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

18 

19 NUMBER_OF_BEACONS_IN_SOLUTION_PRESENT = 0x0001 

20 NUMBER_OF_BEACONS_IN_VIEW_PRESENT = 0x0002 

21 TIME_TO_FIRST_FIX_PRESENT = 0x0004 

22 EHPE_PRESENT = 0x0008 

23 EVPE_PRESENT = 0x0010 

24 HDOP_PRESENT = 0x0020 

25 VDOP_PRESENT = 0x0040 

26 

27 

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

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

30 

31 flags: PositionQualityFlags 

32 number_of_beacons_in_solution: int | None = None 

33 number_of_beacons_in_view: int | None = None 

34 time_to_first_fix: float | None = None 

35 ehpe: float | None = None 

36 evpe: float | None = None 

37 hdop: float | None = None 

38 vdop: float | None = None 

39 position_status: PositionStatus | None = None 

40 

41 

42class PositionQualityCharacteristic(BaseCharacteristic[PositionQualityData]): 

43 """Position Quality characteristic. 

44 

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

46 """ 

47 

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

49 

50 min_length = 2 # Flags(2) minimum 

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

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

53 allow_variable_length: bool = True # Variable optional fields 

54 

55 # Bit masks and shifts for status information in flags 

56 POSITION_STATUS_MASK = 0x0180 

57 POSITION_STATUS_SHIFT = 7 

58 

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

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

61 

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

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

64 

65 Args: 

66 data: Raw bytearray from BLE characteristic 

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

68 

69 Returns: 

70 PositionQualityData containing parsed position quality data 

71 

72 """ 

73 if len(data) < 2: 

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

75 

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

77 

78 # Extract status information from flags 

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

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

81 

82 # Parse optional fields 

83 number_of_beacons_in_solution: int | None = None 

84 number_of_beacons_in_view: int | None = None 

85 time_to_first_fix: float | None = None 

86 ehpe: float | None = None 

87 evpe: float | None = None 

88 hdop: float | None = None 

89 vdop: float | None = None 

90 offset = 2 

91 

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

93 number_of_beacons_in_solution = data[offset] 

94 offset += 1 

95 

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

97 number_of_beacons_in_view = data[offset] 

98 offset += 1 

99 

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

101 # Unit is 1/10 seconds 

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

103 offset += 2 

104 

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

106 # Unit is 1/100 m 

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

108 offset += 4 

109 

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

111 # Unit is 1/100 m 

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

113 offset += 4 

114 

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

116 # Unit is 2*10^-1 

117 hdop = data[offset] / 2.0 

118 offset += 1 

119 

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

121 # Unit is 2*10^-1 

122 vdop = data[offset] / 2.0 

123 

124 return PositionQualityData( 

125 flags=flags, 

126 number_of_beacons_in_solution=number_of_beacons_in_solution, 

127 number_of_beacons_in_view=number_of_beacons_in_view, 

128 time_to_first_fix=time_to_first_fix, 

129 ehpe=ehpe, 

130 evpe=evpe, 

131 hdop=hdop, 

132 vdop=vdop, 

133 position_status=position_status, 

134 ) 

135 

136 def _encode_value(self, data: PositionQualityData) -> bytearray: 

137 """Encode PositionQualityData back to bytes. 

138 

139 Args: 

140 data: PositionQualityData instance to encode 

141 

142 Returns: 

143 Encoded bytes representing the position quality data 

144 

145 """ 

146 result = bytearray() 

147 

148 flags = int(data.flags) 

149 

150 # Set status bits in flags 

151 if data.position_status is not None: 

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

153 

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

155 

156 if data.number_of_beacons_in_solution is not None: 

157 result.append(data.number_of_beacons_in_solution) 

158 

159 if data.number_of_beacons_in_view is not None: 

160 result.append(data.number_of_beacons_in_view) 

161 

162 if data.time_to_first_fix is not None: 

163 time_value = int(data.time_to_first_fix * 10) 

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

165 

166 if data.ehpe is not None: 

167 ehpe_value = int(data.ehpe * 100) 

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

169 

170 if data.evpe is not None: 

171 evpe_value = int(data.evpe * 100) 

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

173 

174 if data.hdop is not None: 

175 hdop_value = int(data.hdop * 2) 

176 result.append(hdop_value) 

177 

178 if data.vdop is not None: 

179 vdop_value = int(data.vdop * 2) 

180 result.append(vdop_value) 

181 

182 return result