Merge pull request #44 from hostcc/feature/duration-sensor

feat: Pseudo-sensor reflecting the duration of the meter cycle

README.rst

@@ -124,6 +124,8 @@ Configuration file is in YAML format and supports following elements:
               # (number) - optional: Zero-based index to pick an entry from
               #  multi-value response to meter's parameter
               response_idx:
+              # (string) - optional: Category of the HASS sensor entity
+              entity_category:
 
 
 Interpolation expressions
@@ -167,7 +169,7 @@ directory.
 Docker support
 ==============
 
-There are Docker images available if you would like to run it as Docker container - you could use 
+There are Docker images available if you would like to run it as Docker container - you could use
 ``ghcr.io/hostcc/energomera-hass-mqtt:latest`` or
 ``ghcr.io/hostcc/energomera-hass-mqtt:<release version>``.
 
@@ -187,7 +189,7 @@ Then, assuming the directory is called ``config`` and resides relative to
 current directory, and the serial port the meter is connected to is
 ``/dev/ttyUSB0`` the following command will run it
 
-.. code:: 
+.. code::
 
   $ docker run --device /dev/ttyUSB0 -v `pwd`/config:/etc/energomera/ \
     ghcr.io/hostcc/energomera-hass-mqtt:latest

pyproject.toml

