Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Fix suggested UOM cannot be set for dsmr entities (#102134)

Browse Source 
* Supply dsmr entities jit on first telegram

* Stale docstr

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Simplify tuple type

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Jan Bouwhuis
2023-10-19 19:22:03 +02:00
committed by GitHub
parent c574cefc30
commit 9db9f1b8a9
2 changed files with 185 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,8 +23,6 @@ from homeassistant.const import (
ATTR_FRIENDLY_NAME,
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfEnergy,
UnitOfPower,
UnitOfVolume,
@@ -84,6 +82,14 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
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()
registry = er.async_get(hass)
entry = registry.async_get("sensor.electricity_meter_power_consumption")
@@ -94,11 +100,9 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
assert entry
assert entry.unique_id == "5678_gas_meter_reading"
telegram_callback = connection_factory.call_args_list[0][0][2]
# make sure entities have been created and return 'unavailable' state
# make sure entities are initialized
power_consumption = hass.states.get("sensor.electricity_meter_power_consumption")
assert power_consumption.state == STATE_UNAVAILABLE
assert power_consumption.state == "0.0"
assert (
power_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.POWER
)
@@ -107,7 +111,24 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
power_consumption.attributes.get(ATTR_STATE_CLASS)
== SensorStateClass.MEASUREMENT
)
assert power_consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
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)
@@ -117,7 +138,7 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
# ensure entities have new state value after incoming telegram
power_consumption = hass.states.get("sensor.electricity_meter_power_consumption")
assert power_consumption.state == "0.0"
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
@@ -131,11 +152,11 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
)
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) == ""
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.state == "745.701"
assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS
assert (
gas_consumption.attributes.get(ATTR_FRIENDLY_NAME)
@@ -153,6 +174,14 @@ async def test_default_setup(hass: HomeAssistant, dsmr_connection_fixture) -> No
async def test_setup_only_energy(hass: HomeAssistant, 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",
@@ -160,9 +189,22 @@ async def test_setup_only_energy(hass: HomeAssistant, dsmr_connection_fixture) -
"reconnect_interval": 30,
"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
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
)
mock_entry.add_to_hass(hass)
@@ -170,6 +212,14 @@ async def test_setup_only_energy(hass: HomeAssistant, dsmr_connection_fixture) -
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()
registry = er.async_get(hass)
entry = registry.async_get("sensor.electricity_meter_power_consumption")
@@ -229,8 +279,8 @@ async def test_v4_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
# 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)
# 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")
@@ -239,7 +289,7 @@ async def test_v4_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
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) == ""
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")
@@ -308,8 +358,8 @@ async def test_v5_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
# 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)
# 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")
@@ -318,7 +368,7 @@ async def test_v5_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
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) == ""
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")
@@ -389,8 +439,8 @@ async def test_luxembourg_meter(hass: HomeAssistant, dsmr_connection_fixture) ->
# 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)
# 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"
@@ -472,8 +522,8 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
# 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)
# 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")
@@ -482,7 +532,7 @@ async def test_belgian_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
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) == ""
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")
@@ -537,8 +587,8 @@ async def test_belgian_meter_low(hass: HomeAssistant, dsmr_connection_fixture) -
# 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)
# 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")
@@ -547,7 +597,7 @@ async def test_belgian_meter_low(hass: HomeAssistant, dsmr_connection_fixture) -
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) == ""
assert active_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
async def test_swedish_meter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
@@ -597,8 +647,8 @@ async def test_swedish_meter(hass: HomeAssistant, dsmr_connection_fixture) -> No
# 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)
# 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"
@@ -675,8 +725,8 @@ async def test_easymeter(hass: HomeAssistant, dsmr_connection_fixture) -> None:
# 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)
# 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.6316"
@@ -800,6 +850,12 @@ async def test_connection_errors_retry(
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 = {
@@ -810,6 +866,19 @@ async def test_reconnect(hass: HomeAssistant, dsmr_connection_fixture) -> None:
"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()
@@ -823,7 +892,7 @@ async def test_reconnect(hass: HomeAssistant, dsmr_connection_fixture) -> None:
protocol.wait_closed = wait_closed
mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options
)
mock_entry.add_to_hass(hass)
@@ -831,11 +900,19 @@ async def test_reconnect(hass: HomeAssistant, dsmr_connection_fixture) -> None:
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 == STATE_UNKNOWN
assert state.state == "35.0"
# indicate disconnect, release wait lock and allow reconnect to happen
closed.set()
@@ -897,7 +974,7 @@ async def test_gas_meter_providing_energy_reading(
telegram_callback = connection_factory.call_args_list[0][0][2]
telegram_callback(telegram)
await asyncio.sleep(0)
await hass.async_block_till_done()
gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption")
assert gas_consumption.state == "123.456"