Coverage for src/ramses_rf/device/__init__.py: 36%

1#!/usr/bin/env python3 

2"""RAMSES RF - Heating devices (e.g. CTL, OTB, BDR, TRV).""" 

3 

4from __future__ import annotations 

5 

6import logging 

7from typing import TYPE_CHECKING, Any 

8 

9from ramses_rf.const import DEV_TYPE_MAP 

10from ramses_tx.const import DevType 

11from ramses_tx.schemas import SZ_CLASS, SZ_FAKED 

12 

13from .base import ( # noqa: F401, isort: skip, pylint: disable=unused-import 

14 BASE_CLASS_BY_SLUG as _BASE_CLASS_BY_SLUG, 

15 Device, 

16 Fakeable, 

17 DeviceHeat, 

18 HgiGateway, 

19 DeviceHvac, 

20) 

21 

22 

23from .heat import ( # noqa: F401, isort: skip, pylint: disable=unused-import 

24 HEAT_CLASS_BY_SLUG as _HEAT_CLASS_BY_SLUG, 

25 BdrSwitch, 

26 Controller, 

27 DhwSensor, 

28 OtbGateway, 

29 OutSensor, 

30 Temperature, 

31 Thermostat, 

32 TrvActuator, 

33 UfhCircuit, 

34 UfhController, 

35 class_dev_heat, 

36) 

37 

38 

39from .hvac import ( # noqa: F401, isort: skip, pylint: disable=unused-import 

40 HVAC_CLASS_BY_SLUG as _HVAC_CLASS_BY_SLUG, 

41 HvacCarbonDioxideSensor, 

42 HvacHumiditySensor, 

43 HvacRemote, 

44 HvacVentilator, 

45 RfsGateway, 

46 class_dev_hvac, 

47) 

48 

49if TYPE_CHECKING: 

50 from ramses_rf import Gateway 

51 from ramses_tx import Address, Message 

52 

53 

54__all__ = [ 

55 # .base 

56 "Device", 

57 "Fakeable", 

58 "DeviceHeat", 

59 "HgiGateway", 

60 "DeviceHvac", 

61 # .heat 

62 "BdrSwitch", 

63 "Controller", 

64 "DhwSensor", 

65 "OtbGateway", 

66 "OutSensor", 

67 "Temperature", 

68 "Thermostat", 

69 "TrvActuator", 

70 "UfhCircuit", 

71 "UfhController", 

72 "class_dev_heat", 

73 # .hvac 

74 "HvacCarbonDioxideSensor", 

75 "HvacHumiditySensor", 

76 "HvacRemote", 

77 "HvacVentilator", 

78 "RfsGateway", 

79 "class_dev_hvac", 

80 # 

81 "best_dev_role", 

82 "device_factory", 

83] 

84 

85_LOGGER = logging.getLogger(__name__) 

86 

87 

88_CLASS_BY_SLUG = _BASE_CLASS_BY_SLUG | _HEAT_CLASS_BY_SLUG | _HVAC_CLASS_BY_SLUG 

89 

90HEAT_DEV_CLASS_BY_SLUG = { 

91 k: v for k, v in _HEAT_CLASS_BY_SLUG.items() if k is not DevType.HEA 

92} 

93HVAC_DEV_CLASS_BY_SLUG = { 

94 k: v for k, v in _HVAC_CLASS_BY_SLUG.items() if k is not DevType.HVC 

95} 

96 

97 

98def best_dev_role( 

99 dev_addr: Address, 

100 *, 

101 msg: Message | None = None, 

102 eavesdrop: bool = False, 

103 **schema: Any, 

104) -> type[Device]: 

105 """Return the best device role (object class) for a given device id/msg/schema. 

106 

107 Heat (CH/DHW) devices can reliably be determined by their address type (e.g. '04:'). 

108 Any device without a known Heat type is considered a HVAC device. 

109 

110 HVAC devices must be explicitly typed, or fingerprinted/eavesdropped. 

111 The generic HVAC class can be promoted later on, when more information is available. 

112 """ 

113 

114 cls: type[Device] 

115 slug: str 

116 

117 try: # convert (say) 'dhw_sensor' to DHW 

118 slug = DEV_TYPE_MAP.slug(schema.get(SZ_CLASS)) # type: ignore[arg-type] 

119 except KeyError: 

120 slug = schema.get(SZ_CLASS) 

121 

122 # a specified device class always takes precedence (even if it is wrong)... 

123 if slug in _CLASS_BY_SLUG: 

124 cls = _CLASS_BY_SLUG[slug] 

125 _LOGGER.debug( 

126 f"Using an explicitly-defined class for: {dev_addr!r} ({cls._SLUG})" 

127 ) 

128 return cls 

129 

130 if dev_addr.type == DEV_TYPE_MAP.HGI: 

131 _LOGGER.debug(f"Using the default class for: {dev_addr!r} ({HgiGateway._SLUG})") 

132 return HgiGateway 

133 

134 try: # or, is it a well-known CH/DHW class, derived from the device type... 

135 if cls := class_dev_heat(dev_addr, msg=msg, eavesdrop=eavesdrop): 

136 _LOGGER.debug( 

137 f"Using the default Heat class for: {dev_addr!r} ({cls._SLUG})" 

138 ) 

139 return cls 

140 except TypeError: 

141 pass 

142 

143 try: # or, a HVAC class, eavesdropped from the message code/payload... 

144 if cls := class_dev_hvac(dev_addr, msg=msg, eavesdrop=eavesdrop): 

145 _LOGGER.debug( 

146 f"Using eavesdropped HVAC class for: {dev_addr!r} ({cls._SLUG})" 

147 ) 

148 return cls # includes DeviceHvac 

149 except TypeError: 

150 pass 

151 

152 # otherwise, use the default device class... 

153 _LOGGER.debug( 

154 f"Using a promotable HVAC class for: {dev_addr!r} ({DeviceHvac._SLUG})" 

155 ) 

156 return DeviceHvac 

157 

158 

159def device_factory( 

160 gwy: Gateway, dev_addr: Address, *, msg: Message | None = None, **traits: Any 

161) -> Device: 

162 """Return the initial device class for a given device id/msg/traits. 

163 

164 Devices of certain classes are promotable to a compatible sub class. 

165 """ 

166 

167 cls: type[Device] = best_dev_role( 

168 dev_addr, 

169 msg=msg, 

170 eavesdrop=gwy.config.enable_eavesdrop, 

171 **traits, 

172 ) 

173 

174 if ( 

175 isinstance(cls, DeviceHvac) 

176 and traits.get(SZ_CLASS) in (DEV_TYPE_MAP.HVC, None) 

177 and traits.get(SZ_FAKED) 

178 ): 

179 raise TypeError( 

180 "Faked devices from the HVAC domain must have an explicit class: {dev_addr}" 

181 ) 

182 

183 return cls.create_from_schema(gwy, dev_addr, **traits)