Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use entity descriptions classes in DSMR (#53549)

pull/53556/head
Franck Nijhof 2021-07-27 14:44:58 +02:00 committed by GitHub
parent b51c666817
commit 27e69037d4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 92 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
homeassistant/components/dsmr/const.py
View File

@ -14,7 +14,7 @@ from homeassistant.const import (
)
from homeassistant.util import dt
from .models import DSMRSensor
from .models import DSMRSensorEntityDescription
DOMAIN = "dsmr"
@ -41,217 +41,217 @@ DATA_TASK = "task"
DEVICE_NAME_ENERGY = "Energy Meter"
DEVICE_NAME_GAS = "Gas Meter"
SENSORS: list[DSMRSensor] = [
DSMRSensor(
SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
DSMRSensorEntityDescription(
key=obis_references.CURRENT_ELECTRICITY_USAGE,
name="Power Consumption",
obis_reference=obis_references.CURRENT_ELECTRICITY_USAGE,
device_class=DEVICE_CLASS_POWER,
force_update=True,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.CURRENT_ELECTRICITY_DELIVERY,
name="Power Production",
obis_reference=obis_references.CURRENT_ELECTRICITY_DELIVERY,
device_class=DEVICE_CLASS_POWER,
force_update=True,
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_ACTIVE_TARIFF,
name="Power Tariff",
obis_reference=obis_references.ELECTRICITY_ACTIVE_TARIFF,
icon="mdi:flash",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_USED_TARIFF_1,
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(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_USED_TARIFF_2,
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(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_DELIVERED_TARIFF_1,
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(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_DELIVERED_TARIFF_2,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE,
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(
DSMRSensorEntityDescription(
key=obis_references.SHORT_POWER_FAILURE_COUNT,
name="Short Power Failure Count",
obis_reference=obis_references.SHORT_POWER_FAILURE_COUNT,
entity_registry_enabled_default=False,
icon="mdi:flash-off",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.LONG_POWER_FAILURE_COUNT,
name="Long Power Failure Count",
obis_reference=obis_references.LONG_POWER_FAILURE_COUNT,
entity_registry_enabled_default=False,
icon="mdi:flash-off",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SAG_L1_COUNT,
name="Voltage Sags Phase L1",
obis_reference=obis_references.VOLTAGE_SAG_L1_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SAG_L2_COUNT,
name="Voltage Sags Phase L2",
obis_reference=obis_references.VOLTAGE_SAG_L2_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SAG_L3_COUNT,
name="Voltage Sags Phase L3",
obis_reference=obis_references.VOLTAGE_SAG_L3_COUNT,
entity_registry_enabled_default=False,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SWELL_L1_COUNT,
name="Voltage Swells Phase L1",
obis_reference=obis_references.VOLTAGE_SWELL_L1_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SWELL_L2_COUNT,
name="Voltage Swells Phase L2",
obis_reference=obis_references.VOLTAGE_SWELL_L2_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.VOLTAGE_SWELL_L3_COUNT,
name="Voltage Swells Phase L3",
obis_reference=obis_references.VOLTAGE_SWELL_L3_COUNT,
entity_registry_enabled_default=False,
icon="mdi:pulse",
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_VOLTAGE_L1,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_VOLTAGE_L2,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_VOLTAGE_L3,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_CURRENT_L1,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_CURRENT_L2,
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(
DSMRSensorEntityDescription(
key=obis_references.INSTANTANEOUS_CURRENT_L3,
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(
DSMRSensorEntityDescription(
key=obis_references.LUXEMBOURG_ELECTRICITY_USED_TARIFF_GLOBAL,
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(
DSMRSensorEntityDescription(
key=obis_references.LUXEMBOURG_ELECTRICITY_DELIVERED_TARIFF_GLOBAL,
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(
DSMRSensorEntityDescription(
key=obis_references.ELECTRICITY_IMPORTED_TOTAL,
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(
DSMRSensorEntityDescription(
key=obis_references.HOURLY_GAS_METER_READING,
name="Gas Consumption",
obis_reference=obis_references.HOURLY_GAS_METER_READING,
dsmr_versions={"4", "5", "5L"},
is_gas=True,
force_update=True,
@ -259,9 +259,9 @@ SENSORS: list[DSMRSensor] = [
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.BELGIUM_HOURLY_GAS_METER_READING,
name="Gas Consumption",
obis_reference=obis_references.BELGIUM_HOURLY_GAS_METER_READING,
dsmr_versions={"5B"},
is_gas=True,
force_update=True,
@ -269,9 +269,9 @@ SENSORS: list[DSMRSensor] = [
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
DSMRSensor(
DSMRSensorEntityDescription(
key=obis_references.GAS_METER_READING,
name="Gas Consumption",
obis_reference=obis_references.GAS_METER_READING,
dsmr_versions={"2.2"},
is_gas=True,
force_update=True,
@ -279,4 +279,4 @@ SENSORS: list[DSMRSensor] = [
last_reset=dt.utc_from_timestamp(0),
state_class=STATE_CLASS_MEASUREMENT,
),
]
)

14
homeassistant/components/dsmr/models.py
View File

@ -2,21 +2,13 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from homeassistant.components.sensor import SensorEntityDescription
@dataclass
class DSMRSensor:
class DSMRSensorEntityDescription(SensorEntityDescription):
"""Represents an DSMR Sensor."""
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

41
homeassistant/components/dsmr/sensor.py
View File

@ -42,7 +42,7 @@ from .const import (
LOGGER,
SENSORS,
)
from .models import DSMRSensor
from .models import DSMRSensorEntityDescription
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
@ -83,10 +83,13 @@ async def async_setup_entry(
"""Set up the DSMR sensor."""
dsmr_version = entry.data[CONF_DSMR_VERSION]
entities = [
DSMREntity(sensor, entry)
for sensor in SENSORS
if (sensor.dsmr_versions is None or dsmr_version in sensor.dsmr_versions)
and (not sensor.is_gas or CONF_SERIAL_ID_GAS in entry.data)
DSMREntity(description, entry)
for description in SENSORS
if (
description.dsmr_versions is None
or dsmr_version in description.dsmr_versions
)
and (not description.is_gas or CONF_SERIAL_ID_GAS in entry.data)
]
async_add_entities(entities)
@ -184,50 +187,46 @@ async def async_setup_entry(
class DSMREntity(SensorEntity):
"""Entity reading values from DSMR telegram."""
entity_description: DSMRSensorEntityDescription
_attr_should_poll = False
def __init__(self, sensor: DSMRSensor, entry: ConfigEntry) -> None:
def __init__(
self, entity_description: DSMRSensorEntityDescription, entry: ConfigEntry
) -> None:
"""Initialize entity."""
self._sensor = sensor
self.entity_description = entity_description
self._entry = entry
self.telegram: dict[str, DSMRObject] = {}
device_serial = entry.data[CONF_SERIAL_ID]
device_name = DEVICE_NAME_ENERGY
if sensor.is_gas:
if entity_description.is_gas:
device_serial = entry.data[CONF_SERIAL_ID_GAS]
device_name = DEVICE_NAME_GAS
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_unique_id = f"{device_serial}_{entity_description.name}".replace(
" ", "_"
)
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:
"""Update data."""
self.telegram = telegram
if self.hass and self._sensor.obis_reference in self.telegram:
if self.hass and self.entity_description.key in self.telegram:
self.async_write_ha_state()
def get_dsmr_object_attr(self, attribute: str) -> str | None:
"""Read attribute from last received telegram for this DSMR object."""
# Make sure telegram contains an object for this entities obis
if self._sensor.obis_reference not in self.telegram:
if self.entity_description.key not in self.telegram:
return None
# Get the attribute value if the object has it
dsmr_object = self.telegram[self._sensor.obis_reference]
dsmr_object = self.telegram[self.entity_description.key]
attr: str | None = getattr(dsmr_object, attribute)
return attr
@ -238,7 +237,7 @@ class DSMREntity(SensorEntity):
if value is None:
return None
if self._sensor.obis_reference == obis_ref.ELECTRICITY_ACTIVE_TARIFF:
if self.entity_description.key == obis_ref.ELECTRICITY_ACTIVE_TARIFF:
return self.translate_tariff(value, self._entry.data[CONF_DSMR_VERSION])
with suppress(TypeError):