Coverage for src / bluetooth_sig / gatt / characteristics / plx_spot_check_measurement.py: 61%

1"""PLX Spot-Check Measurement characteristic implementation.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from enum import IntFlag 

7 

8import msgspec 

9 

10from ...types.gatt_enums import CharacteristicName 

11from ..context import CharacteristicContext 

12from .base import BaseCharacteristic 

13from .plx_features import PLXFeatureFlags, PLXFeaturesCharacteristic 

14from .utils import DataParser, IEEE11073Parser 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class PLXSpotCheckFlags(IntFlag): 

20 """PLX Spot-Check measurement flags.""" 

21 

22 SPO2PR_FAST = 0x01 

23 MEASUREMENT_STATUS_PRESENT = 0x02 

24 DEVICE_AND_SENSOR_STATUS_PRESENT = 0x04 

25 PULSE_AMPLITUDE_INDEX_PRESENT = 0x08 

26 

27 

28class PLXMeasurementStatus(IntFlag): 

29 """PLX Measurement Status flags (16-bit).""" 

30 

31 MEASUREMENT_ONGOING = 0x0001 

32 EARLY_ESTIMATED_DATA = 0x0002 

33 VALIDATED_DATA = 0x0004 

34 FULLY_QUALIFIED_DATA = 0x0008 

35 DATA_FROM_MEASUREMENT_STORAGE = 0x0010 

36 DATA_FOR_DEMONSTRATION = 0x0020 

37 DATA_FROM_TESTING_SIMULATION = 0x0040 

38 DATA_FROM_CALIBRATION_TEST = 0x0080 

39 

40 

41class PLXDeviceAndSensorStatus(IntFlag): 

42 """PLX Device and Sensor Status flags (24-bit).""" 

43 

44 # Device Status (bits 0-15, same as Measurement Status) 

45 DEVICE_MEASUREMENT_ONGOING = 0x000001 

46 DEVICE_EARLY_ESTIMATED_DATA = 0x000002 

47 DEVICE_VALIDATED_DATA = 0x000004 

48 DEVICE_FULLY_QUALIFIED_DATA = 0x000008 

49 DEVICE_DATA_FROM_MEASUREMENT_STORAGE = 0x000010 

50 DEVICE_DATA_FOR_DEMONSTRATION = 0x000020 

51 DEVICE_DATA_FROM_TESTING_SIMULATION = 0x000040 

52 DEVICE_DATA_FROM_CALIBRATION_TEST = 0x000080 

53 

54 # Sensor Status (bits 16-23) 

55 SENSOR_OPERATIONAL = 0x000100 

56 SENSOR_DEFECTIVE = 0x000200 

57 SENSOR_DISCONNECTED = 0x000400 

58 SENSOR_MALFUNCTIONING = 0x000800 

59 SENSOR_UNCALIBRATED = 0x001000 

60 SENSOR_NOT_OPERATIONAL = 0x002000 

61 

62 

63class PLXSpotCheckData(msgspec.Struct, frozen=True, kw_only=True): # pylint: disable=too-few-public-methods 

64 """Parsed PLX spot-check measurement data.""" 

65 

66 spot_check_flags: PLXSpotCheckFlags # PLX spot-check measurement flags 

67 spo2: float # Blood oxygen saturation percentage (SpO2) 

68 pulse_rate: float # Pulse rate in beats per minute 

69 measurement_status: PLXMeasurementStatus | None = None # Optional measurement status flags 

70 device_and_sensor_status: PLXDeviceAndSensorStatus | None = None # Optional device and sensor status flags 

71 pulse_amplitude_index: float | None = None # Optional pulse amplitude index value 

72 supported_features: PLXFeatureFlags | None = None # Optional PLX features from context (PLXFeatureFlags enum) 

73 

74 

75class PLXSpotCheckMeasurementCharacteristic(BaseCharacteristic[PLXSpotCheckData]): 

76 """PLX Spot-Check Measurement characteristic (0x2A5E). 

77 

78 Used to transmit single SpO2 (blood oxygen saturation) and pulse rate 

79 measurements from spot-check readings. 

80 """ 

81 

82 _characteristic_name: str = "PLX Spot-Check Measurement" 

83 

84 _optional_dependencies = [PLXFeaturesCharacteristic] 

85 

86 # Declarative validation (automatic) 

87 min_length: int | None = 5 # Flags(1) + SpO2(2) + PulseRate(2) minimum 

88 max_length: int | None = 12 # + MeasurementStatus(2) + DeviceAndSensorStatus(3) + PulseAmplitudeIndex(2) maximum 

89 allow_variable_length: bool = True # Variable optional fields 

90 

91 def _decode_value( # pylint: disable=too-many-locals,too-many-branches # Complexity needed for spec parsing 

92 self, data: bytearray, ctx: CharacteristicContext | None = None 

93 ) -> PLXSpotCheckData: 

94 """Parse PLX spot-check measurement data according to Bluetooth specification. 

