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

Fix 5B Gas meter in dsmr (#103506)

Browse source 
* Fix 5B Gas meter in dsmr

In commit 1b73219 the gas meter broke for 5B.
As the change can't be reverted easily without removing the peak usage
sensors, we implement a workaround.

The first MBUS_METER_READING2 value will contain the gas meter data just
like the previous BELGIUM_5MIN_GAS_METER_READING did.
But this without the need to touch dsmr_parser (version).

Fixes: #103306, #103293

* Use parametrize

* Apply suggestions from code review

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

* Add additional tests + typo fix

---------

Co-authored-by: Jan Bouwhuis <jbouwh@users.noreply.github.com>
This commit is contained in:
dupondje 2023-11-08 09:13:51 +01:00 • committed by GitHub
parent a0f19f26c4
commit 4f11ee6e0b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 179 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -34,6 +34,3 @@ DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S", "Q3D"}
DSMR_PROTOCOL = "dsmr_protocol"
RFXTRX_DSMR_PROTOCOL = "rfxtrx_dsmr_protocol"
# Temp obis until sensors replaced by mbus variants
BELGIUM_5MIN_GAS_METER_READING = r"\d-\d:24\.2\.3.+?\r\n"

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

@ -44,7 +44,6 @@ from homeassistant.helpers.typing import StateType
from homeassistant.util import Throttle
from .const import (
BELGIUM_5MIN_GAS_METER_READING,
CONF_DSMR_VERSION,
CONF_PRECISION,
CONF_PROTOCOL,
@ -382,16 +381,6 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.GAS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
DSMRSensorEntityDescription(
key="belgium_5min_gas_meter_reading",
translation_key="gas_meter_reading",
obis_reference=BELGIUM_5MIN_GAS_METER_READING,
dsmr_versions={"5B"},
is_gas=True,
force_update=True,
device_class=SensorDeviceClass.GAS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
DSMRSensorEntityDescription(
key="gas_meter_reading",
translation_key="gas_meter_reading",
@ -405,6 +394,31 @@ 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,
force_update=True,
device_class=SensorDeviceClass.GAS,
state_class=SensorStateClass.TOTAL_INCREASING,
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
@ -438,6 +452,10 @@ async def async_setup_entry(
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),)
entities.extend(
[
DSMREntity(
@ -448,7 +466,7 @@ async def async_setup_entry(
telegram, description
), # type: ignore[arg-type]
)
for description in SENSORS
for description in all_sensors
if (
description.dsmr_versions is None
or dsmr_version in description.dsmr_versions

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

@ -8,10 +8,22 @@ 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.dsmr.const import BELGIUM_5MIN_GAS_METER_READING
from homeassistant.components.sensor import (
ATTR_OPTIONS,
ATTR_STATE_CLASS,
@ -483,6 +495,10 @@ 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_METER_READING2,
BELGIUM_MBUS2_METER_READING2,
BELGIUM_MBUS3_METER_READING2,
BELGIUM_MBUS4_METER_READING2,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject, MBusObject
@ -500,13 +516,34 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
}
telegram = {
BELGIUM_5MIN_GAS_METER_READING: MBusObject(
BELGIUM_5MIN_GAS_METER_READING,
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"}],
@ -577,6 +614,115 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
)
@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:
"""Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.objects import MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
key1: MBusObject(
key1,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
key2: MBusObject(
key2,
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(745.696), "unit": "m3"},
],
),
key3: MBusObject(
key3,
[
{"value": datetime.datetime.fromtimestamp(1551642215)},
{"value": Decimal(745.697), "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()
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == gas_value
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_low(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture