WIP first draft of streaming channel

Rol-la · Rol-la · commit 75e228055e81 · 2026-01-21T17:58:40.000+01:00
diff --git a/sensirion_driver_adapters/channel.py b/sensirion_driver_adapters/channel.py
@@ -2,15 +2,60 @@
 # (c) Copyright 2021 Sensirion AG, Switzerland
 
 import abc
-from typing import Any, Iterable, Optional, Tuple
+from dataclasses import dataclass
+from typing import Any, Iterable, Optional, Tuple, Iterator, Protocol
 
 from sensirion_driver_adapters.rx_tx_data import RxData
 
 
+@dataclass
+class MeasurementPacket:
+    timestamp: float
+    data: Optional[bytes]
+
+
+@dataclass
+class Measurement:
+    stream_id: int  # the id of the stream that produced this measurement
+    timestamp: float  # the timestamp of the measurement
+    data: Tuple[Any, ...]  # the measurement data
+
+
+class StreamingChannel(Protocol):
+    def open_measurement_stream(self, tx_bytes: bytes,
+                                payload_offset: int,
+                                response: RxData,
+                                measurement_interval_us: int,
+                                time_before_first_read_us: int = 1000,
+                                repeat_tx_data: bool = False,
+                                slave_address: Optional[int] = None) -> Iterator[Measurement]:
+        """Start a measurement stream.
+
+        :param tx_bytes: The command and its parameters to trigger the measurement on the sensor.
+        :param payload_offset: The bytes up to the payload_offset represent the command id
+        :param response: The response is an object that is able to unpack a raw response.
+        :param measurement_interval_us: The interval between two measurements in microseconds.
+        :param time_before_first_read_us: The time to wait before the first measurement is taken.
+        :param repeat_tx_data: If True, the tx_bytes will be repeated for each measurement until the stream is closed.
+            Otherwise, the tx_bytes will be transmitted only once to trigger the measurement.
+        :param slave_address: Overwrite the i2c-address of the channel
+        """
+
+    def close_measurement_stream(self, stream_id: int) -> None:
+        """
+        """
+
+
 class TxRxChannel(abc.ABC):
     """
-    This is the abstract base class for any channel. A channel is a transportation medium to transfer data from any
-    source to any destination.
+    Defines an abstract base class for bidirectional communication with a sensor device.
+
+    This class provides an interface for transmitting and receiving data with a sensor.
+    It also allows streaming measurements, stripping protocol-level data, and enforcing a
+    timeout property for communication.
+
+    Attributes:
+        timeout (float): The maximum time duration for a communication attempt in seconds.
     """
 
     @abc.abstractmethod
@@ -21,16 +66,16 @@ def write_read(self, tx_bytes: Iterable, payload_offset: int,
                    slave_address: Optional[int] = None,
                    ignore_errors: bool = False) -> Optional[Tuple[Any, ...]]:
         """
-        Transfers the data to and from sensor.
+        Transfers the data to and from a sensor.
 
         :param tx_bytes:
             Raw bytes to be transmitted
         :param payload_offset:
-            The data my contain a header that needs to be left untouched, pushing the date through the protocol stack.
+            The data may contain a header that needs to be left untouched, pushing the date through the protocol stack.
             The Payload offset points to the end of the header and the beginning of the data
         :param response:
             The response is an object that is able to unpack a raw response.
-            It has to provide a method 'interpret_response.
+            It has to provide a method 'interpret_response'.
         :param device_busy_delay:
             Indication how long the receiver of the message will be busy until processing of the data has been
             completed.
@@ -61,7 +106,7 @@ def timeout(self) -> float:
 
 class AbstractMultiChannel(TxRxChannel):
     """
