jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Correct deprecated telegram usage in dsmr (#121847)

Browse Source
This commit is contained in:
Joakim Plate 2024-07-12 16:00:27 +02:00 committed by GitHub
parent 244af091b8
commit a0d0e0f4a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 399 additions and 197 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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 (

48
tests/components/dsmr/test_mbus_migration.py
View File

@ -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(
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_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(
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_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()

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

@ -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(
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(
"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(
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(
"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(
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(
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(
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(
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(
"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_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(
telegram = Telegram()
telegram.add(
BELGIUM_CURRENT_AVERAGE_DEMAND,
CosemObject(
(0, 0),
[{"value": Decimal(1.75), "unit": "kW"}],
),
BELGIUM_MAXIMUM_DEMAND_MONTH: MBusObject(
"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_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_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_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_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_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_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_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_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(
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_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_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_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_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_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_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_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(
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_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_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_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_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(
telegram = Telegram()
telegram.add(
ELECTRICITY_IMPORTED_TOTAL,
CosemObject(
(0, 0),
[{"value": Decimal(123.456), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
ELECTRICITY_EXPORTED_TOTAL: CosemObject(
"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(
telegram = Telegram()
telegram.add(
ELECTRICITY_IMPORTED_TOTAL,
CosemObject(
(0, 0),
[{"value": Decimal(54184.6316), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
ELECTRICITY_EXPORTED_TOTAL: CosemObject(
"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(
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(
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