From 3cc03bd3d83fc3b12d924fdd35522a33f395d75f Mon Sep 17 00:00:00 2001 From: Thomas Vreys Date: Tue, 14 Jul 2026 16:25:27 +0200 Subject: [PATCH 1/2] Add RTSP Video Streaming support Adds support for streaming video from IP video sources over RTSP. This work was sponsored by OIP Sensor Systems. Signed-off-by: Thomas Vreys --- doc/configuration.rst | 34 ++++++++++++++++++++ labgrid/driver/__init__.py | 1 + labgrid/driver/rtspvideodriver.py | 48 +++++++++++++++++++++++++++++ labgrid/remote/client.py | 3 ++ labgrid/remote/exporter.py | 18 +++++++++++ labgrid/resource/__init__.py | 1 + labgrid/resource/rtspvideostream.py | 10 ++++++ tests/test_rtspvideo.py | 19 ++++++++++++ 8 files changed, 134 insertions(+) create mode 100644 labgrid/driver/rtspvideodriver.py create mode 100644 labgrid/resource/rtspvideostream.py create mode 100644 tests/test_rtspvideo.py diff --git a/doc/configuration.rst b/doc/configuration.rst index 57b23f13c..20e66b430 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -1387,6 +1387,21 @@ Arguments: Used by: - `HTTPVideoDriver`_ +RTSPVideoStream +~~~~~~~~~~~~~~~ +An :any:`RTSPVideoStream` resource describes an IP video stream over RTSP. + +.. code-block:: yaml + + RTSPVideoStream: + url: 'rtsp://192.168.110.11/stream1' + +Arguments: + - url (str): URI of the IP video stream + +Used by: + - `RTSPVideoDriver`_ + USBHub ~~~~~~ @@ -3448,6 +3463,25 @@ Although the driver can be used from Python code by calling the ``stream()`` method, it is currently mainly useful for the ``video`` subcommand of ``labgrid-client``. +RTSPVideoDriver +~~~~~~~~~~~~~~~ +The :any:`RTSPVideoDriver` is used to show a video stream over RTSP +from a remote IP video source in a local window. + +Binds to: + video: + - `RTSPVideoStream`_ + +Implements: + - :any:`VideoProtocol` + +Arguments: + - latency (int, default=100): rtspsrc jitterbuffer size in milliseconds + +Although the driver can be used from Python code by calling the ``stream()`` +method, it is currently mainly useful for the ``video`` subcommand of +``labgrid-client``. + ========== ========================================================= Key Description ========== ========================================================= diff --git a/labgrid/driver/__init__.py b/labgrid/driver/__init__.py index d3cd6f55e..753ff79fc 100644 --- a/labgrid/driver/__init__.py +++ b/labgrid/driver/__init__.py @@ -40,6 +40,7 @@ from .usbaudiodriver import USBAudioInputDriver from .usbvideodriver import USBVideoDriver from .httpvideodriver import HTTPVideoDriver +from .rtspvideodriver import RTSPVideoDriver from .networkinterfacedriver import NetworkInterfaceDriver from .provider import HTTPProviderDriver, NFSProviderDriver, TFTPProviderDriver from .rawnetworkinterfacedriver import RawNetworkInterfaceDriver diff --git a/labgrid/driver/rtspvideodriver.py b/labgrid/driver/rtspvideodriver.py new file mode 100644 index 000000000..46951fedf --- /dev/null +++ b/labgrid/driver/rtspvideodriver.py @@ -0,0 +1,48 @@ +import subprocess +import sys +from urllib.parse import urlsplit + +import attr + +from .common import Driver +from ..factory import target_factory +from ..util.proxy import proxymanager +from ..protocol import VideoProtocol + + +@target_factory.reg_driver +@attr.s(eq=False) +class RTSPVideoDriver(Driver, VideoProtocol): + bindings = { + "video": "RTSPVideoStream", + } + + latency = attr.ib(default=100, validator=attr.validators.instance_of(int)) + + def get_qualities(self): + return ("high", [("high", None)]) + + @Driver.check_active + def stream(self, quality_hint=None, controls=None): + s = urlsplit(self.video.url) + if s.scheme != "rtsp": + print(f"Unknown scheme: {s.scheme}", file=sys.stderr) + return + + url = proxymanager.get_url(self.video.url, default_port=554) + pipeline = [ + "gst-launch-1.0", + "rtspsrc", + f"location={url}", + f"latency={self.latency}", + "!", + "decodebin", + "!", + "autovideoconvert", + "!", + "autovideosink", + "sync=false", + ] + + sub = subprocess.run(pipeline) + return sub.returncode diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 069240b9e..e2cecc0f2 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -1373,6 +1373,7 @@ def video(self): target = self._get_target(place) name = self.args.name from ..resource.httpvideostream import HTTPVideoStream + from ..resource.rtspvideostream import RTSPVideoStream from ..resource.udev import USBVideo from ..resource.remote import NetworkUSBVideo @@ -1387,6 +1388,8 @@ def video(self): drv = self._get_driver_or_new(target, "USBVideoDriver", name=name) elif isinstance(resource, HTTPVideoStream): drv = self._get_driver_or_new(target, "HTTPVideoDriver", name=name) + elif isinstance(resource, RTSPVideoStream): + drv = self._get_driver_or_new(target, "RTSPVideoDriver", name=name) if drv: break if not drv: diff --git a/labgrid/remote/exporter.py b/labgrid/remote/exporter.py index 82f1da6b6..7e96ac6f6 100755 --- a/labgrid/remote/exporter.py +++ b/labgrid/remote/exporter.py @@ -749,6 +749,24 @@ def _get_params(self): exports["HTTPVideoStream"] = HTTPVideoStreamExport +@attr.s +class RTSPVideoStreamExport(ResourceExport): + """ResourceExport for an RTSPVideoStream""" + + def __attrs_post_init__(self): + super().__attrs_post_init__() + from ..resource.rtspvideostream import RTSPVideoStream + + self.data["cls"] = "RTSPVideoStream" + self.local = RTSPVideoStream(target=None, name=None, **self.local_params) + + def _get_params(self): + return self.local_params + + +exports["RTSPVideoStream"] = RTSPVideoStreamExport + + @attr.s(eq=False) class LXAIOBusNodeExport(ResourceExport): """ResourceExport for LXAIOBusNode devices accessed via the HTTP API""" diff --git a/labgrid/resource/__init__.py b/labgrid/resource/__init__.py index 53d88f458..eed48535a 100644 --- a/labgrid/resource/__init__.py +++ b/labgrid/resource/__init__.py @@ -44,6 +44,7 @@ from .provider import TFTPProvider, NFSProvider, HTTPProvider from .mqtt import TasmotaPowerPort from .httpvideostream import HTTPVideoStream +from .rtspvideostream import RTSPVideoStream from .dediprogflasher import DediprogFlasher, NetworkDediprogFlasher from .httpdigitalout import HttpDigitalOutput from .sigrok import SigrokDevice diff --git a/labgrid/resource/rtspvideostream.py b/labgrid/resource/rtspvideostream.py new file mode 100644 index 000000000..46941ec1d --- /dev/null +++ b/labgrid/resource/rtspvideostream.py @@ -0,0 +1,10 @@ +import attr + +from ..factory import target_factory +from .common import Resource + + +@target_factory.reg_resource +@attr.s(eq=False) +class RTSPVideoStream(Resource): + url = attr.ib(validator=attr.validators.instance_of(str)) diff --git a/tests/test_rtspvideo.py b/tests/test_rtspvideo.py new file mode 100644 index 000000000..414b5b094 --- /dev/null +++ b/tests/test_rtspvideo.py @@ -0,0 +1,19 @@ +from labgrid.resource.rtspvideostream import RTSPVideoStream +from labgrid.driver.rtspvideodriver import RTSPVideoDriver + + +def test_ipvideo_create_rtsp(target): + r = RTSPVideoStream(target, name=None, url="rtsp://localhost/stream1") + d = RTSPVideoDriver(target, name=None) + assert isinstance(d, RTSPVideoDriver) + +def test_ipvideo_create_with_port(target): + r = RTSPVideoStream(target, name=None, url="rtsp://localhost:8554/stream1") + d = RTSPVideoDriver(target, name=None) + assert isinstance(d, RTSPVideoDriver) + +def test_ipvideo_create_with_latency(target): + r = RTSPVideoStream(target, name=None, url="rtsp://localhost/stream1") + d = RTSPVideoDriver(target, name=None, latency=500) + assert isinstance(d, RTSPVideoDriver) + assert d.latency == 500 From 370bacd1e76722aae219c917ef88693828693518 Mon Sep 17 00:00:00 2001 From: Thomas Vreys Date: Tue, 14 Jul 2026 16:29:10 +0200 Subject: [PATCH 2/2] Add stream introspection Add get_stream_info(), get_resolution() and get_framerate(), which decode the stream into a fakesink and report the negotiated caps (resolution, pixel format) and measured frame rate. This allows stream properties to be asserted in automated tests without a display. It was manually tested by running an RTSP server locally. This work was sponsored by OIP Sensor Systems. Signed-off-by: Thomas Vreys --- doc/configuration.rst | 19 ++- labgrid/driver/rtspvideodriver.py | 228 +++++++++++++++++++++++++++++- tests/test_rtspvideo.py | 73 +++++++++- 3 files changed, 312 insertions(+), 8 deletions(-) diff --git a/doc/configuration.rst b/doc/configuration.rst index 20e66b430..430688758 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -3478,9 +3478,22 @@ Implements: Arguments: - latency (int, default=100): rtspsrc jitterbuffer size in milliseconds -Although the driver can be used from Python code by calling the ``stream()`` -method, it is currently mainly useful for the ``video`` subcommand of -``labgrid-client``. +Besides showing the stream with the ``stream()`` method (used by the +``video`` subcommand of ``labgrid-client``), the driver can inspect the +stream without a display, e.g. for automated tests: + +- ``get_stream_info()`` decodes the stream into a fakesink and returns a + dict with ``width``, ``height``, ``format``, ``framerate`` (nominal, from + the caps), ``measured_fps`` (measured over ``measure_time`` seconds) and + ``caps`` (the full negotiated caps string). It additionally queries + ``gst-discoverer-1.0`` for ``codec`` (the encoded stream's codec string) + and ``depth`` (colour depth in bits per pixel), which are not part of the + decoded caps. +- ``get_resolution()`` returns a ``(width, height)`` tuple. +- ``get_framerate()`` returns the measured frame rate in frames per second. + +``get_resolution()`` and ``get_framerate()`` only decode the stream and do +not run ``gst-discoverer-1.0``. ========== ========================================================= Key Description diff --git a/labgrid/driver/rtspvideodriver.py b/labgrid/driver/rtspvideodriver.py index 46951fedf..d62fe752e 100644 --- a/labgrid/driver/rtspvideodriver.py +++ b/labgrid/driver/rtspvideodriver.py @@ -1,15 +1,70 @@ +import re +import select import subprocess import sys +import time from urllib.parse import urlsplit import attr from .common import Driver +from .exception import ExecutionError from ..factory import target_factory from ..util.proxy import proxymanager from ..protocol import VideoProtocol +def parse_video_caps(caps): + """Parse a GStreamer video caps string into a dict. + + Example input: + video/x-raw, format=(string)I420, width=(int)640, height=(int)480, + framerate=(fraction)30/1 + + Returns a dict with the keys width (int), height (int), format (str) and + framerate (float, 0.0 if not signalled) for the fields present in the + caps, plus the raw caps string under "caps". + """ + info = {"caps": caps.strip()} + m = re.search(r"width=\(int\)(\d+)", caps) + if m: + info["width"] = int(m.group(1)) + m = re.search(r"height=\(int\)(\d+)", caps) + if m: + info["height"] = int(m.group(1)) + m = re.search(r"format=\(string\)([A-Za-z0-9_-]+)", caps) + if m: + info["format"] = m.group(1) + m = re.search(r"framerate=\(fraction\)(\d+)/(\d+)", caps) + if m and int(m.group(2)): + info["framerate"] = int(m.group(1)) / int(m.group(2)) + else: + info["framerate"] = 0.0 + return info + + +def parse_discoverer_info(text): + """Parse the output of `gst-discoverer-1.0 --verbose` into a dict. + + Example input (abridged): + Codec: + image/jpeg, parsed=(boolean)true, width=(int)640, height=(int)480 + Depth: 24 + + Returns a dict with the keys codec (str) and depth (int) for the fields + present in the output. The codec string is kept verbatim, as which fields + it contains differs per codec. + """ + info = {} + m = re.search(r"^\s*Codec:\s*\n\s*(\S.*)$", text, re.MULTILINE) + if m: + info["codec"] = m.group(1).strip() + m = re.search(r"^\s*Depth:\s*(\d+)\s*$", text, re.MULTILINE) + if m: + info["depth"] = int(m.group(1)) + return info + + @target_factory.reg_driver @attr.s(eq=False) class RTSPVideoDriver(Driver, VideoProtocol): @@ -22,14 +77,20 @@ class RTSPVideoDriver(Driver, VideoProtocol): def get_qualities(self): return ("high", [("high", None)]) - @Driver.check_active - def stream(self, quality_hint=None, controls=None): + def _get_url(self): s = urlsplit(self.video.url) if s.scheme != "rtsp": - print(f"Unknown scheme: {s.scheme}", file=sys.stderr) + raise ExecutionError(f"Unknown scheme: {s.scheme}") + return proxymanager.get_url(self.video.url, default_port=554) + + @Driver.check_active + def stream(self, quality_hint=None, controls=None): + try: + url = self._get_url() + except ExecutionError as e: + print(e.msg, file=sys.stderr) return - url = proxymanager.get_url(self.video.url, default_port=554) pipeline = [ "gst-launch-1.0", "rtspsrc", @@ -46,3 +107,162 @@ def stream(self, quality_hint=None, controls=None): sub = subprocess.run(pipeline) return sub.returncode + + def _probe(self, timeout=10.0, measure_time=3.0): + """Run the fakesink pipeline once and return the stream properties. + + Shared engine behind get_stream_info(), get_resolution() and + get_framerate(): parses the negotiated caps and, when measure_time > 0, + counts decoded frames for that many seconds. Kept intentionally lean so + the lightweight getters do not pay for work they do not need; any + heavier probing added later belongs in get_stream_info() around this + call, not here. + """ + url = self._get_url() + + pipeline = [ + "gst-launch-1.0", + "-v", + "rtspsrc", + f"location={url}", + f"latency={self.latency}", + "!", + "decodebin", + "!", + "fakesink", + "sync=false", + "silent=false", + ] + + sub = subprocess.Popen( + pipeline, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True + ) + info = None + frames = 0 + first_frame = None + last_frame = None + caps_deadline = time.monotonic() + timeout + measure_end = None + try: + while True: + now = time.monotonic() + if info is None and now > caps_deadline: + raise ExecutionError( + f"could not determine stream caps within {timeout} seconds" + ) + if measure_end is not None and now >= measure_end: + break + ready, _, _ = select.select([sub.stdout], [], [], 0.25) + if not ready: + if sub.poll() is not None: + raise ExecutionError( + f"gst-launch-1.0 exited with {sub.returncode} before " + "the stream caps could be determined" + ) + continue + line = sub.stdout.readline() + if not line: + if info is not None: + break + raise ExecutionError( + f"gst-launch-1.0 exited with {sub.wait()} before " + "the stream caps could be determined" + ) + if ( + info is None + and "GstFakeSink" in line + and ".GstPad:sink: caps = video/x-raw" in line + ): + info = parse_video_caps(line.split(" caps = ", 1)[1]) + if measure_time <= 0: + break + measure_end = time.monotonic() + measure_time + elif info is not None and "last-message = chain" in line: + frames += 1 + if first_frame is None: + first_frame = now + last_frame = now + finally: + sub.terminate() + try: + sub.wait(timeout=2) + except subprocess.TimeoutExpired: + sub.kill() + sub.wait() + + info["measured_fps"] = ( + (frames - 1) / (last_frame - first_frame) if frames >= 2 else 0.0 + ) + return info + + def _discover(self, timeout=10.0): + """Run gst-discoverer-1.0 once and return codec-level properties. + + _probe() reads the caps at the fakesink, i.e. after decodebin, where + the stream is already raw video: the codec is gone and the colour depth + was never part of the caps. gst-discoverer inspects the encoded stream + instead, so it can report both. + """ + url = self._get_url() + + try: + sub = subprocess.run( + ["gst-discoverer-1.0", "--verbose", url], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + raise ExecutionError( + f"gst-discoverer-1.0 did not finish within {timeout} seconds" + ) from None + if sub.returncode != 0: + raise ExecutionError( + f"gst-discoverer-1.0 exited with {sub.returncode}" + ) + + return parse_discoverer_info(sub.stdout) + + @Driver.check_active + def get_stream_info(self, timeout=10.0, measure_time=3.0): + """Connect to the stream without a display and return its properties. + + Decodes the stream into a fakesink and parses the negotiated caps, + counts the decoded frames for measure_time seconds, and additionally + queries gst-discoverer for the properties that are not part of the + decoded caps. + + Args: + timeout (float): maximum time in seconds to wait for the caps + measure_time (float): how long to count frames for measured_fps; + 0 skips the measurement (faster, measured_fps will be 0.0) + + Returns a dict with: + width, height (int): frame size from the negotiated caps + format (str): pixel format from the negotiated caps + framerate (float): nominal framerate from the caps; often 0.0 + for RTSP, as the SDP usually does not signal one + measured_fps (float): frame rate measured over measure_time + caps (str): the full negotiated caps string + codec (str): the encoded stream's codec, e.g. "image/jpeg, ..." + depth (int): colour depth in bits per pixel + """ + info = self._probe(timeout=timeout, measure_time=measure_time) + info.update(self._discover(timeout=timeout)) + return info + + @Driver.check_active + def get_resolution(self): + """Return the stream resolution as a (width, height) tuple. + + Only probes the negotiated caps; the frame rate is not measured. + """ + info = self._probe(measure_time=0) + return (info["width"], info["height"]) + + @Driver.check_active + def get_framerate(self, measure_time=3.0): + """Return the measured frame rate of the stream in frames per second.""" + info = self._probe(measure_time=measure_time) + return info["measured_fps"] diff --git a/tests/test_rtspvideo.py b/tests/test_rtspvideo.py index 414b5b094..7bfef013b 100644 --- a/tests/test_rtspvideo.py +++ b/tests/test_rtspvideo.py @@ -1,5 +1,9 @@ from labgrid.resource.rtspvideostream import RTSPVideoStream -from labgrid.driver.rtspvideodriver import RTSPVideoDriver +from labgrid.driver.rtspvideodriver import ( + RTSPVideoDriver, + parse_video_caps, + parse_discoverer_info, +) def test_ipvideo_create_rtsp(target): @@ -17,3 +21,70 @@ def test_ipvideo_create_with_latency(target): d = RTSPVideoDriver(target, name=None, latency=500) assert isinstance(d, RTSPVideoDriver) assert d.latency == 500 + +def test_parse_video_caps(): + caps = ("video/x-raw, format=(string)I420, width=(int)640, height=(int)480, " + "interlace-mode=(string)progressive, framerate=(fraction)30/1") + info = parse_video_caps(caps) + assert info["width"] == 640 + assert info["height"] == 480 + assert info["format"] == "I420" + assert info["framerate"] == 30.0 + +def test_parse_video_caps_no_framerate(): + caps = "video/x-raw, format=(string)NV12, width=(int)1920, height=(int)1080, framerate=(fraction)0/1" + info = parse_video_caps(caps) + assert info["width"] == 1920 + assert info["height"] == 1080 + assert info["framerate"] == 0.0 + +def test_parse_video_caps_garabge_framerate(): + caps = "video/x-raw, format=(string)NV12, width=(int)1920, height=(int)1080, framerate=(fraction)30/0" + info = parse_video_caps(caps) + assert info["width"] == 1920 + assert info["height"] == 1080 + assert info["framerate"] == 0.0 + +def test_parse_video_caps_missing_fields(): + info = parse_video_caps("video/x-raw") + assert "width" not in info + assert "height" not in info + assert info["framerate"] == 0.0 + +DISCOVERER_OUTPUT = """\ +Analyzing rtsp://127.0.0.1:8554/test +Done discovering rtsp://127.0.0.1:8554/test + +Properties: + Duration: 99:99:99.999999999 + Seekable: no + Live: yes + video #1: image/jpeg, parsed=(boolean)true, framerate=(fraction)30/1, width=(int)640, height=(int)480 + Tags: + None + + Codec: + image/jpeg, parsed=(boolean)true, framerate=(fraction)30/1, width=(int)640, height=(int)480 + Stream ID: 359314d7d4bba383223927d7e57d4244d0800e629c626be81c505055c62170e2/video:0:0:RTP:AVP:26 + Width: 640 + Height: 480 + Depth: 24 + Frame rate: 30/1 + Pixel aspect ratio: 1/1 + Interlaced: false + Bitrate: 0 + Max bitrate: 0 +""" + +def test_parse_discoverer_info(): + info = parse_discoverer_info(DISCOVERER_OUTPUT) + assert info["codec"] == ( + "image/jpeg, parsed=(boolean)true, framerate=(fraction)30/1, " + "width=(int)640, height=(int)480" + ) + assert info["depth"] == 24 + +def test_parse_discoverer_info_missing_fields(): + info = parse_discoverer_info("Analyzing rtsp://localhost/test\n") + assert "codec" not in info + assert "depth" not in info