Support DSMR data read via RFXtrx with integrated P1 reader (#63529)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/dsmr/config_flow.py

@@ -9,6 +9,10 @@ from typing import Any
 from async_timeout import timeout
 from dsmr_parser import obis_references as obis_ref
 from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
+from dsmr_parser.clients.rfxtrx_protocol import (
+    create_rfxtrx_dsmr_reader,
+    create_rfxtrx_tcp_dsmr_reader,
+)
 from dsmr_parser.objects import DSMRObject
 import serial
 import serial.tools.list_ports
@@ -22,13 +26,16 @@ from homeassistant.data_entry_flow import FlowResult
 
 from .const import (
     CONF_DSMR_VERSION,
+    CONF_PROTOCOL,
     CONF_SERIAL_ID,
     CONF_SERIAL_ID_GAS,
     CONF_TIME_BETWEEN_UPDATE,
     DEFAULT_TIME_BETWEEN_UPDATE,
     DOMAIN,
+    DSMR_PROTOCOL,
     DSMR_VERSIONS,
     LOGGER,
+    RFXTRX_DSMR_PROTOCOL,
 )
 
 CONF_MANUAL_PATH = "Enter Manually"
@@ -37,11 +44,14 @@ CONF_MANUAL_PATH = "Enter Manually"
 class DSMRConnection:
     """Test the connection to DSMR and receive telegram to read serial ids."""
 
-    def __init__(self, host: str | None, port: int, dsmr_version: str) -> None:
+    def __init__(
+        self, host: str | None, port: int, dsmr_version: str, protocol: str
+    ) -> None:
         """Initialize."""
         self._host = host
         self._port = port
         self._dsmr_version = dsmr_version
+        self._protocol = protocol
         self._telegram: dict[str, DSMRObject] = {}
         self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER
         if dsmr_version == "5L":
@@ -78,16 +88,24 @@ class DSMRConnection:
                 transport.close()
 
         if self._host is None:
+            if self._protocol == DSMR_PROTOCOL:
+                create_reader = create_dsmr_reader
+            else:
+                create_reader = create_rfxtrx_dsmr_reader
             reader_factory = partial(
-                create_dsmr_reader,
+                create_reader,
                 self._port,
                 self._dsmr_version,
                 update_telegram,
                 loop=hass.loop,
             )
         else:
+            if self._protocol == DSMR_PROTOCOL:
+                create_reader = create_tcp_dsmr_reader
+            else:
+                create_reader = create_rfxtrx_tcp_dsmr_reader
             reader_factory = partial(
-                create_tcp_dsmr_reader,
+                create_reader,
                 self._host,
                 self._port,
                 self._dsmr_version,
@@ -113,10 +131,15 @@ class DSMRConnection:
 
 
 async def _validate_dsmr_connection(
-    hass: core.HomeAssistant, data: dict[str, Any]
+    hass: core.HomeAssistant, data: dict[str, Any], protocol: str
 ) -> dict[str, str | None]:
     """Validate the user input allows us to connect."""
-    conn = DSMRConnection(data.get(CONF_HOST), data[CONF_PORT], data[CONF_DSMR_VERSION])
+    conn = DSMRConnection(
+        data.get(CONF_HOST),
+        data[CONF_PORT],
+        data[CONF_DSMR_VERSION],
+        protocol,
+    )
 
     if not await conn.validate_connect(hass):
         raise CannotConnect
@@ -260,9 +283,14 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         data = input_data
 
         try:
-            info = await _validate_dsmr_connection(self.hass, data)
+            try:
+                protocol = DSMR_PROTOCOL
+                info = await _validate_dsmr_connection(self.hass, data, protocol)
+            except CannotCommunicate:
+                protocol = RFXTRX_DSMR_PROTOCOL
+                info = await _validate_dsmr_connection(self.hass, data, protocol)
 
-            data = {**data, **info}
+            data = {**data, **info, CONF_PROTOCOL: protocol}
 
             if info[CONF_SERIAL_ID]:
                 await self.async_set_unique_id(info[CONF_SERIAL_ID])

homeassistant/components/dsmr/const.py

@@ -17,6 +17,7 @@ LOGGER = logging.getLogger(__package__)
 
 PLATFORMS = [Platform.SENSOR]
 CONF_DSMR_VERSION = "dsmr_version"
+CONF_PROTOCOL = "protocol"
 CONF_RECONNECT_INTERVAL = "reconnect_interval"
 CONF_PRECISION = "precision"
 CONF_TIME_BETWEEN_UPDATE = "time_between_update"
@@ -37,6 +38,9 @@ DEVICE_NAME_GAS = "Gas Meter"
 
 DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S", "Q3D"}
 
+DSMR_PROTOCOL = "dsmr_protocol"
+RFXTRX_DSMR_PROTOCOL = "rfxtrx_dsmr_protocol"
+
 SENSORS: tuple[DSMRSensorEntityDescription, ...] = (
     DSMRSensorEntityDescription(
         key=obis_references.CURRENT_ELECTRICITY_USAGE,

homeassistant/components/dsmr/sensor.py

@@ -9,6 +9,10 @@ from functools import partial
 
 from dsmr_parser import obis_references as obis_ref
 from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
+from dsmr_parser.clients.rfxtrx_protocol import (
+    create_rfxtrx_dsmr_reader,
+    create_rfxtrx_tcp_dsmr_reader,
+)
 from dsmr_parser.objects import DSMRObject
 import serial
 
@@ -29,6 +33,7 @@ from homeassistant.util import Throttle
 from .const import (
     CONF_DSMR_VERSION,
     CONF_PRECISION,
+    CONF_PROTOCOL,
     CONF_RECONNECT_INTERVAL,
     CONF_SERIAL_ID,
     CONF_SERIAL_ID_GAS,
@@ -40,6 +45,7 @@ from .const import (
     DEVICE_NAME_ELECTRICITY,
     DEVICE_NAME_GAS,
     DOMAIN,
+    DSMR_PROTOCOL,
     LOGGER,
     SENSORS,
 )
@@ -77,9 +83,14 @@ async def async_setup_entry(
 
     # Creates an asyncio.Protocol factory for reading DSMR telegrams from
     # serial and calls update_entities_telegram to update entities on arrival
+    protocol = entry.data.get(CONF_PROTOCOL, DSMR_PROTOCOL)
     if CONF_HOST in entry.data:
+        if protocol == DSMR_PROTOCOL:
+            create_reader = create_tcp_dsmr_reader
+        else:
+            create_reader = create_rfxtrx_tcp_dsmr_reader
         reader_factory = partial(
-            create_tcp_dsmr_reader,
+            create_reader,
             entry.data[CONF_HOST],
             entry.data[CONF_PORT],
             dsmr_version,
@@ -88,8 +99,12 @@ async def async_setup_entry(
             keep_alive_interval=60,
         )
     else:
+        if protocol == DSMR_PROTOCOL:
+            create_reader = create_dsmr_reader
+        else:
+            create_reader = create_rfxtrx_dsmr_reader
         reader_factory = partial(
-            create_dsmr_reader,
+            create_reader,
             entry.data[CONF_PORT],
             dsmr_version,
             update_entities_telegram,

tests/components/dsmr/conftest.py

@@ -3,6 +3,7 @@ import asyncio
 from unittest.mock import MagicMock, patch
 
 from dsmr_parser.clients.protocol import DSMRProtocol
+from dsmr_parser.clients.rfxtrx_protocol import RFXtrxDSMRProtocol
 from dsmr_parser.obis_references import (
     EQUIPMENT_IDENTIFIER,
     EQUIPMENT_IDENTIFIER_GAS,
@@ -36,6 +37,29 @@ async def dsmr_connection_fixture(hass):
         yield (connection_factory, transport, protocol)
 
 
+@pytest.fixture
+async def rfxtrx_dsmr_connection_fixture(hass):
+    """Fixture that mocks RFXtrx connection."""
+
+    transport = MagicMock(spec=asyncio.Transport)
+    protocol = MagicMock(spec=RFXtrxDSMRProtocol)
+
+    async def connection_factory(*args, **kwargs):
+        """Return mocked out Asyncio classes."""
+        return (transport, protocol)
+
+    connection_factory = MagicMock(wraps=connection_factory)
+
+    with patch(
+        "homeassistant.components.dsmr.sensor.create_rfxtrx_dsmr_reader",
+        connection_factory,
+    ), patch(
+        "homeassistant.components.dsmr.sensor.create_rfxtrx_tcp_dsmr_reader",
+        connection_factory,
+    ):
+        yield (connection_factory, transport, protocol)
+
+
 @pytest.fixture
 async def dsmr_connection_send_validate_fixture(hass):
     """Fixture that mocks serial connection."""
@@ -95,3 +119,43 @@ async def dsmr_connection_send_validate_fixture(hass):
         connection_factory,
     ):
         yield (connection_factory, transport, protocol)
+
+
+@pytest.fixture
+async def rfxtrx_dsmr_connection_send_validate_fixture(hass):
+    """Fixture that mocks serial connection."""
+
+    transport = MagicMock(spec=asyncio.Transport)
+    protocol = MagicMock(spec=RFXtrxDSMRProtocol)
+
+    protocol.telegram = {
+        EQUIPMENT_IDENTIFIER: CosemObject([{"value": "12345678", "unit": ""}]),
+        EQUIPMENT_IDENTIFIER_GAS: CosemObject([{"value": "123456789", "unit": ""}]),
+        P1_MESSAGE_TIMESTAMP: CosemObject([{"value": "12345678", "unit": ""}]),
+    }
+
+    async def connection_factory(*args, **kwargs):
+        return (transport, protocol)
+
+    connection_factory = MagicMock(wraps=connection_factory)
+
+    async def wait_closed():
+        if isinstance(connection_factory.call_args_list[0][0][2], str):
+            # TCP
+            telegram_callback = connection_factory.call_args_list[0][0][3]
+        else:
+            # Serial
+            telegram_callback = connection_factory.call_args_list[0][0][2]
+
+        telegram_callback(protocol.telegram)
+
+    protocol.wait_closed = wait_closed
+
+    with patch(
+        "homeassistant.components.dsmr.config_flow.create_rfxtrx_dsmr_reader",
+        connection_factory,
+    ), patch(
+        "homeassistant.components.dsmr.config_flow.create_rfxtrx_tcp_dsmr_reader",
+        connection_factory,
+    ):
+        yield (connection_factory, transport, protocol)

tests/components/dsmr/test_config_flow.py

@@ -49,13 +49,70 @@ async def test_setup_network(hass, dsmr_connection_send_validate_fixture):
     with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"],
-            {"host": "10.10.0.1", "port": 1234, "dsmr_version": "2.2"},
+            {
+                "host": "10.10.0.1",
+                "port": 1234,
+                "dsmr_version": "2.2",
+            },
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "host": "10.10.0.1",
         "port": 1234,
         "dsmr_version": "2.2",
+        "protocol": "dsmr_protocol",
+    }
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == "10.10.0.1:1234"
+    assert result["data"] == {**entry_data, **SERIAL_DATA}
+
+
+async def test_setup_network_rfxtrx(
+    hass,
+    dsmr_connection_send_validate_fixture,
+    rfxtrx_dsmr_connection_send_validate_fixture,
+):
+    """Test we can setup network."""
+    (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
+
+    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": "Network"},
+    )
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "setup_network"
+    assert result["errors"] == {}
+
+    # set-up DSMRProtocol to yield no valid telegram, this will retry with RFXtrxDSMRProtocol
+    protocol.telegram = {}
+
+    with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "10.10.0.1",
+                "port": 1234,
+                "dsmr_version": "2.2",
+            },
+        )
+        await hass.async_block_till_done()
+
+    entry_data = {
+        "host": "10.10.0.1",
+        "port": 1234,
+        "dsmr_version": "2.2",
+        "protocol": "rfxtrx_dsmr_protocol",
     }
 
     assert result["type"] == "create_entry"
@@ -87,12 +144,65 @@ async def test_setup_serial(com_mock, hass, dsmr_connection_send_validate_fixtur
 
     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": "2.2"}
+            result["flow_id"],
+            {"port": port.device, "dsmr_version": "2.2"},
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "port": port.device,
         "dsmr_version": "2.2",
+        "protocol": "dsmr_protocol",
+    }
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == port.device
+    assert result["data"] == {**entry_data, **SERIAL_DATA}
+
+
+@patch("serial.tools.list_ports.comports", return_value=[com_port()])
+async def test_setup_serial_rfxtrx(
+    com_mock,
+    hass,
+    dsmr_connection_send_validate_fixture,
+    rfxtrx_dsmr_connection_send_validate_fixture,
+):
+    """Test we can setup serial."""
+    (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
+
+    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"] == {}
+
+    # set-up DSMRProtocol to yield no valid telegram, this will retry with RFXtrxDSMRProtocol
+    protocol.telegram = {}
+
+    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": "2.2"},
+        )
+        await hass.async_block_till_done()
+
+    entry_data = {
+        "port": port.device,
+        "dsmr_version": "2.2",
+        "protocol": "rfxtrx_dsmr_protocol",
     }
 
     assert result["type"] == "create_entry"
@@ -124,12 +234,15 @@ async def test_setup_5L(com_mock, hass, dsmr_connection_send_validate_fixture):
 
     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"}
+            result["flow_id"],
+            {"port": port.device, "dsmr_version": "5L"},
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "port": port.device,
         "dsmr_version": "5L",
+        "protocol": "dsmr_protocol",
         "serial_id": "12345678",
         "serial_id_gas": "123456789",
     }
@@ -165,10 +278,12 @@ async def test_setup_5S(com_mock, hass, dsmr_connection_send_validate_fixture):
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"], {"port": port.device, "dsmr_version": "5S"}
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "port": port.device,
         "dsmr_version": "5S",
+        "protocol": "dsmr_protocol",
         "serial_id": None,
         "serial_id_gas": None,
     }
@@ -202,12 +317,15 @@ async def test_setup_Q3D(com_mock, hass, dsmr_connection_send_validate_fixture):
 
     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"}
+            result["flow_id"],
+            {"port": port.device, "dsmr_version": "Q3D"},
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "port": port.device,
         "dsmr_version": "Q3D",
+        "protocol": "dsmr_protocol",
         "serial_id": "12345678",
         "serial_id_gas": None,
     }
@@ -240,7 +358,8 @@ async def test_setup_serial_manual(
     assert result["errors"] == {}
 
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"port": "Enter Manually", "dsmr_version": "2.2"}
+        result["flow_id"],
+        {"port": "Enter Manually", "dsmr_version": "2.2"},
     )
 
     assert result["type"] == "form"
@@ -251,10 +370,12 @@ async def test_setup_serial_manual(
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"], {"port": "/dev/ttyUSB0"}
         )
+        await hass.async_block_till_done()
 
     entry_data = {
         "port": "/dev/ttyUSB0",
         "dsmr_version": "2.2",
+        "protocol": "dsmr_protocol",
     }
 
     assert result["type"] == "create_entry"
@@ -297,7 +418,8 @@ async def test_setup_serial_fail(com_mock, hass, dsmr_connection_send_validate_f
         first_fail_connection_factory,
     ):
         result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], {"port": port.device, "dsmr_version": "2.2"}