95 

96 Format: Flags(1) + SpO2(2) + Pulse Rate(2) + [Measurement Status(2)] + 

97 [Device and Sensor Status(3)] + [Pulse Amplitude Index(2)] 

98 SpO2 and Pulse Rate are IEEE-11073 16-bit SFLOAT. 

99 

100 Context Enhancement: 

101 If ctx is provided, this method will attempt to enhance the parsed data with: 

102 - PLX Features (0x2A60): Device capabilities and supported measurement types 

103 

104 Args: 

105 data: Raw bytearray from BLE characteristic. 

106 ctx: Optional CharacteristicContext providing surrounding context (may be None). 

107 

108 Returns: 

109 PLXSpotCheckData containing parsed PLX spot-check data with optional 

110 context-enhanced information. 

111 

112 """ 

113 if len(data) < 5: 

114 raise ValueError("PLX Spot-Check Measurement data must be at least 5 bytes") 

115 

116 flags = PLXSpotCheckFlags(data[0]) 

117 

118 # Parse SpO2 and pulse rate using IEEE-11073 SFLOAT format 

119 spo2 = IEEE11073Parser.parse_sfloat(data, 1) 

120 pulse_rate = IEEE11073Parser.parse_sfloat(data, 3) 

121 

122 # Parse optional fields 

123 measurement_status: PLXMeasurementStatus | None = None 

124 device_and_sensor_status: PLXDeviceAndSensorStatus | None = None 

125 pulse_amplitude_index: float | None = None 

126 offset = 5 

127 

128 if PLXSpotCheckFlags.MEASUREMENT_STATUS_PRESENT in flags and len(data) >= offset + 2: 

129 measurement_status = PLXMeasurementStatus(DataParser.parse_int16(data, offset, signed=False)) 

130 offset += 2 

131 

132 if PLXSpotCheckFlags.DEVICE_AND_SENSOR_STATUS_PRESENT in flags and len(data) >= offset + 3: 

133 device_and_sensor_status = PLXDeviceAndSensorStatus( 

134 DataParser.parse_int32(data[offset : offset + 3] + b"\x00", 0, signed=False) 

135 ) # Pad to 4 bytes 

136 offset += 3 

137 

138 if PLXSpotCheckFlags.PULSE_AMPLITUDE_INDEX_PRESENT in flags and len(data) >= offset + 2: 

139 pulse_amplitude_index = IEEE11073Parser.parse_sfloat(data, offset) 

140 

141 # Context enhancement: PLX Features 

142 supported_features: PLXFeatureFlags | None = None 

143 if ctx: 

144 plx_features_value = self.get_context_characteristic(ctx, CharacteristicName.PLX_FEATURES) 

145 if plx_features_value is not None: 

146 # PLX Features returns PLXFeatureFlags enum 

147 supported_features = plx_features_value 

148 

149 # Create immutable struct with all values 

150 return PLXSpotCheckData( 

151 spot_check_flags=flags, 

152 spo2=spo2, 

153 pulse_rate=pulse_rate, 

154 measurement_status=measurement_status, 

155 device_and_sensor_status=device_and_sensor_status, 

156 pulse_amplitude_index=pulse_amplitude_index, 

157 supported_features=supported_features, 

158 ) 

159 

160 def _encode_value(self, data: PLXSpotCheckData) -> bytearray: 

161 """Encode PLX spot-check measurement value back to bytes. 

162 

163 Args: 

164 data: PLXSpotCheckData instance to encode 

165 

166 Returns: 

167 Encoded bytes representing the measurement 

168 

169 """ 

170 # Build flags 

171 flags = data.spot_check_flags 

172 

173 # Build result 

174 result = bytearray([int(flags)]) 

175 result.extend(IEEE11073Parser.encode_sfloat(data.spo2)) 

176 result.extend(IEEE11073Parser.encode_sfloat(data.pulse_rate)) 

177 

178 # Encode optional measurement status 

179 if data.measurement_status is not None: 

180 result.extend(DataParser.encode_int16(int(data.measurement_status), signed=False)) 

181 

182 # Encode optional device and sensor status 

183 if data.device_and_sensor_status is not None: 

184 # Device and sensor status is 3 bytes (24-bit value) 

185 device_status_bytes = DataParser.encode_int32(int(data.device_and_sensor_status), signed=False)[:3] 

186 result.extend(device_status_bytes) 

187 

188 # Encode optional pulse amplitude index 

189 if data.pulse_amplitude_index is not None: 

190 result.extend(IEEE11073Parser.encode_sfloat(data.pulse_amplitude_index)) 

191 

192 return result