Refactor zwave_js discovery schema foundation (#151146)

This commit is contained in:
Martin Hjelmare
2025-08-27 12:01:34 +02:00
committed by GitHub
parent 20e4d37cc6
commit 81a5b4a684
9 changed files with 489 additions and 239 deletions

View File

@@ -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

View File

@@ -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,
),
]

View File

@@ -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

View File

@@ -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,7 +1380,25 @@ def async_discover_single_value(
)
# all checks passed, this value belongs to an entity
yield ZwaveDiscoveryInfo(
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,
@@ -1471,6 +1411,8 @@ def async_discover_single_value(
entity_category=schema.entity_category,
)
yield discovery_info
# prevent re-discovery of the (primary) value if not allowed
if not schema.allow_multi:
discovered_value_ids[device.id].add(value.value_id)
@@ -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

View File

@@ -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."""

View File

@@ -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)},

View File

@@ -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."""

View File

@@ -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)

View File

@@ -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)