Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 16:36:08 +01:00
Code Activity

Refactor Z-Wave discovery schemas for sensor platform (#165254)

Browse Source 
Co-authored-by: AlCalzone <d.griesel@gmx.net>
This commit is contained in:
Martin Hjelmare
2026-03-23 11:43:30 +01:00
committed by GitHub
parent a2c64f65e1
commit e214ce690a
8 changed files with 472 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/zwave_js/const.py
View File

@@ -142,7 +142,6 @@ ATTR_TWIST_ASSIST = "twist_assist"
ADDON_SLUG = "core_zwave_js"
# Sensor entity description constants
ENTITY_DESC_KEY_BATTERY_LEVEL = "battery_level"
ENTITY_DESC_KEY_BATTERY_LIST_STATE = "battery_list_state"
ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY = "battery_maximum_capacity"
ENTITY_DESC_KEY_BATTERY_TEMPERATURE = "battery_temperature"
@@ -158,13 +157,11 @@ ENTITY_DESC_KEY_HUMIDITY = "humidity"
ENTITY_DESC_KEY_ILLUMINANCE = "illuminance"
ENTITY_DESC_KEY_PRESSURE = "pressure"
ENTITY_DESC_KEY_SIGNAL_STRENGTH = "signal_strength"
ENTITY_DESC_KEY_TEMPERATURE = "temperature"
ENTITY_DESC_KEY_TARGET_TEMPERATURE = "target_temperature"
ENTITY_DESC_KEY_UV_INDEX = "uv_index"
ENTITY_DESC_KEY_MEASUREMENT = "measurement"
ENTITY_DESC_KEY_TOTAL_INCREASING = "total_increasing"
ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER = "energy_production_power"
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME = "energy_production_time"
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL = "energy_production_total"
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY = "energy_production_today"

19
homeassistant/components/zwave_js/discovery.py
View File

@@ -75,10 +75,12 @@ from .models import (
ZWaveValueDiscoverySchema,
ZwaveValueID,
)
from .sensor import DISCOVERY_SCHEMAS as SENSOR_SCHEMAS
NEW_DISCOVERY_SCHEMAS: dict[Platform, list[NewZWaveDiscoverySchema]] = {
Platform.BINARY_SENSOR: BINARY_SENSOR_SCHEMAS,
Platform.EVENT: EVENT_SCHEMAS,
Platform.SENSOR: SENSOR_SCHEMAS,
}
SUPPORTED_PLATFORMS = tuple(NEW_DISCOVERY_SCHEMAS)
@@ -878,7 +880,7 @@ DISCOVERY_SCHEMAS = [
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.BATTERY},
type={ValueType.NUMBER},
property={"level", "maximumCapacity"},
property={"maximumCapacity"},
),
data_template=NumericSensorDataTemplate(),
),
@@ -1564,10 +1566,17 @@ def check_value(
):
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(
if schema.all_available_cc_specific is not None and (
value.metadata.cc_specific is None
or not all(
key in value.metadata.cc_specific and value.metadata.cc_specific[key] == val
for key, val in schema.all_available_cc_specific
)
):
return False
if schema.any_available_cc_specific is not None and (
value.metadata.cc_specific is None
or not any(
key in value.metadata.cc_specific and value.metadata.cc_specific[key] == val
for key, val in schema.any_available_cc_specific
)

10
homeassistant/components/zwave_js/discovery_data_template.py
View File

@@ -43,7 +43,6 @@ from zwave_js_server.const.command_class.multilevel_sensor import (
POWER_SENSORS,
PRESSURE_SENSORS,
SIGNAL_STRENGTH_SENSORS,
TEMPERATURE_SENSORS,
UNIT_A_WEIGHTED_DECIBELS,
UNIT_AMPERE as SENSOR_UNIT_AMPERE,
UNIT_BTU_H,
@@ -131,7 +130,6 @@ from homeassistant.const import (
)
from .const import (
ENTITY_DESC_KEY_BATTERY_LEVEL,
ENTITY_DESC_KEY_BATTERY_LIST_STATE,
ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY,
ENTITY_DESC_KEY_BATTERY_TEMPERATURE,
@@ -139,7 +137,6 @@ from .const import (
ENTITY_DESC_KEY_CO2,
ENTITY_DESC_KEY_CURRENT,
ENTITY_DESC_KEY_ENERGY_MEASUREMENT,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
@@ -152,7 +149,6 @@ from .const import (
ENTITY_DESC_KEY_PRESSURE,
ENTITY_DESC_KEY_SIGNAL_STRENGTH,
ENTITY_DESC_KEY_TARGET_TEMPERATURE,
ENTITY_DESC_KEY_TEMPERATURE,
ENTITY_DESC_KEY_TOTAL_INCREASING,
ENTITY_DESC_KEY_UV_INDEX,
ENTITY_DESC_KEY_VOLTAGE,
@@ -167,7 +163,6 @@ ENERGY_PRODUCTION_DEVICE_CLASS_MAP: dict[str, list[EnergyProductionParameter]] =
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL: [
EnergyProductionParameter.TOTAL_PRODUCTION
],
ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER: [EnergyProductionParameter.POWER],
}
@@ -189,7 +184,6 @@ MULTILEVEL_SENSOR_DEVICE_CLASS_MAP: dict[str, list[MultilevelSensorType]] = {
ENTITY_DESC_KEY_POWER: POWER_SENSORS,
ENTITY_DESC_KEY_PRESSURE: PRESSURE_SENSORS,
ENTITY_DESC_KEY_SIGNAL_STRENGTH: SIGNAL_STRENGTH_SENSORS,
ENTITY_DESC_KEY_TEMPERATURE: TEMPERATURE_SENSORS,
ENTITY_DESC_KEY_VOLTAGE: VOLTAGE_SENSORS,
ENTITY_DESC_KEY_UV_INDEX: [MultilevelSensorType.ULTRAVIOLET],
}
@@ -338,10 +332,6 @@ class NumericSensorDataTemplate(BaseDiscoverySchemaDataTemplate):
def resolve_data(self, value: ZwaveValue) -> NumericSensorDataTemplateData:
"""Resolve helper class data for a discovered value."""
if value.command_class == CommandClass.BATTERY and value.property_ == "level":
return NumericSensorDataTemplateData(
ENTITY_DESC_KEY_BATTERY_LEVEL, PERCENTAGE
)
if value.command_class == CommandClass.BATTERY and value.property_ in (
"chargingStatus",
"rechargeOrReplace",

2
homeassistant/components/zwave_js/models.py
View File

@@ -149,6 +149,8 @@ class ZWaveValueDiscoverySchema(DataclassMustHaveAtLeastOne):
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 ALL of these key/value pairs
all_available_cc_specific: set[tuple[Any, Any]] | 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

290
homeassistant/components/zwave_js/sensor.py
View File

@@ -4,21 +4,53 @@ from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from enum import IntEnum
from typing import Any, cast
import voluptuous as vol
from zwave_js_server.const import CommandClass, RssiError
from zwave_js_server.const.command_class.energy_production import (
CC_SPECIFIC_PARAMETER,
CC_SPECIFIC_SCALE as ENERGY_PRODUCTION_CC_SPECIFIC_SCALE,
EnergyProductionParameter,
PowerScale,
)
from zwave_js_server.const.command_class.meter import (
CC_SPECIFIC_METER_TYPE,
CC_SPECIFIC_SCALE as METER_CC_SPECIFIC_SCALE,
RESET_METER_OPTION_TARGET_VALUE,
RESET_METER_OPTION_TYPE,
VALUE_PROPERTY,
ElectricScale,
MeterType,
)
from zwave_js_server.const.command_class.multilevel_sensor import (
CC_SPECIFIC_SCALE as MULTILEVEL_SENSOR_CC_SPECIFIC_SCALE,
CC_SPECIFIC_SENSOR_TYPE,
TEMPERATURE_SENSORS,
TemperatureScale,
)
from zwave_js_server.exceptions import (
BaseZwaveJSServerError,
RssiErrorReceived,
UnknownValueData,
)
from zwave_js_server.exceptions import BaseZwaveJSServerError, RssiErrorReceived
from zwave_js_server.model.controller import Controller
from zwave_js_server.model.controller.statistics import ControllerStatistics
from zwave_js_server.model.driver import Driver
from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.node.statistics import NodeStatistics
from zwave_js_server.util.command_class.meter import get_meter_type
from zwave_js_server.model.value import Value as ZwaveValue
from zwave_js_server.util.command_class.energy_production import (
get_energy_production_scale_type,
)
from zwave_js_server.util.command_class.meter import (
get_meter_scale_type,
get_meter_type,
)
from zwave_js_server.util.command_class.multilevel_sensor import (
get_multilevel_sensor_scale_type,
)
from homeassistant.components.sensor import (
DOMAIN as SENSOR_DOMAIN,
@@ -27,6 +59,7 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONCENTRATION_PARTS_PER_MILLION,
LIGHT_LUX,
@@ -34,6 +67,7 @@ from homeassistant.const import (
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
UV_INDEX,
EntityCategory,
Platform,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
@@ -55,7 +89,6 @@ from .const import (
ATTR_METER_TYPE_NAME,
ATTR_VALUE,
DOMAIN,
ENTITY_DESC_KEY_BATTERY_LEVEL,
ENTITY_DESC_KEY_BATTERY_LIST_STATE,
ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY,
ENTITY_DESC_KEY_BATTERY_TEMPERATURE,
@@ -63,7 +96,6 @@ from .const import (
ENTITY_DESC_KEY_CO2,
ENTITY_DESC_KEY_CURRENT,
ENTITY_DESC_KEY_ENERGY_MEASUREMENT,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
@@ -76,35 +108,32 @@ from .const import (
ENTITY_DESC_KEY_PRESSURE,
ENTITY_DESC_KEY_SIGNAL_STRENGTH,
ENTITY_DESC_KEY_TARGET_TEMPERATURE,
ENTITY_DESC_KEY_TEMPERATURE,
ENTITY_DESC_KEY_TOTAL_INCREASING,
ENTITY_DESC_KEY_UV_INDEX,
ENTITY_DESC_KEY_VOLTAGE,
LOGGER,
SERVICE_RESET_METER,
)
from .discovery import ZwaveDiscoveryInfo
from .discovery_data_template import (
NumericSensorDataTemplate,
NumericSensorDataTemplateData,
)
from .entity import ZWaveBaseEntity
from .entity import NewZwaveDiscoveryInfo, ZWaveBaseEntity
from .helpers import get_device_info, get_valueless_base_unique_id
from .migrate import async_migrate_statistics_sensors
from .models import ZwaveJSConfigEntry
from .models import (
NewZWaveDiscoverySchema,
ValueType,
ZwaveDiscoveryInfo,
ZwaveJSConfigEntry,
ZWaveValueDiscoverySchema,
)
PARALLEL_UPDATES = 0
# These descriptions should have a non None unit of measurement.
ENTITY_DESCRIPTION_KEY_UNIT_MAP: dict[tuple[str, str], SensorEntityDescription] = {
(ENTITY_DESC_KEY_BATTERY_LEVEL, PERCENTAGE): SensorEntityDescription(
key=ENTITY_DESC_KEY_BATTERY_LEVEL,
device_class=SensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
),
(ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY, PERCENTAGE): SensorEntityDescription(
key=ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY,
entity_category=EntityCategory.DIAGNOSTIC,
@@ -224,21 +253,6 @@ ENTITY_DESCRIPTION_KEY_UNIT_MAP: dict[tuple[str, str], SensorEntityDescription]
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
),
(ENTITY_DESC_KEY_TEMPERATURE, UnitOfTemperature.CELSIUS): SensorEntityDescription(
key=ENTITY_DESC_KEY_TEMPERATURE,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
(
ENTITY_DESC_KEY_TEMPERATURE,
UnitOfTemperature.FAHRENHEIT,
): SensorEntityDescription(
key=ENTITY_DESC_KEY_TEMPERATURE,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
),
(
ENTITY_DESC_KEY_TARGET_TEMPERATURE,
UnitOfTemperature.CELSIUS,
@@ -289,16 +303,6 @@ ENTITY_DESCRIPTION_KEY_UNIT_MAP: dict[tuple[str, str], SensorEntityDescription]
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
),
(
ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
UnitOfPower.WATT,
): SensorEntityDescription(
key=ENTITY_DESC_KEY_POWER,
name="Energy production power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
),
}
# These descriptions are without unit of measurement.
@@ -601,7 +605,7 @@ async def async_setup_entry(
assert driver is not None # Driver is ready before platforms are loaded.
@callback
def async_add_sensor(info: ZwaveDiscoveryInfo) -> None:
def async_add_sensor(info: ZwaveDiscoveryInfo | NewZwaveDiscoveryInfo) -> None:
"""Add Z-Wave Sensor."""
entities: list[ZWaveBaseEntity] = []
@@ -612,7 +616,13 @@ async def async_setup_entry(
entity_description = get_entity_description(data)
if info.platform_hint == "numeric_sensor":
if isinstance(info, NewZwaveDiscoveryInfo) and (
entity_class := info.entity_class
) in (NewZWaveNumericSensor, NewZWaveMeterSensor):
entities.append(entity_class(config_entry, driver, info))
elif isinstance(info, NewZwaveDiscoveryInfo):
pass # other entity classes are not migrated yet
elif info.platform_hint == "numeric_sensor":
entities.append(
ZWaveNumericSensor(
config_entry,
@@ -802,6 +812,62 @@ class ZWaveNumericSensor(ZwaveSensor):
return float(self.info.primary_value.value)
class NewZWaveNumericSensor(ZWaveBaseEntity, SensorEntity):
"""Representation of a Z-Wave Numeric sensor."""
_attr_force_update = True
def __init__(
self,
config_entry: ConfigEntry,
driver: Driver,
info: NewZwaveDiscoveryInfo,
) -> None:
"""Initialize the entity."""
super().__init__(config_entry, driver, info)
entity_description = info.entity_description
if not entity_description.name or entity_description.name is UNDEFINED:
self._attr_name = self.generate_name(include_value_name=True)
self._scale_type = self._get_scale_type()
def _get_scale_type(self) -> IntEnum | None:
"""Return the scale type of the value."""
primary_value = self.info.primary_value
scale_type_function: Callable[[ZwaveValue], IntEnum] | None
match primary_value.command_class:
case CommandClass.METER:
scale_type_function = get_meter_scale_type
case CommandClass.SENSOR_MULTILEVEL:
scale_type_function = get_multilevel_sensor_scale_type
case CommandClass.ENERGY_PRODUCTION:
scale_type_function = get_energy_production_scale_type
case _:
scale_type_function = None
if scale_type_function is None:
return None
try:
scale_type = scale_type_function(primary_value)
except UnknownValueData:
return None
return scale_type
@callback
def on_value_update(self) -> None:
"""Handle scale changes for this value on value updated event."""
# TODO: Try to limit this to metadata updated event. # pylint: disable=fixme
scale_type = self._get_scale_type()
if scale_type is not self._scale_type:
self.hass.config_entries.async_schedule_reload(self.config_entry.entry_id)
@property
def native_value(self) -> float | None:
"""Return state of the sensor."""
if self.info.primary_value.value is None:
return None
return float(self.info.primary_value.value)
class ZWaveMeterSensor(ZWaveNumericSensor):
"""Representation of a Z-Wave Meter CC sensor."""
@@ -842,6 +908,46 @@ class ZWaveMeterSensor(ZWaveNumericSensor):
)
class NewZWaveMeterSensor(NewZWaveNumericSensor):
"""Representation of a Z-Wave Meter CC sensor."""
@property
def extra_state_attributes(self) -> Mapping[str, int | str] | None:
"""Return extra state attributes."""
meter_type = get_meter_type(self.info.primary_value)
return {
ATTR_METER_TYPE: meter_type.value,
ATTR_METER_TYPE_NAME: meter_type.name,
}
async def async_reset_meter(
self, meter_type: int | None = None, value: int | None = None
) -> None:
"""Reset meter(s) on device."""
node = self.info.node
endpoint = self.info.primary_value.endpoint or 0
options = {}
if meter_type is not None:
options[RESET_METER_OPTION_TYPE] = meter_type
if value is not None:
options[RESET_METER_OPTION_TARGET_VALUE] = value
args = [options] if options else []
try:
await node.endpoints[endpoint].async_invoke_cc_api(
CommandClass.METER, "reset", *args, wait_for_result=False
)
except BaseZwaveJSServerError as err:
raise HomeAssistantError(
f"Failed to reset meters on node {node} endpoint {endpoint}: {err}"
) from err
LOGGER.debug(
"Meters on node %s endpoint %s reset with the following options: %s",
node,
endpoint,
options,
)
class ZWaveListSensor(ZwaveSensor):
"""Representation of a Z-Wave Numeric sensor with multiple states."""
@@ -1135,3 +1241,103 @@ class ZWaveStatisticsSensor(SensorEntity):
# Set initial state
self._set_statistics(self.statistics_src.statistics)
DISCOVERY_SCHEMAS: list[NewZWaveDiscoverySchema] = [
NewZWaveDiscoverySchema(
platform=Platform.SENSOR,
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.BATTERY},
type={ValueType.NUMBER},
property={"level"},
),
entity_class=NewZWaveNumericSensor,
entity_description=SensorEntityDescription(
key="battery_level",
device_class=SensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
),
),
NewZWaveDiscoverySchema(
platform=Platform.SENSOR,
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.SENSOR_MULTILEVEL},
type={ValueType.NUMBER},
all_available_cc_specific={
(MULTILEVEL_SENSOR_CC_SPECIFIC_SCALE, TemperatureScale.CELSIUS),
},
any_available_cc_specific={
(CC_SPECIFIC_SENSOR_TYPE, sensor_type)
for sensor_type in TEMPERATURE_SENSORS
},
),
entity_class=NewZWaveNumericSensor,
entity_description=SensorEntityDescription(
key="temperature_celsius",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
),
NewZWaveDiscoverySchema(
platform=Platform.SENSOR,
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.SENSOR_MULTILEVEL},
type={ValueType.NUMBER},
all_available_cc_specific={
(MULTILEVEL_SENSOR_CC_SPECIFIC_SCALE, TemperatureScale.FAHRENHEIT),
},
any_available_cc_specific={
(CC_SPECIFIC_SENSOR_TYPE, sensor_type)
for sensor_type in TEMPERATURE_SENSORS
},
),
entity_class=NewZWaveNumericSensor,
entity_description=SensorEntityDescription(
key="temperature_fahrenheit",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
),
),
NewZWaveDiscoverySchema(
platform=Platform.SENSOR,
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.METER},
type={ValueType.NUMBER},
property={VALUE_PROPERTY},
all_available_cc_specific={
(CC_SPECIFIC_METER_TYPE, MeterType.ELECTRIC),
(METER_CC_SPECIFIC_SCALE, ElectricScale.AMPERE),
},
),
entity_class=NewZWaveMeterSensor,
entity_description=SensorEntityDescription(
key="meter_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
),
),
NewZWaveDiscoverySchema(
platform=Platform.SENSOR,
primary_value=ZWaveValueDiscoverySchema(
command_class={CommandClass.ENERGY_PRODUCTION},
type={ValueType.NUMBER},
all_available_cc_specific={
(CC_SPECIFIC_PARAMETER, EnergyProductionParameter.POWER),
(ENERGY_PRODUCTION_CC_SPECIFIC_SCALE, PowerScale.WATTS),
},
),
entity_class=NewZWaveNumericSensor,
entity_description=SensorEntityDescription(
key="energy_production_power",
name="Energy production power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
),
),
]

1
tests/components/zwave_js/common.py
View File

@@ -41,7 +41,6 @@ EATON_RF9640_ENTITY = "light.allloaddimmer"
AEON_SMART_SWITCH_LIGHT_ENTITY = "light.smart_switch_6"
SCHLAGE_BE469_LOCK_ENTITY = "lock.touchscreen_deadbolt"
ZEN_31_ENTITY = "light.kitchen_under_cabinet_lights"
METER_ENERGY_SENSOR = "sensor.smart_switch_6_electric_consumed_kwh"
METER_VOLTAGE_SENSOR = "sensor.smart_switch_6_electric_consumed_v"
HUMIDIFIER_ADC_T3000_ENTITY = "humidifier.adc_t3000_humidifier"
DEHUMIDIFIER_ADC_T3000_ENTITY = "humidifier.adc_t3000_dehumidifier"

79
tests/components/zwave_js/test_discovery.py
View File

@@ -23,6 +23,7 @@ from homeassistant.components.zwave_js.discovery import (
FirmwareVersionRange,
ZWaveDiscoverySchema,
ZWaveValueDiscoverySchema,
check_value,
)
from homeassistant.components.zwave_js.discovery_data_template import (
DynamicCurrentTempClimateDataTemplate,
@@ -549,6 +550,84 @@ async def test_nabu_casa_zwa2(
)
def _make_mock_value(cc_specific: dict | None = None) -> MagicMock:
"""Create a base mock ZwaveValue for check_value tests."""
value = MagicMock()
value.command_class = 49
value.endpoint = 0
value.property_ = "Air temperature"
value.property_name = "Air temperature"
value.property_key = None
value.metadata.type = "number"
value.metadata.readable = True
value.metadata.writeable = False
value.metadata.states = None
value.metadata.cc_specific = cc_specific
value.metadata.stateful = None
value.value = 9
return value
def test_check_value_all_available_cc_specific_match() -> None:
"""Test check_value matches when all cc_specific key/value pairs are present."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
all_available_cc_specific={("scale", 0), ("sensorType", 1)},
)
value = _make_mock_value({"scale": 0, "sensorType": 1})
assert check_value(value, schema) is True
def test_check_value_all_available_cc_specific_partial_match() -> None:
"""Test check_value fails when not all cc_specific key/value pairs match."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
all_available_cc_specific={("scale", 0), ("sensorType", 1)},
)
value = _make_mock_value({"scale": 0, "sensorType": 5})
assert check_value(value, schema) is False
def test_check_value_all_available_cc_specific_none() -> None:
"""Test check_value fails when cc_specific is None and all_available is set."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
all_available_cc_specific={("scale", 0)},
)
value = _make_mock_value()
assert check_value(value, schema) is False
def test_check_value_any_available_cc_specific_match() -> None:
"""Test check_value matches when any cc_specific key/value pair is present."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
any_available_cc_specific={("sensorType", 1), ("sensorType", 3)},
)
value = _make_mock_value({"sensorType": 1, "scale": 0})
assert check_value(value, schema) is True
def test_check_value_any_available_cc_specific_no_match() -> None:
"""Test check_value fails when no cc_specific key/value pair matches."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
any_available_cc_specific={("sensorType", 1), ("sensorType", 3)},
)
value = _make_mock_value({"sensorType": 5, "scale": 0})
assert check_value(value, schema) is False
def test_check_value_any_available_cc_specific_none() -> None:
"""Test check_value fails when cc_specific is None and any_available is set."""
schema = ZWaveValueDiscoverySchema(
command_class={49},
any_available_cc_specific={("sensorType", 1)},
)
value = _make_mock_value()
assert check_value(value, schema) is False
async def test_nabu_casa_zwa2_legacy(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,

141
tests/components/zwave_js/test_sensor.py
View File

@@ -2,6 +2,7 @@
import copy
from datetime import timedelta
from unittest.mock import patch
import pytest
from zwave_js_server.const.command_class.meter import MeterType
@@ -57,7 +58,6 @@ from .common import (
CURRENT_SENSOR,
ENERGY_SENSOR,
HUMIDITY_SENSOR,
METER_ENERGY_SENSOR,
POWER_SENSOR,
VOLTAGE_SENSOR,
)
@@ -509,8 +509,19 @@ async def test_node_status_sensor_not_ready(
assert "There is no value to refresh for this entity" in caplog.text
@pytest.mark.parametrize(
"entity_id",
[
"sensor.smart_switch_6_electric_consumed_kwh",
"sensor.smart_switch_6_electric_consumed_a",
],
)
async def test_reset_meter(
hass: HomeAssistant, client, aeon_smart_switch_6, integration
hass: HomeAssistant,
client,
aeon_smart_switch_6,
integration,
entity_id: str,
) -> None:
"""Test reset_meter service."""
client.async_send_command.return_value = {}
@@ -521,7 +532,7 @@ async def test_reset_meter(
DOMAIN,
SERVICE_RESET_METER,
{
ATTR_ENTITY_ID: METER_ENERGY_SENSOR,
ATTR_ENTITY_ID: entity_id,
},
blocking=True,
)
@@ -540,7 +551,7 @@ async def test_reset_meter(
DOMAIN,
SERVICE_RESET_METER,
{
ATTR_ENTITY_ID: METER_ENERGY_SENSOR,
ATTR_ENTITY_ID: entity_id,
ATTR_METER_TYPE: 1,
ATTR_VALUE: 2,
},
@@ -564,7 +575,7 @@ async def test_reset_meter(
await hass.services.async_call(
DOMAIN,
SERVICE_RESET_METER,
{ATTR_ENTITY_ID: METER_ENERGY_SENSOR},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
@@ -574,20 +585,57 @@ async def test_reset_meter(
)
@pytest.mark.parametrize(
("entity_id", "device_class", "state_class", "unit_of_measurement"),
[
(
"sensor.smart_switch_6_electric_consumed_kwh",
SensorDeviceClass.ENERGY,
SensorStateClass.TOTAL_INCREASING,
UnitOfEnergy.KILO_WATT_HOUR,
),
(
"sensor.smart_switch_6_electric_consumed_a",
SensorDeviceClass.CURRENT,
SensorStateClass.MEASUREMENT,
UnitOfElectricCurrent.AMPERE,
),
],
)
async def test_meter_attributes(
hass: HomeAssistant, client, aeon_smart_switch_6, integration
hass: HomeAssistant,
client,
aeon_smart_switch_6,
integration,
entity_id: str,
device_class: SensorDeviceClass,
state_class: SensorStateClass,
unit_of_measurement: str,
) -> None:
"""Test meter entity attributes."""
state = hass.states.get(METER_ENERGY_SENSOR)
state = hass.states.get(entity_id)
assert state
assert state.attributes[ATTR_METER_TYPE] == MeterType.ELECTRIC.value
assert state.attributes[ATTR_METER_TYPE_NAME] == MeterType.ELECTRIC.name
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.ENERGY
assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL_INCREASING
assert state.attributes[ATTR_DEVICE_CLASS] == device_class
assert state.attributes[ATTR_STATE_CLASS] is state_class
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == unit_of_measurement
@pytest.mark.parametrize(
("entity_id", "property_key_name"),
[
("sensor.smart_switch_6_electric_consumed_kwh", "Electric_kWh_Consumed"),
("sensor.smart_switch_6_electric_consumed_a", "Electric_A_Consumed"),
],
)
async def test_invalid_meter_scale(
hass: HomeAssistant, client, aeon_smart_switch_6_state, integration
hass: HomeAssistant,
client,
aeon_smart_switch_6_state,
integration,
entity_id: str,
property_key_name: str,
) -> None:
"""Test a meter sensor with an invalid scale."""
node_state = copy.deepcopy(aeon_smart_switch_6_state)
@@ -596,7 +644,7 @@ async def test_invalid_meter_scale(
for value in node_state["values"]
if value["commandClass"] == 50
and value["property"] == "value"
and value["propertyKey"] == 65537
and value["propertyKeyName"] == property_key_name
)
value["metadata"]["ccSpecific"]["scale"] = -1
value["metadata"]["unit"] = None
@@ -613,7 +661,7 @@ async def test_invalid_meter_scale(
client.driver.controller.receive_event(event)
await hass.async_block_till_done()
state = hass.states.get(METER_ENERGY_SENSOR)
state = hass.states.get(entity_id)
assert state
assert state.attributes[ATTR_METER_TYPE] == MeterType.ELECTRIC.value
assert state.attributes[ATTR_METER_TYPE_NAME] == MeterType.ELECTRIC.name
@@ -778,6 +826,75 @@ async def test_unit_change(hass: HomeAssistant, zp3111, client, integration) ->
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE
async def test_new_sensor_invalid_scale(
hass: HomeAssistant, multisensor_6, client, integration
) -> None:
"""Test new-style numeric sensor handles UnknownValueData from invalid scale."""
entity_id = AIR_TEMPERATURE_SENSOR
state = hass.states.get(entity_id)
assert state
assert state.state == "9.0"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfTemperature.CELSIUS
# Update the metadata to an invalid scale (255) to trigger UnknownValueData
# in _get_scale_type on the next value update
event = Event(
"metadata updated",
{
"source": "node",
"event": "metadata updated",
"nodeId": multisensor_6.node_id,
"args": {
"commandClassName": "Multilevel Sensor",
"commandClass": 49,
"endpoint": 0,
"property": "Air temperature",
"metadata": {
"type": "number",
"readable": True,
"writeable": False,
"label": "Air temperature",
"ccSpecific": {"sensorType": 1, "scale": 255},
"unit": None,
},
"propertyName": "Air temperature",
"nodeId": multisensor_6.node_id,
},
},
)
multisensor_6.receive_event(event)
await hass.async_block_till_done()
# Fire a value updated event to trigger on_value_update which calls
# _get_scale_type - the invalid scale should raise UnknownValueData
# which is caught and returns None, triggering a reload since
# None != original TemperatureScale.CELSIUS
with patch.object(
hass.config_entries, "async_schedule_reload"
) as mock_schedule_reload:
event = Event(
"value updated",
{
"source": "node",
"event": "value updated",
"nodeId": multisensor_6.node_id,
"args": {
"commandClassName": "Multilevel Sensor",
"commandClass": 49,
"endpoint": 0,
"property": "Air temperature",
"newValue": 68,
"prevValue": 9,
"propertyName": "Air temperature",
},
},
)
multisensor_6.receive_event(event)
await hass.async_block_till_done()
mock_schedule_reload.assert_called_once_with(integration.entry_id)
async def test_opening_state_sensor(
hass: HomeAssistant,
client,