Coverage for src / bluetooth_sig / gatt / characteristics / intermediate_cuff_pressure.py: 90%

1"""Intermediate Cuff Pressure characteristic implementation.""" 

2 

3from __future__ import annotations 

4 

5import msgspec 

6 

7from bluetooth_sig.types.units import PressureUnit 

8 

9from ..context import CharacteristicContext 

10from .blood_pressure_common import ( 

11 BLOOD_PRESSURE_MAX_KPA, 

12 BLOOD_PRESSURE_MAX_MMHG, 

13 BaseBloodPressureCharacteristic, 

14 BloodPressureFlags, 

15 BloodPressureOptionalFields, 

16) 

17from .utils import IEEE11073Parser 

18 

19 

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

21 """Parsed data from Intermediate Cuff Pressure characteristic.""" 

22 

23 current_cuff_pressure: float 

24 unit: PressureUnit 

25 optional_fields: BloodPressureOptionalFields = BloodPressureOptionalFields() 

26 flags: BloodPressureFlags = BloodPressureFlags(0) 

27 

28 def __post_init__(self) -> None: 

29 """Validate intermediate cuff pressure data.""" 

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

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

32 

33 valid_range = (0, BLOOD_PRESSURE_MAX_MMHG) if self.unit == PressureUnit.MMHG else (0, BLOOD_PRESSURE_MAX_KPA) 

34 

35 if not valid_range[0] <= self.current_cuff_pressure <= valid_range[1]: 

36 raise ValueError( 

37 f"Current cuff pressure {self.current_cuff_pressure} {self.unit.value} is outside valid range " 

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

39 ) 

40 

41 

42class IntermediateCuffPressureCharacteristic(BaseBloodPressureCharacteristic): 

43 """Intermediate Cuff Pressure characteristic (0x2A36). 

44 

45 Used to transmit intermediate cuff pressure values during a blood 

46 pressure measurement process. 

47 

48 SIG Specification Pattern: 

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

50 which status flags are supported by the device. 

51 """ 

52 

53 _is_base_class = False # This is a concrete characteristic class 

54 min_length: int = 7 # Flags(1) + Current Cuff Pressure(2) + Unused(2) + Unused(2) 

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

56 

57 def _decode_value( 

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

59 ) -> IntermediateCuffPressureData: # pylint: disable=too-many-locals 

60 """Parse intermediate cuff pressure data according to Bluetooth specification. 

61 

62 Format: Flags(1) + Current Cuff Pressure(2) + Unused(2) + Unused(2) + [Timestamp(7)] + 

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

64 All pressure values are IEEE-11073 16-bit SFLOAT. Unused fields are set to NaN. 

65 

66 Args: 

67 data: Raw bytearray from BLE characteristic 

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

69 for validating which measurement status flags are supported 

70 validate: Whether to validate ranges (default True) 

71 

72 Returns: 

73 IntermediateCuffPressureData containing parsed cuff pressure data with metadata 

74 

75 SIG Pattern: 

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

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

78 

79 """ 

80 flags = self._parse_blood_pressure_flags(data) 

81 

82 # Parse required fields 

83 current_cuff_pressure = IEEE11073Parser.parse_sfloat(data, 1) 

84 unit = self._parse_blood_pressure_unit(flags) 

85 

86 # Skip unused fields (bytes 3-6, should be NaN but we don't validate here) 

87 

88 # Parse optional fields 

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

90 

91 # Create immutable struct with all values 

92 return IntermediateCuffPressureData( # pylint: disable=duplicate-code # Similar structure in blood_pressure_measurement (same optional fields by spec) 

93 current_cuff_pressure=current_cuff_pressure, 

94 unit=unit, 

95 optional_fields=BloodPressureOptionalFields( 

96 timestamp=timestamp, 

97 pulse_rate=pulse_rate, 

98 user_id=user_id, 

99 measurement_status=measurement_status, 

100 ), 

101 flags=flags, 

102 ) 

103 

104 def _encode_value(self, data: IntermediateCuffPressureData) -> bytearray: 

105 """Encode IntermediateCuffPressureData back to bytes. 

106 

107 Args: 

108 data: IntermediateCuffPressureData instance to encode 

109 

110 Returns: 

111 Encoded bytes representing the intermediate cuff pressure 

112 

113 """ 

114 # Intermediate cuff pressure only uses current pressure, other fields are NaN 

115 return self._encode_blood_pressure_base( 

116 data, 

117 data.optional_fields, 

118 [data.current_cuff_pressure, float("nan"), float("nan")], 

119 )