1 """Data template classes for discovery used to generate additional data for setup."""
3 from __future__
import annotations
5 from collections.abc
import Iterable, Mapping
6 from dataclasses
import dataclass, field
9 from typing
import Any, cast
11 from zwave_js_server.const
import CommandClass
12 from zwave_js_server.const.command_class.energy_production
import (
13 EnergyProductionParameter,
14 EnergyProductionScaleType,
16 TodaysProductionScale,
20 from zwave_js_server.const.command_class.meter
import (
22 ENERGY_TOTAL_INCREASING_METER_TYPES,
23 POWER_FACTOR_METER_TYPES,
25 UNIT_AMPERE
as METER_UNIT_AMPERE,
27 UNIT_CUBIC_METER
as METER_UNIT_CUBIC_METER,
30 UNIT_VOLT
as METER_UNIT_VOLT,
31 UNIT_WATT
as METER_UNIT_WATT,
36 from zwave_js_server.const.command_class.multilevel_sensor
import (
40 ENERGY_MEASUREMENT_SENSORS,
45 SIGNAL_STRENGTH_SENSORS,
47 UNIT_A_WEIGHTED_DECIBELS,
48 UNIT_AMPERE
as SENSOR_UNIT_AMPERE,
52 UNIT_CUBIC_FEET_PER_MINUTE,
53 UNIT_CUBIC_METER
as SENSOR_UNIT_CUBIC_METER,
54 UNIT_CUBIC_METER_PER_HOUR,
62 UNIT_INCHES_OF_MERCURY,
71 UNIT_MICROGRAM_PER_CUBIC_METER,
77 UNIT_PERCENTAGE_VALUE,
78 UNIT_POUND_PER_SQUARE_INCH,
85 UNIT_VOLT
as SENSOR_UNIT_VOLT,
86 UNIT_WATT
as SENSOR_UNIT_WATT,
87 UNIT_WATT_PER_SQUARE_METER,
89 MultilevelSensorScaleType,
92 from zwave_js_server.exceptions
import UnknownValueData
93 from zwave_js_server.model.node
import Node
as ZwaveNode
94 from zwave_js_server.model.value
import (
95 ConfigurationValue
as ZwaveConfigurationValue,
99 from zwave_js_server.util.command_class.energy_production
import (
100 get_energy_production_parameter,
101 get_energy_production_scale_type,
103 from zwave_js_server.util.command_class.meter
import get_meter_scale_type
104 from zwave_js_server.util.command_class.multilevel_sensor
import (
105 get_multilevel_sensor_scale_type,
106 get_multilevel_sensor_type,
110 CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
111 CONCENTRATION_PARTS_PER_MILLION,
115 SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
117 UnitOfElectricCurrent,
118 UnitOfElectricPotential,
131 UnitOfVolumeFlowRate,
132 UnitOfVolumetricFlux,
136 ENTITY_DESC_KEY_BATTERY,
139 ENTITY_DESC_KEY_CURRENT,
140 ENTITY_DESC_KEY_ENERGY_MEASUREMENT,
141 ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
142 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
143 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
144 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
145 ENTITY_DESC_KEY_ENERGY_TOTAL_INCREASING,
146 ENTITY_DESC_KEY_HUMIDITY,
147 ENTITY_DESC_KEY_ILLUMINANCE,
148 ENTITY_DESC_KEY_MEASUREMENT,
149 ENTITY_DESC_KEY_POWER,
150 ENTITY_DESC_KEY_POWER_FACTOR,
151 ENTITY_DESC_KEY_PRESSURE,
152 ENTITY_DESC_KEY_SIGNAL_STRENGTH,
153 ENTITY_DESC_KEY_TARGET_TEMPERATURE,
154 ENTITY_DESC_KEY_TEMPERATURE,
155 ENTITY_DESC_KEY_TOTAL_INCREASING,
156 ENTITY_DESC_KEY_UV_INDEX,
157 ENTITY_DESC_KEY_VOLTAGE,
159 from .helpers
import ZwaveValueID
161 ENERGY_PRODUCTION_DEVICE_CLASS_MAP: dict[str, list[EnergyProductionParameter]] = {
162 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME: [EnergyProductionParameter.TOTAL_TIME],
163 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY: [
164 EnergyProductionParameter.TODAYS_PRODUCTION
166 ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL: [
167 EnergyProductionParameter.TOTAL_PRODUCTION
169 ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER: [EnergyProductionParameter.POWER],
173 METER_DEVICE_CLASS_MAP: dict[str, list[MeterScaleType]] = {
174 ENTITY_DESC_KEY_CURRENT: CURRENT_METER_TYPES,
175 ENTITY_DESC_KEY_VOLTAGE: VOLTAGE_METER_TYPES,
176 ENTITY_DESC_KEY_ENERGY_TOTAL_INCREASING: ENERGY_TOTAL_INCREASING_METER_TYPES,
177 ENTITY_DESC_KEY_POWER: POWER_METER_TYPES,
178 ENTITY_DESC_KEY_POWER_FACTOR: POWER_FACTOR_METER_TYPES,
181 MULTILEVEL_SENSOR_DEVICE_CLASS_MAP: dict[str, list[MultilevelSensorType]] = {
182 ENTITY_DESC_KEY_CO: CO_SENSORS,
183 ENTITY_DESC_KEY_CO2: CO2_SENSORS,
184 ENTITY_DESC_KEY_CURRENT: CURRENT_SENSORS,
185 ENTITY_DESC_KEY_ENERGY_MEASUREMENT: ENERGY_MEASUREMENT_SENSORS,
186 ENTITY_DESC_KEY_HUMIDITY: HUMIDITY_SENSORS,
187 ENTITY_DESC_KEY_ILLUMINANCE: ILLUMINANCE_SENSORS,
188 ENTITY_DESC_KEY_POWER: POWER_SENSORS,
189 ENTITY_DESC_KEY_PRESSURE: PRESSURE_SENSORS,
190 ENTITY_DESC_KEY_SIGNAL_STRENGTH: SIGNAL_STRENGTH_SENSORS,
191 ENTITY_DESC_KEY_TEMPERATURE: TEMPERATURE_SENSORS,
192 ENTITY_DESC_KEY_VOLTAGE: VOLTAGE_SENSORS,
193 ENTITY_DESC_KEY_UV_INDEX: [MultilevelSensorType.ULTRAVIOLET],
196 ENERGY_PRODUCTION_UNIT_MAP: dict[str, list[EnergyProductionScaleType]] = {
197 UnitOfEnergy.WATT_HOUR: [
198 TotalProductionScale.WATT_HOURS,
199 TodaysProductionScale.WATT_HOURS,
201 UnitOfPower.WATT: [PowerScale.WATTS],
202 UnitOfTime.SECONDS: [TotalTimeScale.SECONDS],
203 UnitOfTime.HOURS: [TotalTimeScale.HOURS],
206 METER_UNIT_MAP: dict[str, list[MeterScaleType]] = {
207 UnitOfElectricCurrent.AMPERE: METER_UNIT_AMPERE,
208 UnitOfVolume.CUBIC_FEET: UNIT_CUBIC_FEET,
209 UnitOfVolume.CUBIC_METERS: METER_UNIT_CUBIC_METER,
210 UnitOfVolume.GALLONS: UNIT_US_GALLON,
211 UnitOfEnergy.KILO_WATT_HOUR: UNIT_KILOWATT_HOUR,
212 UnitOfElectricPotential.VOLT: METER_UNIT_VOLT,
213 UnitOfPower.WATT: METER_UNIT_WATT,
216 MULTILEVEL_SENSOR_UNIT_MAP: dict[str, list[MultilevelSensorScaleType]] = {
217 UnitOfElectricCurrent.AMPERE: SENSOR_UNIT_AMPERE,
218 UnitOfPower.BTU_PER_HOUR: UNIT_BTU_H,
219 UnitOfTemperature.CELSIUS: UNIT_CELSIUS,
220 UnitOfLength.CENTIMETERS: UNIT_CENTIMETER,
221 UnitOfVolumeFlowRate.CUBIC_FEET_PER_MINUTE: UNIT_CUBIC_FEET_PER_MINUTE,
222 UnitOfVolume.CUBIC_METERS: SENSOR_UNIT_CUBIC_METER,
223 UnitOfVolumeFlowRate.CUBIC_METERS_PER_HOUR: UNIT_CUBIC_METER_PER_HOUR,
224 UnitOfSoundPressure.DECIBEL: UNIT_DECIBEL,
225 UnitOfSoundPressure.WEIGHTED_DECIBEL_A: UNIT_A_WEIGHTED_DECIBELS,
226 DEGREE: UNIT_DEGREES,
227 CONCENTRATION_MICROGRAMS_PER_CUBIC_METER: [
229 *UNIT_MICROGRAM_PER_CUBIC_METER,
231 UnitOfTemperature.FAHRENHEIT: UNIT_FAHRENHEIT,
232 UnitOfLength.FEET: UNIT_FEET,
233 UnitOfVolume.GALLONS: UNIT_GALLONS,
234 UnitOfFrequency.HERTZ: UNIT_HERTZ,
235 UnitOfPressure.INHG: UNIT_INCHES_OF_MERCURY,
236 UnitOfPressure.KPA: UNIT_KILOPASCAL,
237 UnitOfVolumetricFlux.INCHES_PER_HOUR: UNIT_INCHES_PER_HOUR,
238 UnitOfMass.KILOGRAMS: UNIT_KILOGRAM,
239 UnitOfFrequency.KILOHERTZ: UNIT_KILOHERTZ,
240 UnitOfVolume.LITERS: UNIT_LITER,
242 UnitOfLength.METERS: UNIT_METER,
243 UnitOfElectricCurrent.MILLIAMPERE: UNIT_MILLIAMPERE,
244 UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR: UNIT_MILLIMETER_HOUR,
245 UnitOfElectricPotential.MILLIVOLT: UNIT_MILLIVOLT,
246 UnitOfSpeed.MILES_PER_HOUR: UNIT_MPH,
247 UnitOfSpeed.METERS_PER_SECOND: UNIT_M_S,
248 CONCENTRATION_PARTS_PER_MILLION: UNIT_PARTS_MILLION,
249 PERCENTAGE: [*UNIT_PERCENTAGE_VALUE, *UNIT_RSSI],
250 UnitOfMass.POUNDS: UNIT_POUNDS,
251 UnitOfPressure.PSI: UNIT_POUND_PER_SQUARE_INCH,
252 SIGNAL_STRENGTH_DECIBELS_MILLIWATT: UNIT_POWER_LEVEL,
253 UnitOfTime.SECONDS: UNIT_SECOND,
254 UnitOfPressure.MMHG: UNIT_SYSTOLIC,
255 UnitOfElectricPotential.VOLT: SENSOR_UNIT_VOLT,
256 UnitOfPower.WATT: SENSOR_UNIT_WATT,
257 UnitOfIrradiance.WATTS_PER_SQUARE_METER: UNIT_WATT_PER_SQUARE_METER,
258 UV_INDEX: UNIT_UV_INDEX,
261 _LOGGER = logging.getLogger(__name__)
266 """Base class for discovery schema data templates."""
268 static_data: Any |
None =
None
271 """Resolve helper class data for a discovered value.
273 Can optionally be implemented by subclasses if input data needs to be
274 transformed once discovered Value is available.
279 """Return list of all ZwaveValues resolved by helper that should be watched.
281 Should be implemented by subclasses only if there are values to watch.
286 """Return list of all Value IDs resolved by helper that should be watched.
288 Not to be overwritten by subclasses.
290 return {val.value_id
for val
in self.
values_to_watchvalues_to_watch(resolved_data)
if val}
294 node: ZwaveNode, value_id_obj: ZwaveValueID
295 ) -> ZwaveValue | ZwaveConfigurationValue |
None:
296 """Get a ZwaveValue from a node using a ZwaveValueDict."""
297 value_id = get_value_id_str(
299 value_id_obj.command_class,
300 value_id_obj.property_,
301 endpoint=value_id_obj.endpoint,
302 property_key=value_id_obj.property_key,
304 return node.values.get(value_id)
309 """Data template class for Z-Wave JS Climate entities with dynamic current temps."""
311 lookup_table: dict[str | int, ZwaveValueID] = field(default_factory=dict)
312 dependent_value: ZwaveValueID |
None =
None
315 """Resolve helper class data for a discovered value."""
316 if not self.lookup_table
or not self.dependent_value:
317 raise ValueError(
"Invalid discovery data template")
318 data: dict[str, Any] = {
321 value.node, self.dependent_value
324 for key, value_id
in self.lookup_table.items():
325 data[
"lookup_table"][key] = self.
_get_value_from_id_get_value_from_id(value.node, value_id)
330 self, resolved_data: dict[str, Any]
331 ) -> Iterable[ZwaveValue |
None]:
332 """Return list of all ZwaveValues resolved by helper that should be watched."""
334 *resolved_data[
"lookup_table"].values(),
335 resolved_data[
"dependent_value"],
340 """Get current temperature ZwaveValue from resolved data."""
341 lookup_table: dict[str | int, ZwaveValue |
None] = resolved_data[
"lookup_table"]
342 dependent_value: ZwaveValue |
None = resolved_data[
"dependent_value"]
344 if dependent_value
and dependent_value.value
is not None:
345 lookup_key = dependent_value.metadata.states[
346 str(dependent_value.value)
348 return lookup_table.get(lookup_key)
355 """Class to represent returned data from NumericSensorDataTemplate."""
357 entity_description_key: str |
None =
None
358 unit_of_measurement: str |
None =
None
362 """Data template class for Z-Wave Sensor entities."""
365 def find_key_from_matching_set[_T: Enum](
366 enum_value: _T, set_map: Mapping[str, list[_T]]
368 """Find a key in a set map that matches a given enum value."""
369 for key, value_set
in set_map.items():
370 for value_in_set
in value_set:
374 value_in_set.__class__ == enum_value.__class__
375 and value_in_set == enum_value
380 def resolve_data(self, value: ZwaveValue) -> NumericSensorDataTemplateData:
381 """Resolve helper class data for a discovered value."""
383 if value.command_class == CommandClass.BATTERY:
386 if value.command_class == CommandClass.METER:
388 meter_scale_type = get_meter_scale_type(value)
389 except UnknownValueData:
392 unit = self.find_key_from_matching_set(meter_scale_type, METER_UNIT_MAP)
395 if meter_scale_type
in (
396 ElectricScale.PULSE_COUNT,
397 ElectricScale.KILOVOLT_AMPERE_HOUR,
398 ElectricScale.KILOVOLT_AMPERE_REACTIVE_HOUR,
401 ENTITY_DESC_KEY_TOTAL_INCREASING, unit
405 if meter_scale_type == ElectricScale.KILOVOLT_AMPERE_REACTIVE:
409 self.find_key_from_matching_set(
410 meter_scale_type, METER_DEVICE_CLASS_MAP
415 if value.command_class == CommandClass.SENSOR_MULTILEVEL:
417 sensor_type = get_multilevel_sensor_type(value)
418 multilevel_sensor_scale_type = get_multilevel_sensor_scale_type(value)
419 except UnknownValueData:
421 unit = self.find_key_from_matching_set(
422 multilevel_sensor_scale_type, MULTILEVEL_SENSOR_UNIT_MAP
424 if sensor_type == MultilevelSensorType.TARGET_TEMPERATURE:
426 ENTITY_DESC_KEY_TARGET_TEMPERATURE, unit
428 key = self.find_key_from_matching_set(
429 sensor_type, MULTILEVEL_SENSOR_DEVICE_CLASS_MAP
434 if value.command_class == CommandClass.ENERGY_PRODUCTION:
435 energy_production_parameter = get_energy_production_parameter(value)
436 energy_production_scale_type = get_energy_production_scale_type(value)
437 unit = self.find_key_from_matching_set(
438 energy_production_scale_type, ENERGY_PRODUCTION_UNIT_MAP
440 key = self.find_key_from_matching_set(
441 energy_production_parameter, ENERGY_PRODUCTION_DEVICE_CLASS_MAP
451 """Mixin data class for the current_tilt_value and target_tilt_value."""
453 current_tilt_value_id: ZwaveValueID
454 target_tilt_value_id: ZwaveValueID
459 """Tilt data template class for Z-Wave Cover entities."""
462 """Resolve helper class data for a discovered value."""
464 value.node, self.current_tilt_value_id
466 assert current_tilt_value
468 value.node, self.target_tilt_value_id
470 assert target_tilt_value
472 "current_tilt_value": current_tilt_value,
473 "target_tilt_value": target_tilt_value,
477 self, resolved_data: dict[str, Any]
478 ) -> Iterable[ZwaveValue |
None]:
479 """Return list of all ZwaveValues resolved by helper that should be watched."""
480 return [resolved_data[
"current_tilt_value"], resolved_data[
"target_tilt_value"]]
484 """Get current tilt ZwaveValue from resolved data."""
485 return resolved_data[
"current_tilt_value"]
489 """Get target tilt ZwaveValue from resolved data."""
490 return resolved_data[
"target_tilt_value"]
495 """Data class to represent how a fan's values map to features."""
497 presets: dict[int, str] = field(default_factory=dict)
498 speeds: list[tuple[int, int]] = field(default_factory=list)
503 These inputs are hardcoded in `discovery.py`, so these checks should
504 only fail due to developer error.
506 assert len(self.speeds) > 0,
"At least one speed must be specified"
507 for speed_range
in self.speeds:
508 (low, high) = speed_range
509 assert high >= low,
"Speed range values must be ordered"
514 """Mixin to define `get_fan_value_mapping`."""
517 self, resolved_data: dict[str, Any]
518 ) -> FanValueMapping |
None:
519 """Get the value mappings for this device."""
520 raise NotImplementedError
525 """Mixin data class for defining fan properties that change based on a device configuration option."""
527 configuration_option: ZwaveValueID
528 configuration_value_to_fan_value_mapping: dict[int, FanValueMapping]
533 BaseDiscoverySchemaDataTemplate,
534 FanValueMappingDataTemplate,
535 ConfigurableFanValueMappingValueMix,
537 """Gets fan speeds based on a configuration value.
540 ZWaveDiscoverySchema(
542 hint="has_fan_value_mapping",
544 data_template=ConfigurableFanValueMappingDataTemplate(
545 configuration_option=ZwaveValueID(
546 property_=5, command_class=CommandClass.CONFIGURATION, endpoint=0
548 configuration_value_to_fan_value_mapping={
549 0: FanValueMapping(speeds=[(1,33), (34,66), (67,99)]),
550 1: FanValueMapping(speeds=[(1,24), (25,49), (50,74), (75,99)]),
554 `configuration_option` is a reference to the setting that determines which
555 value mapping to use (e.g., 3 speeds or 4 speeds).
557 `configuration_value_to_fan_value_mapping` maps the values from
558 `configuration_option` to the value mapping object.
563 self, value: ZwaveValue
564 ) -> dict[str, ZwaveConfigurationValue |
None]:
565 """Resolve helper class data for a discovered value."""
567 ZwaveConfigurationValue |
None,
570 return {
"configuration_value": zwave_value}
573 self, resolved_data: dict[str, ZwaveConfigurationValue |
None]
574 ) -> Iterable[ZwaveConfigurationValue |
None]:
575 """Return list of all ZwaveValues that should be watched."""
577 resolved_data[
"configuration_value"],
581 self, resolved_data: dict[str, ZwaveConfigurationValue |
None]
582 ) -> FanValueMapping |
None:
583 """Get current fan properties from resolved data."""
584 zwave_value = resolved_data[
"configuration_value"]
586 if zwave_value
is None:
587 _LOGGER.warning(
"Unable to read device configuration value")
590 if zwave_value.value
is None:
591 _LOGGER.warning(
"Fan configuration value is missing")
594 fan_value_mapping = self.configuration_value_to_fan_value_mapping.
get(
597 if fan_value_mapping
is None:
598 _LOGGER.warning(
"Unrecognized fan configuration value")
601 return fan_value_mapping
606 """Mixin data class for defining supported fan speeds."""
608 fan_value_mapping: FanValueMapping
613 BaseDiscoverySchemaDataTemplate,
614 FanValueMappingDataTemplate,
615 FixedFanValueMappingValueMix,
617 """Specifies a fixed set of properties for a fan.
620 ZWaveDiscoverySchema(
622 hint="has_fan_value_mapping",
624 data_template=FixedFanValueMappingDataTemplate(
625 config=FanValueMapping(
626 speeds=[(1, 32), (33, 65), (66, 99)]
634 self, resolved_data: dict[str, ZwaveConfigurationValue]
635 ) -> FanValueMapping:
636 """Get the fan properties for this device."""
637 return self.fan_value_mapping
set[str] value_ids_to_watch(self, Any resolved_data)
Iterable[ZwaveValue|None] values_to_watch(self, Any resolved_data)
Any resolve_data(self, ZwaveValue value)
ZwaveValue|ZwaveConfigurationValue|None _get_value_from_id(ZwaveNode node, ZwaveValueID value_id_obj)
Iterable[ZwaveConfigurationValue|None] values_to_watch(self, dict[str, ZwaveConfigurationValue|None] resolved_data)
dict[str, ZwaveConfigurationValue|None] resolve_data(self, ZwaveValue value)
FanValueMapping|None get_fan_value_mapping(self, dict[str, ZwaveConfigurationValue|None] resolved_data)
Iterable[ZwaveValue|None] values_to_watch(self, dict[str, Any] resolved_data)
ZwaveValue current_tilt_value(dict[str, ZwaveValue] resolved_data)
dict[str, ZwaveValue] resolve_data(self, ZwaveValue value)
ZwaveValue target_tilt_value(dict[str, ZwaveValue] resolved_data)
Iterable[ZwaveValue|None] values_to_watch(self, dict[str, Any] resolved_data)
ZwaveValue|None current_temperature_value(dict[str, Any] resolved_data)
dict[str, Any] resolve_data(self, ZwaveValue value)
FanValueMapping|None get_fan_value_mapping(self, dict[str, Any] resolved_data)
FanValueMapping get_fan_value_mapping(self, dict[str, ZwaveConfigurationValue] resolved_data)
NumericSensorDataTemplateData resolve_data(self, ZwaveValue value)
web.Response get(self, web.Request request, str config_key)