Coverage for src/bluetooth_sig/gatt/characteristics/cooking_common.py: 98%

1"""Shared helpers and structs for Cooking/Cookware characteristics.""" 

2 

3from __future__ import annotations 

4 

5from enum import IntFlag 

6 

7import msgspec 

8 

9from ..constants import SIZE_UINT8, SIZE_UINT16 

10from .cooking_temperature import CookingTemperatureCharacteristic 

11from .cooking_zone_perceived_power import CookingZonePerceivedPowerCharacteristic 

12from .humidity import HumidityCharacteristic 

13from .kitchen_appliance_airflow import KitchenApplianceAirflowCharacteristic 

14from .percentage_8 import Percentage8Characteristic 

15from .power import PowerCharacteristic 

16from .utils import DataParser 

17 

18COOKING_TEMPERATURE = CookingTemperatureCharacteristic() 

19COOKING_ZONE_PERCEIVED_POWER = CookingZonePerceivedPowerCharacteristic() 

20HUMIDITY = HumidityCharacteristic() 

21KITCHEN_APPLIANCE_AIRFLOW = KitchenApplianceAirflowCharacteristic() 

22PERCENTAGE_8 = Percentage8Characteristic() 

23POWER = PowerCharacteristic() 

24 

25 

26class CookingConditionsFlags(IntFlag): 

27 """Bit flags for Cooking Conditions payload presence.""" 

28 

29 POWER_LEVEL_PRESENT = 1 << 0 

30 TEMPERATURE_PRESENT = 1 << 1 

31 HUMIDITY_PRESENT = 1 << 2 

32 BLOWER_FAN_SPEED_PRESENT = 1 << 3 

33 MANUFACTURER_DATA_PRESENT = 1 << 4 

34 

35 

36class CookingConditionsData(msgspec.Struct, frozen=True, kw_only=True): 

37 """Shared parsed data model for Cooking Conditions characteristics.""" 

38 

39 flags: CookingConditionsFlags 

40 power_level: float | None = None 

41 temperature: float | None = None 

42 humidity: float | None = None 

43 blower_fan_speed: float | None = None 

44 manufacturer_specific_data: bytes | None = None 

45 

46 

47def parse_cooking_conditions(data: bytearray) -> CookingConditionsData: 

48 """Parse a Cooking Conditions-compatible byte payload.""" 

49 flags = CookingConditionsFlags(DataParser.parse_int16(data, 0, signed=False)) 

50 validate_flags(flags, CookingConditionsFlags, "Cooking Conditions Flags") 

51 offset = 2 

52 

53 power_level = None 

54 if flags & CookingConditionsFlags.POWER_LEVEL_PRESENT: 

55 power_level = COOKING_ZONE_PERCEIVED_POWER.parse_value(bytearray(data[offset : offset + SIZE_UINT16])) 

56 offset += SIZE_UINT16 

57 

58 temperature = None 

59 if flags & CookingConditionsFlags.TEMPERATURE_PRESENT: 

60 temperature = COOKING_TEMPERATURE.parse_value(bytearray(data[offset : offset + SIZE_UINT16])) 

61 offset += SIZE_UINT16 

62 

63 humidity = None 

64 if flags & CookingConditionsFlags.HUMIDITY_PRESENT: 

65 humidity = HUMIDITY.parse_value(bytearray(data[offset : offset + SIZE_UINT16])) 

66 offset += SIZE_UINT16 

67 

68 blower_fan_speed = None 

69 if flags & CookingConditionsFlags.BLOWER_FAN_SPEED_PRESENT: 

70 blower_fan_speed = PERCENTAGE_8.parse_value(bytearray(data[offset : offset + SIZE_UINT8])) 

71 offset += SIZE_UINT8 

72 

73 manufacturer_data = None 

74 if flags & CookingConditionsFlags.MANUFACTURER_DATA_PRESENT: 

75 manufacturer_data = bytes(data[offset:]) 

76 

77 return CookingConditionsData( 

78 flags=flags, 

79 power_level=power_level, 

80 temperature=temperature, 

81 humidity=humidity, 

82 blower_fan_speed=blower_fan_speed, 

83 manufacturer_specific_data=manufacturer_data, 

84 ) 

85 

86 

87def encode_cooking_conditions(value: CookingConditionsData) -> bytearray: 

88 """Encode shared Cooking Conditions data model into bytes.""" 

89 validate_flags(value.flags, CookingConditionsFlags, "Cooking Conditions Flags") 

90 result = bytearray() 

91 result.extend(DataParser.encode_int16(int(value.flags), signed=False)) 

92 

93 if value.flags & CookingConditionsFlags.POWER_LEVEL_PRESENT: 

94 if value.power_level is None: 

95 raise ValueError("power_level is required when POWER_LEVEL_PRESENT is set") 

96 result.extend(COOKING_ZONE_PERCEIVED_POWER.build_value(value.power_level)) 

97 

98 if value.flags & CookingConditionsFlags.TEMPERATURE_PRESENT: 

99 if value.temperature is None: 

100 raise ValueError("temperature is required when TEMPERATURE_PRESENT is set") 

101 result.extend(COOKING_TEMPERATURE.build_value(value.temperature)) 

102 

103 if value.flags & CookingConditionsFlags.HUMIDITY_PRESENT: 

104 if value.humidity is None: 

105 raise ValueError("humidity is required when HUMIDITY_PRESENT is set") 

106 result.extend(HUMIDITY.build_value(value.humidity)) 

107 

108 if value.flags & CookingConditionsFlags.BLOWER_FAN_SPEED_PRESENT: 

109 if value.blower_fan_speed is None: 

110 raise ValueError("blower_fan_speed is required when BLOWER_FAN_SPEED_PRESENT is set") 

111 result.extend(PERCENTAGE_8.build_value(value.blower_fan_speed)) 

112 

113 if value.flags & CookingConditionsFlags.MANUFACTURER_DATA_PRESENT: 

114 if value.manufacturer_specific_data is None: 

115 raise ValueError("manufacturer_specific_data is required when MANUFACTURER_DATA_PRESENT is set") 

116 result.extend(value.manufacturer_specific_data) 

117 

118 return result 

119 

120 

121def validate_flags(flags: IntFlag, flag_type: type[IntFlag], field_name: str) -> None: 

122 """Reject RFU bits for manually parsed IntFlag fields.""" 

123 valid_mask = 0 

124 for member in flag_type: 

125 valid_mask |= int(member) 

126 if int(flags) & ~valid_mask: 

127 raise ValueError(f"{field_name} contains reserved bits")