Coverage for src / bluetooth_sig / gatt / characteristics / enhanced_blood_pressure_measurement.py: 100%

1"""Enhanced Blood Pressure Measurement characteristic implementation. 

2 

3Implements the Enhanced Blood Pressure Measurement characteristic (0x2B34). 

4Extends the regular Blood Pressure Measurement with uint32 timestamps 

5(seconds since epoch), a User Facing Time field, and an Epoch Start flag. 

6 

7Flag-bit assignments (from GSS YAML): 

8 Bit 0: Units (0=mmHg, 1=kPa) 

9 Bit 1: Time Stamp present (uint32 seconds since epoch) 

10 Bit 2: Pulse Rate present (medfloat16) 

11 Bit 3: User ID present (uint8) 

12 Bit 4: Measurement Status present (boolean[16]) 

13 Bit 5: User Facing Time present (uint32 seconds since epoch) 

14 Bit 6: Epoch Start 2000 (0=1900, 1=2000) 

15 Bit 7: Reserved 

16 

17References: 

18 Bluetooth SIG Blood Pressure Service 1.1 

19 org.bluetooth.characteristic.enhanced_blood_pressure_measurement (GSS YAML) 

20""" 

21 

22from __future__ import annotations 

23 

24from enum import IntEnum, IntFlag 

25 

26import msgspec 

27 

28from bluetooth_sig.types.units import PressureUnit 

29 

30from ..context import CharacteristicContext 

31from .base import BaseCharacteristic 

32from .blood_pressure_measurement import BloodPressureMeasurementStatus 

33from .utils import DataParser, IEEE11073Parser 

34 

35 

36class EpochYear(IntEnum): 

37 """Epoch start year for Enhanced Blood Pressure timestamps.""" 

38 

39 EPOCH_1900 = 1900 

40 EPOCH_2000 = 2000 

41 

42 

43class EnhancedBloodPressureFlags(IntFlag): 

44 """Enhanced Blood Pressure Measurement flags.""" 

45 

46 UNITS_KPA = 0x01 

47 TIMESTAMP_PRESENT = 0x02 

48 PULSE_RATE_PRESENT = 0x04 

49 USER_ID_PRESENT = 0x08 

50 MEASUREMENT_STATUS_PRESENT = 0x10 

51 USER_FACING_TIME_PRESENT = 0x20 

52 EPOCH_START_2000 = 0x40 

53 

54 

55class EnhancedBloodPressureData(msgspec.Struct, frozen=True, kw_only=True): 

56 """Parsed data from Enhanced Blood Pressure Measurement characteristic. 

57 

58 Attributes: 

59 flags: Raw 8-bit flags field. 

60 systolic: Systolic pressure value. 

61 diastolic: Diastolic pressure value. 

62 mean_arterial_pressure: Mean arterial pressure value. 

63 unit: Pressure unit (mmHg or kPa). 

64 timestamp: Seconds since epoch start. None if absent. 

65 pulse_rate: Pulse rate in BPM. None if absent. 

66 user_id: User ID (0-255). None if absent. 

67 measurement_status: 16-bit measurement status flags. None if absent. 

68 user_facing_time: User-facing time in seconds since epoch. None if absent. 

69 epoch_year: Epoch start year (1900 or 2000). 

70 

71 """ 

72 

73 flags: EnhancedBloodPressureFlags 

74 systolic: float 

75 diastolic: float 

76 mean_arterial_pressure: float 

77 unit: PressureUnit 

78 timestamp: int | None = None 

79 pulse_rate: float | None = None 

80 user_id: int | None = None 

81 measurement_status: BloodPressureMeasurementStatus | None = None 

82 user_facing_time: int | None = None 

83 epoch_year: EpochYear = EpochYear.EPOCH_1900 

84 

85 

86class EnhancedBloodPressureMeasurementCharacteristic( 

87 BaseCharacteristic[EnhancedBloodPressureData], 

88): 

89 """Enhanced Blood Pressure Measurement characteristic (0x2B34). 

90 

91 Enhanced variant of Blood Pressure Measurement with uint32 timestamps 

92 (seconds since epoch) instead of 7-byte DateTime, plus a new User Facing 

93 Time field and Epoch Start 2000 flag. 

94 """ 

95 

96 expected_type = EnhancedBloodPressureData 

97 min_length: int = 7 # flags(1) + compound value(6) 

98 allow_variable_length: bool = True 

99 

100 def _decode_value( 

101 self, 

102 data: bytearray, 

103 ctx: CharacteristicContext | None = None, 

104 *, 

105 validate: bool = True, 

106 ) -> EnhancedBloodPressureData: 

107 """Parse Enhanced Blood Pressure Measurement from raw BLE bytes. 

108 

109 Args: 

110 data: Raw bytearray from BLE characteristic. 

111 ctx: Optional context (unused). 

112 validate: Whether to validate ranges. 

113 

114 Returns: 

115 EnhancedBloodPressureData with all present fields populated. 

116 

117 """ 