@@ -36,3 +36,8 @@ signature-mutators = [
 # No typing support in the module
 module = "iec62056_21.*"
 ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# No typing support in the module
+module = "callee.*"
+ignore_missing_imports = true

requirements_dev.txt

@@ -9,3 +9,4 @@ freezegun==1.5.1
 mypy==1.11.2
 types-python-dateutil==2.8.19.14
 types-PyYAML==6.0.12
+callee==0.3.1

src/energomera_hass_mqtt/extra_sensors.py

@@ -22,61 +22,58 @@
 Package to provide additional sensors on top of `IecToHassSensor`.
 """
 from __future__ import annotations
-from typing import Dict, Union, Any
+from typing import Any, Generic, TypeVar, Optional
+from iec62056_21.messages import DataSet as IecDataSet
 from .iec_hass_sensor import IecToHassSensor
 
+T = TypeVar('T')
 
-class IecToHassBinarySensor(IecToHassSensor):
+
+class PseudoSensor(IecToHassSensor, Generic[T]):
     """
-    Represents HASS binary sensor.
+    Represents a pseudo-sensor, i.e. one doesn't exist on meter.
     """
-    _mqtt_topic_base = 'binary_sensor'
+    def __init__(self, value: T, *args: Any, **kwargs: Any):
+        self._state_last_will_payload: Optional[T] = None
+        # Invoke the parent constructor providing `iec_item` from the
+        # pseudo-sensor value
+        kwargs['iec_item'] = [
+            IecDataSet(value=PseudoSensor._format_value(value))
+        ]
+        super().__init__(*args, **kwargs)
 
-    def hass_state_payload(
-        self, value: Union[str, str]
-    ) -> Dict[str, str]:  # pylint: disable=no-self-use
+    @staticmethod
+    def _format_value(value: T) -> Optional[str]:
         """
-        Transforms the binary sensor payload to the format understood by HASS
-        MQTT discovery for the sensor type.
+        Formats the sensor's value according to its type.
 
         :param value: The sensor value
+        :return: The formatted value
         """
+        result = None
         if isinstance(value, bool):
-            b_value = value
-        elif isinstance(value, str):
-            b_value = value.lower() == 'true'
+            result = 'ON' if value else 'OFF'
         else:
-            raise TypeError(
-                f'Unsupported argument type to {__name__}: {type(value)}'
-            )
-
-        return dict(
-            value='ON' if b_value else 'OFF'
-        )
-
+            result = str(value)
+        return result
 
-class PseudoBinarySensor(IecToHassBinarySensor):
-    """
-    Represents a pseudo-sensor, i.e. one doesn't exist on meter.
+    @property
+    def state_last_will_payload(self) -> Optional[str]:
+        """
+        Stores the value of the last will payload for the item, i.e. sent by
+        the MQTT broker if the client disconnects uncleanly.
 
-    :param value: The sensor value
-    :param args: Parameters passed through to parent constructor
-    :param kwargs: Keyword parameters passed through to parent constructor
-    """
-    def __init__(self, value: bool, *args: Any, **kwargs: Any):
-        class PseudoValue:  # pylint: disable=too-few-public-methods
-            """
-            Mimics `class`:iec62056_21.messages.AnswerDataMessage to store the
-            sensor value in a conmpatible manner.
+        :param value: Value for last will payload
+        """
+        return PseudoSensor._format_value(self._state_last_will_payload)
 
-            :param value: Sensor value
-            """
-            value = None
+    @state_last_will_payload.setter
+    def state_last_will_payload(self, value: T) -> None:
+        self._state_last_will_payload = value
 
-            def __init__(self, value: bool) -> None:
-                self.value = value
 
-        # Invoke the parent constructor providing `iec_item` from the
-        # pseudo-sensor value
-        kwargs['iec_item'] = [PseudoValue(value)]
-        super().__init__(*args, **kwargs)
+class PseudoBinarySensor(PseudoSensor[bool]):
+    """
+    Represents a binary pseudo-sensor.
+    """
+    _mqtt_topic_base = 'binary_sensor'

src/energomera_hass_mqtt/hass_mqtt.py

@@ -27,14 +27,15 @@
 import logging
 import ssl
 from os import getenv
+from time import time
 
 from iec62056_21.messages import CommandMessage, DataSet as IecDataSet
 from iec62056_21.client import Iec6205621Client
 from iec62056_21.transports import SerialTransport
 from iec62056_21 import utils
 from .mqtt_client import MqttClient
 from .iec_hass_sensor import IecToHassSensor
-from .extra_sensors import PseudoBinarySensor
+from .extra_sensors import PseudoBinarySensor, PseudoSensor
 from .schema import ConfigParameterSchema
 from .exceptions import EnergomeraMeterError
 if TYPE_CHECKING:
@@ -177,6 +178,7 @@ async def iec_read_admin(self) -> None:
         Primary method to loop over the parameters requested and process them.
         """
 
+        start = time()
         try:
             _LOGGER.debug('Opening connection with meter')
             self._client.connect()
@@ -221,12 +223,13 @@ async def iec_read_admin(self) -> None:
             # End the session
             _LOGGER.debug('Closing session with meter')
             self._client.send_break()
-
         except TimeoutError as exc:
             await self.set_online_sensor(False)
             raise exc
         else:
             await self.set_online_sensor(True)
+            duration = time() - start
+            await self.set_duration_sensor(duration)
         finally:
             # Disconnect serial client ignoring possible
             # exceptions - it might have not been connected yet
@@ -278,5 +281,26 @@ async def set_online_sensor(
             serial_number=self._serial_number
         )
         # Set the last will of the MQTT client to the `state=False`
-        hass_item.set_state_last_will_payload(value=False)
+        hass_item.state_last_will_payload = False
         await hass_item.process(setup_only)
+
+    async def set_duration_sensor(self, value: float) -> None:
+        """
+        Adds a pseudo-sensor to HASS reflecting the duration of the meter
+        cycle.
+        """
+        _LOGGER.debug('Cycle duration: %s s', value)
+        # Add a pseudo-sensor
+        param = ConfigParameterSchema(
+            address='CYCLE_DURATION',
+            name='Meter cycle duration',
+            unit='s',
+            entity_category='diagnostic',
+        )
+        hass_item = PseudoSensor[float](
+            mqtt_config=self._config.of.mqtt,
+            mqtt_client=self._mqtt_client, config_param=param,
+            value=value, model=self._model, sw_version=self._sw_version,
+            serial_number=self._serial_number
+        )
+        await hass_item.process()

src/energomera_hass_mqtt/iec_hass_sensor.py

@@ -75,16 +75,15 @@ def __init__(  # pylint: disable=too-many-arguments
         self._model = model
         self._sw_version = sw_version
         self._serial_number = serial_number
-        self._state_last_will_payload: Optional[bool] = None
 
-    def set_state_last_will_payload(self, value: bool) -> None:
+    @property
+    def state_last_will_payload(self) -> Optional[str]:
         """
-        Stores there value of the last will payload for the item, i.e. sent by
-        the MQTT broker if the client disconnects uncleanly.
+        Stores value of last will payload to be set for MQTT client.
 
-        :param value: Value for last will payload
+        Should be implemented by subclasses.
         """
-        self._state_last_will_payload = value
+        return None
 
     @property
     def iec_item(self) -> List[IecDataSet]:
@@ -202,6 +201,7 @@ def hass_config_payload(
             unit_of_measurement=self._config_param.unit,
             state_class=self._config_param.state_class,
             state_topic=self._hass_state_topic,
+            entity_category=self._config_param.entity_category,
             value_template='{{ value_json.value }}',
         )
         # Skip empty values
@@ -245,9 +245,9 @@ async def process(self, setup_only: bool = False) -> None:
                               " at '%s' address",
                               self._hass_item_name, self._config_param.address)
                 # Set last will for MQTT if specified for the item
-                if self._state_last_will_payload is not None:
+                if self.state_last_will_payload is not None:
                     will_payload = self.hass_state_payload(
-                        value=str(self._state_last_will_payload)
+                        value=self.state_last_will_payload
                     )
                     json_will_payload = json.dumps(will_payload)
 

src/energomera_hass_mqtt/schema.py

@@ -89,12 +89,13 @@ class ConfigParameterSchema(BaseModel):
     """
     address: str
     name: Union[str, List[str]]
-    device_class: str
+    device_class: Optional[str] = None
     state_class: Optional[str] = None
     unit: Optional[str] = None
     additional_data: Optional[str] = None
     entity_name: Optional[str] = None
     response_idx: Optional[int] = None
+    entity_category: Optional[str] = None
 
 
 class ConfigSchema(BaseModel):

tests/conftest.py

@@ -28,6 +28,7 @@
 import sys
 from functools import reduce
 from unittest.mock import patch, call, DEFAULT, AsyncMock, Mock
+from callee import Regex as CallRegexMatcher
 import pytest
 from pytest import FixtureRequest
 import iec62056_21.transports
@@ -36,6 +37,7 @@
 MockMqttT = Dict[str, Mock]
 MockSerialT = Dict[str, Mock]
 
+
 SERIAL_EXCHANGE_BASE = [
     {
         'receive_bytes': b'/?!\r\n',
@@ -875,6 +877,34 @@
         '/CE301_00123456_IS_ONLINE/state',
         payload=json.dumps({'value': 'ON'}),
     ),
+    call(
+        topic='homeassistant/sensor/CE301_00123456/'
+        'CE301_00123456_CYCLE_DURATION/config',
+        payload=json.dumps(
+            {
+                'name': 'Meter cycle duration',
+                'device': {
+                    'name': '00123456',
+                    'ids': 'CE301_00123456',
+                    'model': 'CE301',
+                    'sw_version': '12'
+                },
+                'unique_id': 'CE301_00123456_CYCLE_DURATION',
+                'object_id': 'CE301_00123456_CYCLE_DURATION',
+                'unit_of_measurement': 's',
+                'state_topic': 'homeassistant/sensor/CE301_00123456/'
+                               'CE301_00123456_CYCLE_DURATION/state',
+                'entity_category': 'diagnostic',
+                'value_template': '{{ value_json.value }}'
+            }
+        ),
+        retain=True
+    ),
+    call(
+        topic='homeassistant/sensor/CE301_00123456/'
+        'CE301_00123456_CYCLE_DURATION/state',
+        payload=CallRegexMatcher('{"value": "[0-9.]+"}'),
+    ),
     call(
         topic='homeassistant/binary_sensor/CE301_00123456'
         '/CE301_00123456_IS_ONLINE/state',