add MqttRlDspSink based on VSE industry document RL-DSP;

with a modified MQTT topic and payload structure

Co-authored-by: Copilot <copilot@github.com>

Author: raymar9

smartmeter_datacollector/sinks/mqtt_sink.py

@@ -192,3 +192,23 @@ def data_point_to_mqtt_json(data_point: MeterDataPoint, timestamp: int) -> str:
                 "obis": data_point.obis.to_short_str(),
             }
         )
+
+
+class MqttSinkRlDsp(MqttDataSink):
+    def __init__(self, config: MqttConfig) -> None:
+        super().__init__(config)
+        self._group = "building"
+
+    async def send(self, data_bundle: MeterDataBundle) -> None:
+        topic = self.build_topic_name(data_bundle)
+        payload = self.to_mqtt_payload(data_bundle)
+        await self._publish_with_retries(topic, payload, retries=self.RETRIES)
+
+    def build_topic_name(self, data_bundle: MeterDataBundle) -> str:
+        return f"dt/{self._group}/{data_bundle.source}/ds"
+
+    @staticmethod
+    def to_mqtt_payload(data_bundle: MeterDataBundle) -> str:
+        meter: dict[str, str | float] = {"ts": data_bundle.timestamp.isoformat()}
+        meter.update({dp.obis.to_short_str(): dp.value for dp in data_bundle.data_points})
+        return json.dumps({"meter": meter})

tests/test_mqtt_sink.py

@@ -15,12 +15,13 @@
 from aiomqtt import MqttCodeError
 from paho.mqtt.client import MQTT_ERR_NO_CONN
 
-from smartmeter_datacollector.sinks.mqtt_sink import MqttConfig, MqttDataSink
+from smartmeter_datacollector.sinks.mqtt_sink import MqttConfig, MqttDataSink, MqttSinkRlDsp
 from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle, MeterDataPoint, MeterDataPointType
 from smartmeter_datacollector.smartmeter.obis import OBISCode
 
 TEST_DATA_POINT_TYPE = MeterDataPointType("TEST_TYPE", "test type", "unit")
 TEST_OBIS = OBISCode(0, 1, 2, 3, 4, 5)
+TEST_OBIS_2 = OBISCode(1, 1, 2, 8, 0, 255)
 
 
 @pytest.fixture(autouse=True)
@@ -190,3 +191,84 @@ def test_mqtt_config_encrypted_authorized_client_cert(_: mock.MagicMock):
 
     with mock.patch("smartmeter_datacollector.sinks.mqtt_sink.ssl"):
         sink = MqttDataSink(cfg)
+
+
+@pytest.mark.asyncio
+async def test_mqtt_sink_rldsp_send_publishes_single_message(mocked_mqtt_client: mock.MagicMock):
+    config = MqttConfig("localhost")
+    sink = MqttSinkRlDsp(config)
+
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_point = MeterDataPoint(TEST_DATA_POINT_TYPE, 42.0, TEST_OBIS)
+    data_bundle = MeterDataBundle("meter1", timestamp, [data_point])
+
+    await sink.send(data_bundle)
+
+    assert mocked_mqtt_client.publish.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_mqtt_sink_rldsp_send_uses_correct_topic(mocked_mqtt_client: mock.MagicMock):
+    config = MqttConfig("localhost")
+    sink = MqttSinkRlDsp(config)
+
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_point = MeterDataPoint(TEST_DATA_POINT_TYPE, 42.0, TEST_OBIS)
+    data_bundle = MeterDataBundle("meter1", timestamp, [data_point])
+    expected_topic = "dt/building/meter1/ds"
+
+    await sink.send(data_bundle)
+
+    mocked_mqtt_client.publish.assert_awaited_once_with(expected_topic, mock.ANY)
+
+
+@pytest.mark.asyncio
+async def test_mqtt_sink_rldsp_send_multiple_datapoints_single_publish(mocked_mqtt_client: mock.MagicMock):
+    config = MqttConfig("localhost")
+    sink = MqttSinkRlDsp(config)
+
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_points = [
+        MeterDataPoint(TEST_DATA_POINT_TYPE, 100.0, TEST_OBIS),
+        MeterDataPoint(TEST_DATA_POINT_TYPE, 200.0, TEST_OBIS_2),
+    ]
+    data_bundle = MeterDataBundle("meter1", timestamp, data_points)
+
+    await sink.send(data_bundle)
+
+    assert mocked_mqtt_client.publish.await_count == 1
+
+
+def test_mqtt_sink_rldsp_to_mqtt_payload_contains_timestamp():
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_point = MeterDataPoint(TEST_DATA_POINT_TYPE, 42.0, TEST_OBIS)
+    data_bundle = MeterDataBundle("meter1", timestamp, [data_point])
+
+    payload = json.loads(MqttSinkRlDsp.to_mqtt_payload(data_bundle))
+
+    assert payload["meter"]["ts"] == timestamp.isoformat()
+
+
+def test_mqtt_sink_rldsp_to_mqtt_payload_contains_obis_values():
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_points = [
+        MeterDataPoint(TEST_DATA_POINT_TYPE, 100.0, TEST_OBIS),
+        MeterDataPoint(TEST_DATA_POINT_TYPE, 200.0, TEST_OBIS_2),
+    ]
+    data_bundle = MeterDataBundle("meter1", timestamp, data_points)
+
+    payload = json.loads(MqttSinkRlDsp.to_mqtt_payload(data_bundle))
+
+    assert payload["meter"][TEST_OBIS.to_short_str()] == 100.0
+    assert payload["meter"][TEST_OBIS_2.to_short_str()] == 200.0
+
+
+def test_mqtt_sink_rldsp_build_topic_name():
+    config = MqttConfig("localhost")
+    sink = MqttSinkRlDsp(config)
+    timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc)
+    data_bundle = MeterDataBundle("my_meter", timestamp, [])
+
+    topic = sink.build_topic_name(data_bundle)
+
+    assert topic == "dt/building/my_meter/ds"