-    This is the base class for any multi channel implementation. A multi channel is used to mimic simultaneous
+    This is the base class for any multichannel implementation. A multichannel is used to mimic simultaneous
     communication with several sensors and is used by the MultiDeviceDecorator.
     """
 
diff --git a/sensirion_driver_adapters/i2c_adapter/i2c_streaming_channel.py b/sensirion_driver_adapters/i2c_adapter/i2c_streaming_channel.py
@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+# (c) Copyright 2026 Sensirion AG, Switzerland
+import datetime
+import logging
+import time
+from typing import Any, Iterator, Optional, Tuple, Callable, Iterable
+
+from sensirion_driver_adapters.channel import StreamingChannel, Measurement, MeasurementPacket, TxRxChannel
+from sensirion_driver_adapters.i2c_adapter.i2c_channel import I2cChannel
+from sensirion_driver_adapters.rx_tx_data import RxData
+
+log = logging.getLogger(__name__)
+
+
+class I2cStreamingChannel(StreamingChannel, TxRxChannel):
+    def __init__(self,
+                 start_repeated_measurement: Callable[[int, int, bytes, bool, int], int],
+                 get_next_measurement_data: Callable[[int, RxData, float, int], Iterator[MeasurementPacket]],
+                 stop_repeated_measurement: Callable[[int], None],
+                 i2c_channel: I2cChannel) -> None:
+
+        self._i2_channel = i2c_channel
+        self._start_repeated_measurement = start_repeated_measurement
+        self._get_next_measurement_data = get_next_measurement_data
+        self._stop_repeated_measurement = stop_repeated_measurement
+        self._get_next_measurement = self.default_get_next_measurement
+        self._stream_id = -1
+
+    def open_measurement_stream(self, tx_bytes: bytes,
+                                payload_offset: int,
+                                response: RxData,
+                                measurement_interval_us: int,
+                                time_before_first_read_us: int = 1000,
+                                repeat_tx_data: bool = False,
+                                slave_address: Optional[int] = None,
+                                _get_next_measurement: Optional[Callable] = None) -> Iterator[Measurement]:
+
+        if _get_next_measurement:
+            # allow overriding the default get_next_measurement function
+            self._get_next_measurement = _get_next_measurement
+        tx_data = self._i2_channel.build_tx_data(tx_bytes, payload_offset, self._i2_channel._crc)
+        i2c_address = self._i2_channel._slave_address
+        rx_len = response.rx_length * 3 // 2 if self._i2_channel._crc else response.rx_length
+        stream_id = self._start_repeated_measurement(measurement_interval_us,
+                                                     i2c_address,
+                                                     tx_data,
+                                                     repeat_tx_data,
+                                                     rx_len)
+        return self._get_next_measurement(stream_id, self._get_next_measurement_data, response, measurement_interval_us)
+
+    def close_measurement_stream(self, stream_id: int) -> None:
+        self._stop_repeated_measurement(stream_id)
+
+    def default_get_next_measurement(self, stream_id: int,
+                                     get_next_measurement_data: Callable[[int, RxData, float, int],
+                                     Iterator[MeasurementPacket]],
+                                     response: RxData,
+                                     measurement_interval_us: int) -> Iterator[Measurement]:
+
+        start_time: datetime.datetime = datetime.datetime.now()
+        measurement_interval_s = measurement_interval_us / 1000000.0
+        time.sleep(measurement_interval_s * 2)  # wait for two samples before starting
+        while True:
+            nr_of_packets = 0
+            time_stamp = start_time.timestamp()
+            for new_packet in get_next_measurement_data(stream_id, response, time_stamp, measurement_interval_us):
+                raw_data = self._i2_channel.strip_protocol(new_packet.data)
+                next_measurement = Measurement(stream_id=stream_id,
+                                               timestamp=new_packet.timestamp,
+                                               data=response.unpack(raw_data))
+                time_stamp = new_packet.timestamp
+                yield next_measurement
+                nr_of_packets += 1
+            if nr_of_packets == 0:
+                time.sleep(measurement_interval_s * 2)  # wait for approx 50 samples
+                start_time = datetime.datetime.now()
+                continue
+            next_measurement_time = datetime.datetime.fromtimestamp(time_stamp + measurement_interval_s)
+            current_time = datetime.datetime.now()
+            sleep_time = (next_measurement_time - current_time).total_seconds()
+            if sleep_time > 0:
+                time.sleep(sleep_time)
+                start_time = datetime.datetime.now()
+            start_time = next_measurement_time
+
+    def write_read(self, tx_bytes: Iterable, payload_offset: int,
+                   response: RxData,
+                   device_busy_delay: float = 0.0,
+                   post_processing_delay: Optional[float] = None,
+                   slave_address: Optional[int] = None,
+                   ignore_errors: bool = False) -> Optional[Tuple[Any, ...]]:
+        return self._i2_channel.write_read(tx_bytes, payload_offset, response, device_busy_delay,
+                                           post_processing_delay, slave_address, ignore_errors)
+
+    def strip_protocol(self, data) -> None:
+        """"""
+        self._i2_channel.strip_protocol(data)
+
+    def timeout(self) -> float:
+        return self._i2_channel.timeout
diff --git a/sensirion_driver_adapters/i2c_adapter/sensor_bridge_i2c_channel_provider.py b/sensirion_driver_adapters/i2c_adapter/sensor_bridge_i2c_channel_provider.py
@@ -2,17 +2,19 @@
 # (c) Copyright 2023 Sensirion AG, Switzerland
 
 import time
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Iterator, Dict
 
 from sensirion_i2c_driver import I2cConnection
 from sensirion_shdlc_driver import ShdlcSerialPort, ShdlcConnection
 from sensirion_shdlc_sensorbridge import (SensorBridgePort,
                                           SensorBridgeShdlcDevice,
                                           SensorBridgeI2cProxy)
+from sensirion_shdlc_sensorbridge.device import ReadBufferResponse, RepeatedTransceiveHandle
 
-from sensirion_driver_adapters.channel import TxRxChannel
+from sensirion_driver_adapters.channel import TxRxChannel, MeasurementPacket
 from sensirion_driver_adapters.channel_provider import I2cChannelProvider
-from sensirion_driver_adapters.i2c_adapter.i2c_channel import I2cChannel
+from sensirion_driver_adapters.i2c_adapter.i2c_channel import I2cChannel, RxData
+from sensirion_driver_adapters.i2c_adapter.i2c_streaming_channel import I2cStreamingChannel
 
 
 class SensorBridgeI2cChannelProvider(I2cChannelProvider):
@@ -40,6 +42,7 @@ def __init__(self, sensor_bridge_port: SensorBridgePort,
         self._shdlc_port: Optional[ShdlcSerialPort] = None
         self._sensor_bridge: Optional[SensorBridgeShdlcDevice] = None
         self._i2c_transceiver: Optional[SensorBridgeI2cProxy] = None
+        self._streams: Dict[int, RepeatedTransceiveHandle] = {}
 
     def release_channel_resources(self):
         """
@@ -49,6 +52,8 @@ def release_channel_resources(self):
         """
         if self._sensor_bridge is None:
             return
+        for stream_id in self._streams.keys():
+            self._stop_repeated_measurement(stream_id)
         assert self._sensor_bridge_port is not None, "Illegal state: SensorBridgePort is None!"
         assert self._shdlc_port is not None, "Illegal state: Shdlc port is None!"
         self._sensor_bridge.switch_supply_off(self._sensor_bridge_port)
@@ -80,6 +85,62 @@ def get_channel(self, slave_address: int,
             The crc calculator that can compute the crc checksum of the byte stream
         """
 
-        return I2cChannel(I2cConnection(self._i2c_transceiver),
-                          slave_address=slave_address,
-                          crc=self.try_create_crc_calculator(crc_parameters))
+        i2c_channel = I2cChannel(I2cConnection(self._i2c_transceiver),
+                                 slave_address=slave_address,
+                                 crc=self.try_create_crc_calculator(crc_parameters))
+        channel = I2cStreamingChannel(
+            start_repeated_measurement=self._start_repeated_i2c_transceive,
+            get_next_measurement_data=self._get_next_measurement_data,
+            stop_repeated_measurement=self._stop_repeated_measurement,
+            i2c_channel=i2c_channel)
+        return channel
+
+    def _start_repeated_i2c_transceive(self, measurement_interval_us: int,
+                                       i2c_address: int,
+                                       tx_data: bytes,
+                                       _: bool,
+                                       rx_length: int) -> int:
+        """Adapter to for sensor bridge start_repeated_i2c_transceive function."""
+        handle: RepeatedTransceiveHandle = self._sensor_bridge.start_repeated_i2c_transceive(self._sensor_bridge_port,
+                                                                                             measurement_interval_us,
+                                                                                             i2c_address, tx_data,
+                                                                                             rx_length, 1000000)
+        self._streams[handle.raw_handle] = handle
+        return handle.raw_handle
+
+    def _stop_repeated_measurement(self, stream_id) -> None:
+        """Adapter to for sensor bridge stop_repeated_measurement function."""
+        handle = self._streams.get(stream_id)
+        if handle is None:
+            raise ValueError(f"No stream with id {stream_id} found!")
+        self._sensor_bridge.stop_repeated_i2c_transceive(handle)
+
+    def _get_next_measurement_data(self, stream_id: int, rx_descriptor: RxData,
+                                   start_time:
+                                   float, interval_us: int) -> Iterator[MeasurementPacket]:
+
+        rx_length = rx_descriptor.rx_length * 3 // 2
+
+        def handle_buffer_response(response: ReadBufferResponse, start_time: float) -> Iterator[MeasurementPacket]:
+            lost_packets = response.lost_bytes % rx_length
+            for i in range(lost_packets):
+                p = MeasurementPacket(timestamp=start_time, data=None)
+                start_time += interval_us / 1000000
+                yield p
+            for d in response.values:
+                p = MeasurementPacket(timestamp=start_time,
+                                      data=None if d.error else d.data)
+                start_time += interval_us / 1000000
+                yield p
+
+        handle = self._streams.get(stream_id)
+        if handle is None:
+            raise ValueError(f"No stream with id {stream_id} found!")
+        buffer_response: ReadBufferResponse = self._sensor_bridge.read_buffer(handle)
+        for p in handle_buffer_response(buffer_response, start_time):
+            yield p
+
+        while buffer_response.remaining_bytes > 0:
+            buffer_response = self._sensor_bridge.read_buffer(handle)
+            for p in handle_buffer_response(buffer_response, start_time):
+                yield p
