Add DSMR config options for EasyMeter/Q3D

homeassistant/components/dsmr/config_flow.py

@@ -46,6 +46,8 @@ def __init__(self, host: str | None, port: int, dsmr_version: str) -> None:
         self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER
         if dsmr_version == "5L":
             self._equipment_identifier = obis_ref.LUXEMBOURG_EQUIPMENT_IDENTIFIER
+        if dsmr_version == "Q3D":
+            self._equipment_identifier = obis_ref.Q3D_EQUIPMENT_IDENTIFIER
 
     def equipment_identifier(self) -> str | None:
         """Equipment identifier."""

homeassistant/components/dsmr/const.py

@@ -35,7 +35,7 @@
 DEVICE_NAME_ENERGY = "Energy Meter"
 DEVICE_NAME_GAS = "Gas Meter"
 
-DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S"}
+DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S", "Q3D"}
 
 SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
     DSMRSensorEntityDescription(
@@ -244,15 +244,15 @@
     DSMRSensorEntityDescription(
         key=obis_references.ELECTRICITY_IMPORTED_TOTAL,
         name="Energy Consumption (total)",
-        dsmr_versions={"5", "5B", "5L", "5S"},
+        dsmr_versions={"5", "5B", "5L", "5S", "Q3D"},
         force_update=True,
         device_class=SensorDeviceClass.ENERGY,
         state_class=SensorStateClass.TOTAL_INCREASING,
     ),
     DSMRSensorEntityDescription(
         key=obis_references.ELECTRICITY_EXPORTED_TOTAL,
         name="Energy Production (total)",
-        dsmr_versions={"5L", "5S"},
+        dsmr_versions={"5L", "5S", "Q3D"},
         force_update=True,
         device_class=SensorDeviceClass.ENERGY,
         state_class=SensorStateClass.TOTAL_INCREASING,

tests/components/dsmr/conftest.py

@@ -8,6 +8,7 @@
     EQUIPMENT_IDENTIFIER_GAS,
     LUXEMBOURG_EQUIPMENT_IDENTIFIER,
     P1_MESSAGE_TIMESTAMP,
+    Q3D_EQUIPMENT_IDENTIFIER,
 )
 from dsmr_parser.objects import CosemObject
 import pytest
@@ -63,6 +64,12 @@ async def connection_factory(*args, **kwargs):
             protocol.telegram = {
                 P1_MESSAGE_TIMESTAMP: CosemObject([{"value": "12345678", "unit": ""}]),
             }
+        if args[1] == "Q3D":
+            protocol.telegram = {
+                Q3D_EQUIPMENT_IDENTIFIER: CosemObject(
+                    [{"value": "12345678", "unit": ""}]
+                ),
+            }
 
         return (transport, protocol)
 

tests/components/dsmr/test_config_flow.py

@@ -99,6 +99,84 @@ async def test_setup_serial(com_mock, hass, dsmr_connection_send_validate_fixtur
     assert result["data"] == {**entry_data, **SERIAL_DATA}
 
 
+@patch("serial.tools.list_ports.comports", return_value=[com_port()])
+async def test_setup_5L(com_mock, hass, dsmr_connection_send_validate_fixture):
+    """Test we can setup serial."""
+    port = com_port()
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"type": "Serial"},
+    )
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "setup_serial"
+    assert result["errors"] == {}
+
+    with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"port": port.device, "dsmr_version": "5L"}
+        )
+
+    entry_data = {
+        "port": port.device,
+        "dsmr_version": "5L",
+        "serial_id": "12345678",
+        "serial_id_gas": "123456789",
+    }
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == port.device
+    assert result["data"] == entry_data
+
+
+@patch("serial.tools.list_ports.comports", return_value=[com_port()])
+async def test_setup_Q3D(com_mock, hass, dsmr_connection_send_validate_fixture):
+    """Test we can setup serial."""
+    port = com_port()
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"type": "Serial"},
+    )
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "setup_serial"
+    assert result["errors"] == {}
+
+    with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"port": port.device, "dsmr_version": "Q3D"}
+        )
+
+    entry_data = {
+        "port": port.device,
+        "dsmr_version": "Q3D",
+        "serial_id": "12345678",
+        "serial_id_gas": None,
+    }
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == port.device
+    assert result["data"] == entry_data
+
+
 @patch("serial.tools.list_ports.comports", return_value=[com_port()])
 async def test_setup_serial_manual(
     com_mock, hass, dsmr_connection_send_validate_fixture

tests/components/dsmr/test_sensor.py

@@ -576,6 +576,80 @@ async def test_swedish_meter(hass, dsmr_connection_fixture):
     )
 
 
+async def test_easymeter(hass, dsmr_connection_fixture):
+    """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",
+        "precision": 4,
+        "reconnect_interval": 30,
+        "serial_id": None,
+        "serial_id_gas": None,
+    }
+    entry_options = {
+        "time_between_update": 0,
+    }
+
+    telegram = {
+        ELECTRICITY_IMPORTED_TOTAL: CosemObject(
+            [{"value": Decimal(54184.6316), "unit": ENERGY_KILO_WATT_HOUR}]
+        ),
+        ELECTRICITY_EXPORTED_TOTAL: CosemObject(
+            [{"value": Decimal(19981.1069), "unit": ENERGY_KILO_WATT_HOUR}]
+        ),
+    }
+
+    mock_entry = MockConfigEntry(
+        domain="dsmr",
+        unique_id="/dev/ttyUSB0",
+        data=entry_data,
+        options=entry_options,
+    )
+
+    mock_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+
+    telegram_callback = connection_factory.call_args_list[0][0][2]
+
+    # simulate a telegram pushed from the smartmeter and parsed by dsmr_parser
+    telegram_callback(telegram)
+
+    # after receiving telegram entities need to have the chance to update
+    await asyncio.sleep(0)
+
+    power_tariff = hass.states.get("sensor.energy_consumption_total")
+    assert power_tariff.state == "54184.6316"
+    assert power_tariff.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
+    assert power_tariff.attributes.get(ATTR_ICON) is None
+    assert (
+        power_tariff.attributes.get(ATTR_STATE_CLASS)
+        == SensorStateClass.TOTAL_INCREASING
+    )
+    assert (
+        power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ENERGY_KILO_WATT_HOUR
+    )
+
+    power_tariff = hass.states.get("sensor.energy_production_total")
+    assert power_tariff.state == "19981.1069"
+    assert (
+        power_tariff.attributes.get(ATTR_STATE_CLASS)
+        == SensorStateClass.TOTAL_INCREASING
+    )
+    assert (
+        power_tariff.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == ENERGY_KILO_WATT_HOUR
+    )
+
+
 async def test_tcp(hass, dsmr_connection_fixture):
     """If proper config provided TCP connection should be made."""
     (connection_factory, transport, protocol) = dsmr_connection_fixture