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

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 valid_range = (0, BLOOD_PRESSURE_MAX_MMHG) if self.unit == PressureUnit.MMHG else (0, BLOOD_PRESSURE_MAX_KPA) 

48 

49 for name, value in [ 

50 ("systolic", self.systolic), 

51 ("diastolic", self.diastolic), 

52 ("mean_arterial_pressure", self.mean_arterial_pressure), 

53 ]: 

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

55 raise ValueError( 

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

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

58 ) 

59 

60 

61class BloodPressureMeasurementCharacteristic(BaseBloodPressureCharacteristic): 

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

63 

64 Used to transmit blood pressure measurements with systolic, 

65 diastolic and mean arterial pressure. 

66 

67 SIG Specification Pattern: 

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

69 which status flags are supported by the device. 

70 """ 

71 

72 _is_base_class = False # This is a concrete characteristic class 

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

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

75 

76 def _decode_value( 

77 self, data: bytearray, ctx: CharacteristicContext | None = None, *, validate: bool = True 

78 ) -> BloodPressureData: # pylint: disable=too-many-locals 

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

80 

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

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

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

84 

85 Args: 

86 data: Raw bytearray from BLE characteristic 

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

88 validate: Whether to validate ranges (default True) 

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 flags = self._parse_blood_pressure_flags(data) 

100 

101 # Parse required fields 

102 systolic = IEEE11073Parser.parse_sfloat(data, 1) 

103 diastolic = IEEE11073Parser.parse_sfloat(data, 3) 

104 mean_arterial_pressure = IEEE11073Parser.parse_sfloat(data, 5) 

105 unit = self._parse_blood_pressure_unit(flags) 

106 

107 # Parse optional fields 

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

109 

110 # Create immutable struct with all values 

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

112 systolic=systolic, 

113 diastolic=diastolic, 

114 mean_arterial_pressure=mean_arterial_pressure, 

115 unit=unit, 

116 optional_fields=BloodPressureOptionalFields( 

117 timestamp=timestamp, 

118 pulse_rate=pulse_rate, 

119 user_id=user_id, 

120 measurement_status=measurement_status, 

121 ), 

122 flags=flags, 

123 ) 

124 

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

126 """Encode BloodPressureData back to bytes. 

127 

128 Args: 

129 data: BloodPressureData instance to encode 

130 

131 Returns: 

132 Encoded bytes representing the blood pressure measurement 

133 

134 """ 

135 return self._encode_blood_pressure_base( 

136 data, 

137 data.optional_fields, 

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

139 )