evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

DSMR: Device/state classes, icons, less common disabled by default (#52159)

Browse source
This commit is contained in:
Franck Nijhof 2021-06-24 18:48:51 +02:00 • committed by GitHub
parent 4533a77597
commit 34a317b847
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 201 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

105
homeassistant/components/dsmr/const.py
View file

@ -5,6 +5,15 @@ import logging
from dsmr_parser import obis_references
from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT
from homeassistant.const import (
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_VOLTAGE,
)
from homeassistant.util import dt
from .models import DSMRSensor
DOMAIN = "dsmr"
@ -33,164 +42,242 @@ DATA_TASK = "task"
DEVICE_NAME_ENERGY = "Energy Meter"
DEVICE_NAME_GAS = "Gas Meter"
ICON_GAS = "mdi:fire"
ICON_POWER = "mdi:flash"
ICON_POWER_FAILURE = "mdi:flash-off"
ICON_SWELL_SAG = "mdi:pulse"
SENSORS: list[DSMRSensor] = [
DSMRSensor(
name="Power Consumption",
obis_reference=obis_references.CURRENT_ELECTRICITY_USAGE,
device_class=DEVICE_CLASS_POWER,
force_update=True,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Production",
obis_reference=obis_references.CURRENT_ELECTRICITY_DELIVERY,
device_class=DEVICE_CLASS_POWER,
force_update=True,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Tariff",
obis_reference=obis_references.ELECTRICITY_ACTIVE_TARIFF,
icon="mdi:flash",
),
DSMRSensor(
name="Energy Consumption (tarif 1)",
obis_reference=obis_references.ELECTRICITY_USED_TARIFF_1,
device_class=DEVICE_CLASS_ENERGY,
force_update=True,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Consumption (tarif 2)",
obis_reference=obis_references.ELECTRICITY_USED_TARIFF_2,
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Production (tarif 1)",
obis_reference=obis_references.ELECTRICITY_DELIVERED_TARIFF_1,
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Production (tarif 2)",
obis_reference=obis_references.ELECTRICITY_DELIVERED_TARIFF_2,
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Consumption Phase L1",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Consumption Phase L2",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Consumption Phase L3",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Production Phase L1",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Production Phase L2",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Power Production Phase L3",
obis_reference=obis_references.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE,
device_class=DEVICE_CLASS_POWER,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Short Power Failure Count",
obis_reference=obis_references.SHORT_POWER_FAILURE_COUNT,
entity_registry_enabled_default=False,
icon="mdi:flash-off",
),
DSMRSensor(
name="Long Power Failure Count",
obis_reference=obis_references.LONG_POWER_FAILURE_COUNT,
entity_registry_enabled_default=False,
icon="mdi:flash-off",
),
DSMRSensor(
name="Voltage Sags Phase L1",
obis_reference=obis_references.VOLTAGE_SAG_L1_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
name="Voltage Sags Phase L2",
obis_reference=obis_references.VOLTAGE_SAG_L2_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
name="Voltage Sags Phase L3",
obis_reference=obis_references.VOLTAGE_SAG_L3_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
name="Voltage Swells Phase L1",
obis_reference=obis_references.VOLTAGE_SWELL_L1_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
name="Voltage Swells Phase L2",
obis_reference=obis_references.VOLTAGE_SWELL_L2_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
name="Voltage Swells Phase L3",
obis_reference=obis_references.VOLTAGE_SWELL_L3_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
name="Voltage Phase L1",
obis_reference=obis_references.INSTANTANEOUS_VOLTAGE_L1,
device_class=DEVICE_CLASS_VOLTAGE,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Voltage Phase L2",
obis_reference=obis_references.INSTANTANEOUS_VOLTAGE_L2,
device_class=DEVICE_CLASS_VOLTAGE,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Voltage Phase L3",
obis_reference=obis_references.INSTANTANEOUS_VOLTAGE_L3,
device_class=DEVICE_CLASS_VOLTAGE,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Current Phase L1",
obis_reference=obis_references.INSTANTANEOUS_CURRENT_L1,
device_class=DEVICE_CLASS_CURRENT,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Current Phase L2",
obis_reference=obis_references.INSTANTANEOUS_CURRENT_L2,
device_class=DEVICE_CLASS_CURRENT,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Current Phase L3",
obis_reference=obis_references.INSTANTANEOUS_CURRENT_L3,
device_class=DEVICE_CLASS_CURRENT,
entity_registry_enabled_default=False,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Consumption (total)",
obis_reference=obis_references.LUXEMBOURG_ELECTRICITY_USED_TARIFF_GLOBAL,
dsmr_versions={"5L"},
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Production (total)",
obis_reference=obis_references.LUXEMBOURG_ELECTRICITY_DELIVERED_TARIFF_GLOBAL,
dsmr_versions={"5L"},
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Energy Consumption (total)",
obis_reference=obis_references.ELECTRICITY_IMPORTED_TOTAL,
dsmr_versions={"2.2", "4", "5", "5B"},
force_update=True,
device_class=DEVICE_CLASS_ENERGY,
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Gas Consumption",
obis_reference=obis_references.HOURLY_GAS_METER_READING,
dsmr_versions={"4", "5", "5L"},
force_update=True,
is_gas=True,
force_update=True,
icon="mdi:fire",
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Gas Consumption",
obis_reference=obis_references.BELGIUM_HOURLY_GAS_METER_READING,
dsmr_versions={"5B"},
force_update=True,
is_gas=True,
force_update=True,
icon="mdi:fire",
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
name="Gas Consumption",
obis_reference=obis_references.GAS_METER_READING,
dsmr_versions={"2.2"},
force_update=True,
is_gas=True,
force_update=True,
icon="mdi:fire",
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
]

6
homeassistant/components/dsmr/models.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
@dataclass
@ -11,6 +12,11 @@ class DSMRSensor:
name: str
obis_reference: str
device_class: str | None = None
dsmr_versions: set[str] | None = None
entity_registry_enabled_default: bool = True
force_update: bool = False
icon: str | None = None
is_gas: bool = False
last_reset: datetime | None = None
state_class: str | None = None

32
homeassistant/components/dsmr/sensor.py
View file

@ -39,10 +39,6 @@ from .const import (
DEVICE_NAME_ENERGY,
DEVICE_NAME_GAS,
DOMAIN,
ICON_GAS,
ICON_POWER,
ICON_POWER_FAILURE,
ICON_SWELL_SAG,
LOGGER,
SENSORS,
)
@ -196,13 +192,20 @@ class DSMREntity(SensorEntity):
device_serial = entry.data[CONF_SERIAL_ID_GAS]
device_name = DEVICE_NAME_GAS
self._attr_name = sensor.name
self._attr_force_update = sensor.force_update
self._attr_unique_id = f"{device_serial}_{sensor.name}".replace(" ", "_")
self._attr_device_class = sensor.device_class
self._attr_device_info = {
"identifiers": {(DOMAIN, device_serial)},
"name": device_name,
}
self._attr_entity_registry_enabled_default = (
sensor.entity_registry_enabled_default
)
self._attr_force_update = sensor.force_update
self._attr_icon = sensor.icon
self._attr_last_reset = sensor.last_reset
self._attr_name = sensor.name
self._attr_state_class = sensor.state_class
self._attr_unique_id = f"{device_serial}_{sensor.name}".replace(" ", "_")
@callback
def update_data(self, telegram: dict[str, DSMRObject]) -> None:
@ -221,21 +224,6 @@ class DSMREntity(SensorEntity):
dsmr_object = self.telegram[self._sensor.obis_reference]
return getattr(dsmr_object, attribute, None)
@property
def icon(self) -> str | None:
"""Icon to use in the frontend, if any."""
if not self.name:
return None
if "Sags" in self.name or "Swells" in self.name:
return ICON_SWELL_SAG
if "Failure" in self.name:
return ICON_POWER_FAILURE
if "Power" in self.name:
return ICON_POWER
if "Gas" in self.name:
return ICON_GAS
return None
@property
def state(self) -> StateType:
"""Return the state of sensor, if available, translate if needed."""

103
tests/components/dsmr/test_sensor.py
View file

@ -13,8 +13,22 @@ from unittest.mock import DEFAULT, MagicMock
from homeassistant import config_entries
from homeassistant.components.dsmr.const import DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import ENERGY_KILO_WATT_HOUR, VOLUME_CUBIC_METERS
from homeassistant.components.sensor import (
ATTR_LAST_RESET,
ATTR_STATE_CLASS,
DOMAIN as SENSOR_DOMAIN,
STATE_CLASS_MEASUREMENT,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
ENERGY_KILO_WATT_HOUR,
STATE_UNKNOWN,
VOLUME_CUBIC_METERS,
)
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
@ -118,8 +132,12 @@ async def test_default_setup(hass, dsmr_connection_fixture):
# make sure entities have been created and return 'unknown' state
power_consumption = hass.states.get("sensor.power_consumption")
assert power_consumption.state == "unknown"
assert power_consumption.attributes.get("unit_of_measurement") is None
assert power_consumption.state == STATE_UNKNOWN
assert power_consumption.attributes.get(ATTR_DEVICE_CLASS) == DEVICE_CLASS_POWER
assert power_consumption.attributes.get(ATTR_ICON) is None
assert power_consumption.attributes.get(ATTR_LAST_RESET) is None
assert power_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert power_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
@ -137,12 +155,22 @@ async def test_default_setup(hass, dsmr_connection_fixture):
# tariff should be translated in human readable and have no unit
power_tariff = hass.states.get("sensor.power_tariff")
assert power_tariff.state == "low"
assert power_tariff.attributes.get("unit_of_measurement") == ""
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) is None
assert power_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert power_tariff.attributes.get(ATTR_LAST_RESET) is None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ""
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) is None
assert gas_consumption.attributes.get(ATTR_ICON) == "mdi:fire"
assert gas_consumption.attributes.get(ATTR_LAST_RESET) is not None
assert gas_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == VOLUME_CUBIC_METERS
)
async def test_setup_only_energy(hass, dsmr_connection_fixture):
@ -226,12 +254,23 @@ async def test_v4_meter(hass, dsmr_connection_fixture):
# tariff should be translated in human readable and have no unit
power_tariff = hass.states.get("sensor.power_tariff")
assert power_tariff.state == "low"
assert power_tariff.attributes.get("unit_of_measurement") == ""
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) is None
assert power_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert power_tariff.attributes.get(ATTR_LAST_RESET) is None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ""
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) is None
assert gas_consumption.attributes.get(ATTR_ICON) == "mdi:fire"
assert gas_consumption.attributes.get(ATTR_LAST_RESET) is not None
assert gas_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == VOLUME_CUBIC_METERS
)
async def test_v5_meter(hass, dsmr_connection_fixture):
@ -286,12 +325,22 @@ async def test_v5_meter(hass, dsmr_connection_fixture):
# tariff should be translated in human readable and have no unit
power_tariff = hass.states.get("sensor.power_tariff")
assert power_tariff.state == "low"
assert power_tariff.attributes.get("unit_of_measurement") == ""
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) is None
assert power_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert power_tariff.attributes.get(ATTR_LAST_RESET) is None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ""
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) is None
assert gas_consumption.attributes.get(ATTR_ICON) == "mdi:fire"
assert gas_consumption.attributes.get(ATTR_LAST_RESET) is not None
assert gas_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == VOLUME_CUBIC_METERS
)
async def test_luxembourg_meter(hass, dsmr_connection_fixture):
@ -351,7 +400,13 @@ async def test_luxembourg_meter(hass, dsmr_connection_fixture):
power_tariff = hass.states.get("sensor.energy_consumption_total")
assert power_tariff.state == "123.456"
assert power_tariff.attributes.get("unit_of_measurement") == ENERGY_KILO_WATT_HOUR
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) == DEVICE_CLASS_ENERGY
assert power_tariff.attributes.get(ATTR_ICON) is None
assert power_tariff.attributes.get(ATTR_LAST_RESET) is not None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ENERGY_KILO_WATT_HOUR
)
power_tariff = hass.states.get("sensor.energy_production_total")
assert power_tariff.state == "654.321"
@ -360,7 +415,13 @@ async def test_luxembourg_meter(hass, dsmr_connection_fixture):
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) is None
assert gas_consumption.attributes.get(ATTR_ICON) == "mdi:fire"
assert gas_consumption.attributes.get(ATTR_LAST_RESET) is not None
assert gas_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == VOLUME_CUBIC_METERS
)
async def test_belgian_meter(hass, dsmr_connection_fixture):
@ -415,12 +476,22 @@ async def test_belgian_meter(hass, dsmr_connection_fixture):
# tariff should be translated in human readable and have no unit
power_tariff = hass.states.get("sensor.power_tariff")
assert power_tariff.state == "normal"
assert power_tariff.attributes.get("unit_of_measurement") == ""
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) is None
assert power_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert power_tariff.attributes.get(ATTR_LAST_RESET) is None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ""
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) is None
assert gas_consumption.attributes.get(ATTR_ICON) == "mdi:fire"
assert gas_consumption.attributes.get(ATTR_LAST_RESET) is not None
assert gas_consumption.attributes.get(ATTR_STATE_CLASS) == STATE_CLASS_MEASUREMENT
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == VOLUME_CUBIC_METERS
)
async def test_belgian_meter_low(hass, dsmr_connection_fixture):
@ -464,7 +535,11 @@ async def test_belgian_meter_low(hass, dsmr_connection_fixture):
# tariff should be translated in human readable and have no unit
power_tariff = hass.states.get("sensor.power_tariff")
assert power_tariff.state == "low"
assert power_tariff.attributes.get("unit_of_measurement") == ""
assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) is None
assert power_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert power_tariff.attributes.get(ATTR_LAST_RESET) is None
assert power_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ""
async def test_tcp(hass, dsmr_connection_fixture):