Handle invalid scale for zwave_js multilevel/meter sensors (#101173)

* Handle invalid scale for zwave_js multilevel/meter sensors

* Remove logging statement
This commit is contained in:
Raman Gupta 2023-10-04 22:55:18 -04:00 committed by GitHub
parent c951c03447
commit 383c63000e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 95 additions and 25 deletions

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Iterable, Mapping from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field from dataclasses import dataclass, field
import logging import logging
from typing import Any, cast from typing import Any, TypeVar, cast
from zwave_js_server.const import CommandClass from zwave_js_server.const import CommandClass
from zwave_js_server.const.command_class.energy_production import ( from zwave_js_server.const.command_class.energy_production import (
@ -87,6 +87,7 @@ from zwave_js_server.const.command_class.multilevel_sensor import (
MultilevelSensorScaleType, MultilevelSensorScaleType,
MultilevelSensorType, MultilevelSensorType,
) )
from zwave_js_server.exceptions import UnknownValueData
from zwave_js_server.model.node import Node as ZwaveNode from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.value import ( from zwave_js_server.model.value import (
ConfigurationValue as ZwaveConfigurationValue, ConfigurationValue as ZwaveConfigurationValue,
@ -355,24 +356,22 @@ class NumericSensorDataTemplateData:
unit_of_measurement: str | None = None unit_of_measurement: str | None = None
T = TypeVar(
"T",
MultilevelSensorType,
MultilevelSensorScaleType,
MeterScaleType,
EnergyProductionParameter,
EnergyProductionScaleType,
)
class NumericSensorDataTemplate(BaseDiscoverySchemaDataTemplate): class NumericSensorDataTemplate(BaseDiscoverySchemaDataTemplate):
"""Data template class for Z-Wave Sensor entities.""" """Data template class for Z-Wave Sensor entities."""
@staticmethod @staticmethod
def find_key_from_matching_set( def find_key_from_matching_set(
enum_value: MultilevelSensorType enum_value: T, set_map: Mapping[str, list[T]]
| MultilevelSensorScaleType
| MeterScaleType
| EnergyProductionParameter
| EnergyProductionScaleType,
set_map: Mapping[
str,
list[MultilevelSensorType]
| list[MultilevelSensorScaleType]
| list[MeterScaleType]
| list[EnergyProductionScaleType]
| list[EnergyProductionParameter],
],
) -> str | None: ) -> str | None:
"""Find a key in a set map that matches a given enum value.""" """Find a key in a set map that matches a given enum value."""
for key, value_set in set_map.items(): for key, value_set in set_map.items():
@ -393,7 +392,11 @@ class NumericSensorDataTemplate(BaseDiscoverySchemaDataTemplate):
return NumericSensorDataTemplateData(ENTITY_DESC_KEY_BATTERY, PERCENTAGE) return NumericSensorDataTemplateData(ENTITY_DESC_KEY_BATTERY, PERCENTAGE)
if value.command_class == CommandClass.METER: if value.command_class == CommandClass.METER:
meter_scale_type = get_meter_scale_type(value) try:
meter_scale_type = get_meter_scale_type(value)
except UnknownValueData:
return NumericSensorDataTemplateData()
unit = self.find_key_from_matching_set(meter_scale_type, METER_UNIT_MAP) unit = self.find_key_from_matching_set(meter_scale_type, METER_UNIT_MAP)
# We do this because even though these are energy scales, they don't meet # We do this because even though these are energy scales, they don't meet
# the unit requirements for the energy device class. # the unit requirements for the energy device class.
@ -418,8 +421,11 @@ class NumericSensorDataTemplate(BaseDiscoverySchemaDataTemplate):
) )
if value.command_class == CommandClass.SENSOR_MULTILEVEL: if value.command_class == CommandClass.SENSOR_MULTILEVEL:
sensor_type = get_multilevel_sensor_type(value) try:
multilevel_sensor_scale_type = get_multilevel_sensor_scale_type(value) sensor_type = get_multilevel_sensor_type(value)
multilevel_sensor_scale_type = get_multilevel_sensor_scale_type(value)
except UnknownValueData:
return NumericSensorDataTemplateData()
unit = self.find_key_from_matching_set( unit = self.find_key_from_matching_set(
multilevel_sensor_scale_type, MULTILEVEL_SENSOR_UNIT_MAP multilevel_sensor_scale_type, MULTILEVEL_SENSOR_UNIT_MAP
) )

View File

@ -130,6 +130,40 @@ async def test_numeric_sensor(
assert state.state == "0" assert state.state == "0"
async def test_invalid_multilevel_sensor_scale(
hass: HomeAssistant, client, multisensor_6_state, integration
) -> None:
"""Test a multilevel sensor with an invalid scale."""
node_state = copy.deepcopy(multisensor_6_state)
value = next(
value
for value in node_state["values"]
if value["commandClass"] == 49 and value["property"] == "Air temperature"
)
value["metadata"]["ccSpecific"]["scale"] = -1
value["metadata"]["unit"] = None
event = Event(
"node added",
{
"source": "controller",
"event": "node added",
"node": node_state,
"result": "",
},
)
client.driver.controller.receive_event(event)
await hass.async_block_till_done()
state = hass.states.get(AIR_TEMPERATURE_SENSOR)
assert state
assert state.state == "9.0"
assert ATTR_UNIT_OF_MEASUREMENT not in state.attributes
assert ATTR_DEVICE_CLASS not in state.attributes
assert ATTR_STATE_CLASS not in state.attributes
async def test_energy_sensors( async def test_energy_sensors(
hass: HomeAssistant, hank_binary_switch, integration hass: HomeAssistant, hank_binary_switch, integration
) -> None: ) -> None:
@ -424,10 +458,7 @@ async def test_node_status_sensor_not_ready(
async def test_reset_meter( async def test_reset_meter(
hass: HomeAssistant, hass: HomeAssistant, client, aeon_smart_switch_6, integration
client,
aeon_smart_switch_6,
integration,
) -> None: ) -> None:
"""Test reset_meter service.""" """Test reset_meter service."""
client.async_send_command.return_value = {} client.async_send_command.return_value = {}
@ -487,10 +518,7 @@ async def test_reset_meter(
async def test_meter_attributes( async def test_meter_attributes(
hass: HomeAssistant, hass: HomeAssistant, client, aeon_smart_switch_6, integration
client,
aeon_smart_switch_6,
integration,
) -> None: ) -> None:
"""Test meter entity attributes.""" """Test meter entity attributes."""
state = hass.states.get(METER_ENERGY_SENSOR) state = hass.states.get(METER_ENERGY_SENSOR)
@ -501,6 +529,42 @@ async def test_meter_attributes(
assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL_INCREASING assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL_INCREASING
async def test_invalid_meter_scale(
hass: HomeAssistant, client, aeon_smart_switch_6_state, integration
) -> None:
"""Test a meter sensor with an invalid scale."""
node_state = copy.deepcopy(aeon_smart_switch_6_state)
value = next(
value
for value in node_state["values"]
if value["commandClass"] == 50
and value["property"] == "value"
and value["propertyKey"] == 65537
)
value["metadata"]["ccSpecific"]["scale"] = -1
value["metadata"]["unit"] = None
event = Event(
"node added",
{
"source": "controller",
"event": "node added",
"node": node_state,
"result": "",
},
)
client.driver.controller.receive_event(event)
await hass.async_block_till_done()
state = hass.states.get(METER_ENERGY_SENSOR)
assert state
assert state.attributes[ATTR_METER_TYPE] == MeterType.ELECTRIC.value
assert state.attributes[ATTR_METER_TYPE_NAME] == MeterType.ELECTRIC.name
assert ATTR_DEVICE_CLASS not in state.attributes
assert ATTR_STATE_CLASS not in state.attributes
assert ATTR_UNIT_OF_MEASUREMENT not in state.attributes
async def test_special_meters( async def test_special_meters(
hass: HomeAssistant, aeon_smart_switch_6_state, client, integration hass: HomeAssistant, aeon_smart_switch_6_state, client, integration
) -> None: ) -> None: