Coverage for src / bluetooth_sig / gatt / characteristics / blood_pressure_measurement.py: 93%

1"""Blood Pressure Measurement characteristic implementation.""" 

2 

3from __future__ import annotations 

4 

5from enum import IntFlag 

6 

7import msgspec 

8 

9from bluetooth_sig.types.units import PressureUnit 

10 

11from ..context import CharacteristicContext 

12from .blood_pressure_common import ( 

13 BLOOD_PRESSURE_MAX_KPA, 

14 BLOOD_PRESSURE_MAX_MMHG, 

15 BaseBloodPressureCharacteristic, 

16 BloodPressureFlags, 

17 BloodPressureOptionalFields, 

18) 

19from .utils import IEEE11073Parser 

20 

21 

22class BloodPressureMeasurementStatus(IntFlag): 

23 """Blood Pressure Measurement Status flags as per Bluetooth SIG specification.""" 

24 

25 BODY_MOVEMENT_DETECTED = 0x0001 

26 CUFF_TOO_LOOSE = 0x0002 

27 IRREGULAR_PULSE_DETECTED = 0x0004 

28 PULSE_RATE_OUT_OF_RANGE = 0x0008 

29 IMPROPER_MEASUREMENT_POSITION = 0x0010 

30 

31 

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

33 """Parsed data from Blood Pressure Measurement characteristic.""" 

34 

35 systolic: float 

36 diastolic: float 

37 mean_arterial_pressure: float 

38 unit: PressureUnit 

39 optional_fields: BloodPressureOptionalFields = BloodPressureOptionalFields() 

40 flags: BloodPressureFlags = BloodPressureFlags(0) 

41 

42 def __post_init__(self) -> None: 

43 """Validate blood pressure data.""" 

44 if self.unit not in (PressureUnit.MMHG, PressureUnit.KPA): 

45 raise ValueError(f"Blood pressure unit must be MMHG or KPA, got {self.unit}") 

46 

47 if self.unit == PressureUnit.MMHG: 

48 valid_range = (0, BLOOD_PRESSURE_MAX_MMHG) 

49 else: # kPa 

50 valid_range = (0, BLOOD_PRESSURE_MAX_KPA) 

51 

52 for name, value in [ 

53 ("systolic", self.systolic), 

54 ("diastolic", self.diastolic), 

55 ("mean_arterial_pressure", self.mean_arterial_pressure), 

56 ]: 

57 if not valid_range[0] <= value <= valid_range[1]: 

58 raise ValueError( 

59 f"{name} pressure {value} {self.unit.value} is outside valid range " 

60 f"({valid_range[0]}-{valid_range[1]} {self.unit.value})" 

61 ) 

62 

63 

64class BloodPressureMeasurementCharacteristic(BaseBloodPressureCharacteristic): 

65 """Blood Pressure Measurement characteristic (0x2A35). 

66 

67 Used to transmit blood pressure measurements with systolic, 

68 diastolic and mean arterial pressure. 

69 

70 SIG Specification Pattern: 

71 This characteristic can use Blood Pressure Feature (0x2A49) to interpret 

72 which status flags are supported by the device. 

73 """ 

74 

75 _is_base_class = False # This is a concrete characteristic class 

76 min_length: int = 7 # Flags(1) + Systolic(2) + Diastolic(2) + MAP(2) 

77 allow_variable_length: bool = True # Optional timestamp, pulse rate, user ID, status 

78 

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

80 """Parse blood pressure measurement data according to Bluetooth specification. 

81 

82 Format: Flags(1) + Systolic(2) + Diastolic(2) + MAP(2) + [Timestamp(7)] + 

83 [Pulse Rate(2)] + [User ID(1)] + [Measurement Status(2)]. 

84 All pressure values are IEEE-11073 16-bit SFLOAT. 

85 

86 Args: 

87 data: Raw bytearray from BLE characteristic 

88 ctx: Optional context providing access to Blood Pressure Feature characteristic 

89 for validating which measurement status flags are supported 

90 

91 Returns: 

92 BloodPressureData containing parsed blood pressure data with metadata 

93 

94 SIG Pattern: 

95 When context is available, can validate that measurement status flags are 

96 within the device's supported features as indicated by Blood Pressure Feature. 

97 

98 """ 

99 if len(data) < 7: 

100 raise ValueError("Blood Pressure Measurement data must be at least 7 bytes") 

101 

102 flags = self._parse_blood_pressure_flags(data) 

103 

104 # Parse required fields 

105 systolic = IEEE11073Parser.parse_sfloat(data, 1) 

106 diastolic = IEEE11073Parser.parse_sfloat(data, 3) 

107 mean_arterial_pressure = IEEE11073Parser.parse_sfloat(data, 5) 

108 unit = self._parse_blood_pressure_unit(flags) 

109 

110 # Parse optional fields 

111 timestamp, pulse_rate, user_id, measurement_status = self._parse_optional_fields(data, flags) 

112 

113 # Create immutable struct with all values 

114 return BloodPressureData( # pylint: disable=duplicate-code # Similar structure in intermediate_cuff_pressure (same optional fields by spec) 

115 systolic=systolic, 

116 diastolic=diastolic, 

117 mean_arterial_pressure=mean_arterial_pressure, 

118 unit=unit, 

119 optional_fields=BloodPressureOptionalFields( 

120 timestamp=timestamp, 

121 pulse_rate=pulse_rate, 

122 user_id=user_id, 

123 measurement_status=measurement_status, 

124 ), 

125 flags=flags, 

126 ) 

127 

128 def _encode_value(self, data: BloodPressureData) -> bytearray: 

129 """Encode BloodPressureData back to bytes. 

130 

131 Args: 

132 data: BloodPressureData instance to encode 

133 

134 Returns: 

135 Encoded bytes representing the blood pressure measurement 

136 

137 """ 

138 return self._encode_blood_pressure_base( 

139 data, 

140 data.optional_fields, 

141 [data.systolic, data.diastolic, data.mean_arterial_pressure], 

142 )