evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
hass-core/tests/components/dsmr/test_sensor.py
dupondje 0aaa517217
add missing state class to dsmr sensors (#110712) 
current_average_demand and maximum_demand_current_month are missing a state class, which gives errors in the statistics tab in HA.
So adding them to the sensor.
2024-02-16 12:08:09 +01:00

1392 lines
47 KiB
Python
Raw Blame History

"""Test for DSMR components.
Tests setup of the DSMR component and ensure incoming telegrams cause
Entity to be updated with new values.
"""
import asyncio
import datetime
from decimal import Decimal
from itertools import chain, repeat
from unittest.mock import DEFAULT, MagicMock
import pytest
from homeassistant import config_entries
from homeassistant.components.sensor import (
ATTR_OPTIONS,
ATTR_STATE_CLASS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_FRIENDLY_NAME,
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNKNOWN,
UnitOfEnergy,
UnitOfPower,
UnitOfVolume,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, patch
async def test_default_setup(
hass: HomeAssistant, entity_registry: er.EntityRegistry, dsmr_connection_fixture
) -> None:
"""Test the default setup."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
CURRENT_ELECTRICITY_USAGE,
ELECTRICITY_ACTIVE_TARIFF,
GAS_METER_READING,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
CURRENT_ELECTRICITY_USAGE: CosemObject(
CURRENT_ELECTRICITY_USAGE,
[{"value": Decimal("0.0"), "unit": UnitOfPower.WATT}],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
GAS_METER_READING: MBusObject(
GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": UnitOfVolume.CUBIC_METERS},
],
),
}
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()
entry = entity_registry.async_get("sensor.electricity_meter_power_consumption")
assert entry
assert entry.unique_id == "1234_current_electricity_usage"
entry = entity_registry.async_get("sensor.gas_meter_gas_consumption")
assert entry
assert entry.unique_id == "5678_gas_meter_reading"
# make sure entities are initialized
power_consumption = hass.states.get("sensor.electricity_meter_power_consumption")
assert power_consumption.state == "0.0"
assert (
power_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.POWER
)
assert power_consumption.attributes.get(ATTR_ICON) is None
assert (
power_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.MEASUREMENT
)
assert power_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "W"
telegram = {
CURRENT_ELECTRICITY_USAGE: CosemObject(
CURRENT_ELECTRICITY_USAGE,
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
GAS_METER_READING: MBusObject(
GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642214)},
{"value": Decimal(745.701), "unit": UnitOfVolume.CUBIC_METERS},
],
),
}
# 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()
# ensure entities have new state value after incoming telegram
power_consumption = hass.states.get("sensor.electricity_meter_power_consumption")
assert power_consumption.state == "35.0"
assert power_consumption.attributes.get("unit_of_measurement") == UnitOfPower.WATT
# 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 == "low"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENUM
assert active_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert (
active_tariff.attributes.get(ATTR_FRIENDLY_NAME)
== "Electricity Meter Active tariff"
)
assert active_tariff.attributes.get(ATTR_OPTIONS) == ["low", "normal"]
assert active_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == "745.701"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
assert (
gas_consumption.attributes.get(ATTR_FRIENDLY_NAME)
== "Gas Meter Gas consumption"
)
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_setup_only_energy(
hass: HomeAssistant, entity_registry: er.EntityRegistry, dsmr_connection_fixture
) -> None:
"""Test the default setup."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
CURRENT_ELECTRICITY_USAGE,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
"serial_id": "1234",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
CURRENT_ELECTRICITY_USAGE: CosemObject(
CURRENT_ELECTRICITY_USAGE,
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
}
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()
entry = entity_registry.async_get("sensor.electricity_meter_power_consumption")
assert entry
assert entry.unique_id == "1234_current_electricity_usage"
entry = entity_registry.async_get("sensor.gas_meter_gas_consumption")
assert not entry
async def test_v4_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if v4 meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
ELECTRICITY_ACTIVE_TARIFF,
HOURLY_GAS_METER_READING,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "4",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
HOURLY_GAS_METER_READING: MBusObject(
HOURLY_GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
}
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 == "low"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENUM
assert active_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert active_tariff.attributes.get(ATTR_OPTIONS) == ["low", "normal"]
assert active_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
# check if gas consumption 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
assert (
gas_consumption.attributes.get("unit_of_measurement")
== UnitOfVolume.CUBIC_METERS
)
assert (
gas_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfVolume.CUBIC_METERS
)
@pytest.mark.parametrize(
("value", "state"),
[
(Decimal(745.690), "745.69"),
(Decimal(745.695), "745.695"),
(Decimal(0.000), STATE_UNKNOWN),
],
)
async def test_v5_meter(
hass: HomeAssistant, dsmr_connection_fixture, value: Decimal, state: str
) -> None:
"""Test if v5 meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
ELECTRICITY_ACTIVE_TARIFF,
HOURLY_GAS_METER_READING,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
HOURLY_GAS_METER_READING: MBusObject(
HOURLY_GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": value, "unit": "m3"},
],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
}
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 == "low"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENUM
assert active_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert active_tariff.attributes.get(ATTR_OPTIONS) == ["low", "normal"]
assert active_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
# check if gas consumption is parsed correctly
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == state
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_luxembourg_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if v5 meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
ELECTRICITY_EXPORTED_TOTAL,
ELECTRICITY_IMPORTED_TOTAL,
HOURLY_GAS_METER_READING,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5L",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
HOURLY_GAS_METER_READING: MBusObject(
HOURLY_GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(745.695), "unit": "m3"},
],
),
ELECTRICITY_IMPORTED_TOTAL: CosemObject(
ELECTRICITY_IMPORTED_TOTAL,
[{"value": Decimal(123.456), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
ELECTRICITY_EXPORTED_TOTAL: CosemObject(
ELECTRICITY_EXPORTED_TOTAL,
[{"value": Decimal(654.321), "unit": UnitOfEnergy.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 be created
await hass.async_block_till_done()
active_tariff = hass.states.get("sensor.electricity_meter_energy_consumption_total")
assert active_tariff.state == "123.456"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
assert active_tariff.attributes.get(ATTR_ICON) is None
assert (
active_tariff.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
active_tariff = hass.states.get("sensor.electricity_meter_energy_production_total")
assert active_tariff.state == "654.321"
assert (
active_tariff.attributes.get("unit_of_measurement")
== UnitOfEnergy.KILO_WATT_HOUR
)
# check if gas consumption 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
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(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_CURRENT_AVERAGE_DEMAND,
BELGIUM_MAXIMUM_DEMAND_MONTH,
BELGIUM_MBUS1_DEVICE_TYPE,
BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS1_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_DEVICE_TYPE,
BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER,
BELGIUM_MBUS4_METER_READING1,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject, MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"serial_id": "1234",
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
BELGIUM_CURRENT_AVERAGE_DEMAND: CosemObject(
BELGIUM_CURRENT_AVERAGE_DEMAND,
[{"value": Decimal(1.75), "unit": "kW"}],
),
BELGIUM_MAXIMUM_DEMAND_MONTH: MBusObject(
BELGIUM_MAXIMUM_DEMAND_MONTH,
[
{"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,
[{"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": ""}]
),
}
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 == "normal"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENUM
assert active_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert active_tariff.attributes.get(ATTR_OPTIONS) == ["low", "normal"]
assert active_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
# check current average demand is parsed correctly
avg_demand = hass.states.get("sensor.electricity_meter_current_average_demand")
assert avg_demand.state == "1.75"
assert avg_demand.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfPower.KILO_WATT
assert avg_demand.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
# check max average demand is parsed correctly
max_demand = hass.states.get(
"sensor.electricity_meter_maximum_demand_current_month"
)
assert max_demand.state == "4.11"
assert max_demand.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfPower.KILO_WATT
assert max_demand.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
# 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
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")
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
)
# 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.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",
"dsmr_version": "5B",
"serial_id": "1234",
"serial_id_gas": None,
}
entry_options = {
"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,
[{"value": "37464C4F32313139303333373331", "unit": ""}],
),
BELGIUM_MBUS1_METER_READING1: MBusObject(
BELGIUM_MBUS1_METER_READING1,
[
{"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,
[{"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"},
],
),
}
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 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 == "678.901"
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 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",
"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."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ELECTRICITY_ACTIVE_TARIFF
from dsmr_parser.objects import CosemObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5B",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0002", "unit": ""}]
)
}
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 == "low"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENUM
assert active_tariff.attributes.get(ATTR_ICON) == "mdi:flash"
assert active_tariff.attributes.get(ATTR_OPTIONS) == ["low", "normal"]
assert active_tariff.attributes.get(ATTR_STATE_CLASS) is None
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
async def test_swedish_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if v5 meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
ELECTRICITY_EXPORTED_TOTAL,
ELECTRICITY_IMPORTED_TOTAL,
)
from dsmr_parser.objects import CosemObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "5S",
"serial_id": None,
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
ELECTRICITY_IMPORTED_TOTAL: CosemObject(
ELECTRICITY_IMPORTED_TOTAL,
[{"value": Decimal(123.456), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
ELECTRICITY_EXPORTED_TOTAL: CosemObject(
ELECTRICITY_EXPORTED_TOTAL,
[{"value": Decimal(654.321), "unit": UnitOfEnergy.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 be created
await hass.async_block_till_done()
active_tariff = hass.states.get("sensor.electricity_meter_energy_consumption_total")
assert active_tariff.state == "123.456"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
assert active_tariff.attributes.get(ATTR_ICON) is None
assert (
active_tariff.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
active_tariff = hass.states.get("sensor.electricity_meter_energy_production_total")
assert active_tariff.state == "654.321"
assert (
active_tariff.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
async def test_easymeter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""Test if Q3D meter is correctly parsed."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import (
ELECTRICITY_EXPORTED_TOTAL,
ELECTRICITY_IMPORTED_TOTAL,
)
from dsmr_parser.objects import CosemObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "Q3D",
"serial_id": None,
"serial_id_gas": None,
}
entry_options = {
"time_between_update": 0,
}
telegram = {
ELECTRICITY_IMPORTED_TOTAL: CosemObject(
ELECTRICITY_IMPORTED_TOTAL,
[{"value": Decimal(54184.6316), "unit": UnitOfEnergy.KILO_WATT_HOUR}],
),
ELECTRICITY_EXPORTED_TOTAL: CosemObject(
ELECTRICITY_EXPORTED_TOTAL,
[{"value": Decimal(19981.1069), "unit": UnitOfEnergy.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 be created
await hass.async_block_till_done()
active_tariff = hass.states.get("sensor.electricity_meter_energy_consumption_total")
assert active_tariff.state == "54184.632"
assert active_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
assert active_tariff.attributes.get(ATTR_ICON) is None
assert (
active_tariff.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
active_tariff = hass.states.get("sensor.electricity_meter_energy_production_total")
assert active_tariff.state == "19981.107"
assert (
active_tariff.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
async def test_tcp(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""If proper config provided TCP connection should be made."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = {
"host": "localhost",
"port": "1234",
"dsmr_version": "2.2",
"protocol": "dsmr_protocol",
"serial_id": "1234",
"serial_id_gas": "5678",
}
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert connection_factory.call_args_list[0][0][0] == "localhost"
assert connection_factory.call_args_list[0][0][1] == "1234"
async def test_rfxtrx_tcp(hass: HomeAssistant, rfxtrx_dsmr_connection_fixture) -> None:
"""If proper config provided RFXtrx TCP connection should be made."""
(connection_factory, transport, protocol) = rfxtrx_dsmr_connection_fixture
entry_data = {
"host": "localhost",
"port": "1234",
"dsmr_version": "2.2",
"protocol": "rfxtrx_dsmr_protocol",
"serial_id": "1234",
"serial_id_gas": "5678",
}
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert connection_factory.call_args_list[0][0][0] == "localhost"
assert connection_factory.call_args_list[0][0][1] == "1234"
@patch("homeassistant.components.dsmr.sensor.DEFAULT_RECONNECT_INTERVAL", 0)
async def test_connection_errors_retry(
hass: HomeAssistant, dsmr_connection_fixture
) -> None:
"""Connection should be retried on error during setup."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
"serial_id": "1234",
"serial_id_gas": "5678",
}
# override the mock to have it fail the first time and succeed after
first_fail_connection_factory = MagicMock(
return_value=(transport, protocol),
side_effect=chain([TimeoutError], repeat(DEFAULT)),
)
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data
)
mock_entry.add_to_hass(hass)
with patch(
"homeassistant.components.dsmr.sensor.create_dsmr_reader",
first_fail_connection_factory,
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# wait for sleep to resolve
await hass.async_block_till_done()
assert first_fail_connection_factory.call_count >= 2, "connecting not retried"
@patch("homeassistant.components.dsmr.sensor.DEFAULT_RECONNECT_INTERVAL", 0)
async def test_reconnect(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"""If transport disconnects, the connection should be retried."""
from dsmr_parser.obis_references import (
CURRENT_ELECTRICITY_USAGE,
ELECTRICITY_ACTIVE_TARIFF,
)
from dsmr_parser.objects import CosemObject
(connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
CURRENT_ELECTRICITY_USAGE: CosemObject(
CURRENT_ELECTRICITY_USAGE,
[{"value": Decimal("35.0"), "unit": UnitOfPower.WATT}],
),
ELECTRICITY_ACTIVE_TARIFF: CosemObject(
ELECTRICITY_ACTIVE_TARIFF, [{"value": "0001", "unit": ""}]
),
}
# mock waiting coroutine while connection lasts
closed = asyncio.Event()
# Handshake so that `hass.async_block_till_done()` doesn't cycle forever
closed2 = asyncio.Event()
async def wait_closed():
await closed.wait()
closed2.set()
protocol.wait_closed = wait_closed
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()
assert connection_factory.call_count == 1
state = hass.states.get("sensor.electricity_meter_power_consumption")
assert state
assert state.state == "35.0"
# indicate disconnect, release wait lock and allow reconnect to happen
closed.set()
# wait for lock set to resolve
await closed2.wait()
closed2.clear()
closed.clear()
await hass.async_block_till_done()
assert connection_factory.call_count >= 2, "connecting not retried"
# setting it so teardown can be successful
closed.set()
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == config_entries.ConfigEntryState.NOT_LOADED
async def test_gas_meter_providing_energy_reading(
hass: HomeAssistant, dsmr_connection_fixture
) -> None:
"""Test that gas providing energy readings use the correct device class."""
(connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import GAS_METER_READING
from dsmr_parser.objects import MBusObject
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
"serial_id": "1234",
"serial_id_gas": "5678",
}
entry_options = {
"time_between_update": 0,
}
telegram = {
GAS_METER_READING: MBusObject(
GAS_METER_READING,
[
{"value": datetime.datetime.fromtimestamp(1551642213)},
{"value": Decimal(123.456), "unit": UnitOfEnergy.GIGA_JOULE},
],
),
}
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]
telegram_callback(telegram)
await hass.async_block_till_done()
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == "123.456"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
assert (
gas_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.TOTAL_INCREASING
)
assert (
gas_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.GIGA_JOULE
)
Reference in a new issue View git blame Copy permalink