From a0d0e0f4a9ddd60e0a50e8885a11c19af04f550b Mon Sep 17 00:00:00 2001
From: Joakim Plate <elupus@ecce.se>
Date: Fri, 12 Jul 2024 16:00:27 +0200
Subject: [PATCH] Correct deprecated telegram usage in dsmr (#121847)
---
homeassistant/components/dsmr/sensor.py | 16 +-
tests/components/dsmr/test_mbus_migration.py | 56 +-
tests/components/dsmr/test_sensor.py | 524 +++++++++++++------
3 files changed, 399 insertions(+), 197 deletions(-)
diff --git a/homeassistant/components/dsmr/sensor.py b/homeassistant/components/dsmr/sensor.py
index 7b2e916529a..d46b2777a34 100644
--- a/homeassistant/components/dsmr/sensor.py
+++ b/homeassistant/components/dsmr/sensor.py
@@ -16,7 +16,7 @@ from dsmr_parser.clients.rfxtrx_protocol import (
create_rfxtrx_dsmr_reader,
create_rfxtrx_tcp_dsmr_reader,
)
-from dsmr_parser.objects import DSMRObject
+from dsmr_parser.objects import DSMRObject, Telegram
import serial
from homeassistant.components.sensor import (
@@ -380,7 +380,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
def create_mbus_entity(
- mbus: int, mtype: int, telegram: dict[str, DSMRObject]
+ mbus: int, mtype: int, telegram: Telegram
) -> DSMRSensorEntityDescription | None:
"""Create a new MBUS Entity."""
if (
@@ -478,7 +478,7 @@ def rename_old_gas_to_mbus(
def create_mbus_entities(
- hass: HomeAssistant, telegram: dict[str, DSMRObject], entry: ConfigEntry
+ hass: HomeAssistant, telegram: Telegram, entry: ConfigEntry
) -> list[DSMREntity]:
"""Create MBUS Entities."""
entities = []
@@ -523,7 +523,7 @@ async def async_setup_entry(
add_entities_handler: Callable[..., None] | None
@callback
- def init_async_add_entities(telegram: dict[str, DSMRObject]) -> None:
+ def init_async_add_entities(telegram: Telegram) -> None:
"""Add the sensor entities after the first telegram was received."""
nonlocal add_entities_handler
assert add_entities_handler is not None
@@ -560,7 +560,7 @@ async def async_setup_entry(
)
@Throttle(min_time_between_updates)
- def update_entities_telegram(telegram: dict[str, DSMRObject] | None) -> None:
+ def update_entities_telegram(telegram: Telegram | None) -> None:
"""Update entities with latest telegram and trigger state update."""
nonlocal initialized
# Make all device entities aware of new telegram
@@ -709,7 +709,7 @@ class DSMREntity(SensorEntity):
self,
entity_description: DSMRSensorEntityDescription,
entry: ConfigEntry,
- telegram: dict[str, DSMRObject],
+ telegram: Telegram,
device_class: SensorDeviceClass,
native_unit_of_measurement: str | None,
serial_id: str = "",
@@ -720,7 +720,7 @@ class DSMREntity(SensorEntity):
self._attr_device_class = device_class
self._attr_native_unit_of_measurement = native_unit_of_measurement
self._entry = entry
- self.telegram: dict[str, DSMRObject] | None = telegram
+ self.telegram: Telegram | None = telegram
device_serial = entry.data[CONF_SERIAL_ID]
device_name = DEVICE_NAME_ELECTRICITY
@@ -750,7 +750,7 @@ class DSMREntity(SensorEntity):
self._attr_unique_id = f"{device_serial}_{entity_description.key}"
@callback
- def update_data(self, telegram: dict[str, DSMRObject] | None) -> None:
+ def update_data(self, telegram: Telegram | None) -> None:
"""Update data."""
self.telegram = telegram
if self.hass and (
diff --git a/tests/components/dsmr/test_mbus_migration.py b/tests/components/dsmr/test_mbus_migration.py
index 18f5e850ecd..a8b7ef9c356 100644
--- a/tests/components/dsmr/test_mbus_migration.py
+++ b/tests/components/dsmr/test_mbus_migration.py
@@ -9,7 +9,7 @@ from dsmr_parser.obis_references import (
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_METER_READING2,
)
-from dsmr_parser.objects import CosemObject, MBusObject
+from dsmr_parser.objects import CosemObject, MBusObject, Telegram
from homeassistant.components.dsmr.const import DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
@@ -65,22 +65,31 @@ async def test_migrate_gas_to_mbus(
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,
+ telegram = Telegram()
+ telegram.add(
+ BELGIUM_MBUS1_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 1),
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
- BELGIUM_MBUS1_METER_READING2: MBusObject(
- BELGIUM_MBUS1_METER_READING2,
+ "BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_METER_READING2,
+ MBusObject(
+ (0, 1),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
- }
+ "BELGIUM_MBUS1_METER_READING2",
+ )
assert await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
@@ -173,22 +182,31 @@ async def test_migrate_gas_to_mbus_exists(
)
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,
+ telegram = Telegram()
+ telegram.add(
+ BELGIUM_MBUS1_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 1),
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
- BELGIUM_MBUS1_METER_READING2: MBusObject(
- BELGIUM_MBUS1_METER_READING2,
+ "BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_METER_READING2,
+ MBusObject(
+ (0, 1),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
- }
+ "BELGIUM_MBUS1_METER_READING2",
+ )
assert await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
diff --git a/tests/components/dsmr/test_sensor.py b/tests/components/dsmr/test_sensor.py
index 435594d4eef..a7c4a98be1e 100644
--- a/tests/components/dsmr/test_sensor.py
+++ b/tests/components/dsmr/test_sensor.py
@@ -37,7 +37,7 @@ from dsmr_parser.obis_references import (
GAS_METER_READING,
HOURLY_GAS_METER_READING,
)
-from dsmr_parser.objects import CosemObject, MBusObject
+from dsmr_parser.objects import CosemObject, MBusObject, Telegram
import pytest
from homeassistant.components.sensor import (
@@ -80,22 +80,31 @@ async def test_default_setup(
"time_between_update": 0,
}
- telegram = {
- CURRENT_ELECTRICITY_USAGE: CosemObject(
- CURRENT_ELECTRICITY_USAGE,
+ telegram = Telegram()
+ telegram.add(
+ CURRENT_ELECTRICITY_USAGE,
+ CosemObject(
+ (0, 0),
[{"value": Decimal("0.0"), "unit": UnitOfPower.WATT}],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- GAS_METER_READING: MBusObject(
- GAS_METER_READING,
+ "CURRENT_ELECTRICITY_USAGE",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
+ telegram.add(
+ GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": UnitOfVolume.CUBIC_METERS},
],
),
- }
+ "GAS_METER_READING",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -134,22 +143,31 @@ async def test_default_setup(
)
assert power_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "W"
- telegram = {
- CURRENT_ELECTRICITY_USAGE: CosemObject(
- CURRENT_ELECTRICITY_USAGE,
+ telegram = Telegram()
+ telegram.add(
+ CURRENT_ELECTRICITY_USAGE,
+ CosemObject(
+ (0, 0),
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- GAS_METER_READING: MBusObject(
- GAS_METER_READING,
+ "CURRENT_ELECTRICITY_USAGE",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
+ telegram.add(
+ GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(745.701), "unit": UnitOfVolume.CUBIC_METERS},
],
),
- }
+ "GAS_METER_READING",
+ )
# simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
telegram_callback(telegram)
@@ -209,15 +227,20 @@ async def test_setup_only_energy(
"time_between_update": 0,
}
- telegram = {
- CURRENT_ELECTRICITY_USAGE: CosemObject(
- CURRENT_ELECTRICITY_USAGE,
+ telegram = Telegram()
+ telegram.add(
+ CURRENT_ELECTRICITY_USAGE,
+ CosemObject(
+ (0, 0),
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- }
+ "CURRENT_ELECTRICITY_USAGE",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -260,18 +283,23 @@ async def test_v4_meter(
"time_between_update": 0,
}
- telegram = {
- HOURLY_GAS_METER_READING: MBusObject(
- HOURLY_GAS_METER_READING,
+ telegram = Telegram()
+ telegram.add(
+ HOURLY_GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- }
+ "HOURLY_GAS_METER_READING",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -343,18 +371,23 @@ async def test_v5_meter(
"time_between_update": 0,
}
- telegram = {
- HOURLY_GAS_METER_READING: MBusObject(
- HOURLY_GAS_METER_READING,
+ telegram = Telegram()
+ telegram.add(
+ HOURLY_GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": value, "unit": "m3"},
],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- }
+ "HOURLY_GAS_METER_READING",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -411,23 +444,34 @@ async def test_luxembourg_meter(
"time_between_update": 0,
}
- telegram = {
- HOURLY_GAS_METER_READING: MBusObject(
- HOURLY_GAS_METER_READING,
+ telegram = Telegram()
+ telegram.add(
+ HOURLY_GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
- ELECTRICITY_IMPORTED_TOTAL: CosemObject(
- ELECTRICITY_IMPORTED_TOTAL,
+ "HOURLY_GAS_METER_READING",
+ )
+ telegram.add(
+ ELECTRICITY_IMPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(123.456), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- ELECTRICITY_EXPORTED_TOTAL: CosemObject(
- ELECTRICITY_EXPORTED_TOTAL,
+ "ELECTRICITY_IMPORTED_TOTAL",
+ )
+ telegram.add(
+ ELECTRICITY_EXPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(654.321), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- }
+ "ELECTRICITY_EXPORTED_TOTAL",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -495,78 +539,127 @@ async def test_belgian_meter(
"time_between_update": 0,
}
- telegram = {
- BELGIUM_CURRENT_AVERAGE_DEMAND: CosemObject(
- BELGIUM_CURRENT_AVERAGE_DEMAND,
+ telegram = Telegram()
+ telegram.add(
+ BELGIUM_CURRENT_AVERAGE_DEMAND,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(1.75), "unit": "kW"}],
),
- BELGIUM_MAXIMUM_DEMAND_MONTH: MBusObject(
- BELGIUM_MAXIMUM_DEMAND_MONTH,
+ "BELGIUM_CURRENT_AVERAGE_DEMAND",
+ )
+ telegram.add(
+ BELGIUM_MAXIMUM_DEMAND_MONTH,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642218)},
{"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,
+ "BELGIUM_MAXIMUM_DEMAND_MONTH",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_DEVICE_TYPE,
+ CosemObject((0, 1), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 1),
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
- BELGIUM_MBUS1_METER_READING2: MBusObject(
- BELGIUM_MBUS1_METER_READING2,
+ "BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_METER_READING2,
+ MBusObject(
+ (0, 1),
[
{"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,
+ "BELGIUM_MBUS1_METER_READING2",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS2_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 2),
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
- BELGIUM_MBUS2_METER_READING1: MBusObject(
- BELGIUM_MBUS2_METER_READING1,
+ "BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_METER_READING1,
+ MBusObject(
+ (0, 2),
[
{"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,
+ "BELGIUM_MBUS2_METER_READING1",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS3_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 3),
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
- BELGIUM_MBUS3_METER_READING2: MBusObject(
- BELGIUM_MBUS3_METER_READING2,
+ "BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_METER_READING2,
+ MBusObject(
+ (0, 3),
[
{"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,
+ "BELGIUM_MBUS3_METER_READING2",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS4_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 4),
[{"value": "37464C4F32313139303333373334", "unit": ""}],
),
- BELGIUM_MBUS4_METER_READING1: MBusObject(
- BELGIUM_MBUS4_METER_READING1,
+ "BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_METER_READING1,
+ MBusObject(
+ (0, 4),
[
{"value": datetime.datetime.fromtimestamp(1551642216)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- }
+ "BELGIUM_MBUS4_METER_READING1",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -680,64 +773,103 @@ async def test_belgian_meter_alt(
"time_between_update": 0,
}
- telegram = {
- BELGIUM_MBUS1_DEVICE_TYPE: CosemObject(
- BELGIUM_MBUS1_DEVICE_TYPE, [{"value": "007", "unit": ""}]
- ),
- BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: CosemObject(
- BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ telegram = Telegram()
+ telegram.add(
+ BELGIUM_MBUS1_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 1),
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
- BELGIUM_MBUS1_METER_READING1: MBusObject(
- BELGIUM_MBUS1_METER_READING1,
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_METER_READING1,
+ MBusObject(
+ (0, 1),
[
{"value": datetime.datetime.fromtimestamp(1551642215)},
{"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,
+ "BELGIUM_MBUS1_METER_READING1",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS2_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 2),
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
- BELGIUM_MBUS2_METER_READING2: MBusObject(
- BELGIUM_MBUS2_METER_READING2,
+ "BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_METER_READING2,
+ MBusObject(
+ (0, 2),
[
{"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,
+ BELGIUM_MBUS2_METER_READING2,
+ )
+ telegram.add(
+ BELGIUM_MBUS3_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS3_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 3),
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
- BELGIUM_MBUS3_METER_READING1: MBusObject(
- BELGIUM_MBUS3_METER_READING1,
+ "BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_METER_READING1,
+ MBusObject(
+ (0, 3),
[
{"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,
+ "BELGIUM_MBUS3_METER_READING1",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS4_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 4),
[{"value": "37464C4F32313139303333373334", "unit": ""}],
),
- BELGIUM_MBUS4_METER_READING2: MBusObject(
- BELGIUM_MBUS4_METER_READING2,
+ "BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_METER_READING2,
+ MBusObject(
+ (0, 4),
[
{"value": datetime.datetime.fromtimestamp(1551642218)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
- }
+ "BELGIUM_MBUS4_METER_READING2",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -829,49 +961,78 @@ async def test_belgian_meter_mbus(
"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,
+ telegram = Telegram()
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0003", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "006", "unit": ""}]),
+ "BELGIUM_MBUS1_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 1),
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
- BELGIUM_MBUS2_DEVICE_TYPE: CosemObject(
- BELGIUM_MBUS2_DEVICE_TYPE, [{"value": "003", "unit": ""}]
- ),
- BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: CosemObject(
- BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
+ "BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "003", "unit": ""}]),
+ "BELGIUM_MBUS2_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 2),
[{"value": "37464C4F32313139303333373332", "unit": ""}],
),
- BELGIUM_MBUS3_DEVICE_TYPE: CosemObject(
- BELGIUM_MBUS3_DEVICE_TYPE, [{"value": "007", "unit": ""}]
- ),
- BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER: CosemObject(
- BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
+ "BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS3_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER,
+ CosemObject(
+ (0, 3),
[{"value": "37464C4F32313139303333373333", "unit": ""}],
),
- BELGIUM_MBUS3_METER_READING2: MBusObject(
- BELGIUM_MBUS3_METER_READING2,
+ "BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER",
+ )
+ telegram.add(
+ BELGIUM_MBUS3_METER_READING2,
+ MBusObject(
+ (0, 3),
[
{"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,
+ "BELGIUM_MBUS3_METER_READING2",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_DEVICE_TYPE,
+ CosemObject((0, 0), [{"value": "007", "unit": ""}]),
+ "BELGIUM_MBUS4_DEVICE_TYPE",
+ )
+ telegram.add(
+ BELGIUM_MBUS4_METER_READING1,
+ MBusObject(
+ (0, 4),
[
{"value": datetime.datetime.fromtimestamp(1551642218)},
{"value": Decimal(13.13), "unit": "m3"},
],
),
- }
+ "BELGIUM_MBUS4_METER_READING1",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -938,11 +1099,12 @@ async def test_belgian_meter_low(
"time_between_update": 0,
}
- telegram = {
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0002", "unit": ""}]
- )
- }
+ telegram = Telegram()
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0002", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -986,16 +1148,23 @@ async def test_swedish_meter(
"time_between_update": 0,
}
- telegram = {
- ELECTRICITY_IMPORTED_TOTAL: CosemObject(
- ELECTRICITY_IMPORTED_TOTAL,
+ telegram = Telegram()
+ telegram.add(
+ ELECTRICITY_IMPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(123.456), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- ELECTRICITY_EXPORTED_TOTAL: CosemObject(
- ELECTRICITY_EXPORTED_TOTAL,
+ "ELECTRICITY_IMPORTED_TOTAL",
+ )
+ telegram.add(
+ ELECTRICITY_EXPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(654.321), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- }
+ "ELECTRICITY_EXPORTED_TOTAL",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
@@ -1054,16 +1223,23 @@ async def test_easymeter(
"time_between_update": 0,
}
- telegram = {
- ELECTRICITY_IMPORTED_TOTAL: CosemObject(
- ELECTRICITY_IMPORTED_TOTAL,
+ telegram = Telegram()
+ telegram.add(
+ ELECTRICITY_IMPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(54184.6316), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- ELECTRICITY_EXPORTED_TOTAL: CosemObject(
- ELECTRICITY_EXPORTED_TOTAL,
+ "ELECTRICITY_IMPORTED_TOTAL",
+ )
+ telegram.add(
+ ELECTRICITY_EXPORTED_TOTAL,
+ CosemObject(
+ (0, 0),
[{"value": Decimal(19981.1069), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
- }
+ "ELECTRICITY_EXPORTED_TOTAL",
+ )
mock_entry = MockConfigEntry(
domain="dsmr",
@@ -1222,15 +1398,20 @@ async def test_reconnect(
"time_between_update": 0,
}
- telegram = {
- CURRENT_ELECTRICITY_USAGE: CosemObject(
- CURRENT_ELECTRICITY_USAGE,
+ telegram = Telegram()
+ telegram.add(
+ CURRENT_ELECTRICITY_USAGE,
+ CosemObject(
+ (0, 0),
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
- ELECTRICITY_ACTIVE_TARIFF: CosemObject(
- ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
- ),
- }
+ "CURRENT_ELECTRICITY_USAGE",
+ )
+ telegram.add(
+ ELECTRICITY_ACTIVE_TARIFF,
+ CosemObject((0, 0), [{"value": "0001", "unit": ""}]),
+ "ELECTRICITY_ACTIVE_TARIFF",
+ )
# mock waiting coroutine while connection lasts
closed = asyncio.Event()
@@ -1300,15 +1481,18 @@ async def test_gas_meter_providing_energy_reading(
"time_between_update": 0,
}
- telegram = {
- GAS_METER_READING: MBusObject(
- GAS_METER_READING,
+ telegram = Telegram()
+ telegram.add(
+ GAS_METER_READING,
+ MBusObject(
+ (0, 0),
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(123.456), "unit": UnitOfEnergy.GIGA_JOULE},
],
),
- }
+ "GAS_METER_READING",
+ )
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
--
GitLab