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

Add explicit support for Luxembourg Smarty meter in dsmr integration (#43975)

Browse source 
* Add support for Luxembourg Smarty meter

* Add config flow test

* Add sensor tests
This commit is contained in:
Rob Bierbooms 2020-12-27 09:39:36 +01:00 • committed by GitHub
parent 1f27fb4644
commit 9531b08f2a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 140 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
homeassistant/components/dsmr/config_flow.py
View file

@ -35,11 +35,15 @@ class DSMRConnection:
self._port = port
self._dsmr_version = dsmr_version
self._telegram = {}
if dsmr_version == "5L":
self._equipment_identifier = obis_ref.LUXEMBOURG_EQUIPMENT_IDENTIFIER
else:
self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER
def equipment_identifier(self):
"""Equipment identifier."""
if obis_ref.EQUIPMENT_IDENTIFIER in self._telegram:
dsmr_object = self._telegram[obis_ref.EQUIPMENT_IDENTIFIER]
if self._equipment_identifier in self._telegram:
dsmr_object = self._telegram[self._equipment_identifier]
return getattr(dsmr_object, "value", None)
def equipment_identifier_gas(self):
@ -52,7 +56,7 @@ class DSMRConnection:
"""Test if we can validate connection with the device."""
def update_telegram(telegram):
if obis_ref.EQUIPMENT_IDENTIFIER in telegram:
if self._equipment_identifier in telegram:
self._telegram = telegram
transport.close()

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

@ -54,7 +54,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.string,
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_DSMR_VERSION, default=DEFAULT_DSMR_VERSION): vol.All(
cv.string, vol.In(["5B", "5", "4", "2.2"])
cv.string, vol.In(["5L", "5B", "5", "4", "2.2"])
),
vol.Optional(CONF_RECONNECT_INTERVAL, default=DEFAULT_RECONNECT_INTERVAL): int,
vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): vol.Coerce(int),
@ -85,7 +85,6 @@ async def async_setup_entry(
["Power Consumption", obis_ref.CURRENT_ELECTRICITY_USAGE],
["Power Production", obis_ref.CURRENT_ELECTRICITY_DELIVERY],
["Power Tariff", obis_ref.ELECTRICITY_ACTIVE_TARIFF],
["Energy Consumption (total)", obis_ref.ELECTRICITY_IMPORTED_TOTAL],
["Energy Consumption (tarif 1)", obis_ref.ELECTRICITY_USED_TARIFF_1],
["Energy Consumption (tarif 2)", obis_ref.ELECTRICITY_USED_TARIFF_2],
["Energy Production (tarif 1)", obis_ref.ELECTRICITY_DELIVERED_TARIFF_1],
@ -112,6 +111,24 @@ async def async_setup_entry(
["Current Phase L3", obis_ref.INSTANTANEOUS_CURRENT_L3],
]
if dsmr_version == "5L":
obis_mapping.extend(
[
[
"Energy Consumption (total)",
obis_ref.LUXEMBOURG_ELECTRICITY_USED_TARIFF_GLOBAL,
],
[
"Energy Production (total)",
obis_ref.LUXEMBOURG_ELECTRICITY_DELIVERED_TARIFF_GLOBAL,
],
]
)
else:
obis_mapping.extend(
[["Energy Consumption (total)", obis_ref.ELECTRICITY_IMPORTED_TOTAL]]
)
# Generate device entities
devices = [
DSMREntity(name, DEVICE_NAME_ENERGY, config[CONF_SERIAL_ID], obis, config)
@ -120,7 +137,7 @@ async def async_setup_entry(
# Protocol version specific obis
if CONF_SERIAL_ID_GAS in config:
if dsmr_version in ("4", "5"):
if dsmr_version in ("4", "5", "5L"):
gas_obis = obis_ref.HOURLY_GAS_METER_READING
elif dsmr_version in ("5B",):
gas_obis = obis_ref.BELGIUM_HOURLY_GAS_METER_READING

28
tests/components/dsmr/conftest.py
View file

@ -2,7 +2,11 @@
import asyncio
from dsmr_parser.clients.protocol import DSMRProtocol
from dsmr_parser.obis_references import EQUIPMENT_IDENTIFIER, EQUIPMENT_IDENTIFIER_GAS
from dsmr_parser.obis_references import (
EQUIPMENT_IDENTIFIER,
EQUIPMENT_IDENTIFIER_GAS,
LUXEMBOURG_EQUIPMENT_IDENTIFIER,
)
from dsmr_parser.objects import CosemObject
import pytest
@ -38,17 +42,27 @@ async def dsmr_connection_send_validate_fixture(hass):
transport = MagicMock(spec=asyncio.Transport)
protocol = MagicMock(spec=DSMRProtocol)
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
return (transport, protocol)
connection_factory = MagicMock(wraps=connection_factory)
protocol.telegram = {
EQUIPMENT_IDENTIFIER: CosemObject([{"value": "12345678", "unit": ""}]),
EQUIPMENT_IDENTIFIER_GAS: CosemObject([{"value": "123456789", "unit": ""}]),
}
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
if args[1] == "5L":
protocol.telegram = {
LUXEMBOURG_EQUIPMENT_IDENTIFIER: CosemObject(
[{"value": "12345678", "unit": ""}]
),
EQUIPMENT_IDENTIFIER_GAS: CosemObject(
[{"value": "123456789", "unit": ""}]
),
}
return (transport, protocol)
connection_factory = MagicMock(wraps=connection_factory)
async def wait_closed():
if isinstance(connection_factory.call_args_list[0][0][2], str):
# TCP

23
tests/components/dsmr/test_config_flow.py
View file

@ -242,3 +242,26 @@ async def test_options_flow(hass):
await hass.async_block_till_done()
assert entry.options == {"time_between_update": 15}
async def test_import_luxembourg(hass, dsmr_connection_send_validate_fixture):
"""Test we can import."""
await setup.async_setup_component(hass, "persistent_notification", {})
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5L",
"precision": 4,
"reconnect_interval": 30,
}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=entry_data,
)
assert result["type"] == "create_entry"
assert result["title"] == "/dev/ttyUSB0"
assert result["data"] == {**entry_data, **SERIAL_DATA}

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

@ -337,6 +337,75 @@ async def test_v5_meter(hass, dsmr_connection_fixture):
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
async def test_luxembourg_meter(hass, dsmr_connection_fixture):
"""Test if v5 meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
HOURLY_GAS_METER_READING,
LUXEMBOURG_ELECTRICITY_DELIVERED_TARIFF_GLOBAL,
LUXEMBOURG_ELECTRICITY_USED_TARIFF_GLOBAL,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5L",
"precision": 4,
"reconnect_interval": 30,
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
HOURLY_GAS_METER_READING: MBusObject(
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": VOLUME_CUBIC_METERS},
]
),
LUXEMBOURG_ELECTRICITY_USED_TARIFF_GLOBAL: CosemObject(
[{"value": Decimal(123.456), "unit": ENERGY_KILO_WATT_HOUR}]
),
LUXEMBOURG_ELECTRICITY_DELIVERED_TARIFF_GLOBAL: CosemObject(
[{"value": Decimal(654.321), "unit": ENERGY_KILO_WATT_HOUR}]
),
}
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 update
await asyncio.sleep(0)
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
power_tariff = hass.states.get("sensor.energy_production_total")
assert power_tariff.state == "654.321"
assert power_tariff.attributes.get("unit_of_measurement") == ENERGY_KILO_WATT_HOUR
# 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
async def test_belgian_meter(hass, dsmr_connection_fixture):
"""Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture