Method stream_data

tests/utils/device_simulator.py:183–210

Stream data continuously. Args: generator: Function that returns frame bytes rate_hz: Streaming rate in Hz duration_seconds: Duration to stream (None = infinite)

(
        self,
        generator: Callable[[], bytes],
        rate_hz: float = 10.0,
        duration_seconds: Optional[float] = None,
    )

                time.sleep(delay)

    def stream_data(
        self,
        generator: Callable[[], bytes],
        rate_hz: float = 10.0,
        duration_seconds: Optional[float] = None,
    ) -> None:
        """
        Stream data continuously.

        Args:
            generator: Function that returns frame bytes
            rate_hz: Streaming rate in Hz
            duration_seconds: Duration to stream (None = infinite)
        """
        interval = 1.0 / rate_hz
        start_time = time.time()

        while self._running:
            if duration_seconds and (time.time() - start_time) >= duration_seconds:
                break

            try:
                frame = generator()
                self.send_frame(frame)
            except Exception:
                break

            time.sleep(interval)

    def wait_for_connection(self, timeout: float = 10.0) -> bool:
        """