diff --git a/homeassistant/components/zwave_js/__init__.py b/homeassistant/components/zwave_js/__init__.py index af42f024e6a..f78c201340a 100644 --- a/homeassistant/components/zwave_js/__init__.py +++ b/homeassistant/components/zwave_js/__init__.py @@ -115,11 +115,7 @@ from .const import ( ZWAVE_JS_VALUE_NOTIFICATION_EVENT, ZWAVE_JS_VALUE_UPDATED_EVENT, ) -from .discovery import ( - ZwaveDiscoveryInfo, - async_discover_node_values, - async_discover_single_value, -) +from .discovery import async_discover_node_values, async_discover_single_value from .helpers import ( async_disable_server_logging_if_needed, async_enable_server_logging_if_needed, @@ -131,7 +127,7 @@ from .helpers import ( get_valueless_base_unique_id, ) from .migrate import async_migrate_discovered_value -from .models import ZwaveJSConfigEntry, ZwaveJSData +from .models import PlatformZwaveDiscoveryInfo, ZwaveJSConfigEntry, ZwaveJSData from .services import async_setup_services CONNECT_TIMEOUT = 10 @@ -776,7 +772,7 @@ class NodeEvents: # Remove any old value ids if this is a reinterview. self.controller_events.discovered_value_ids.pop(device.id, None) - value_updates_disc_info: dict[str, ZwaveDiscoveryInfo] = {} + value_updates_disc_info: dict[str, PlatformZwaveDiscoveryInfo] = {} # run discovery on all node values and create/update entities await asyncio.gather( @@ -858,8 +854,8 @@ class NodeEvents: async def async_handle_discovery_info( self, device: dr.DeviceEntry, - disc_info: ZwaveDiscoveryInfo, - value_updates_disc_info: dict[str, ZwaveDiscoveryInfo], + disc_info: PlatformZwaveDiscoveryInfo, + value_updates_disc_info: dict[str, PlatformZwaveDiscoveryInfo], ) -> None: """Handle discovery info and all dependent tasks.""" platform = disc_info.platform @@ -901,7 +897,9 @@ class NodeEvents: ) async def async_on_value_added( - self, value_updates_disc_info: dict[str, ZwaveDiscoveryInfo], value: Value + self, + value_updates_disc_info: dict[str, PlatformZwaveDiscoveryInfo], + value: Value, ) -> None: """Fire value updated event.""" # If node isn't ready or a device for this node doesn't already exist, we can @@ -1036,7 +1034,9 @@ class NodeEvents: @callback def async_on_value_updated_fire_event( - self, value_updates_disc_info: dict[str, ZwaveDiscoveryInfo], value: Value + self, + value_updates_disc_info: dict[str, PlatformZwaveDiscoveryInfo], + value: Value, ) -> None: """Fire value updated event.""" # Get the discovery info for the value that was updated. If there is diff --git a/homeassistant/components/zwave_js/binary_sensor.py b/homeassistant/components/zwave_js/binary_sensor.py index 5b7fe4f4d7c..2280ba69c01 100644 --- a/homeassistant/components/zwave_js/binary_sensor.py +++ b/homeassistant/components/zwave_js/binary_sensor.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from zwave_js_server.const import CommandClass from zwave_js_server.const.command_class.lock import DOOR_STATUS_PROPERTY @@ -17,15 +17,21 @@ from homeassistant.components.binary_sensor import ( BinarySensorEntity, BinarySensorEntityDescription, ) -from homeassistant.const import EntityCategory +from homeassistant.const import EntityCategory, Platform from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from .const import DOMAIN -from .discovery import ZwaveDiscoveryInfo -from .entity import ZWaveBaseEntity -from .models import ZwaveJSConfigEntry +from .entity import NewZwaveDiscoveryInfo, ZWaveBaseEntity +from .models import ( + NewZWaveDiscoverySchema, + ValueType, + ZwaveDiscoveryInfo, + ZwaveJSConfigEntry, + ZWaveValueDiscoverySchema, +) PARALLEL_UPDATES = 0 @@ -50,11 +56,11 @@ NOTIFICATION_IRRIGATION = "17" NOTIFICATION_GAS = "18" -@dataclass(frozen=True) +@dataclass(frozen=True, kw_only=True) class NotificationZWaveJSEntityDescription(BinarySensorEntityDescription): """Represent a Z-Wave JS binary sensor entity description.""" - off_state: str = "0" + not_states: set[str] = field(default_factory=lambda: {"0"}) states: tuple[str, ...] | None = None @@ -65,6 +71,13 @@ class PropertyZWaveJSEntityDescription(BinarySensorEntityDescription): on_states: tuple[str, ...] +@dataclass(frozen=True, kw_only=True) +class NewNotificationZWaveJSEntityDescription(BinarySensorEntityDescription): + """Represent a Z-Wave JS binary sensor entity description.""" + + state_key: str + + # Mappings for Notification sensors # https://github.com/zwave-js/specs/blob/master/Registries/Notification%20Command%20Class%2C%20list%20of%20assigned%20Notifications.xlsx # @@ -106,24 +119,6 @@ class PropertyZWaveJSEntityDescription(BinarySensorEntityDescription): # - Sump pump failure NOTIFICATION_SENSOR_MAPPINGS: tuple[NotificationZWaveJSEntityDescription, ...] = ( - NotificationZWaveJSEntityDescription( - # NotificationType 1: Smoke Alarm - State Id's 1 and 2 - Smoke detected - key=NOTIFICATION_SMOKE_ALARM, - states=("1", "2"), - device_class=BinarySensorDeviceClass.SMOKE, - ), - NotificationZWaveJSEntityDescription( - # NotificationType 1: Smoke Alarm - State Id's 4, 5, 7, 8 - key=NOTIFICATION_SMOKE_ALARM, - states=("4", "5", "7", "8"), - device_class=BinarySensorDeviceClass.PROBLEM, - entity_category=EntityCategory.DIAGNOSTIC, - ), - NotificationZWaveJSEntityDescription( - # NotificationType 1: Smoke Alarm - All other State Id's - key=NOTIFICATION_SMOKE_ALARM, - entity_category=EntityCategory.DIAGNOSTIC, - ), NotificationZWaveJSEntityDescription( # NotificationType 2: Carbon Monoxide - State Id's 1 and 2 key=NOTIFICATION_CARBON_MONOOXIDE, @@ -212,8 +207,8 @@ NOTIFICATION_SENSOR_MAPPINGS: tuple[NotificationZWaveJSEntityDescription, ...] = NotificationZWaveJSEntityDescription( # NotificationType 6: Access Control - State Id 22 (door/window open) key=NOTIFICATION_ACCESS_CONTROL, - off_state="23", - states=("22", "23"), + not_states={"23"}, + states=("22",), device_class=BinarySensorDeviceClass.DOOR, ), NotificationZWaveJSEntityDescription( @@ -245,8 +240,8 @@ NOTIFICATION_SENSOR_MAPPINGS: tuple[NotificationZWaveJSEntityDescription, ...] = # NotificationType 8: Power Management - # State Id's 2, 3 (Mains status) key=NOTIFICATION_POWER_MANAGEMENT, - off_state="2", - states=("2", "3"), + not_states={"2"}, + states=("3",), device_class=BinarySensorDeviceClass.PLUG, entity_category=EntityCategory.DIAGNOSTIC, ), @@ -353,7 +348,7 @@ BOOLEAN_SENSOR_MAPPINGS: dict[tuple[int, int | str], BinarySensorEntityDescripti @callback def is_valid_notification_binary_sensor( - info: ZwaveDiscoveryInfo, + info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo, ) -> bool | NotificationZWaveJSEntityDescription: """Return if the notification CC Value is valid as binary sensor.""" if not info.primary_value.metadata.states: @@ -370,13 +365,36 @@ async def async_setup_entry( client = config_entry.runtime_data.client @callback - def async_add_binary_sensor(info: ZwaveDiscoveryInfo) -> None: + def async_add_binary_sensor( + info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo, + ) -> None: """Add Z-Wave Binary Sensor.""" driver = client.driver assert driver is not None # Driver is ready before platforms are loaded. - entities: list[BinarySensorEntity] = [] + entities: list[Entity] = [] - if info.platform_hint == "notification": + if ( + isinstance(info, NewZwaveDiscoveryInfo) + and info.entity_class is ZWaveNotificationBinarySensor + and isinstance( + info.entity_description, NotificationZWaveJSEntityDescription + ) + and is_valid_notification_binary_sensor(info) + ): + entities.extend( + ZWaveNotificationBinarySensor( + config_entry, driver, info, state_key, info.entity_description + ) + for state_key in info.primary_value.metadata.states + if state_key not in info.entity_description.not_states + and ( + not info.entity_description.states + or state_key in info.entity_description.states + ) + ) + elif isinstance(info, NewZwaveDiscoveryInfo): + pass # other entity classes are not migrated yet + elif info.platform_hint == "notification": # ensure the notification CC Value is valid as binary sensor if not is_valid_notification_binary_sensor(info): return @@ -401,7 +419,7 @@ async def async_setup_entry( if ( notification_description - and notification_description.off_state == state_key + and state_key in notification_description.not_states ): continue entities.append( @@ -477,7 +495,7 @@ class ZWaveNotificationBinarySensor(ZWaveBaseEntity, BinarySensorEntity): self, config_entry: ZwaveJSConfigEntry, driver: Driver, - info: ZwaveDiscoveryInfo, + info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo, state_key: str, description: NotificationZWaveJSEntityDescription | None = None, ) -> None: @@ -543,3 +561,71 @@ class ZWaveConfigParameterBinarySensor(ZWaveBooleanBinarySensor): alternate_value_name=self.info.primary_value.property_name, additional_info=[property_key_name] if property_key_name else None, ) + + +DISCOVERY_SCHEMAS: list[NewZWaveDiscoverySchema] = [ + NewZWaveDiscoverySchema( + platform=Platform.BINARY_SENSOR, + primary_value=ZWaveValueDiscoverySchema( + command_class={ + CommandClass.NOTIFICATION, + }, + type={ValueType.NUMBER}, + any_available_states_keys={1, 2}, + any_available_cc_specific={(CC_SPECIFIC_NOTIFICATION_TYPE, 1)}, + ), + allow_multi=True, + entity_description=NotificationZWaveJSEntityDescription( + # NotificationType 1: Smoke Alarm - State Id's 1 and 2 - Smoke detected + key=NOTIFICATION_SMOKE_ALARM, + states=("1", "2"), + device_class=BinarySensorDeviceClass.SMOKE, + ), + entity_class=ZWaveNotificationBinarySensor, + ), + NewZWaveDiscoverySchema( + platform=Platform.BINARY_SENSOR, + primary_value=ZWaveValueDiscoverySchema( + command_class={ + CommandClass.NOTIFICATION, + }, + type={ValueType.NUMBER}, + any_available_states_keys={4, 5, 7, 8}, + any_available_cc_specific={(CC_SPECIFIC_NOTIFICATION_TYPE, 1)}, + ), + allow_multi=True, + entity_description=NotificationZWaveJSEntityDescription( + # NotificationType 1: Smoke Alarm - State Id's 4, 5, 7, 8 + key=NOTIFICATION_SMOKE_ALARM, + states=("4", "5", "7", "8"), + device_class=BinarySensorDeviceClass.PROBLEM, + entity_category=EntityCategory.DIAGNOSTIC, + ), + entity_class=ZWaveNotificationBinarySensor, + ), + NewZWaveDiscoverySchema( + platform=Platform.BINARY_SENSOR, + primary_value=ZWaveValueDiscoverySchema( + command_class={ + CommandClass.NOTIFICATION, + }, + type={ValueType.NUMBER}, + any_available_cc_specific={(CC_SPECIFIC_NOTIFICATION_TYPE, 1)}, + ), + allow_multi=True, + entity_description=NotificationZWaveJSEntityDescription( + # NotificationType 1: Smoke Alarm - All other State Id's + key=NOTIFICATION_SMOKE_ALARM, + entity_category=EntityCategory.DIAGNOSTIC, + not_states={ + "1", + "2", + "4", + "5", + "7", + "8", + }, + ), + entity_class=ZWaveNotificationBinarySensor, + ), +] diff --git a/homeassistant/components/zwave_js/cover.py b/homeassistant/components/zwave_js/cover.py index 424fe94b8b9..d468a233f05 100644 --- a/homeassistant/components/zwave_js/cover.py +++ b/homeassistant/components/zwave_js/cover.py @@ -299,11 +299,23 @@ class ZWaveMultilevelSwitchCover(CoverPositionMixin): # Entity class attributes self._attr_device_class = CoverDeviceClass.WINDOW - if self.info.platform_hint and self.info.platform_hint.startswith("shutter"): + if ( + isinstance(self.info, ZwaveDiscoveryInfo) + and self.info.platform_hint + and self.info.platform_hint.startswith("shutter") + ): self._attr_device_class = CoverDeviceClass.SHUTTER - elif self.info.platform_hint and self.info.platform_hint.startswith("blind"): + elif ( + isinstance(self.info, ZwaveDiscoveryInfo) + and self.info.platform_hint + and self.info.platform_hint.startswith("blind") + ): self._attr_device_class = CoverDeviceClass.BLIND - elif self.info.platform_hint and self.info.platform_hint.startswith("gate"): + elif ( + isinstance(self.info, ZwaveDiscoveryInfo) + and self.info.platform_hint + and self.info.platform_hint.startswith("gate") + ): self._attr_device_class = CoverDeviceClass.GATE diff --git a/homeassistant/components/zwave_js/discovery.py b/homeassistant/components/zwave_js/discovery.py index 7030009f5ad..858e4c300b8 100644 --- a/homeassistant/components/zwave_js/discovery.py +++ b/homeassistant/components/zwave_js/discovery.py @@ -3,9 +3,8 @@ from __future__ import annotations from collections.abc import Generator -from dataclasses import asdict, dataclass, field -from enum import StrEnum -from typing import TYPE_CHECKING, Any, cast +from dataclasses import dataclass +from typing import cast from awesomeversion import AwesomeVersion from zwave_js_server.const import ( @@ -55,6 +54,7 @@ from homeassistant.const import EntityCategory, Platform from homeassistant.core import callback from homeassistant.helpers.device_registry import DeviceEntry +from .binary_sensor import DISCOVERY_SCHEMAS as BINARY_SENSOR_SCHEMAS from .const import COVER_POSITION_PROPERTY_KEYS, COVER_TILT_PROPERTY_KEYS, LOGGER from .discovery_data_template import ( BaseDiscoverySchemaDataTemplate, @@ -65,108 +65,20 @@ from .discovery_data_template import ( FixedFanValueMappingDataTemplate, NumericSensorDataTemplate, ) -from .helpers import ZwaveValueID +from .entity import NewZwaveDiscoveryInfo +from .models import ( + FirmwareVersionRange, + NewZWaveDiscoverySchema, + ValueType, + ZwaveDiscoveryInfo, + ZWaveValueDiscoverySchema, + ZwaveValueID, +) -if TYPE_CHECKING: - from _typeshed import DataclassInstance - - -class ValueType(StrEnum): - """Enum with all value types.""" - - ANY = "any" - BOOLEAN = "boolean" - NUMBER = "number" - STRING = "string" - - -class DataclassMustHaveAtLeastOne: - """A dataclass that must have at least one input parameter that is not None.""" - - def __post_init__(self: DataclassInstance) -> None: - """Post dataclass initialization.""" - if all(val is None for val in asdict(self).values()): - raise ValueError("At least one input parameter must not be None") - - -@dataclass -class FirmwareVersionRange(DataclassMustHaveAtLeastOne): - """Firmware version range dictionary.""" - - min: str | None = None - max: str | None = None - min_ver: AwesomeVersion | None = field(default=None, init=False) - max_ver: AwesomeVersion | None = field(default=None, init=False) - - def __post_init__(self) -> None: - """Post dataclass initialization.""" - super().__post_init__() - if self.min: - self.min_ver = AwesomeVersion(self.min) - if self.max: - self.max_ver = AwesomeVersion(self.max) - - -@dataclass -class ZwaveDiscoveryInfo: - """Info discovered from (primary) ZWave Value to create entity.""" - - # node to which the value(s) belongs - node: ZwaveNode - # the value object itself for primary value - primary_value: ZwaveValue - # bool to specify whether state is assumed and events should be fired on value - # update - assumed_state: bool - # the home assistant platform for which an entity should be created - platform: Platform - # helper data to use in platform setup - platform_data: Any - # additional values that need to be watched by entity - additional_value_ids_to_watch: set[str] - # hint for the platform about this discovered entity - platform_hint: str | None = "" - # data template to use in platform logic - platform_data_template: BaseDiscoverySchemaDataTemplate | None = None - # bool to specify whether entity should be enabled by default - entity_registry_enabled_default: bool = True - # the entity category for the discovered entity - entity_category: EntityCategory | None = None - - -@dataclass -class ZWaveValueDiscoverySchema(DataclassMustHaveAtLeastOne): - """Z-Wave Value discovery schema. - - The Z-Wave Value must match these conditions. - Use the Z-Wave specifications to find out the values for these parameters: - https://github.com/zwave-js/specs/tree/master - """ - - # [optional] the value's command class must match ANY of these values - command_class: set[int] | None = None - # [optional] the value's endpoint must match ANY of these values - endpoint: set[int] | None = None - # [optional] the value's property must match ANY of these values - property: set[str | int] | None = None - # [optional] the value's property name must match ANY of these values - property_name: set[str] | None = None - # [optional] the value's property key must match ANY of these values - property_key: set[str | int | None] | None = None - # [optional] the value's property key must NOT match ANY of these values - not_property_key: set[str | int | None] | None = None - # [optional] the value's metadata_type must match ANY of these values - type: set[str] | None = None - # [optional] the value's metadata_readable must match this value - readable: bool | None = None - # [optional] the value's metadata_writeable must match this value - writeable: bool | None = None - # [optional] the value's states map must include ANY of these key/value pairs - any_available_states: set[tuple[int, str]] | None = None - # [optional] the value's value must match this value - value: Any | None = None - # [optional] the value's metadata_stateful must match this value - stateful: bool | None = None +NEW_DISCOVERY_SCHEMAS: dict[Platform, list[NewZWaveDiscoverySchema]] = { + Platform.BINARY_SENSOR: BINARY_SENSOR_SCHEMAS, +} +SUPPORTED_PLATFORMS = tuple(NEW_DISCOVERY_SCHEMAS) @dataclass @@ -1316,7 +1228,7 @@ DISCOVERY_SCHEMAS = [ @callback def async_discover_node_values( node: ZwaveNode, device: DeviceEntry, discovered_value_ids: dict[str, set[str]] -) -> Generator[ZwaveDiscoveryInfo]: +) -> Generator[ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo]: """Run discovery on ZWave node and return matching (primary) values.""" for value in node.values.values(): # We don't need to rediscover an already processed value_id @@ -1327,9 +1239,19 @@ def async_discover_node_values( @callback def async_discover_single_value( value: ZwaveValue, device: DeviceEntry, discovered_value_ids: dict[str, set[str]] -) -> Generator[ZwaveDiscoveryInfo]: +) -> Generator[ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo]: """Run discovery on a single ZWave value and return matching schema info.""" - for schema in DISCOVERY_SCHEMAS: + # Temporary workaround for new schemas + schemas: tuple[ZWaveDiscoverySchema | NewZWaveDiscoverySchema, ...] = ( + *( + new_schema + for _schemas in NEW_DISCOVERY_SCHEMAS.values() + for new_schema in _schemas + ), + *DISCOVERY_SCHEMAS, + ) + + for schema in schemas: # abort if attribute(s) already discovered if value.value_id in discovered_value_ids[device.id]: continue @@ -1458,18 +1380,38 @@ def async_discover_single_value( ) # all checks passed, this value belongs to an entity - yield ZwaveDiscoveryInfo( - node=value.node, - primary_value=value, - assumed_state=schema.assumed_state, - platform=schema.platform, - platform_hint=schema.hint, - platform_data_template=schema.data_template, - platform_data=resolved_data, - additional_value_ids_to_watch=additional_value_ids_to_watch, - entity_registry_enabled_default=schema.entity_registry_enabled_default, - entity_category=schema.entity_category, - ) + + discovery_info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo + + # Temporary workaround for new schemas + if isinstance(schema, NewZWaveDiscoverySchema): + discovery_info = NewZwaveDiscoveryInfo( + node=value.node, + primary_value=value, + assumed_state=schema.assumed_state, + platform=schema.platform, + platform_data_template=schema.data_template, + platform_data=resolved_data, + additional_value_ids_to_watch=additional_value_ids_to_watch, + entity_class=schema.entity_class, + entity_description=schema.entity_description, + ) + + else: + discovery_info = ZwaveDiscoveryInfo( + node=value.node, + primary_value=value, + assumed_state=schema.assumed_state, + platform=schema.platform, + platform_hint=schema.hint, + platform_data_template=schema.data_template, + platform_data=resolved_data, + additional_value_ids_to_watch=additional_value_ids_to_watch, + entity_registry_enabled_default=schema.entity_registry_enabled_default, + entity_category=schema.entity_category, + ) + + yield discovery_info # prevent re-discovery of the (primary) value if not allowed if not schema.allow_multi: @@ -1615,6 +1557,25 @@ def check_value( ) ): return False + if ( + schema.any_available_states_keys is not None + and value.metadata.states is not None + and not any( + str(key) in value.metadata.states + for key in schema.any_available_states_keys + ) + ): + return False + # check available cc specific + if ( + schema.any_available_cc_specific is not None + and value.metadata.cc_specific is not None + and not any( + key in value.metadata.cc_specific and value.metadata.cc_specific[key] == val + for key, val in schema.any_available_cc_specific + ) + ): + return False # check value if schema.value is not None and value.value not in schema.value: return False diff --git a/homeassistant/components/zwave_js/discovery_data_template.py b/homeassistant/components/zwave_js/discovery_data_template.py index 731a786d226..8fbc5f35555 100644 --- a/homeassistant/components/zwave_js/discovery_data_template.py +++ b/homeassistant/components/zwave_js/discovery_data_template.py @@ -90,11 +90,9 @@ from zwave_js_server.const.command_class.multilevel_sensor import ( MultilevelSensorType, ) from zwave_js_server.exceptions import UnknownValueData -from zwave_js_server.model.node import Node as ZwaveNode from zwave_js_server.model.value import ( ConfigurationValue as ZwaveConfigurationValue, Value as ZwaveValue, - get_value_id_str, ) from zwave_js_server.util.command_class.energy_production import ( get_energy_production_parameter, @@ -159,7 +157,7 @@ from .const import ( ENTITY_DESC_KEY_UV_INDEX, ENTITY_DESC_KEY_VOLTAGE, ) -from .helpers import ZwaveValueID +from .models import BaseDiscoverySchemaDataTemplate, ZwaveValueID ENERGY_PRODUCTION_DEVICE_CLASS_MAP: dict[str, list[EnergyProductionParameter]] = { ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME: [EnergyProductionParameter.TOTAL_TIME], @@ -264,49 +262,6 @@ MULTILEVEL_SENSOR_UNIT_MAP: dict[str, list[MultilevelSensorScaleType]] = { _LOGGER = logging.getLogger(__name__) -@dataclass -class BaseDiscoverySchemaDataTemplate: - """Base class for discovery schema data templates.""" - - static_data: Any | None = None - - def resolve_data(self, value: ZwaveValue) -> Any: - """Resolve helper class data for a discovered value. - - Can optionally be implemented by subclasses if input data needs to be - transformed once discovered Value is available. - """ - return {} - - def values_to_watch(self, resolved_data: Any) -> Iterable[ZwaveValue | None]: - """Return list of all ZwaveValues resolved by helper that should be watched. - - Should be implemented by subclasses only if there are values to watch. - """ - return [] - - def value_ids_to_watch(self, resolved_data: Any) -> set[str]: - """Return list of all Value IDs resolved by helper that should be watched. - - Not to be overwritten by subclasses. - """ - return {val.value_id for val in self.values_to_watch(resolved_data) if val} - - @staticmethod - def _get_value_from_id( - node: ZwaveNode, value_id_obj: ZwaveValueID - ) -> ZwaveValue | ZwaveConfigurationValue | None: - """Get a ZwaveValue from a node using a ZwaveValueDict.""" - value_id = get_value_id_str( - node, - value_id_obj.command_class, - value_id_obj.property_, - endpoint=value_id_obj.endpoint, - property_key=value_id_obj.property_key, - ) - return node.values.get(value_id) - - @dataclass class DynamicCurrentTempClimateDataTemplate(BaseDiscoverySchemaDataTemplate): """Data template class for Z-Wave JS Climate entities with dynamic current temps.""" diff --git a/homeassistant/components/zwave_js/entity.py b/homeassistant/components/zwave_js/entity.py index 08a587d8d20..ab892565c0f 100644 --- a/homeassistant/components/zwave_js/entity.py +++ b/homeassistant/components/zwave_js/entity.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Sequence +from dataclasses import dataclass from typing import Any from zwave_js_server.exceptions import BaseZwaveJSServerError @@ -18,16 +19,33 @@ from homeassistant.core import callback from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.dispatcher import async_dispatcher_connect -from homeassistant.helpers.entity import Entity +from homeassistant.helpers.entity import Entity, EntityDescription from homeassistant.helpers.typing import UNDEFINED from .const import DOMAIN, EVENT_VALUE_UPDATED, LOGGER -from .discovery import ZwaveDiscoveryInfo +from .discovery_data_template import BaseDiscoverySchemaDataTemplate from .helpers import get_device_id, get_unique_id, get_valueless_base_unique_id +from .models import PlatformZwaveDiscoveryInfo, ZwaveDiscoveryInfo EVENT_VALUE_REMOVED = "value removed" +@dataclass(kw_only=True) +class NewZwaveDiscoveryInfo(PlatformZwaveDiscoveryInfo): + """Info discovered from (primary) ZWave Value to create entity. + + This is the new discovery info that will replace ZwaveDiscoveryInfo. + """ + + entity_class: type[ZWaveBaseEntity] + # the entity description to use + entity_description: EntityDescription + # helper data to use in platform setup + platform_data: Any = None + # data template to use in platform logic + platform_data_template: BaseDiscoverySchemaDataTemplate | None = None + + class ZWaveBaseEntity(Entity): """Generic Entity Class for a Z-Wave Device.""" @@ -35,7 +53,10 @@ class ZWaveBaseEntity(Entity): _attr_has_entity_name = True def __init__( - self, config_entry: ConfigEntry, driver: Driver, info: ZwaveDiscoveryInfo + self, + config_entry: ConfigEntry, + driver: Driver, + info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo, ) -> None: """Initialize a generic Z-Wave device entity.""" self.config_entry = config_entry @@ -52,12 +73,14 @@ class ZWaveBaseEntity(Entity): # Entity class attributes self._attr_name = self.generate_name() self._attr_unique_id = get_unique_id(driver, self.info.primary_value.value_id) - if self.info.entity_registry_enabled_default is False: - self._attr_entity_registry_enabled_default = False - if self.info.entity_category is not None: - self._attr_entity_category = self.info.entity_category - if self.info.assumed_state: - self._attr_assumed_state = True + if isinstance(info, NewZwaveDiscoveryInfo): + self.entity_description = info.entity_description + else: + if (enabled_default := info.entity_registry_enabled_default) is False: + self._attr_entity_registry_enabled_default = enabled_default + if (entity_category := info.entity_category) is not None: + self._attr_entity_category = entity_category + self._attr_assumed_state = self.info.assumed_state # device is precreated in main handler self._attr_device_info = DeviceInfo( identifiers={get_device_id(driver, self.info.node)}, diff --git a/homeassistant/components/zwave_js/helpers.py b/homeassistant/components/zwave_js/helpers.py index 17f4909662c..dc415c157b6 100644 --- a/homeassistant/components/zwave_js/helpers.py +++ b/homeassistant/components/zwave_js/helpers.py @@ -60,16 +60,6 @@ DRIVER_READY_EVENT_TIMEOUT = 60 SERVER_VERSION_TIMEOUT = 10 -@dataclass -class ZwaveValueID: - """Class to represent a value ID.""" - - property_: str | int - command_class: int - endpoint: int | None = None - property_key: str | int | None = None - - @dataclass class ZwaveValueMatcher: """Class to allow matching a Z-Wave Value.""" diff --git a/homeassistant/components/zwave_js/migrate.py b/homeassistant/components/zwave_js/migrate.py index ac749cb516b..e4cd414a2bb 100644 --- a/homeassistant/components/zwave_js/migrate.py +++ b/homeassistant/components/zwave_js/migrate.py @@ -5,6 +5,7 @@ from __future__ import annotations from dataclasses import dataclass import logging +from zwave_js_server.const import CommandClass from zwave_js_server.model.driver import Driver from zwave_js_server.model.node import Node from zwave_js_server.model.value import Value as ZwaveValue @@ -14,8 +15,8 @@ from homeassistant.core import HomeAssistant, callback from homeassistant.helpers import device_registry as dr, entity_registry as er from .const import DOMAIN -from .discovery import ZwaveDiscoveryInfo from .helpers import get_unique_id, get_valueless_base_unique_id +from .models import PlatformZwaveDiscoveryInfo _LOGGER = logging.getLogger(__name__) @@ -140,7 +141,7 @@ def async_migrate_discovered_value( registered_unique_ids: set[str], device: dr.DeviceEntry, driver: Driver, - disc_info: ZwaveDiscoveryInfo, + disc_info: PlatformZwaveDiscoveryInfo, ) -> None: """Migrate unique ID for entity/entities tied to discovered value.""" @@ -162,7 +163,7 @@ def async_migrate_discovered_value( if ( disc_info.platform == Platform.BINARY_SENSOR - and disc_info.platform_hint == "notification" + and disc_info.primary_value.command_class == CommandClass.NOTIFICATION ): for state_key in disc_info.primary_value.metadata.states: # ignore idle key (0) diff --git a/homeassistant/components/zwave_js/models.py b/homeassistant/components/zwave_js/models.py index 63f77871c14..ba93be7a554 100644 --- a/homeassistant/components/zwave_js/models.py +++ b/homeassistant/components/zwave_js/models.py @@ -1,15 +1,27 @@ -"""Type definitions for Z-Wave JS integration.""" +"""Provide models for the Z-Wave integration.""" from __future__ import annotations -from dataclasses import dataclass -from typing import TYPE_CHECKING +from collections.abc import Iterable +from dataclasses import asdict, dataclass, field +from enum import StrEnum +from typing import TYPE_CHECKING, Any +from awesomeversion import AwesomeVersion from zwave_js_server.const import LogLevel +from zwave_js_server.model.node import Node as ZwaveNode +from zwave_js_server.model.value import ( + ConfigurationValue as ZwaveConfigurationValue, + Value as ZwaveValue, + get_value_id_str, +) from homeassistant.config_entries import ConfigEntry +from homeassistant.const import EntityCategory, Platform +from homeassistant.helpers.entity import EntityDescription if TYPE_CHECKING: + from _typeshed import DataclassInstance from zwave_js_server.client import Client as ZwaveClient from . import DriverEvents @@ -25,3 +37,213 @@ class ZwaveJSData: type ZwaveJSConfigEntry = ConfigEntry[ZwaveJSData] + + +@dataclass +class ZwaveValueID: + """Class to represent a value ID.""" + + property_: str | int + command_class: int + endpoint: int | None = None + property_key: str | int | None = None + + +class ValueType(StrEnum): + """Enum with all value types.""" + + ANY = "any" + BOOLEAN = "boolean" + NUMBER = "number" + STRING = "string" + + +class DataclassMustHaveAtLeastOne: + """A dataclass that must have at least one input parameter that is not None.""" + + def __post_init__(self: DataclassInstance) -> None: + """Post dataclass initialization.""" + if all(val is None for val in asdict(self).values()): + raise ValueError("At least one input parameter must not be None") + + +@dataclass +class FirmwareVersionRange(DataclassMustHaveAtLeastOne): + """Firmware version range dictionary.""" + + min: str | None = None + max: str | None = None + min_ver: AwesomeVersion | None = field(default=None, init=False) + max_ver: AwesomeVersion | None = field(default=None, init=False) + + def __post_init__(self) -> None: + """Post dataclass initialization.""" + super().__post_init__() + if self.min: + self.min_ver = AwesomeVersion(self.min) + if self.max: + self.max_ver = AwesomeVersion(self.max) + + +@dataclass +class PlatformZwaveDiscoveryInfo: + """Info discovered from (primary) ZWave Value to create entity.""" + + # node to which the value(s) belongs + node: ZwaveNode + # the value object itself for primary value + primary_value: ZwaveValue + # bool to specify whether state is assumed and events should be fired on value + # update + assumed_state: bool + # the home assistant platform for which an entity should be created + platform: Platform + # additional values that need to be watched by entity + additional_value_ids_to_watch: set[str] + + +@dataclass +class ZwaveDiscoveryInfo(PlatformZwaveDiscoveryInfo): + """Info discovered from (primary) ZWave Value to create entity.""" + + # helper data to use in platform setup + platform_data: Any = None + # data template to use in platform logic + platform_data_template: BaseDiscoverySchemaDataTemplate | None = None + # hint for the platform about this discovered entity + platform_hint: str | None = "" + # bool to specify whether entity should be enabled by default + entity_registry_enabled_default: bool = True + # the entity category for the discovered entity + entity_category: EntityCategory | None = None + + +@dataclass +class ZWaveValueDiscoverySchema(DataclassMustHaveAtLeastOne): + """Z-Wave Value discovery schema. + + The Z-Wave Value must match these conditions. + Use the Z-Wave specifications to find out the values for these parameters: + https://github.com/zwave-js/specs/tree/master + """ + + # [optional] the value's command class must match ANY of these values + command_class: set[int] | None = None + # [optional] the value's endpoint must match ANY of these values + endpoint: set[int] | None = None + # [optional] the value's property must match ANY of these values + property: set[str | int] | None = None + # [optional] the value's property name must match ANY of these values + property_name: set[str] | None = None + # [optional] the value's property key must match ANY of these values + property_key: set[str | int | None] | None = None + # [optional] the value's property key must NOT match ANY of these values + not_property_key: set[str | int | None] | None = None + # [optional] the value's metadata_type must match ANY of these values + type: set[str] | None = None + # [optional] the value's metadata_readable must match this value + readable: bool | None = None + # [optional] the value's metadata_writeable must match this value + writeable: bool | None = None + # [optional] the value's states map must include ANY of these key/value pairs + any_available_states: set[tuple[int, str]] | None = None + # [optional] the value's states map must include ANY of these keys + any_available_states_keys: set[int] | None = None + # [optional] the value's cc specific map must include ANY of these key/value pairs + any_available_cc_specific: set[tuple[Any, Any]] | None = None + # [optional] the value's value must match this value + value: Any | None = None + # [optional] the value's metadata_stateful must match this value + stateful: bool | None = None + + +@dataclass +class NewZWaveDiscoverySchema: + """Z-Wave discovery schema. + + The Z-Wave node and it's (primary) value for an entity must match these conditions. + Use the Z-Wave specifications to find out the values for these parameters: + https://github.com/zwave-js/node-zwave-js/tree/master/specs + """ + + # specify the hass platform for which this scheme applies (e.g. light, sensor) + platform: Platform + # platform-specific entity description + entity_description: EntityDescription + # entity class to use to instantiate the entity + entity_class: type + # primary value belonging to this discovery scheme + primary_value: ZWaveValueDiscoverySchema + # [optional] template to generate platform specific data to use in setup + data_template: BaseDiscoverySchemaDataTemplate | None = None + # [optional] the node's manufacturer_id must match ANY of these values + manufacturer_id: set[int] | None = None + # [optional] the node's product_id must match ANY of these values + product_id: set[int] | None = None + # [optional] the node's product_type must match ANY of these values + product_type: set[int] | None = None + # [optional] the node's firmware_version must be within this range + firmware_version_range: FirmwareVersionRange | None = None + # [optional] the node's firmware_version must match ANY of these values + firmware_version: set[str] | None = None + # [optional] the node's basic device class must match ANY of these values + device_class_basic: set[str | int] | None = None + # [optional] the node's generic device class must match ANY of these values + device_class_generic: set[str | int] | None = None + # [optional] the node's specific device class must match ANY of these values + device_class_specific: set[str | int] | None = None + # [optional] additional values that ALL need to be present + # on the node for this scheme to pass + required_values: list[ZWaveValueDiscoverySchema] | None = None + # [optional] additional values that MAY NOT be present + # on the node for this scheme to pass + absent_values: list[ZWaveValueDiscoverySchema] | None = None + # [optional] bool to specify if this primary value may be discovered + # by multiple platforms + allow_multi: bool = False + # [optional] bool to specify whether state is assumed + # and events should be fired on value update + assumed_state: bool = False + + +@dataclass +class BaseDiscoverySchemaDataTemplate: + """Base class for discovery schema data templates.""" + + static_data: Any | None = None + + def resolve_data(self, value: ZwaveValue) -> Any: + """Resolve helper class data for a discovered value. + + Can optionally be implemented by subclasses if input data needs to be + transformed once discovered Value is available. + """ + return {} + + def values_to_watch(self, resolved_data: Any) -> Iterable[ZwaveValue | None]: + """Return list of all ZwaveValues resolved by helper that should be watched. + + Should be implemented by subclasses only if there are values to watch. + """ + return [] + + def value_ids_to_watch(self, resolved_data: Any) -> set[str]: + """Return list of all Value IDs resolved by helper that should be watched. + + Not to be overwritten by subclasses. + """ + return {val.value_id for val in self.values_to_watch(resolved_data) if val} + + @staticmethod + def _get_value_from_id( + node: ZwaveNode, value_id_obj: ZwaveValueID + ) -> ZwaveValue | ZwaveConfigurationValue | None: + """Get a ZwaveValue from a node using a ZwaveValueDict.""" + value_id = get_value_id_str( + node, + value_id_obj.command_class, + value_id_obj.property_, + endpoint=value_id_obj.endpoint, + property_key=value_id_obj.property_key, + ) + return node.values.get(value_id)