118 flags = EnhancedBloodPressureFlags(data[0]) 

119 unit = PressureUnit.KPA if flags & EnhancedBloodPressureFlags.UNITS_KPA else PressureUnit.MMHG 

120 

121 # Compound pressure value: 3 x medfloat16 (6 bytes) 

122 systolic = IEEE11073Parser.parse_sfloat(data, 1) 

123 diastolic = IEEE11073Parser.parse_sfloat(data, 3) 

124 mean_arterial_pressure = IEEE11073Parser.parse_sfloat(data, 5) 

125 offset = 7 

126 

127 epoch_year = ( 

128 EpochYear.EPOCH_2000 if flags & EnhancedBloodPressureFlags.EPOCH_START_2000 else EpochYear.EPOCH_1900 

129 ) 

130 

131 timestamp: int | None = None 

132 if flags & EnhancedBloodPressureFlags.TIMESTAMP_PRESENT: 

133 timestamp = DataParser.parse_int32(data, offset, signed=False) 

134 offset += 4 

135 

136 pulse_rate: float | None = None 

137 if flags & EnhancedBloodPressureFlags.PULSE_RATE_PRESENT: 

138 pulse_rate = IEEE11073Parser.parse_sfloat(data, offset) 

139 offset += 2 

140 

141 user_id: int | None = None 

142 if flags & EnhancedBloodPressureFlags.USER_ID_PRESENT: 

143 user_id = data[offset] 

144 offset += 1 

145 

146 measurement_status: BloodPressureMeasurementStatus | None = None 

147 if flags & EnhancedBloodPressureFlags.MEASUREMENT_STATUS_PRESENT: 

148 measurement_status = BloodPressureMeasurementStatus(DataParser.parse_int16(data, offset, signed=False)) 

149 offset += 2 

150 

151 user_facing_time: int | None = None 

152 if flags & EnhancedBloodPressureFlags.USER_FACING_TIME_PRESENT: 

153 user_facing_time = DataParser.parse_int32(data, offset, signed=False) 

154 offset += 4 

155 

156 return EnhancedBloodPressureData( 

157 flags=flags, 

158 systolic=systolic, 

159 diastolic=diastolic, 

160 mean_arterial_pressure=mean_arterial_pressure, 

161 unit=unit, 

162 timestamp=timestamp, 

163 pulse_rate=pulse_rate, 

164 user_id=user_id, 

165 measurement_status=measurement_status, 

166 user_facing_time=user_facing_time, 

167 epoch_year=epoch_year, 

168 ) 

169 

170 def _encode_value(self, data: EnhancedBloodPressureData) -> bytearray: 

171 """Encode EnhancedBloodPressureData back to BLE bytes. 

172 

173 Args: 

174 data: EnhancedBloodPressureData instance. 

175 

176 Returns: 

177 Encoded bytearray matching the BLE wire format. 

178 

179 """ 

180 flags = EnhancedBloodPressureFlags(0) 

181 if data.unit == PressureUnit.KPA: 

182 flags |= EnhancedBloodPressureFlags.UNITS_KPA 

183 if data.timestamp is not None: 

184 flags |= EnhancedBloodPressureFlags.TIMESTAMP_PRESENT 

185 if data.pulse_rate is not None: 

186 flags |= EnhancedBloodPressureFlags.PULSE_RATE_PRESENT 

187 if data.user_id is not None: 

188 flags |= EnhancedBloodPressureFlags.USER_ID_PRESENT 

189 if data.measurement_status is not None: 

190 flags |= EnhancedBloodPressureFlags.MEASUREMENT_STATUS_PRESENT 

191 if data.user_facing_time is not None: 

192 flags |= EnhancedBloodPressureFlags.USER_FACING_TIME_PRESENT 

193 if data.epoch_year == EpochYear.EPOCH_2000: 

194 flags |= EnhancedBloodPressureFlags.EPOCH_START_2000 

195 

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

197 result.extend(IEEE11073Parser.encode_sfloat(data.systolic)) 

198 result.extend(IEEE11073Parser.encode_sfloat(data.diastolic)) 

199 result.extend(IEEE11073Parser.encode_sfloat(data.mean_arterial_pressure)) 

200 

201 if data.timestamp is not None: 

202 result.extend(DataParser.encode_int32(data.timestamp, signed=False)) 

203 

204 if data.pulse_rate is not None: 

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

206 

207 if data.user_id is not None: 

208 result.append(data.user_id) 

209 

210 if data.measurement_status is not None: 

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

212 

213 if data.user_facing_time is not None: 

214 result.extend(DataParser.encode_int32(data.user_facing_time, signed=False)) 

215 

216 return result