Serial Stream with HA and message delimiter

config.example.yml

@@ -42,7 +42,7 @@ digital_inputs:
     off_payload: "OFF"
     pullup: no
     pulldown: yes
-    
+
   - name: button_left
     module: raspberrypi
     pin: 23
@@ -76,3 +76,16 @@ sensor_inputs:
     module: ds18b22
     interval: 10
     digits: 2
+
+stream_modules:
+  - name: mmwave_uart
+    module: serial
+    device: /dev/serial0
+    baud: 115200
+    delimiter: "\r\n"
+    reset_before_read: True
+    read_interval: 5
+    timeout: 60
+    ha_discovery:
+      unique_id: mmwave_uart
+      name: MMWave UART

mqtt_io/home_assistant.py

@@ -8,7 +8,9 @@
 import logging
 from typing import Any, Dict
 
-from .constants import INPUT_TOPIC, OUTPUT_TOPIC, SENSOR_TOPIC, SET_SUFFIX
+from .constants import (
+    INPUT_TOPIC, OUTPUT_TOPIC, SENSOR_TOPIC, STREAM_TOPIC,
+    SET_SUFFIX, SEND_SUFFIX)
 from .mqtt import MQTTClientOptions, MQTTMessageSend
 from .types import ConfigType
 from . import VERSION
@@ -148,3 +150,34 @@ def hass_announce_sensor_input(
         json.dumps(sensor_config).encode("utf8"),
         retain=True,
     )
+
+def hass_announce_stream(
+    in_conf: ConfigType, mqtt_conf: ConfigType, mqtt_options: MQTTClientOptions
+) -> MQTTMessageSend:
+    """
+    Create a message which announces stream as text to Home Assistant.
+    """
+    disco_conf: ConfigType = mqtt_conf["ha_discovery"]
+    name: str = in_conf["name"]
+    disco_prefix: str = disco_conf["prefix"]
+    stream_config = get_common_config(in_conf, mqtt_conf, mqtt_options)
+    stream_config.update(
+        dict(
+            unique_id=f"{mqtt_options.client_id}_{in_conf['module']}_input_{name}",
+            state_topic="/".join((mqtt_conf["topic_prefix"], STREAM_TOPIC, name)),
+            command_topic="/".join((mqtt_conf["topic_prefix"], STREAM_TOPIC, name, SEND_SUFFIX))
+        )
+    )
+    return MQTTMessageSend(
+        "/".join(
+            (
+                disco_prefix,
+                stream_config.pop("component", "text"),
+                mqtt_options.client_id,
+                name,
+                "config",
+            )
+        ),
+        json.dumps(stream_config).encode("utf8"),
+        retain=True,
+    )

mqtt_io/modules/stream/serial.py

@@ -37,6 +37,21 @@
         "empty": False,
         "allowed": STOPBITS_CHOICES,
     },
+    "delimiter": {
+        "type": "string",
+        "required": False,
+        "empty": True
+    },
+    "reset_before_read": {
+        "type": "boolean",
+        "required": False,
+        "empty": False,
+        "default": False
+    },
+    "ha_discovery": {
+        "type": "dict",
+        "allow_unknown": True
+    }
 }
 
 # pylint: disable=no-member
@@ -80,9 +95,22 @@ def setup_module(self) -> None:
         self.ser.flushInput()
 
     def read(self) -> Optional[bytes]:
-        return self.ser.read(self.ser.in_waiting) or None
+        data = None
+        if "delimiter" in self.config:
+            if "reset_before_read" in self.config:
+                if self.config["reset_before_read"]:
+                    self.ser.reset_input_buffer()
+            data = self.ser.read_until(self.config["delimiter"].encode("utf-8"))
+            if data:
+                data = data[:-len(self.config["delimiter"])]
+        else:
+            data = self.ser.read(self.ser.in_waiting) or None
+
+        return data
 
     def write(self, data: bytes) -> None:
+        if "delimiter" in self.config:
+            data = data + self.config["delimiter"].encode("utf-8")
         self.ser.write(data)
 
     def cleanup(self) -> None:

mqtt_io/server.py

@@ -56,6 +56,7 @@
     hass_announce_digital_input,
     hass_announce_digital_output,
     hass_announce_sensor_input,
+    hass_announce_stream
 )
 from .modules import install_missing_module_requirements
 from .modules.gpio import GenericGPIO, InterruptEdge, InterruptSupport, PinDirection
@@ -636,6 +637,12 @@ def _ha_discovery_announce(self) -> None:
                     sens_conf, mqtt_config, self.mqtt_client_options
                 )
             )
+        for stream_conf in self.stream_configs.values():
+            messages.append(
+                hass_announce_stream(
+                    stream_conf, mqtt_config, self.mqtt_client_options
+                )
+            )
         for msg in messages:
             self.mqtt_task_queue.put_nowait(
                 PriorityCoro(self._mqtt_publish(msg), MQTT_ANNOUNCE_PRIORITY)