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

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 if self.unit == PressureUnit.MMHG: 

34 valid_range = (0, BLOOD_PRESSURE_MAX_MMHG) 

35 else: # kPa 

36 valid_range = (0, BLOOD_PRESSURE_MAX_KPA) 

37 

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

39 raise ValueError( 

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

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

42 ) 

43 

44 

45class IntermediateCuffPressureCharacteristic(BaseBloodPressureCharacteristic): 

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

47 

48 Used to transmit intermediate cuff pressure values during a blood 

49 pressure measurement process. 

50 

51 SIG Specification Pattern: 

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

53 which status flags are supported by the device. 

54 """ 

55 

56 _is_base_class = False # This is a concrete characteristic class 

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

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

59 

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

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

62 

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

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

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

66 

67 Args: 

68 data: Raw bytearray from BLE characteristic 

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

70 for validating which measurement status flags are supported 

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 if len(data) < 7: 

81 raise ValueError("Intermediate Cuff Pressure data must be at least 7 bytes") 

82 

83 flags = self._parse_blood_pressure_flags(data) 

84 

85 # Parse required fields 

86 current_cuff_pressure = IEEE11073Parser.parse_sfloat(data, 1) 

87 unit = self._parse_blood_pressure_unit(flags) 

88 

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

90 

91 # Parse optional fields 

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

93 

94 # Create immutable struct with all values 

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

96 current_cuff_pressure=current_cuff_pressure, 

97 unit=unit, 

98 optional_fields=BloodPressureOptionalFields( 

99 timestamp=timestamp, 

100 pulse_rate=pulse_rate, 

101 user_id=user_id, 

102 measurement_status=measurement_status, 

103 ), 

104 flags=flags, 

105 ) 

106 

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

108 """Encode IntermediateCuffPressureData back to bytes. 

109 

110 Args: 

111 data: IntermediateCuffPressureData instance to encode 

112 

113 Returns: 

114 Encoded bytes representing the intermediate cuff pressure 

115 

116 """ 

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

118 return self._encode_blood_pressure_base( 

119 data, 

120 data.optional_fields, 

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

122 )