+            result["flow_id"],
+            {"port": port.device, "dsmr_version": "2.2"},
         )
 
     assert result["type"] == "form"
@@ -307,10 +429,18 @@ async def test_setup_serial_fail(com_mock, hass, dsmr_connection_send_validate_f
 
 @patch("serial.tools.list_ports.comports", return_value=[com_port()])
 async def test_setup_serial_timeout(
-    com_mock, hass, dsmr_connection_send_validate_fixture
+    com_mock,
+    hass,
+    dsmr_connection_send_validate_fixture,
+    rfxtrx_dsmr_connection_send_validate_fixture,
 ):
     """Test failed serial connection."""
     (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
+    (
+        connection_factory,
+        transport,
+        rfxtrx_protocol,
+    ) = rfxtrx_dsmr_connection_send_validate_fixture
 
     port = com_port()
 
@@ -324,6 +454,12 @@ async def test_setup_serial_timeout(
     )
     protocol.wait_closed = first_timeout_wait_closed
 
+    first_timeout_wait_closed = AsyncMock(
+        return_value=True,
+        side_effect=chain([asyncio.TimeoutError], repeat(DEFAULT)),
+    )
+    rfxtrx_protocol.wait_closed = first_timeout_wait_closed
+
     assert result["type"] == "form"
     assert result["step_id"] == "user"
     assert result["errors"] is None
@@ -349,10 +485,18 @@ async def test_setup_serial_timeout(
 
 @patch("serial.tools.list_ports.comports", return_value=[com_port()])
 async def test_setup_serial_wrong_telegram(
-    com_mock, hass, dsmr_connection_send_validate_fixture
+    com_mock,
+    hass,
+    dsmr_connection_send_validate_fixture,
+    rfxtrx_dsmr_connection_send_validate_fixture,
 ):
     """Test failed telegram data."""
     (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
+    (
+        rfxtrx_connection_factory,
+        transport,
+        rfxtrx_protocol,
+    ) = rfxtrx_dsmr_connection_send_validate_fixture
 
     port = com_port()
 
@@ -360,8 +504,6 @@ async def test_setup_serial_wrong_telegram(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
 
-    protocol.telegram = {}
-
     assert result["type"] == "form"
     assert result["step_id"] == "user"
     assert result["errors"] is None
@@ -375,8 +517,12 @@ async def test_setup_serial_wrong_telegram(
     assert result["step_id"] == "setup_serial"
     assert result["errors"] == {}
 
+    protocol.telegram = {}
+    rfxtrx_protocol.telegram = {}
+
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"port": port.device, "dsmr_version": "2.2"}
+        result["flow_id"],
+        {"port": port.device, "dsmr_version": "2.2"},
     )
 
     assert result["type"] == "form"

tests/components/dsmr/test_sensor.py

@@ -658,6 +658,35 @@ async def test_tcp(hass, dsmr_connection_fixture):
         "host": "localhost",
         "port": "1234",
         "dsmr_version": "2.2",
+        "protocol": "dsmr_protocol",
+        "precision": 4,
+        "reconnect_interval": 30,
+        "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, rfxtrx_dsmr_connection_fixture):
+    """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",
         "precision": 4,
         "reconnect_interval": 30,
         "serial_id": "1234",