jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for multiple mbus devices in dsmr (#84097)

Browse Source 
* Add support for multiple mbus devices in dsmr

A dsmr meter can have 4 mbus devices.
Support them all and also add support for a water meter on the mbus
device.

* Apply suggestions from code review

Co-authored-by: Jan Bouwhuis <jbouwh@users.noreply.github.com>

* Rewrite old gas sensor to new mbus sensor

* No force updates + fix mbus entity unique_id

* Remove old gas device

* Add additional tests

* Fix remarks from last review + move migrated 5b gas meter to new device_id

* Fix ruff error

* Last fixes

---------

Co-authored-by: Jan Bouwhuis <jbouwh@users.noreply.github.com>
This commit is contained in:
dupondje 2023-11-29 15:41:58 +01:00 committed by GitHub
parent 36eb858d0a
commit ba481001c3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 720 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -29,6 +29,7 @@ DATA_TASK = "task"
DEVICE_NAME_ELECTRICITY = "Electricity Meter"
DEVICE_NAME_GAS = "Gas Meter"
DEVICE_NAME_WATER = "Water Meter"
DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S", "Q3D"}

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

@ -34,6 +34,7 @@ from homeassistant.const import (
UnitOfVolume,
)
from homeassistant.core import CoreState, Event, HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
@ -57,6 +58,7 @@ from .const import (
DEFAULT_TIME_BETWEEN_UPDATE,
DEVICE_NAME_ELECTRICITY,
DEVICE_NAME_GAS,
DEVICE_NAME_WATER,
DOMAIN,
DSMR_PROTOCOL,
LOGGER,
@ -73,6 +75,7 @@ class DSMRSensorEntityDescription(SensorEntityDescription):
dsmr_versions: set[str] | None = None
is_gas: bool = False
is_water: bool = False
obis_reference: str
@ -374,28 +377,138 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
)
def add_gas_sensor_5B(telegram: dict[str, DSMRObject]) -> DSMRSensorEntityDescription:
"""Return correct entity for 5B Gas meter."""
ref = None
if obis_references.BELGIUM_MBUS1_METER_READING2 in telegram:
ref = obis_references.BELGIUM_MBUS1_METER_READING2
elif obis_references.BELGIUM_MBUS2_METER_READING2 in telegram:
ref = obis_references.BELGIUM_MBUS2_METER_READING2
elif obis_references.BELGIUM_MBUS3_METER_READING2 in telegram:
ref = obis_references.BELGIUM_MBUS3_METER_READING2
elif obis_references.BELGIUM_MBUS4_METER_READING2 in telegram:
ref = obis_references.BELGIUM_MBUS4_METER_READING2
elif ref is None:
ref = obis_references.BELGIUM_MBUS1_METER_READING2
return DSMRSensorEntityDescription(
key="belgium_5min_gas_meter_reading",
translation_key="gas_meter_reading",
obis_reference=ref,
dsmr_versions={"5B"},
is_gas=True,
device_class=SensorDeviceClass.GAS,
state_class=SensorStateClass.TOTAL_INCREASING,
)
def create_mbus_entity(
mbus: int, mtype: int, telegram: dict[str, DSMRObject]
) -> DSMRSensorEntityDescription | None:
"""Create a new MBUS Entity."""
if (
mtype == 3
and (
obis_reference := getattr(
obis_references, f"BELGIUM_MBUS{mbus}_METER_READING2"
)
)
in telegram
):
return DSMRSensorEntityDescription(
key=f"mbus{mbus}_gas_reading",
translation_key="gas_meter_reading",
obis_reference=obis_reference,
is_gas=True,
device_class=SensorDeviceClass.GAS,
state_class=SensorStateClass.TOTAL_INCREASING,
)
if (
mtype == 7
and (
obis_reference := getattr(
obis_references, f"BELGIUM_MBUS{mbus}_METER_READING1"
)
)
in telegram
):
return DSMRSensorEntityDescription(
key=f"mbus{mbus}_water_reading",
translation_key="water_meter_reading",
obis_reference=obis_reference,
is_water=True,
device_class=SensorDeviceClass.WATER,
state_class=SensorStateClass.TOTAL_INCREASING,
)
return None
def device_class_and_uom(
telegram: dict[str, DSMRObject],
entity_description: DSMRSensorEntityDescription,
) -> tuple[SensorDeviceClass | None, str | None]:
"""Get native unit of measurement from telegram,."""
dsmr_object = telegram[entity_description.obis_reference]
uom: str | None = getattr(dsmr_object, "unit") or None
with suppress(ValueError):
if entity_description.device_class == SensorDeviceClass.GAS and (
enery_uom := UnitOfEnergy(str(uom))
):
return (SensorDeviceClass.ENERGY, enery_uom)
if uom in UNIT_CONVERSION:
return (entity_description.device_class, UNIT_CONVERSION[uom])
return (entity_description.device_class, uom)
def rename_old_gas_to_mbus(
hass: HomeAssistant, entry: ConfigEntry, mbus_device_id: str
) -> None:
"""Rename old gas sensor to mbus variant."""
dev_reg = dr.async_get(hass)
device_entry_v1 = dev_reg.async_get_device(identifiers={(DOMAIN, entry.entry_id)})
if device_entry_v1 is not None:
device_id = device_entry_v1.id
ent_reg = er.async_get(hass)
entries = er.async_entries_for_device(ent_reg, device_id)
for entity in entries:
if entity.unique_id.endswith("belgium_5min_gas_meter_reading"):
try:
ent_reg.async_update_entity(
entity.entity_id,
new_unique_id=mbus_device_id,
device_id=mbus_device_id,
)
except ValueError:
LOGGER.warning(
"Skip migration of %s because it already exists",
entity.entity_id,
)
else:
LOGGER.info(
"Migrated entity %s from unique id %s to %s",
entity.entity_id,
entity.unique_id,
mbus_device_id,
)
# Cleanup old device
dev_entities = er.async_entries_for_device(
ent_reg, device_id, include_disabled_entities=True
)
if not dev_entities:
dev_reg.async_remove_device(device_id)
def create_mbus_entities(
hass: HomeAssistant, telegram: dict[str, DSMRObject], entry: ConfigEntry
) -> list[DSMREntity]:
"""Create MBUS Entities."""
entities = []
for idx in range(1, 5):
if (
device_type := getattr(obis_references, f"BELGIUM_MBUS{idx}_DEVICE_TYPE")
) not in telegram:
continue
if (type_ := int(telegram[device_type].value)) not in (3, 7):
continue
if (
identifier := getattr(
obis_references,
f"BELGIUM_MBUS{idx}_EQUIPMENT_IDENTIFIER",
)
) in telegram:
serial_ = telegram[identifier].value
rename_old_gas_to_mbus(hass, entry, serial_)
else:
serial_ = ""
if description := create_mbus_entity(idx, type_, telegram):
entities.append(
DSMREntity(
description,
entry,
telegram,
*device_class_and_uom(telegram, description), # type: ignore[arg-type]
serial_,
idx,
)
)
return entities
async def async_setup_entry(
@ -415,25 +528,10 @@ async def async_setup_entry(
add_entities_handler()
add_entities_handler = None
def device_class_and_uom(
telegram: dict[str, DSMRObject],
entity_description: DSMRSensorEntityDescription,
) -> tuple[SensorDeviceClass | None, str | None]:
"""Get native unit of measurement from telegram,."""
dsmr_object = telegram[entity_description.obis_reference]
uom: str | None = getattr(dsmr_object, "unit") or None
with suppress(ValueError):
if entity_description.device_class == SensorDeviceClass.GAS and (
enery_uom := UnitOfEnergy(str(uom))
):
return (SensorDeviceClass.ENERGY, enery_uom)
if uom in UNIT_CONVERSION:
return (entity_description.device_class, UNIT_CONVERSION[uom])
return (entity_description.device_class, uom)
all_sensors = SENSORS
if dsmr_version == "5B":
all_sensors += (add_gas_sensor_5B(telegram),)
mbus_entities = create_mbus_entities(hass, telegram, entry)
for mbus_entity in mbus_entities:
entities.append(mbus_entity)
entities.extend(
[
@ -443,7 +541,7 @@ async def async_setup_entry(
telegram,
*device_class_and_uom(telegram, description), # type: ignore[arg-type]
)
for description in all_sensors
for description in SENSORS
if (
description.dsmr_versions is None
or dsmr_version in description.dsmr_versions
@ -618,6 +716,8 @@ class DSMREntity(SensorEntity):
telegram: dict[str, DSMRObject],
device_class: SensorDeviceClass,
native_unit_of_measurement: str | None,
serial_id: str = "",
mbus_id: int = 0,
) -> None:
"""Initialize entity."""
self.entity_description = entity_description
@ -629,8 +729,15 @@ class DSMREntity(SensorEntity):
device_serial = entry.data[CONF_SERIAL_ID]
device_name = DEVICE_NAME_ELECTRICITY
if entity_description.is_gas:
device_serial = entry.data[CONF_SERIAL_ID_GAS]
if serial_id:
device_serial = serial_id
else:
device_serial = entry.data[CONF_SERIAL_ID_GAS]
device_name = DEVICE_NAME_GAS
if entity_description.is_water:
if serial_id:
device_serial = serial_id
device_name = DEVICE_NAME_WATER
if device_serial is None:
device_serial = entry.entry_id
@ -638,7 +745,13 @@ class DSMREntity(SensorEntity):
identifiers={(DOMAIN, device_serial)},
name=device_name,
)
self._attr_unique_id = f"{device_serial}_{entity_description.key}"
if mbus_id != 0:
if serial_id:
self._attr_unique_id = f"{device_serial}"
else:
self._attr_unique_id = f"{device_serial}_{mbus_id}"
else:
self._attr_unique_id = f"{device_serial}_{entity_description.key}"
@callback
def update_data(self, telegram: dict[str, DSMRObject] | None) -> None:

3
homeassistant/components/dsmr/strings.json
View File

@ -147,6 +147,9 @@
},
"voltage_swell_l3_count": {
"name": "Voltage swells phase L3"
},
"water_meter_reading": {
"name": "Water consumption"
}
}
},

212
tests/components/dsmr/test_mbus_migration.py Normal file
View File

@ -0,0 +1,212 @@
"""Tests for the DSMR integration."""
import datetime
from decimal import Decimal
from homeassistant.components.dsmr.const import DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def test_migrate_gas_to_mbus(
hass: HomeAssistant, entity_registry: er.EntityRegistry, dsmr_connection_fixture
) -> None:
"""Test migration of unique_id."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_METER_READING2,
)
from dsmr_parser.objects import CosemObject, MBusObject
mock_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="/dev/ttyUSB0",
data={
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "37464C4F32313139303333373331",
},
options={
"time_between_update": 0,
},
)
mock_entry.add_to_hass(hass)
old_unique_id = "37464C4F32313139303333373331_belgium_5min_gas_meter_reading"
device_registry = hass.helpers.device_registry.async_get(hass)
device = device_registry.async_get_or_create(
config_entry_id=mock_entry.entry_id,
identifiers={(DOMAIN, mock_entry.entry_id)},
name="Gas Meter",
)
await hass.async_block_till_done()
entity: er.RegistryEntry = entity_registry.async_get_or_create(
suggested_object_id="gas_meter_reading",
disabled_by=None,
domain=SENSOR_DOMAIN,
platform=DOMAIN,
device_id=device.id,
unique_id=old_unique_id,
config_entry=mock_entry,
)
assert entity.unique_id == old_unique_id
await hass.async_block_till_done()
telegram = {
BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
BELGIUM_MBUS1_METER_READING2: MBusObject(
BELGIUM_MBUS1_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
}
assert await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
telegram_callback = connection_factory.call_args_list[0][0][2]
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
# after receiving telegram entities need to have the chance to be created
await hass.async_block_till_done()
dev_entities = er.async_entries_for_device(
entity_registry, device.id, include_disabled_entities=True
)
assert not dev_entities
assert (
entity_registry.async_get_entity_id(SENSOR_DOMAIN, DOMAIN, old_unique_id)
is None
)
assert (
entity_registry.async_get_entity_id(
SENSOR_DOMAIN, DOMAIN, "37464C4F32313139303333373331"
)
== "sensor.gas_meter_reading"
)
async def test_migrate_gas_to_mbus_exists(
hass: HomeAssistant, entity_registry: er.EntityRegistry, dsmr_connection_fixture
) -> None:
"""Test migration of unique_id."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_METER_READING2,
)
from dsmr_parser.objects import CosemObject, MBusObject
mock_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="/dev/ttyUSB0",
data={
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "37464C4F32313139303333373331",
},
options={
"time_between_update": 0,
},
)
mock_entry.add_to_hass(hass)
old_unique_id = "37464C4F32313139303333373331_belgium_5min_gas_meter_reading"
device_registry = hass.helpers.device_registry.async_get(hass)
device = device_registry.async_get_or_create(
config_entry_id=mock_entry.entry_id,
identifiers={(DOMAIN, mock_entry.entry_id)},
name="Gas Meter",
)
await hass.async_block_till_done()
entity: er.RegistryEntry = entity_registry.async_get_or_create(
suggested_object_id="gas_meter_reading",
disabled_by=None,
domain=SENSOR_DOMAIN,
platform=DOMAIN,
device_id=device.id,
unique_id=old_unique_id,
config_entry=mock_entry,
)
assert entity.unique_id == old_unique_id
device2 = device_registry.async_get_or_create(
config_entry_id=mock_entry.entry_id,
identifiers={(DOMAIN, "37464C4F32313139303333373331")},
name="Gas Meter",
)
await hass.async_block_till_done()
entity_registry.async_get_or_create(
suggested_object_id="gas_meter_reading_alt",
disabled_by=None,
domain=SENSOR_DOMAIN,
platform=DOMAIN,
device_id=device2.id,
unique_id="37464C4F32313139303333373331",
config_entry=mock_entry,
)
await hass.async_block_till_done()
telegram = {
BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
BELGIUM_MBUS1_METER_READING2: MBusObject(
BELGIUM_MBUS1_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
}
assert await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
telegram_callback = connection_factory.call_args_list[0][0][2]
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
# after receiving telegram entities need to have the chance to be created
await hass.async_block_till_done()
assert (
entity_registry.async_get_entity_id(SENSOR_DOMAIN, DOMAIN, old_unique_id)
== "sensor.gas_meter_reading"
)

451
tests/components/dsmr/test_sensor.py
View File

@ -8,21 +8,8 @@ import asyncio
import datetime
from decimal import Decimal
from itertools import chain, repeat
from typing import Literal
from unittest.mock import DEFAULT, MagicMock
from dsmr_parser.obis_references import (
BELGIUM_MBUS1_METER_READING1,
BELGIUM_MBUS1_METER_READING2,
BELGIUM_MBUS2_METER_READING1,
BELGIUM_MBUS2_METER_READING2,
BELGIUM_MBUS3_METER_READING1,
BELGIUM_MBUS3_METER_READING2,
BELGIUM_MBUS4_METER_READING1,
BELGIUM_MBUS4_METER_READING2,
)
import pytest
from homeassistant import config_entries
from homeassistant.components.sensor import (
ATTR_OPTIONS,
@ -145,8 +132,8 @@ async def test_default_setup(
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
# after receiving telegram entities need to have the chance to update
await asyncio.sleep(0)
# after receiving telegram entities need to have the chance to be created
await hass.async_block_till_done()
# ensure entities have new state value after incoming telegram
power_consumption = hass.states.get("sensor.electricity_meter_power_consumption")
@ -495,10 +482,18 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
from dsmr_parser.obis_references import (
BELGIUM_CURRENT_AVERAGE_DEMAND,
BELGIUM_MAXIMUM_DEMAND_MONTH,
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_METER_READING2,
BELGIUM_MBUS2_METER_READING2,
BELGIUM_MBUS2_DEVICE_TYPE,
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS2_METER_READING1,
BELGIUM_MBUS3_DEVICE_TYPE,
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS3_METER_READING2,
BELGIUM_MBUS4_METER_READING2,
BELGIUM_MBUS4_DEVICE_TYPE,
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS4_METER_READING1,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject, MBusObject
@ -509,41 +504,13 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "5678",
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
BELGIUM_MBUS1_METER_READING2: MBusObject(
BELGIUM_MBUS1_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
BELGIUM_MBUS2_METER_READING2: MBusObject(
BELGIUM_MBUS2_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(745.696), "unit": "m3"},
],
),
BELGIUM_MBUS3_METER_READING2: MBusObject(
BELGIUM_MBUS3_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642215)},
{"value": Decimal(745.697), "unit": "m3"},
],
),
BELGIUM_MBUS4_METER_READING2: MBusObject(
BELGIUM_MBUS4_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642216)},
{"value": Decimal(745.698), "unit": "m3"},
],
),
BELGIUM_CURRENT_AVERAGE_DEMAND: CosemObject(
BELGIUM_CURRENT_AVERAGE_DEMAND,
[{"value": Decimal(1.75), "unit": "kW"}],
@ -555,6 +522,62 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
{"value": Decimal(4.11), "unit": "kW"},
],
),
BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
BELGIUM_MBUS1_METER_READING2: MBusObject(
BELGIUM_MBUS1_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
BELGIUM_MBUS2_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS2_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
BELGIUM_MBUS2_METER_READING1: MBusObject(
BELGIUM_MBUS2_METER_READING1,
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(678.695), "unit": "m3"},
],
),
BELGIUM_MBUS3_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS3_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
BELGIUM_MBUS3_METER_READING2: MBusObject(
BELGIUM_MBUS3_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642215)},
{"value": Decimal(12.12), "unit": "m3"},
],
),
BELGIUM_MBUS4_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS4_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373334", "unit": ""}],
),
BELGIUM_MBUS4_METER_READING1: MBusObject(
BELGIUM_MBUS4_METER_READING1,
[
{"value": datetime.datetime.fromtimestamp(1551642216)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
@ -600,7 +623,7 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
assert max_demand.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfPower.KILO_WATT
assert max_demand.attributes.get(ATTR_STATE_CLASS) is None
# check if gas consumption is parsed correctly
# check if gas consumption mbus1 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
@ -613,48 +636,69 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
== UnitOfVolume.CUBIC_METERS
)
# check if water usage mbus2 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption")
assert water_consumption.state == "678.695"
assert (
water_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.WATER
)
assert (
water_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
water_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
@pytest.mark.parametrize(
("key1", "key2", "key3", "gas_value"),
[
(
BELGIUM_MBUS1_METER_READING1,
BELGIUM_MBUS2_METER_READING2,
BELGIUM_MBUS3_METER_READING1,
"745.696",
),
(
BELGIUM_MBUS1_METER_READING2,
BELGIUM_MBUS2_METER_READING1,
BELGIUM_MBUS3_METER_READING2,
"745.695",
),
(
BELGIUM_MBUS4_METER_READING2,
BELGIUM_MBUS2_METER_READING1,
BELGIUM_MBUS3_METER_READING1,
"745.695",
),
(
BELGIUM_MBUS4_METER_READING1,
BELGIUM_MBUS2_METER_READING1,
BELGIUM_MBUS3_METER_READING2,
"745.697",
),
],
)
async def test_belgian_meter_alt(
hass: HomeAssistant,
dsmr_connection_fixture,
key1: Literal,
key2: Literal,
key3: Literal,
gas_value: str,
) -> None:
# check if gas consumption mbus1 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption_2")
assert gas_consumption.state == "12.12"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
assert (
gas_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
# check if water usage mbus2 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption_2")
assert water_consumption.state == "13.13"
assert (
water_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.WATER
)
assert (
water_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
water_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
async def test_belgian_meter_alt(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.objects import MBusObject
from dsmr_parser.obis_references import (
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_METER_READING1,
BELGIUM_MBUS2_DEVICE_TYPE,
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS2_METER_READING2,
BELGIUM_MBUS3_DEVICE_TYPE,
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS3_METER_READING1,
BELGIUM_MBUS4_DEVICE_TYPE,
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS4_METER_READING2,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
@ -662,32 +706,67 @@ async def test_belgian_meter_alt(
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "5678",
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
key1: MBusObject(
key1,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
key2: MBusObject(
key2,
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(745.696), "unit": "m3"},
],
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
key3: MBusObject(
key3,
BELGIUM_MBUS1_METER_READING1: MBusObject(
BELGIUM_MBUS1_METER_READING1,
[
{"value": datetime.datetime.fromtimestamp(1551642215)},
{"value": Decimal(745.697), "unit": "m3"},
{"value": Decimal(123.456), "unit": "m3"},
],
),
BELGIUM_MBUS2_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS2_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
BELGIUM_MBUS2_METER_READING2: MBusObject(
BELGIUM_MBUS2_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642216)},
{"value": Decimal(678.901), "unit": "m3"},
],
),
BELGIUM_MBUS3_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS3_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
BELGIUM_MBUS3_METER_READING1: MBusObject(
BELGIUM_MBUS3_METER_READING1,
[
{"value": datetime.datetime.fromtimestamp(1551642217)},
{"value": Decimal(12.12), "unit": "m3"},
],
),
BELGIUM_MBUS4_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS4_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373334", "unit": ""}],
),
BELGIUM_MBUS4_METER_READING2: MBusObject(
BELGIUM_MBUS4_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642218)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
}
@ -709,9 +788,24 @@ async def test_belgian_meter_alt(
# after receiving telegram entities need to have the chance to be created
await hass.async_block_till_done()
# check if gas consumption is parsed correctly
# check if water usage mbus1 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption")
assert water_consumption.state == "123.456"
assert (
water_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.WATER
)
assert (
water_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
water_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
# check if gas consumption mbus2 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == gas_value
assert gas_consumption.state == "678.901"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
assert (
gas_consumption.attributes.get(ATTR_STATE_CLASS)
@ -722,6 +816,157 @@ async def test_belgian_meter_alt(
== UnitOfVolume.CUBIC_METERS
)
# check if water usage mbus3 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption_2")
assert water_consumption.state == "12.12"
assert (
water_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.WATER
)
assert (
water_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
water_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
# check if gas consumption mbus4 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption_2")
assert gas_consumption.state == "13.13"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
assert (
gas_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
async def test_belgian_meter_mbus(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS2_DEVICE_TYPE,
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS3_DEVICE_TYPE,
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS3_METER_READING2,
BELGIUM_MBUS4_DEVICE_TYPE,
BELGIUM_MBUS4_METER_READING1,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0003", "unit": ""}]
),
BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "006", "unit": ""}]
),
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
BELGIUM_MBUS2_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS2_DEVICE_TYPE, [{"value": "003", "unit": ""}]
),
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
BELGIUM_MBUS3_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS3_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER: CosemObject(
BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
BELGIUM_MBUS3_METER_READING2: MBusObject(
BELGIUM_MBUS3_METER_READING2,
[
{"value": datetime.datetime.fromtimestamp(1551642217)},
{"value": Decimal(12.12), "unit": "m3"},
],
),
BELGIUM_MBUS4_DEVICE_TYPE: CosemObject(
BELGIUM_MBUS4_DEVICE_TYPE, [{"value": "007", "unit": ""}]
),
BELGIUM_MBUS4_METER_READING1: MBusObject(
BELGIUM_MBUS4_METER_READING1,
[
{"value": datetime.datetime.fromtimestamp(1551642218)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
}
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
telegram_callback = connection_factory.call_args_list[0][0][2]
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
# after receiving telegram entities need to have the chance to be created
await hass.async_block_till_done()
# tariff should be translated in human readable and have no unit
active_tariff = hass.states.get("sensor.electricity_meter_active_tariff")
assert active_tariff.state == "unknown"
# check if gas consumption mbus2 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption is None
# check if water usage mbus3 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption_2")
assert water_consumption is None
# check if gas consumption mbus4 is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption_2")
assert gas_consumption is None
# check if gas consumption mbus4 is parsed correctly
water_consumption = hass.states.get("sensor.water_meter_water_consumption")
assert water_consumption.state == "13.13"
assert (
water_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.WATER
)
assert (
water_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
water_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
async def test_belgian_meter_low(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if Belgian meter is correctly parsed."""