Merge pull request #88 from scs/feature/vse-standardization

VSE RL-DSP standardization

Author: raymar9

README.md

@@ -22,15 +22,15 @@ The following smart meters are supported (see [Wiki/Home](https://github.com/scs
 * Landis+Gyr E570: \
   Data pushed by smart meter over CII interface (wired M-Bus, HDLC, DLMS/COSEM)
 * Landis+Gyr E360: \
-  Data pushed by smart meter over P1 interface (HDLC, DLMS/COSEM only, no DSMR)
+  Data pushed by smart meter over P1 interface (P1, HDLC, DLMS/COSEM only, no DSMR)
 * Iskraemeco AM550: \
-  Data pushed by smart meter over P1 interface (HDLC, DLMS/COSEM only, no DSMR)
+  Data pushed by smart meter over P1 interface (P1, HDLC, DLMS/COSEM only, no DSMR)
 * Kamstrup OMNIPOWER with HAN-NVE: \
   Data pushed by smart meter over inserted [HAN-NVE module](https://www.kamstrup.com/en-en/electricity-solutions/meters-devices/modules/hannve) (wired M-Bus, HDLC, DLMS/COSEM)
 * Siemens TD-351x: \
   Data fetched over bidirectional IR interface (IEC 62056-21, Mode C, unencrypted)
 
-Note: All smart meters integrated so far push binary data encoded with HDLC (IEC 62056-46) and DLMS/COSEM. Both unencrypted and encrypted DLMS messages are accepted by the software.
+Note: All smart meters integrated so far push binary data encoded with HDLC (IEC 62056-46) and DLMS/COSEM. Both unencrypted and encrypted DLMS messages are accepted by the software. The P1-DSMR ASCII-based protocol is not supported.
 
 The following data sinks are implemented:
 * MQTT (v3.1.1):
@@ -43,6 +43,8 @@ The following data sinks are implemented:
     * Unauthenticated
     * Authenticated with username / password
     * Authenticated with client certificate
+* MQTT RL-DSP: \
+like MQTT sink but with topic- and payload-format specified in VSE RL-DSP CH2024 document
 * Logger to `stdout`
 
 `smartmeter-datacollector` is fully configurable through a `.ini` configuration file. The [`smartmeter-datacollector-configurator`](https://github.com/scs/smartmeter-datacollector-configurator) web interface can help to create and modify the configuration.
@@ -122,12 +124,12 @@ pipx install smartmeter-datacollector-configurator
 
 ## Method 3: Debian package
 
-`smartmeter-datacollector` is also available as a Debian (`.deb`) package from the [releases](https://github.com/scs/smartmeter-datacollector/releases) which installs the application bundled with its Python dependencies in a zipapp / `.pyz` file. The Debian package includes a systemd service file which enables `smartmeter-datacollector` to automatically start after booting the system.
+`smartmeter-datacollector` is also available as a Debian (`.deb`) package from [releases](https://github.com/scs/smartmeter-datacollector/releases) which installs the application bundled with its Python dependencies in a zipapp / `.pyz` file. The Debian package includes a systemd service file which enables `smartmeter-datacollector` to automatically start after booting the system.
 
 ### Requirements
 
 * Distribution: Debian based (like Debian, Ubuntu, ..)
-* Release: bookworm/12 (trixie/13)
+* Release: >= bookworm/12
 * CPU architecture: independent
 
 ### Installation

smartmeter_datacollector/collector.py

@@ -11,7 +11,7 @@
 from typing import List
 
 from smartmeter_datacollector.sinks.data_sink import DataSink
-from smartmeter_datacollector.smartmeter.meter_data import MeterDataPoint
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle
 
 LOGGER = logging.getLogger("collector")
 
@@ -25,16 +25,15 @@ def register_sink(self, sink: DataSink) -> None:
         assert isinstance(sink, DataSink)
         self._data_sinks.append(sink)
 
-    def notify(self, reader_data_points: List[MeterDataPoint]) -> None:
-        for point in reader_data_points:
-            try:
-                self._queue.put_nowait(point)
-            except QueueFull:
-                LOGGER.warning("Queue is full. Current data points are dropped.")
-                return
+    def notify(self, reader_data_bundle: MeterDataBundle) -> None:
+        try:
+            self._queue.put_nowait(reader_data_bundle)
+        except QueueFull:
+            LOGGER.warning("Queue is full. Current data points are dropped.")
+            return
 
     async def process_queue(self) -> None:
         while True:
-            data_point: MeterDataPoint = await self._queue.get()
+            data_bundle: MeterDataBundle = await self._queue.get()
             for sink in self._data_sinks:
-                await sink.send(data_point)
+                await sink.send(data_bundle)

smartmeter_datacollector/config.py

@@ -27,7 +27,7 @@ class InvalidConfigError(Exception):
     },
     'sink1': {
         'type': "mqtt",
-        'host': "localhost",
+        'host': "127.0.0.1",
         'port': 1883,
         'tls': False,
         'ca_file_path': "",

smartmeter_datacollector/factory.py

@@ -13,7 +13,7 @@
 from smartmeter_datacollector.config import InvalidConfigError
 from smartmeter_datacollector.sinks.data_sink import DataSink
 from smartmeter_datacollector.sinks.logger_sink import LoggerSink
-from smartmeter_datacollector.sinks.mqtt_sink import MqttConfig, MqttDataSink
+from smartmeter_datacollector.sinks.mqtt_sink import MqttConfig, MqttDataSink, MqttSinkRlDsp
 from smartmeter_datacollector.smartmeter.iskraam550 import IskraAM550
 from smartmeter_datacollector.smartmeter.kamstrup_han import KamstrupHAN
 from smartmeter_datacollector.smartmeter.lge360 import LGE360
@@ -91,6 +91,9 @@ def build_sinks(config: ConfigParser) -> List[DataSink]:
         elif sink_type == "mqtt":
             mqtt_config = MqttConfig.from_sink_config(sink_config)
             sinks.append(MqttDataSink(mqtt_config))
+        elif sink_type == "mqttrldsp":
+            mqtt_config = MqttConfig.from_sink_config(sink_config)
+            sinks.append(MqttSinkRlDsp(mqtt_config))
         else:
             raise InvalidConfigError(f"'type' is invalid or missing: {sink_type}")
     return sinks

smartmeter_datacollector/sinks/data_sink.py

@@ -7,7 +7,7 @@
 #
 from abc import ABC, abstractmethod
 
-from smartmeter_datacollector.smartmeter.meter_data import MeterDataPoint
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle
 
 
 class DataSink(ABC):
@@ -20,5 +20,5 @@ async def stop(self) -> None:
         raise NotImplementedError()
 
     @abstractmethod
-    async def send(self, data_point: MeterDataPoint) -> None:
+    async def send(self, data_bundle: MeterDataBundle) -> None:
         raise NotImplementedError()

smartmeter_datacollector/sinks/logger_sink.py

@@ -8,7 +8,7 @@
 import logging
 
 from smartmeter_datacollector.sinks.data_sink import DataSink
-from smartmeter_datacollector.smartmeter.meter_data import MeterDataPoint
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle
 
 
 class LoggerSink(DataSink):
@@ -22,5 +22,5 @@ async def start(self) -> None:
     async def stop(self) -> None:
         pass
 
-    async def send(self, data_point: MeterDataPoint) -> None:
-        self._logger.info(str(data_point))
+    async def send(self, data_bundle: MeterDataBundle) -> None:
+        self._logger.info(str(data_bundle))

smartmeter_datacollector/sinks/mqtt_sink.py

@@ -16,7 +16,7 @@
 from aiomqtt import Client, MqttCodeError, MqttError
 
 from smartmeter_datacollector.sinks.data_sink import DataSink
-from smartmeter_datacollector.smartmeter.meter_data import MeterDataPoint
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle, MeterDataPoint
 
 LOGGER = logging.getLogger("sink")
 
@@ -33,6 +33,7 @@ class MqttConfig:
     check_hostname: bool = True
     client_cert_path: Optional[str] = None
     client_key_path: Optional[str] = None
+    topic_group: str = "building"
 
     def with_tls(
         self, ca_cert_path: Optional[str] = None, check_hostname: bool = True
@@ -72,6 +73,9 @@ def from_sink_config(config: SectionProxy) -> "MqttConfig":
         client_key_path = config.get("client_key_path")
         if client_cert_path is not None and client_key_path is not None:
             mqtt_cfg.with_client_cert_auth(str(client_cert_path), str(client_key_path))
+        topic_group = config.get("topic_group")
+        if topic_group and topic_group.strip().isalnum():
+            mqtt_cfg.topic_group = topic_group.strip()
         return mqtt_cfg
 
 
@@ -139,11 +143,11 @@ async def stop(self) -> None:
         self._client_task = None
         LOGGER.info("Disconnected from MQTT broker")
 
-    async def send(self, data_point: MeterDataPoint) -> None:
-        topic = MqttDataSink.get_topic_name_for_datapoint(data_point)
-        dp_json = self.data_point_to_mqtt_json(data_point)
-
-        await self._publish_with_retries(topic, dp_json, retries=self.RETRIES)
+    async def send(self, data_bundle: MeterDataBundle) -> None:
+        for data_point in data_bundle.data_points:
+            topic = MqttDataSink.get_topic_name_for_datapoint(data_point, data_bundle.source)
+            dp_json = self.data_point_to_mqtt_json(data_point, int(data_bundle.timestamp.timestamp()))
+            await self._publish_with_retries(topic, dp_json, retries=self.RETRIES)
 
     async def _connection_handler(self) -> None:
         while True:
@@ -180,14 +184,36 @@ async def _publish_with_retries(self, topic: str, payload: str, retries: int = 1
                      topic, retries + 1)
 
     @staticmethod
-    def get_topic_name_for_datapoint(data_point: MeterDataPoint) -> str:
-        return f"smartmeter/{data_point.source}/{data_point.type.identifier}"
+    def get_topic_name_for_datapoint(data_point: MeterDataPoint, source: str) -> str:
+        return f"smartmeter/{source}/{data_point.type.identifier}"
 
     @staticmethod
-    def data_point_to_mqtt_json(data_point: MeterDataPoint) -> str:
+    def data_point_to_mqtt_json(data_point: MeterDataPoint, timestamp: int) -> str:
         return json.dumps(
             {
                 "value": data_point.value,
-                "timestamp": int(data_point.timestamp.timestamp()),
+                "timestamp": timestamp,
+                "obis": data_point.obis.to_short_str(),
             }
         )
+
+
+class MqttSinkRlDsp(MqttDataSink):
+    def __init__(self, config: MqttConfig) -> None:
+        super().__init__(config)
+
+        self._group = config.topic_group
+
+    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})

smartmeter_datacollector/smartmeter/cosem.py

@@ -27,7 +27,7 @@ class RegisterCosem:
     scaling: float = 1.0
 
 
-DEFAULT_REGISTER_MAPPING = [
+DEFAULT_REGISTER_MAP = [
     RegisterCosem(OBISCode(1, 0, 1, 7, 0), MeterDataPointTypes.ACTIVE_POWER_P.value),
     RegisterCosem(OBISCode(1, 0, 2, 7, 0), MeterDataPointTypes.ACTIVE_POWER_N.value),
     RegisterCosem(OBISCode(1, 0, 3, 7, 0), MeterDataPointTypes.REACTIVE_POWER_P.value),
@@ -118,10 +118,9 @@ def __init__(self,
             LOGGER.warning("Empty fallback ID. Setting to random UUID %s.", fallback_id)
         self._fallback_id = fallback_id
         self._id_obis_override = id_obis_override if id_obis_override else []
-        registers = DEFAULT_REGISTER_MAPPING
+        self._register_obis = {r.obis: r for r in DEFAULT_REGISTER_MAP}
         if register_obis_extended:
-            registers += register_obis_extended
-        self._register_obis = {r.obis: r for r in registers}
+            self._register_obis.update({r.obis: r for r in register_obis_extended})
         self._id_detect_countdown = Cosem.OBJECT_DETECT_ATTEMPTS
 
     def retrieve_id(self, dlms_objects: Dict[OBISCode, Any]) -> str:

smartmeter_datacollector/smartmeter/hdlc_dlms_parser.py

@@ -16,7 +16,7 @@
 from gurux_dlms.secure import GXDLMSSecureClient
 
 from smartmeter_datacollector.smartmeter.cosem import Cosem
-from smartmeter_datacollector.smartmeter.meter_data import MeterDataPoint
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataBundle, MeterDataPoint
 from smartmeter_datacollector.smartmeter.obis import OBISCode
 
 LOGGER = logging.getLogger("smartmeter")
@@ -112,8 +112,8 @@ def parse_to_dlms_objects(self) -> List[GXDLMSObject]:
         return dlms_objects
 
     def convert_dlms_bundle_to_reader_data(self, dlms_objects: List[GXDLMSObject],
-                                           message_time: Optional[datetime] = None) -> List[MeterDataPoint]:
-        obis_obj_pairs = {}
+                                           message_time: Optional[datetime] = None) -> MeterDataBundle:
+        obis_obj_pairs: dict[OBISCode, GXDLMSObject] = {}
         for obj in dlms_objects:
             try:
                 obis = OBISCode.from_string(str(obj.logicalName))
@@ -158,8 +158,8 @@ def convert_dlms_bundle_to_reader_data(self, dlms_objects: List[GXDLMSObject],
                 except (TypeError, ValueError, OverflowError):
                     LOGGER.warning("Invalid register value '%s'. Skipping register.", str(raw_value))
                     continue
-                data_points.append(MeterDataPoint(data_point_type, value, meter_id, timestamp))
-        return data_points
+                data_points.append(MeterDataPoint(data_point_type, value, obis))
+        return MeterDataBundle(meter_id, timestamp, data_points)
 
     def _parse_dlms_with_push_object_list(self) -> List[GXDLMSObject]:
         parsed_objects: List[Tuple[GXDLMSObject, int]] = []

smartmeter_datacollector/smartmeter/iskraam550.py

@@ -10,8 +10,10 @@
 
 import serial
 
-from smartmeter_datacollector.smartmeter.cosem import Cosem
+from smartmeter_datacollector.smartmeter.cosem import Cosem, RegisterCosem
 from smartmeter_datacollector.smartmeter.meter import MeterError, SerialHdlcDlmsMeter
+from smartmeter_datacollector.smartmeter.meter_data import MeterDataPointTypes
+from smartmeter_datacollector.smartmeter.obis import OBISCode
 from smartmeter_datacollector.smartmeter.reader import ReaderError
 from smartmeter_datacollector.smartmeter.serial_reader import SerialConfig
 
@@ -33,7 +35,12 @@ def __init__(self, port: str,
             stop_bits=serial.STOPBITS_ONE,
             termination=SerialHdlcDlmsMeter.HDLC_FLAG
         )
-        cosem = Cosem(fallback_id=port)
+        voltage_registers = [
+            RegisterCosem(OBISCode(1, 0, 32, 7, 0), MeterDataPointTypes.VOLTAGE_L1.value, 0.1),
+            RegisterCosem(OBISCode(1, 0, 52, 7, 0), MeterDataPointTypes.VOLTAGE_L2.value, 0.1),
+            RegisterCosem(OBISCode(1, 0, 72, 7, 0), MeterDataPointTypes.VOLTAGE_L3.value, 0.1),
+        ]
+        cosem = Cosem(fallback_id=port, register_obis_extended=voltage_registers)
         try:
             super().__init__(serial_config, cosem, decryption_key, use_system_time)
         except ReaderError as ex: