From 53982719fad749a857939472d4f31f146458f8d8 Mon Sep 17 00:00:00 2001 From: "Ilya (Marshal)" Date: Sun, 1 Nov 2026 14:47:43 +0100 Subject: [PATCH] Add aiortc's internal modules and an import alias to the compat layer --- CHANGES.md | 6 + THIRD_PARTY_LICENSES.md | 3 +- docs/gen_api_docs.py | 10 +- docs/source/guides/aiortc.md | 23 +- pyproject.toml | 6 + .../python/webrtc/aiortc/__init__.py | 1 + python-webrtc/python/webrtc/aiortc/_alias.py | 27 + .../python/webrtc/aiortc/_ice_servers.py | 124 +++ python-webrtc/python/webrtc/aiortc/clock.py | 57 ++ .../python/webrtc/aiortc/jitterbuffer.py | 152 ++++ .../python/webrtc/aiortc/rtcdtlstransport.py | 2 + .../python/webrtc/aiortc/rtcicetransport.py | 6 + .../python/webrtc/aiortc/rtcrtpreceiver.py | 83 +- .../python/webrtc/aiortc/rtcsctptransport.py | 2 + python-webrtc/python/webrtc/aiortc/rtp.py | 816 +++++++++++++++++- python-webrtc/python/webrtc/aiortc/utils.py | 82 ++ .../python/webrtc/utils/module_alias.py | 92 ++ tests/aiortc/conftest.py | 8 +- tests/aiortc/test_alias.py | 89 ++ tests/aiortc/test_downstream_names.py | 445 ++++++++++ tests/aiortc/test_media.py | 94 ++ 21 files changed, 2113 insertions(+), 15 deletions(-) create mode 100644 python-webrtc/python/webrtc/aiortc/_alias.py create mode 100644 python-webrtc/python/webrtc/aiortc/_ice_servers.py create mode 100644 python-webrtc/python/webrtc/aiortc/clock.py create mode 100644 python-webrtc/python/webrtc/aiortc/jitterbuffer.py create mode 100644 python-webrtc/python/webrtc/aiortc/utils.py create mode 100644 python-webrtc/python/webrtc/utils/module_alias.py create mode 100644 tests/aiortc/test_alias.py create mode 100644 tests/aiortc/test_downstream_names.py diff --git a/CHANGES.md b/CHANGES.md index d5fa3da..751f2c4 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,11 @@ # Change Log +## Unreleased + +* Add `webrtc.aiortc.install_alias()`, which makes `import aiortc` give the compat layer, for libraries that import aiortc themselves by [@MarshalX](https://github.com/MarshalX) +* Add aiortc's `rtp`, `jitterbuffer`, `utils` and `clock` modules and the ICE server helpers of `rtcicetransport` to the compat layer by [@MarshalX](https://github.com/MarshalX) +* Drop a compat layer track's buffered frames when its receiver's `_enabled` is set to `False`, and add the track's `_queue`, for libraries that skip stale frames by [@MarshalX](https://github.com/MarshalX) + ## Version 0.1.0 **24.10.2026** diff --git a/THIRD_PARTY_LICENSES.md b/THIRD_PARTY_LICENSES.md index 55b1c64..7bcf68c 100644 --- a/THIRD_PARTY_LICENSES.md +++ b/THIRD_PARTY_LICENSES.md @@ -242,7 +242,8 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. # aiortc The code of webrtc/aiortc/sdp.py, mediastreams.py, rtcconfiguration.py, rtcrtpparameters.py, -rtcsessiondescription.py, stats.py, contrib/media.py and contrib/signaling.py, copied from it: +rtcsessiondescription.py, stats.py, rtp.py, jitterbuffer.py, utils.py, clock.py, _ice_servers.py, contrib/media.py and +contrib/signaling.py, copied from it: ``` Copyright (c) Jeremy Lainé. All rights reserved. diff --git a/docs/gen_api_docs.py b/docs/gen_api_docs.py index 1761a97..f08ebf8 100644 --- a/docs/gen_api_docs.py +++ b/docs/gen_api_docs.py @@ -22,7 +22,15 @@ # the internals of the package, except the events every interface inherits TOP_LEVEL = ['enums', 'exceptions', 'streams', 'utils.events', 'base', 'openh264', 'loopback'] # internal webrtc.utils modules, without a page -INTERNAL = ['utils.lifetime', 'utils.loops', 'utils.names', 'utils.operations', 'utils.strings', 'utils.transfer'] +INTERNAL = [ + 'utils.lifetime', + 'utils.loops', + 'utils.module_alias', + 'utils.names', + 'utils.operations', + 'utils.strings', + 'utils.transfer', +] # modules of several public classes, or whose class name isn't the best title TITLES = { diff --git a/docs/source/guides/aiortc.md b/docs/source/guides/aiortc.md index 3050d14..3412980 100644 --- a/docs/source/guides/aiortc.md +++ b/docs/source/guides/aiortc.md @@ -33,6 +33,25 @@ startup, see [H.264](h264.md). Python WebRTC isn't affiliated with or endorsed by aiortc. +## Libraries that import aiortc + +A library that imports `aiortc` itself runs on the layer unchanged. Call `webrtc.aiortc.install_alias()` before +importing it: + +```python +import webrtc.aiortc + +webrtc.aiortc.install_alias() + +import some_library # its aiortc imports now get webrtc.aiortc +``` + +`aiortc.x` then gives `webrtc.aiortc.x`, and a module the layer lacks raises `ModuleNotFoundError`. Call it first, +in each process: it raises `RuntimeError` if `aiortc` is imported already. + +The library's aiortc dependency stays installed but unused. Keep it if the library imports aiortc's dependencies, such +as `aioice` or `cryptography`, directly. + ## What differs Not supported: @@ -42,7 +61,9 @@ Not supported: Transports come from `RTCPeerConnection`. - **Certificate import.** `RTCCertificate(key, cert)` raises `NotImplementedError`. Generated certificates are ECDSA, with one SHA-256 fingerprint. -- **Internal modules.** `rtp`, `jitterbuffer`, `clock`, `rate` and `utils` aren't there, and `codecs` has only the +- **Internal modules.** `rtp`, `jitterbuffer`, `clock` and `utils` are aiortc's code, but the engine's packets never + pass through them, so a jitter buffer given to a receiver does nothing. Setting `rtcsctptransport.USERDATA_MAX_LENGTH` + or `rtcdtlstransport.X509_DIGEST_ALGORITHMS` does nothing either. `rate` isn't there, and `codecs` has only the bitrate constants: no encoder and decoder classes, no `CODECS` or `HEADER_EXTENSIONS` table. `MAX_FRAME_RATE` exists but isn't applied, as in aiortc. - **An explicit `id` on a non-negotiated data channel.** It's ignored, as the specification says. diff --git a/pyproject.toml b/pyproject.toml index 6d2704d..3e2d335 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -153,6 +153,9 @@ extend-exclude = [ "python-webrtc/python/webrtc/aiortc/contrib/media.py", "python-webrtc/python/webrtc/aiortc/contrib/signaling.py", "python-webrtc/python/webrtc/aiortc/rtcconfiguration.py", "python-webrtc/python/webrtc/aiortc/rtcrtpparameters.py", "python-webrtc/python/webrtc/aiortc/rtcsessiondescription.py", "python-webrtc/python/webrtc/aiortc/stats.py", + "python-webrtc/python/webrtc/aiortc/rtp.py", "python-webrtc/python/webrtc/aiortc/jitterbuffer.py", + "python-webrtc/python/webrtc/aiortc/utils.py", "python-webrtc/python/webrtc/aiortc/clock.py", + "python-webrtc/python/webrtc/aiortc/_ice_servers.py", ] [tool.ruff.format] @@ -235,6 +238,9 @@ project-excludes = [ "python-webrtc/python/webrtc/aiortc/contrib/media.py", "python-webrtc/python/webrtc/aiortc/contrib/signaling.py", "python-webrtc/python/webrtc/aiortc/rtcconfiguration.py", "python-webrtc/python/webrtc/aiortc/rtcrtpparameters.py", "python-webrtc/python/webrtc/aiortc/rtcsessiondescription.py", "python-webrtc/python/webrtc/aiortc/stats.py", + "python-webrtc/python/webrtc/aiortc/rtp.py", "python-webrtc/python/webrtc/aiortc/jitterbuffer.py", + "python-webrtc/python/webrtc/aiortc/utils.py", "python-webrtc/python/webrtc/aiortc/clock.py", + "python-webrtc/python/webrtc/aiortc/_ice_servers.py", ] # tests/fuzz runs with its directory on PYTHONPATH (.github/scripts/fuzz.sh) search-path = ["python-webrtc/python", "stubs", ".", "tests/fuzz"] diff --git a/python-webrtc/python/webrtc/aiortc/__init__.py b/python-webrtc/python/webrtc/aiortc/__init__.py index cb4c106..9dd9e8c 100644 --- a/python-webrtc/python/webrtc/aiortc/__init__.py +++ b/python-webrtc/python/webrtc/aiortc/__init__.py @@ -9,6 +9,7 @@ from __future__ import annotations +from webrtc.aiortc._alias import install_alias as install_alias from webrtc.aiortc.events import RTCTrackEvent as RTCTrackEvent from webrtc.aiortc.exceptions import InternalError, InvalidAccessError, InvalidStateError, OperationError from webrtc.aiortc.mediastreams import AudioStreamTrack, MediaStreamError, MediaStreamTrack, VideoStreamTrack diff --git a/python-webrtc/python/webrtc/aiortc/_alias.py b/python-webrtc/python/webrtc/aiortc/_alias.py new file mode 100644 index 0000000..4d98b58 --- /dev/null +++ b/python-webrtc/python/webrtc/aiortc/_alias.py @@ -0,0 +1,27 @@ +# +# Copyright 2022-2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +from __future__ import annotations + +from webrtc.utils import module_alias + + +class _Finder(module_alias.ModuleAlias): + source = 'aiortc' + target = 'webrtc.aiortc' + + +def install_alias() -> None: + """Makes ``import aiortc.x`` give ``webrtc.aiortc.x``, for libraries that import aiortc themselves. + + A module this package lacks raises :obj:`ModuleNotFoundError`. Call it before importing such a library, in each + process. Calling it again does nothing. + + Raises: + RuntimeError: If an ``aiortc`` module is imported already. + """ + module_alias.install(_Finder()) diff --git a/python-webrtc/python/webrtc/aiortc/_ice_servers.py b/python-webrtc/python/webrtc/aiortc/_ice_servers.py new file mode 100644 index 0000000..000dd02 --- /dev/null +++ b/python-webrtc/python/webrtc/aiortc/_ice_servers.py @@ -0,0 +1,124 @@ +# Copied from aiortc 1.15.0 (src/aiortc/rtcicetransport.py), its STUN and TURN URL helpers only. +# +# Copyright (c) Jeremy Lainé. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of aiortc nor the names of its contributors may +# be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import re +from typing import Any + +from .rtcconfiguration import RTCIceServer + +# See https://datatracker.ietf.org/doc/html/rfc7064 +# transport is not defined by RFC 7064 and rejected by browsers. +STUN_REGEX = re.compile( + r"(?Pstun|stuns)\:(?P[^?:]+)(\:(?P[0-9]+?))?" + r"(\?transport=(?P.*))?" +) +# See https://datatracker.ietf.org/doc/html/rfc7065 +TURN_REGEX = re.compile( + r"(?Pturn|turns)\:(?P[^?:]+)(\:(?P[0-9]+?))?" + r"(\?transport=(?P.*))?" +) + + +def connection_kwargs(servers: list[RTCIceServer]) -> dict[str, Any]: + kwargs: dict[str, Any] = {} + + for server in servers: + if isinstance(server.urls, list): + uris = server.urls + else: + uris = [server.urls] + + for uri in uris: + parsed = parse_stun_turn_uri(uri) + + if parsed["scheme"] == "stun": + # only a single STUN server is supported + if "stun_server" in kwargs: + continue + + kwargs["stun_server"] = (parsed["host"], parsed["port"]) + elif parsed["scheme"] in ["turn", "turns"]: + # only a single TURN server is supported + if "turn_server" in kwargs: + continue + + # only 'udp' and 'tcp' transports are supported + if parsed["scheme"] == "turn" and parsed["transport"] not in [ + "udp", + "tcp", + ]: + continue + elif parsed["scheme"] == "turns" and parsed["transport"] != "tcp": + continue + + # only 'password' credentialType is supported + if server.credentialType != "password": + continue + + kwargs["turn_server"] = (parsed["host"], parsed["port"]) + kwargs["turn_ssl"] = parsed["scheme"] == "turns" + kwargs["turn_transport"] = parsed["transport"] + kwargs["turn_username"] = server.username + kwargs["turn_password"] = server.credential + + return kwargs + + +def parse_stun_turn_uri(uri: str) -> dict[str, Any]: + if uri.startswith("stun"): + match = STUN_REGEX.fullmatch(uri) + elif uri.startswith("turn"): + match = TURN_REGEX.fullmatch(uri) + else: + raise ValueError("malformed uri: invalid scheme") + + if not match: + raise ValueError("malformed uri") + + # set port + parsed: dict[str, Any] = match.groupdict() + if parsed["port"]: + parsed["port"] = int(parsed["port"]) + elif parsed["scheme"] in ["stuns", "turns"]: + parsed["port"] = 5349 + else: + parsed["port"] = 3478 + + # set transport + if parsed["scheme"] == "turn" and not parsed["transport"]: + parsed["transport"] = "udp" + elif parsed["scheme"] == "turns" and not parsed["transport"]: + parsed["transport"] = "tcp" + elif parsed["scheme"] in ["stun", "stuns"]: + if parsed["transport"] is not None: + raise ValueError( + "malformed uri: " + parsed["scheme"] + " must not contain transport" + ) + del parsed["transport"] + + return parsed diff --git a/python-webrtc/python/webrtc/aiortc/clock.py b/python-webrtc/python/webrtc/aiortc/clock.py new file mode 100644 index 0000000..7285925 --- /dev/null +++ b/python-webrtc/python/webrtc/aiortc/clock.py @@ -0,0 +1,57 @@ +# Copied from aiortc 1.15.0 (src/aiortc/clock.py), unchanged. +# +# Copyright (c) Jeremy Lainé. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of aiortc nor the names of its contributors may +# be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import datetime + +NTP_EPOCH = datetime.datetime(1900, 1, 1, tzinfo=datetime.timezone.utc) + + +def current_datetime() -> datetime.datetime: + return datetime.datetime.now(datetime.timezone.utc) + + +def current_ms() -> int: + delta = current_datetime() - NTP_EPOCH + return int(delta.total_seconds() * 1000) + + +def current_ntp_time() -> int: + return datetime_to_ntp(current_datetime()) + + +def datetime_from_ntp(ntp: int) -> datetime.datetime: + seconds = ntp >> 32 + microseconds = ((ntp & 0xFFFFFFFF) * 1000000) / (1 << 32) + return NTP_EPOCH + datetime.timedelta(seconds=seconds, microseconds=microseconds) + + +def datetime_to_ntp(dt: datetime.datetime) -> int: + delta = dt - NTP_EPOCH + high = int(delta.total_seconds()) + low = round((delta.microseconds * (1 << 32)) // 1000000) + return (high << 32) | low diff --git a/python-webrtc/python/webrtc/aiortc/jitterbuffer.py b/python-webrtc/python/webrtc/aiortc/jitterbuffer.py new file mode 100644 index 0000000..a73855d --- /dev/null +++ b/python-webrtc/python/webrtc/aiortc/jitterbuffer.py @@ -0,0 +1,152 @@ +# Copied from aiortc 1.15.0 (src/aiortc/jitterbuffer.py), unchanged. +# +# Copyright (c) Jeremy Lainé. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of aiortc nor the names of its contributors may +# be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +from typing import Optional + +from .rtp import RtpPacket +from .utils import uint16_add + +MAX_MISORDER = 100 + + +class JitterFrame: + def __init__(self, data: bytes, timestamp: int) -> None: + self.data = data + self.timestamp = timestamp + + +class JitterBuffer: + def __init__( + self, capacity: int, prefetch: int = 0, is_video: bool = False + ) -> None: + assert capacity & (capacity - 1) == 0, "capacity must be a power of 2" + self._capacity = capacity + self._origin: Optional[int] = None + self._packets: list[Optional[RtpPacket]] = [None for i in range(capacity)] + self._prefetch = prefetch + self._is_video = is_video + + @property + def capacity(self) -> int: + return self._capacity + + def add(self, packet: RtpPacket) -> tuple[bool, Optional[JitterFrame]]: + pli_flag = False + if self._origin is None: + self._origin = packet.sequence_number + delta = 0 + misorder = 0 + else: + delta = uint16_add(packet.sequence_number, -self._origin) + misorder = uint16_add(self._origin, -packet.sequence_number) + + if misorder < delta: + if misorder >= MAX_MISORDER: + self.remove(self.capacity) + self._origin = packet.sequence_number + delta = misorder = 0 + if self._is_video: + pli_flag = True + else: + return pli_flag, None + + if delta >= self.capacity: + # remove just enough frames to fit the received packets + excess = delta - self.capacity + 1 + if self.smart_remove(excess): + self._origin = packet.sequence_number + if self._is_video: + pli_flag = True + + pos = packet.sequence_number % self._capacity + self._packets[pos] = packet + + return pli_flag, self._remove_frame(packet.sequence_number) + + def _remove_frame(self, sequence_number: int) -> Optional[JitterFrame]: + frame = None + frames = 0 + packets: list[RtpPacket] = [] + remove = 0 + timestamp = None + + for count in range(self.capacity): + pos = (self._origin + count) % self._capacity + packet = self._packets[pos] + if packet is None: + break + if timestamp is None: + timestamp = packet.timestamp + elif packet.timestamp != timestamp: + # we now have a complete frame, only store the first one + if frame is None: + frame = JitterFrame( + data=b"".join([x._data for x in packets]), # type: ignore + timestamp=timestamp, + ) + remove = count + + # check we have prefetched enough + frames += 1 + if frames >= self._prefetch: + self.remove(remove) + return frame + + # start a new frame + packets = [] + timestamp = packet.timestamp + + packets.append(packet) + + return None + + def remove(self, count: int) -> None: + assert count <= self._capacity + for i in range(count): + pos = self._origin % self._capacity + self._packets[pos] = None + self._origin = uint16_add(self._origin, 1) + + def smart_remove(self, count: int) -> bool: + """ + Makes sure that all packages belonging to the same frame are removed + to prevent sending corrupted frames to the decoder. + """ + timestamp = None + for i in range(self._capacity): + pos = self._origin % self._capacity + packet = self._packets[pos] + if packet is not None: + if i >= count and timestamp != packet.timestamp: + break + timestamp = packet.timestamp + self._packets[pos] = None + self._origin = uint16_add(self._origin, 1) + if i == self._capacity - 1: + return True + return False diff --git a/python-webrtc/python/webrtc/aiortc/rtcdtlstransport.py b/python-webrtc/python/webrtc/aiortc/rtcdtlstransport.py index 4262168..864b170 100644 --- a/python-webrtc/python/webrtc/aiortc/rtcdtlstransport.py +++ b/python-webrtc/python/webrtc/aiortc/rtcdtlstransport.py @@ -28,6 +28,8 @@ # attributes are never on the first line, and a newline before them is much faster to find than a line start _FINGERPRINT = re.compile(r'\na=fingerprint:(\S+) (\S+)') +# settable but unused: the engine makes one SHA-256 fingerprint +X509_DIGEST_ALGORITHMS = {'sha-256': 'SHA256', 'sha-384': 'SHA384', 'sha-512': 'SHA512'} @dataclass diff --git a/python-webrtc/python/webrtc/aiortc/rtcicetransport.py b/python-webrtc/python/webrtc/aiortc/rtcicetransport.py index abd32a1..1c98b64 100644 --- a/python-webrtc/python/webrtc/aiortc/rtcicetransport.py +++ b/python-webrtc/python/webrtc/aiortc/rtcicetransport.py @@ -13,6 +13,12 @@ from pyee.asyncio import AsyncIOEventEmitter import webrtc +from webrtc.aiortc._ice_servers import ( + STUN_REGEX as STUN_REGEX, + TURN_REGEX as TURN_REGEX, + connection_kwargs as connection_kwargs, + parse_stun_turn_uri as parse_stun_turn_uri, +) from webrtc.aiortc._utils import _emitter, _ice_state, _ReadOnly from webrtc.aiortc.rtcconfiguration import RTCIceServer diff --git a/python-webrtc/python/webrtc/aiortc/rtcrtpreceiver.py b/python-webrtc/python/webrtc/aiortc/rtcrtpreceiver.py index 3ac4262..dc9dcf6 100644 --- a/python-webrtc/python/webrtc/aiortc/rtcrtpreceiver.py +++ b/python-webrtc/python/webrtc/aiortc/rtcrtpreceiver.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import collections import fractions from dataclasses import dataclass @@ -33,7 +34,6 @@ from webrtc.aiortc.sdp import parameters_from_sdp if TYPE_CHECKING: - import asyncio import datetime as dt from av.frame import Frame @@ -184,6 +184,9 @@ def _fill(self, data: webrtc.AudioData) -> list[Frame]: self._pending = None return frames + def drop_partial(self) -> None: + self._pending = None + def _stamped(self, frame: av.AudioFrame) -> av.AudioFrame: frame.pts = self._samples frame.sample_rate = self._rate @@ -216,9 +219,11 @@ def __init__(self, kind: str, id: str | None = None) -> None: if id is not None: self._id = id self._buffered = True + self._processor: webrtc.MediaStreamTrackProcessor | None = None self._reader: webrtc.ReadableStreamDefaultReader[webrtc.VideoFrame | webrtc.AudioData] | None = None self._converter: _VideoFrames | _AudioFrames | None = None self._frames: collections.deque[Frame] = collections.deque() + self._frame_queue: _FrameQueue | None = None @classmethod def _of(cls, native: webrtc.MediaStreamTrack, receiver: RTCRtpReceiver | None, id: str | None = None) -> Self: @@ -242,13 +247,45 @@ def _open(self) -> webrtc.ReadableStreamDefaultReader[webrtc.VideoFrame | webrtc buffer = _AUDIO_BUFFER init = webrtc.MediaStreamTrackProcessorInit(self._native, buffer) - self._reader = webrtc.MediaStreamTrackProcessor(init).readable.get_reader() + self._processor = webrtc.MediaStreamTrackProcessor(init) + self._reader = self._processor.readable.get_reader() + + # a reopened reader keeps the timestamps going + if self._converter is not None: + return self._reader if video: self._converter = _VideoFrames() else: self._converter = _AudioFrames(*_audio_shape(self._receiver)) return self._reader + def _release(self) -> None: + self._frames.clear() + if isinstance(self._converter, _AudioFrames): + self._converter.drop_partial() + if self._reader is not None: + self._reader.cancel().add_done_callback(_retrieved) + self._reader = None + self._processor = None + + def _ready_frames(self) -> int: + """How many frames recv() returns without waiting.""" + processor = self._processor + converter = self._converter + if processor is not None and converter is not None: + native = processor._native_obj + item = native.read() + while item is not None: + self._frames.extend(converter.convert(processor._wrap_media(item))) + item = native.read() + return len(self._frames) + + @property + def _queue(self) -> _FrameQueue: + if self._frame_queue is None: + self._frame_queue = _FrameQueue(self) + return self._frame_queue + @override async def recv(self) -> Frame: if self.readyState != 'live': @@ -257,6 +294,9 @@ async def recv(self) -> Frame: reader = self._reader if self._reader is not None else self._open() media = (await reader.read()).value if media is None or self._converter is None: + if reader is not self._reader and self.readyState == 'live': + # the receiver was disabled while this read waited + continue self.stop() raise MediaStreamError self._frames.extend(self._converter.convert(media)) @@ -265,11 +305,35 @@ async def recv(self) -> Frame: @override def stop(self) -> None: super().stop() - if self._reader is not None: - self._reader.cancel().add_done_callback(_retrieved) + self._release() self._native.stop() +class _FrameQueue(asyncio.Queue['Frame']): + # apps drain it to skip stale frames + def __init__(self, track: RemoteStreamTrack) -> None: + super().__init__() + self._track = track + + @override + def qsize(self) -> int: + return self._track._ready_frames() + + @override + def empty(self) -> bool: + return self.qsize() == 0 + + @override + def get_nowait(self) -> Frame: + if self.empty(): + raise asyncio.QueueEmpty + return self._track._frames.popleft() + + @override + def task_done(self) -> None: + pass + + def _retrieved(future: asyncio.Future[None]) -> None: if not future.cancelled(): _ = future.exception() @@ -319,11 +383,22 @@ async def getStats(self) -> RTCStatsReport: class RTCRtpReceiver(_RtpEndpoint[webrtc.RTCRtpReceiver]): _native_class = webrtc.RTCRtpReceiver _track: RemoteStreamTrack | None = None + _receiving = True track: _ReadOnly[RemoteStreamTrack | None] = _ReadOnly('_track') def __init__(self, kind: str, transport: RTCDtlsTransport) -> None: raise NotImplementedError(_ORTC) + @property + def _enabled(self) -> bool: + return self._receiving + + @_enabled.setter + def _enabled(self, enabled: bool) -> None: + if not enabled and self._track is not None: + self._track._release() + self._receiving = enabled + def getSynchronizationSources(self) -> list[RTCRtpSynchronizationSource]: sources = self._native.get_synchronization_sources() return [RTCRtpSynchronizationSource(_datetime(s.timestamp), s.source) for s in sources] diff --git a/python-webrtc/python/webrtc/aiortc/rtcsctptransport.py b/python-webrtc/python/webrtc/aiortc/rtcsctptransport.py index 33bf434..cab2010 100644 --- a/python-webrtc/python/webrtc/aiortc/rtcsctptransport.py +++ b/python-webrtc/python/webrtc/aiortc/rtcsctptransport.py @@ -23,6 +23,8 @@ # attributes are never on the first line, and a newline before them is much faster to find than a line start _SCTP_PORT = re.compile(r'\na=sctp-port:(\d+)') +# settable but unused: the engine sizes its own chunks +USERDATA_MAX_LENGTH = 1200 @dataclass diff --git a/python-webrtc/python/webrtc/aiortc/rtp.py b/python-webrtc/python/webrtc/aiortc/rtp.py index d5ffcaf..84ed989 100644 --- a/python-webrtc/python/webrtc/aiortc/rtp.py +++ b/python-webrtc/python/webrtc/aiortc/rtp.py @@ -1,10 +1,818 @@ +# Copied from aiortc 1.15.0 (src/aiortc/rtp.py), unchanged. # -# Copyright 2022-2026 Ilya (Marshal) . All rights reserved. +# Copyright (c) Jeremy Lainé. +# All rights reserved. # -# Use of this source code is governed by a BSD-style license -# that can be found in the LICENSE.md file in the root of the project. +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: # +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of aiortc nor the names of its contributors may +# be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import math +import os +import struct +from dataclasses import dataclass, field +from struct import pack, unpack, unpack_from +from typing import Any, Optional, Union + +from av import AudioFrame + +from .rtcrtpparameters import RTCRtpParameters -from __future__ import annotations +# used for NACK and retransmission +RTP_HISTORY_SIZE = 128 +# reserved to avoid confusion with RTCP FORBIDDEN_PAYLOAD_TYPES = range(72, 77) +DYNAMIC_PAYLOAD_TYPES = range(96, 128) + +RTP_HEADER_LENGTH = 12 +RTCP_HEADER_LENGTH = 4 + +PACKETS_LOST_MIN = -(1 << 23) +PACKETS_LOST_MAX = (1 << 23) - 1 + +RTCP_SR = 200 +RTCP_RR = 201 +RTCP_SDES = 202 +RTCP_BYE = 203 +RTCP_RTPFB = 205 +RTCP_PSFB = 206 + +RTCP_RTPFB_NACK = 1 + +RTCP_PSFB_PLI = 1 +RTCP_PSFB_SLI = 2 +RTCP_PSFB_RPSI = 3 +RTCP_PSFB_FIR = 4 +RTCP_PSFB_APP = 15 + + +@dataclass +class HeaderExtensions: + abs_send_time: Optional[int] = None + audio_level: Any = None + mid: Any = None + repaired_rtp_stream_id: Any = None + rtp_stream_id: Any = None + transmission_offset: Optional[int] = None + transport_sequence_number: Optional[int] = None + + +class HeaderExtensionsMap: + def __init__(self) -> None: + self.__ids = HeaderExtensions() + + def configure(self, parameters: RTCRtpParameters) -> None: + for ext in parameters.headerExtensions: + if ext.uri == "urn:ietf:params:rtp-hdrext:sdes:mid": + self.__ids.mid = ext.id + elif ext.uri == "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id": + self.__ids.repaired_rtp_stream_id = ext.id + elif ext.uri == "urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id": + self.__ids.rtp_stream_id = ext.id + elif ( + ext.uri == "http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time" + ): + self.__ids.abs_send_time = ext.id + elif ext.uri == "urn:ietf:params:rtp-hdrext:toffset": + self.__ids.transmission_offset = ext.id + elif ext.uri == "urn:ietf:params:rtp-hdrext:ssrc-audio-level": + self.__ids.audio_level = ext.id + elif ( + ext.uri + == "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01" + ): + self.__ids.transport_sequence_number = ext.id + + def get(self, extension_profile: int, extension_value: bytes) -> HeaderExtensions: + values = HeaderExtensions() + for x_id, x_value in unpack_header_extensions( + extension_profile, extension_value + ): + if x_id == self.__ids.mid: + values.mid = x_value.decode("utf8") + elif x_id == self.__ids.repaired_rtp_stream_id: + values.repaired_rtp_stream_id = x_value.decode("ascii") + elif x_id == self.__ids.rtp_stream_id: + values.rtp_stream_id = x_value.decode("ascii") + elif x_id == self.__ids.abs_send_time: + values.abs_send_time = unpack("!L", b"\00" + x_value)[0] + elif x_id == self.__ids.transmission_offset: + values.transmission_offset = unpack("!l", x_value + b"\00")[0] >> 8 + elif x_id == self.__ids.audio_level: + vad_level = unpack("!B", x_value)[0] + values.audio_level = (vad_level & 0x80 == 0x80, vad_level & 0x7F) + elif x_id == self.__ids.transport_sequence_number: + values.transport_sequence_number = unpack("!H", x_value)[0] + return values + + def set(self, values: HeaderExtensions) -> tuple[int, bytes]: + extensions = [] + if values.mid is not None and self.__ids.mid: + extensions.append((self.__ids.mid, values.mid.encode("utf8"))) + if ( + values.repaired_rtp_stream_id is not None + and self.__ids.repaired_rtp_stream_id + ): + extensions.append( + ( + self.__ids.repaired_rtp_stream_id, + values.repaired_rtp_stream_id.encode("ascii"), + ) + ) + if values.rtp_stream_id is not None and self.__ids.rtp_stream_id: + extensions.append( + (self.__ids.rtp_stream_id, values.rtp_stream_id.encode("ascii")) + ) + if values.abs_send_time is not None and self.__ids.abs_send_time: + extensions.append( + (self.__ids.abs_send_time, pack("!L", values.abs_send_time)[1:]) + ) + if values.transmission_offset is not None and self.__ids.transmission_offset: + extensions.append( + ( + self.__ids.transmission_offset, + pack("!l", values.transmission_offset << 8)[0:2], + ) + ) + if values.audio_level is not None and self.__ids.audio_level: + extensions.append( + ( + self.__ids.audio_level, + pack( + "!B", + (0x80 if values.audio_level[0] else 0) + | (values.audio_level[1] & 0x7F), + ), + ) + ) + if ( + values.transport_sequence_number is not None + and self.__ids.transport_sequence_number + ): + extensions.append( + ( + self.__ids.transport_sequence_number, + pack("!H", values.transport_sequence_number), + ) + ) + return pack_header_extensions(extensions) + + +def clamp_packets_lost(count: int) -> int: + return max(PACKETS_LOST_MIN, min(count, PACKETS_LOST_MAX)) + + +def pack_packets_lost(count: int) -> bytes: + return pack("!l", count)[1:] + + +def unpack_packets_lost(d: bytes) -> int: + if d[0] & 0x80: + d = b"\xff" + d + else: + d = b"\x00" + d + return unpack("!l", d)[0] + + +def pack_rtcp_packet(packet_type: int, count: int, payload: bytes) -> bytes: + assert len(payload) % 4 == 0 + return pack("!BBH", (2 << 6) | count, packet_type, len(payload) // 4) + payload + + +def pack_remb_fci(bitrate: int, ssrcs: list[int]) -> bytes: + """ + Pack the FCI for a Receiver Estimated Maximum Bitrate report. + + https://tools.ietf.org/html/draft-alvestrand-rmcat-remb-03 + """ + data = b"REMB" + exponent = 0 + mantissa = bitrate + while mantissa > 0x3FFFF: + mantissa >>= 1 + exponent += 1 + data += pack( + "!BBH", len(ssrcs), (exponent << 2) | (mantissa >> 16), (mantissa & 0xFFFF) + ) + for ssrc in ssrcs: + data += pack("!L", ssrc) + return data + + +def unpack_remb_fci(data: bytes) -> tuple[int, list[int]]: + """ + Unpack the FCI for a Receiver Estimated Maximum Bitrate report. + + https://tools.ietf.org/html/draft-alvestrand-rmcat-remb-03 + """ + if len(data) < 8 or data[0:4] != b"REMB": + raise ValueError("Invalid REMB prefix") + + exponent = (data[5] & 0xFC) >> 2 + mantissa = ((data[5] & 0x03) << 16) | (data[6] << 8) | data[7] + bitrate = mantissa << exponent + + pos = 8 + ssrcs = [] + for r in range(data[4]): + ssrcs.append(unpack_from("!L", data, pos)[0]) + pos += 4 + + return (bitrate, ssrcs) + + +def is_rtcp(msg: bytes) -> bool: + return len(msg) >= 2 and msg[1] >= 192 and msg[1] <= 208 + + +def padl(length: int) -> int: + """ + Return amount of padding needed for a 4-byte multiple. + """ + return 4 * ((length + 3) // 4) - length + + +def unpack_header_extensions( + extension_profile: int, extension_value: bytes +) -> list[tuple[int, bytes]]: + """ + Parse header extensions according to RFC 5285. + """ + extensions = [] + pos = 0 + + if extension_profile == 0xBEDE: + # One-Byte Header + while pos < len(extension_value): + # skip padding byte + if extension_value[pos] == 0: + pos += 1 + continue + + x_id = (extension_value[pos] & 0xF0) >> 4 + x_length = (extension_value[pos] & 0x0F) + 1 + pos += 1 + + if len(extension_value) < pos + x_length: + raise ValueError("RTP one-byte header extension value is truncated") + x_value = extension_value[pos : pos + x_length] + extensions.append((x_id, x_value)) + pos += x_length + elif extension_profile == 0x1000: + # Two-Byte Header + while pos < len(extension_value): + # skip padding byte + if extension_value[pos] == 0: + pos += 1 + continue + + if len(extension_value) < pos + 2: + raise ValueError("RTP two-byte header extension is truncated") + x_id, x_length = unpack_from("!BB", extension_value, pos) + pos += 2 + + if len(extension_value) < pos + x_length: + raise ValueError("RTP two-byte header extension value is truncated") + x_value = extension_value[pos : pos + x_length] + extensions.append((x_id, x_value)) + pos += x_length + + return extensions + + +def pack_header_extensions(extensions: list[tuple[int, bytes]]) -> tuple[int, bytes]: + """ + Serialize header extensions according to RFC 5285. + """ + extension_profile = 0 + extension_value = b"" + + if not extensions: + return extension_profile, extension_value + + one_byte = True + for x_id, x_value in extensions: + x_length = len(x_value) + assert x_id > 0 and x_id < 256 + assert x_length >= 0 and x_length < 256 + if x_id > 14 or x_length == 0 or x_length > 16: + one_byte = False + + if one_byte: + # One-Byte Header + extension_profile = 0xBEDE + extension_value = b"" + for x_id, x_value in extensions: + x_length = len(x_value) + extension_value += pack("!B", (x_id << 4) | (x_length - 1)) + extension_value += x_value + else: + # Two-Byte Header + extension_profile = 0x1000 + extension_value = b"" + for x_id, x_value in extensions: + x_length = len(x_value) + extension_value += pack("!BB", x_id, x_length) + extension_value += x_value + + extension_value += b"\x00" * padl(len(extension_value)) + return extension_profile, extension_value + + +def compute_audio_level_dbov(frame: AudioFrame) -> int: + """ + Compute the energy level as spelled out in RFC 6465, Appendix A. + """ + MAX_SAMPLE_VALUE = 32767 + MAX_AUDIO_LEVEL = 0 + MIN_AUDIO_LEVEL = -127 + rms = 0.0 + buf = bytes(frame.planes[0]) + s = struct.Struct("h") + for unpacked in s.iter_unpack(buf): + sample = unpacked[0] + rms += sample * sample + rms = math.sqrt(rms / (frame.samples * MAX_SAMPLE_VALUE * MAX_SAMPLE_VALUE)) + if rms > 0: + db = 20 * math.log10(rms) + db = max(db, MIN_AUDIO_LEVEL) + db = min(db, MAX_AUDIO_LEVEL) + else: + db = MIN_AUDIO_LEVEL + return round(db) + + +@dataclass +class RtcpReceiverInfo: + ssrc: int + fraction_lost: int + packets_lost: int + highest_sequence: int + jitter: int + lsr: int + dlsr: int + + def __bytes__(self) -> bytes: + data = pack("!LB", self.ssrc, self.fraction_lost) + data += pack_packets_lost(self.packets_lost) + data += pack("!LLLL", self.highest_sequence, self.jitter, self.lsr, self.dlsr) + return data + + @classmethod + def parse(cls, data: bytes) -> "RtcpReceiverInfo": + ssrc, fraction_lost = unpack("!LB", data[0:5]) + packets_lost = unpack_packets_lost(data[5:8]) + highest_sequence, jitter, lsr, dlsr = unpack("!LLLL", data[8:]) + return cls( + ssrc=ssrc, + fraction_lost=fraction_lost, + packets_lost=packets_lost, + highest_sequence=highest_sequence, + jitter=jitter, + lsr=lsr, + dlsr=dlsr, + ) + + +@dataclass +class RtcpSenderInfo: + ntp_timestamp: int + rtp_timestamp: int + packet_count: int + octet_count: int + + def __bytes__(self) -> bytes: + return pack( + "!QLLL", + self.ntp_timestamp, + self.rtp_timestamp, + self.packet_count, + self.octet_count, + ) + + @classmethod + def parse(cls, data: bytes) -> "RtcpSenderInfo": + ntp_timestamp, rtp_timestamp, packet_count, octet_count = unpack("!QLLL", data) + return cls( + ntp_timestamp=ntp_timestamp, + rtp_timestamp=rtp_timestamp, + packet_count=packet_count, + octet_count=octet_count, + ) + + +@dataclass +class RtcpSourceInfo: + ssrc: int + items: list[tuple[Any, bytes]] + + +@dataclass +class RtcpByePacket: + sources: list[int] + + def __bytes__(self) -> bytes: + payload = b"".join([pack("!L", ssrc) for ssrc in self.sources]) + return pack_rtcp_packet(RTCP_BYE, len(self.sources), payload) + + @classmethod + def parse(cls, data: bytes, count: int) -> "RtcpByePacket": + if len(data) < 4 * count: + raise ValueError("RTCP bye length is invalid") + if count > 0: + sources = list(unpack_from("!" + ("L" * count), data, 0)) + else: + sources = [] + return cls(sources=sources) + + +@dataclass +class RtcpPsfbPacket: + """ + Payload-Specific Feedback Message (RFC 4585). + """ + + fmt: int + ssrc: int + media_ssrc: int + fci: bytes = b"" + + def __bytes__(self) -> bytes: + payload = pack("!LL", self.ssrc, self.media_ssrc) + self.fci + return pack_rtcp_packet(RTCP_PSFB, self.fmt, payload) + + @classmethod + def parse(cls, data: bytes, fmt: int) -> "RtcpPsfbPacket": + if len(data) < 8: + raise ValueError("RTCP payload-specific feedback length is invalid") + + ssrc, media_ssrc = unpack("!LL", data[0:8]) + fci = data[8:] + return cls(fmt=fmt, ssrc=ssrc, media_ssrc=media_ssrc, fci=fci) + + +@dataclass +class RtcpRrPacket: + ssrc: int + reports: list[RtcpReceiverInfo] = field(default_factory=list) + + def __bytes__(self) -> bytes: + payload = pack("!L", self.ssrc) + for report in self.reports: + payload += bytes(report) + return pack_rtcp_packet(RTCP_RR, len(self.reports), payload) + + @classmethod + def parse(cls, data: bytes, count: int) -> "RtcpRrPacket": + if len(data) != 4 + 24 * count: + raise ValueError("RTCP receiver report length is invalid") + + ssrc = unpack("!L", data[0:4])[0] + pos = 4 + reports = [] + for r in range(count): + reports.append(RtcpReceiverInfo.parse(data[pos : pos + 24])) + pos += 24 + return cls(ssrc=ssrc, reports=reports) + + +@dataclass +class RtcpRtpfbPacket: + """ + Generic RTP Feedback Message (RFC 4585). + """ + + fmt: int + ssrc: int + media_ssrc: int + + # generick NACK + lost: list[int] = field(default_factory=list) + + def __bytes__(self) -> bytes: + payload = pack("!LL", self.ssrc, self.media_ssrc) + if self.lost: + pid = self.lost[0] + blp = 0 + for p in self.lost[1:]: + d = p - pid - 1 + if d < 16: + blp |= 1 << d + else: + payload += pack("!HH", pid, blp) + pid = p + blp = 0 + payload += pack("!HH", pid, blp) + return pack_rtcp_packet(RTCP_RTPFB, self.fmt, payload) + + @classmethod + def parse(cls, data: bytes, fmt: int) -> "RtcpRtpfbPacket": + if len(data) < 8 or len(data) % 4: + raise ValueError("RTCP RTP feedback length is invalid") + + ssrc, media_ssrc = unpack("!LL", data[0:8]) + lost = [] + for pos in range(8, len(data), 4): + pid, blp = unpack("!HH", data[pos : pos + 4]) + lost.append(pid) + for d in range(0, 16): + if (blp >> d) & 1: + lost.append(pid + d + 1) + return cls(fmt=fmt, ssrc=ssrc, media_ssrc=media_ssrc, lost=lost) + + +@dataclass +class RtcpSdesPacket: + chunks: list[RtcpSourceInfo] = field(default_factory=list) + + def __bytes__(self) -> bytes: + payload = b"" + for chunk in self.chunks: + payload += pack("!L", chunk.ssrc) + for d_type, d_value in chunk.items: + payload += pack("!BB", d_type, len(d_value)) + d_value + payload += b"\x00\x00" + while len(payload) % 4: + payload += b"\x00" + return pack_rtcp_packet(RTCP_SDES, len(self.chunks), payload) + + @classmethod + def parse(cls, data: bytes, count: int) -> "RtcpSdesPacket": + pos = 0 + chunks = [] + for r in range(count): + if len(data) < pos + 4: + raise ValueError("RTCP SDES source is truncated") + ssrc = unpack_from("!L", data, pos)[0] + pos += 4 + + items = [] + while pos < len(data) - 1: + d_type, d_length = unpack_from("!BB", data, pos) + pos += 2 + + if len(data) < pos + d_length: + raise ValueError("RTCP SDES item is truncated") + d_value = data[pos : pos + d_length] + pos += d_length + if d_type == 0: + break + else: + items.append((d_type, d_value)) + chunks.append(RtcpSourceInfo(ssrc=ssrc, items=items)) + return cls(chunks=chunks) + + +@dataclass +class RtcpSrPacket: + ssrc: int + sender_info: RtcpSenderInfo + reports: list[RtcpReceiverInfo] = field(default_factory=list) + + def __bytes__(self) -> bytes: + payload = pack("!L", self.ssrc) + payload += bytes(self.sender_info) + for report in self.reports: + payload += bytes(report) + return pack_rtcp_packet(RTCP_SR, len(self.reports), payload) + + @classmethod + def parse(cls, data: bytes, count: int) -> "RtcpSrPacket": + if len(data) != 24 + 24 * count: + raise ValueError("RTCP sender report length is invalid") + + ssrc = unpack_from("!L", data)[0] + sender_info = RtcpSenderInfo.parse(data[4:24]) + pos = 24 + reports = [] + for r in range(count): + reports.append(RtcpReceiverInfo.parse(data[pos : pos + 24])) + pos += 24 + return RtcpSrPacket(ssrc=ssrc, sender_info=sender_info, reports=reports) + + +AnyRtcpPacket = Union[ + RtcpByePacket, + RtcpPsfbPacket, + RtcpRrPacket, + RtcpRtpfbPacket, + RtcpSdesPacket, + RtcpSrPacket, +] + + +class RtcpPacket: + @classmethod + def parse(cls, data: bytes) -> list[AnyRtcpPacket]: + pos = 0 + packets: list[AnyRtcpPacket] = [] + + while pos < len(data): + if len(data) < pos + RTCP_HEADER_LENGTH: + raise ValueError( + f"RTCP packet length is less than {RTCP_HEADER_LENGTH} bytes" + ) + + v_p_count, packet_type, length = unpack("!BBH", data[pos : pos + 4]) + version = v_p_count >> 6 + padding = (v_p_count >> 5) & 1 + count = v_p_count & 0x1F + if version != 2: + raise ValueError("RTCP packet has invalid version") + pos += 4 + + end = pos + length * 4 + if len(data) < end: + raise ValueError("RTCP packet is truncated") + payload = data[pos:end] + pos = end + + if padding: + if not payload or not payload[-1] or payload[-1] > len(payload): + raise ValueError("RTCP packet padding length is invalid") + payload = payload[0 : -payload[-1]] + + if packet_type == RTCP_BYE: + packets.append(RtcpByePacket.parse(payload, count)) + elif packet_type == RTCP_SDES: + packets.append(RtcpSdesPacket.parse(payload, count)) + elif packet_type == RTCP_SR: + packets.append(RtcpSrPacket.parse(payload, count)) + elif packet_type == RTCP_RR: + packets.append(RtcpRrPacket.parse(payload, count)) + elif packet_type == RTCP_RTPFB: + packets.append(RtcpRtpfbPacket.parse(payload, count)) + elif packet_type == RTCP_PSFB: + packets.append(RtcpPsfbPacket.parse(payload, count)) + + return packets + + +class RtpPacket: + def __init__( + self, + payload_type: int = 0, + marker: int = 0, + sequence_number: int = 0, + timestamp: int = 0, + ssrc: int = 0, + payload: bytes = b"", + ) -> None: + self.version = 2 + self.marker = marker + self.payload_type = payload_type + self.sequence_number = sequence_number + self.timestamp = timestamp + self.ssrc = ssrc + self.csrc: list[int] = [] + self.extensions = HeaderExtensions() + self.payload = payload + self.padding_size = 0 + + def __repr__(self) -> str: + return ( + f"RtpPacket(seq={self.sequence_number}, ts={self.timestamp}, " + f"marker={self.marker}, payload={self.payload_type}, " + f"{len(self.payload)} bytes)" + ) + + @classmethod + def parse( + cls, data: bytes, extensions_map: HeaderExtensionsMap = HeaderExtensionsMap() + ) -> "RtpPacket": + if len(data) < RTP_HEADER_LENGTH: + raise ValueError( + f"RTP packet length is less than {RTP_HEADER_LENGTH} bytes" + ) + + v_p_x_cc, m_pt, sequence_number, timestamp, ssrc = unpack("!BBHLL", data[0:12]) + version = v_p_x_cc >> 6 + padding = (v_p_x_cc >> 5) & 1 + extension = (v_p_x_cc >> 4) & 1 + cc = v_p_x_cc & 0x0F + if version != 2: + raise ValueError("RTP packet has invalid version") + if len(data) < RTP_HEADER_LENGTH + 4 * cc: + raise ValueError("RTP packet has truncated CSRC") + + packet = cls( + marker=(m_pt >> 7), + payload_type=(m_pt & 0x7F), + sequence_number=sequence_number, + timestamp=timestamp, + ssrc=ssrc, + ) + + pos = RTP_HEADER_LENGTH + for i in range(0, cc): + packet.csrc.append(unpack_from("!L", data, pos)[0]) + pos += 4 + + if extension: + if len(data) < pos + 4: + raise ValueError("RTP packet has truncated extension profile / length") + extension_profile, extension_length = unpack_from("!HH", data, pos) + extension_length *= 4 + pos += 4 + + if len(data) < pos + extension_length: + raise ValueError("RTP packet has truncated extension value") + extension_value = data[pos : pos + extension_length] + pos += extension_length + packet.extensions = extensions_map.get(extension_profile, extension_value) + + if padding: + padding_len = data[-1] + if not padding_len or padding_len > len(data) - pos: + raise ValueError("RTP packet padding length is invalid") + packet.padding_size = padding_len + packet.payload = data[pos:-padding_len] + else: + packet.payload = data[pos:] + + return packet + + def serialize( + self, extensions_map: HeaderExtensionsMap = HeaderExtensionsMap() + ) -> bytes: + extension_profile, extension_value = extensions_map.set(self.extensions) + has_extension = bool(extension_value) + + padding = self.padding_size > 0 + data = pack( + "!BBHLL", + (self.version << 6) + | (padding << 5) + | (has_extension << 4) + | len(self.csrc), + (self.marker << 7) | self.payload_type, + self.sequence_number, + self.timestamp, + self.ssrc, + ) + for csrc in self.csrc: + data += pack("!L", csrc) + if has_extension: + data += pack("!HH", extension_profile, len(extension_value) >> 2) + data += extension_value + data += self.payload + if padding: + data += os.urandom(self.padding_size - 1) + data += bytes([self.padding_size]) + return data + + +def unwrap_rtx(rtx: RtpPacket, payload_type: int, ssrc: int) -> RtpPacket: + """ + Recover initial packet from a retransmission packet. + """ + packet = RtpPacket( + payload_type=payload_type, + marker=rtx.marker, + sequence_number=unpack("!H", rtx.payload[0:2])[0], + timestamp=rtx.timestamp, + ssrc=ssrc, + payload=rtx.payload[2:], + ) + packet.csrc = rtx.csrc + packet.extensions = rtx.extensions + return packet + + +def wrap_rtx( + packet: RtpPacket, payload_type: int, sequence_number: int, ssrc: int +) -> RtpPacket: + """ + Create a retransmission packet from a lost packet. + """ + rtx = RtpPacket( + payload_type=payload_type, + marker=packet.marker, + sequence_number=sequence_number, + timestamp=packet.timestamp, + ssrc=ssrc, + payload=pack("!H", packet.sequence_number) + packet.payload, + ) + rtx.csrc = packet.csrc + rtx.extensions = packet.extensions + return rtx diff --git a/python-webrtc/python/webrtc/aiortc/utils.py b/python-webrtc/python/webrtc/aiortc/utils.py new file mode 100644 index 0000000..df111fe --- /dev/null +++ b/python-webrtc/python/webrtc/aiortc/utils.py @@ -0,0 +1,82 @@ +# Copied from aiortc 1.15.0 (src/aiortc/utils.py), unchanged. +# +# Copyright (c) Jeremy Lainé. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of aiortc nor the names of its contributors may +# be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import os +from struct import unpack + + +def random16() -> int: + return unpack("!H", os.urandom(2))[0] + + +def random32() -> int: + return unpack("!L", os.urandom(4))[0] + + +def uint16_add(a: int, b: int) -> int: + """ + Return a + b. + """ + return (a + b) & 0xFFFF + + +def uint16_gt(a: int, b: int) -> bool: + """ + Return a > b. + """ + half_mod = 0x8000 + return ((a < b) and ((b - a) > half_mod)) or ((a > b) and ((a - b) < half_mod)) + + +def uint16_gte(a: int, b: int) -> bool: + """ + Return a >= b. + """ + return (a == b) or uint16_gt(a, b) + + +def uint32_add(a: int, b: int) -> int: + """ + Return a + b. + """ + return (a + b) & 0xFFFFFFFF + + +def uint32_gt(a: int, b: int) -> bool: + """ + Return a > b. + """ + half_mod = 0x80000000 + return ((a < b) and ((b - a) > half_mod)) or ((a > b) and ((a - b) < half_mod)) + + +def uint32_gte(a: int, b: int) -> bool: + """ + Return a >= b. + """ + return (a == b) or uint32_gt(a, b) diff --git a/python-webrtc/python/webrtc/utils/module_alias.py b/python-webrtc/python/webrtc/utils/module_alias.py new file mode 100644 index 0000000..117d9e0 --- /dev/null +++ b/python-webrtc/python/webrtc/utils/module_alias.py @@ -0,0 +1,92 @@ +# +# Copyright 2022-2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +"""Serves the imports of one package from another.""" + +from __future__ import annotations + +import importlib +import importlib.abc +import importlib.util +import sys +from typing import TYPE_CHECKING, ClassVar + +from typing_extensions import override + +if TYPE_CHECKING: + from collections.abc import Sequence + from importlib.machinery import ModuleSpec + from types import ModuleType + + +def _in_package(name: str, package: str) -> bool: + return name == package or name.startswith(package + '.') + + +class _Loader(importlib.abc.Loader): + def __init__(self, target: str) -> None: + self._target = target + self._spec: ModuleSpec | None = None + + @override + def create_module(self, spec: ModuleSpec) -> ModuleType: + module = importlib.import_module(self._target) + self._spec = module.__spec__ + return module + + @override + def exec_module(self, module: ModuleType) -> None: + # its relative imports read its own spec + module.__spec__ = self._spec + + +class ModuleAlias(importlib.abc.MetaPathFinder): + """Gives ``target.x`` for ``source.x``, never the source's own module; a subclass sets both.""" + + source: ClassVar[str] + target: ClassVar[str] + + @override + def find_spec( + self, fullname: str, path: Sequence[str] | None, target: ModuleType | None = None + ) -> ModuleSpec | None: + if not _in_package(fullname, self.source): + return None + + target_name = self.target + fullname[len(self.source) :] + found = importlib.util.find_spec(target_name) + if found is None: + return None + + is_package = found.submodule_search_locations is not None + return importlib.util.spec_from_loader(fullname, _Loader(target_name), is_package=is_package) + + def imported_originals(self) -> list[str]: + """The source's own modules imported already.""" + imported = list(sys.modules.items()) + return [ + name + for name, module in imported + if _in_package(name, self.source) and not _in_package(getattr(module, '__name__', ''), self.target) + ] + + +def install(alias: ModuleAlias) -> None: + """Puts the alias first on :data:`sys.meta_path`; one of the same type again does nothing. + + Raises: + RuntimeError: If a module of the source is imported already. + """ + if any(isinstance(finder, type(alias)) for finder in sys.meta_path): + return + + imported = alias.imported_originals() + if len(imported) > 0: + msg = f'{", ".join(imported)} imported already: call install_alias() before anything imports {alias.source}' + raise RuntimeError(msg) + + sys.meta_path.insert(0, alias) diff --git a/tests/aiortc/conftest.py b/tests/aiortc/conftest.py index a3c604c..4a45e99 100644 --- a/tests/aiortc/conftest.py +++ b/tests/aiortc/conftest.py @@ -7,8 +7,8 @@ from __future__ import annotations -import pytest +from importlib.util import find_spec -# the aiortc extra -_ = pytest.importorskip('av') -_ = pytest.importorskip('pyee') +# the aiortc extra; a skip here would fail a run of this directory alone +if find_spec('av') is None or find_spec('pyee') is None: + collect_ignore_glob = ['test_*.py'] diff --git a/tests/aiortc/test_alias.py b/tests/aiortc/test_alias.py new file mode 100644 index 0000000..a6655c1 --- /dev/null +++ b/tests/aiortc/test_alias.py @@ -0,0 +1,89 @@ +# +# Copyright 2022-2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +"""``import aiortc`` served by the layer, each in a process of its own.""" + +from __future__ import annotations + +import importlib +import importlib.util +import sys +import types + +import pytest + +import webrtc.aiortc +from tests.isolation import isolated +from webrtc.aiortc import install_alias + + +def alias_finders() -> int: + return sum(type(finder).__module__ == 'webrtc.aiortc._alias' for finder in sys.meta_path) + + +@isolated +def test_modules_are_the_layers() -> None: + install_alias() + + package = importlib.import_module('aiortc') + media = importlib.import_module('aiortc.contrib.media') + jitterbuffer = importlib.import_module('aiortc.jitterbuffer') + + assert package is webrtc.aiortc + assert media is importlib.import_module('webrtc.aiortc.contrib.media') + assert jitterbuffer is importlib.import_module('webrtc.aiortc.jitterbuffer') + assert sys.modules['aiortc.contrib.media'] is media + assert media.__spec__ is not None + assert media.__spec__.name == 'webrtc.aiortc.contrib.media' + + +@isolated +def test_found_without_importing() -> None: + install_alias() + + spec = importlib.util.find_spec('aiortc') + + assert spec is not None + assert spec.submodule_search_locations is not None + assert 'aiortc' not in sys.modules + + +@isolated +def test_a_module_the_layer_lacks_is_not_found() -> None: + install_alias() + + with pytest.raises(ModuleNotFoundError) as error: + _ = importlib.import_module('aiortc.rate') + + assert error.value.name == 'aiortc.rate' + + +@isolated +def test_installing_again_does_nothing() -> None: + install_alias() + _ = importlib.import_module('aiortc') + + install_alias() + + assert alias_finders() == 1 + + +@isolated +def test_fails_once_the_original_is_imported() -> None: + sys.modules['aiortc'] = types.ModuleType('aiortc') + + with pytest.raises(RuntimeError, match='aiortc imported already'): + install_alias() + + assert alias_finders() == 0 + + +@isolated +def test_other_packages_are_left_alone() -> None: + install_alias() + + assert importlib.util.find_spec('aiortc_extras') is None diff --git a/tests/aiortc/test_downstream_names.py b/tests/aiortc/test_downstream_names.py new file mode 100644 index 0000000..ffaebef --- /dev/null +++ b/tests/aiortc/test_downstream_names.py @@ -0,0 +1,445 @@ +# +# Copyright 2022-2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +"""The aiortc names dependents use, each row citing one use as ``project:path:line@commit``. + +A row goes when its project drops the use. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import enum +import importlib +import pkgutil +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Callable + +import pytest + +import webrtc +from tests.aiortc.test_media import CONFIGURATION, write_media +from tests.isolation import isolated +from webrtc.aiortc import RTCIceServer, RTCPeerConnection, RTCRtpTransceiver, install_alias +from webrtc.aiortc.contrib.media import MediaPlayer, MediaRelay +from webrtc.aiortc.jitterbuffer import JitterBuffer, JitterFrame +from webrtc.aiortc.rtcicetransport import STUN_REGEX, TURN_REGEX, connection_kwargs, parse_stun_turn_uri +from webrtc.aiortc.rtcrtpreceiver import RemoteStreamTrack +from webrtc.aiortc.rtp import RtpPacket + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + +@dataclass(frozen=True) +class Name: + """A module, or a name read from one, by its aiortc path.""" + + module: str + name: str | None + use: str + + +class Mode(enum.Enum): + READ = 'read' + WRITE = 'write' + CALL = 'call' + + +class Expected(enum.Enum): + WORKS = 'works' + NO_OP = 'settable no-op' + #: the dependent checks with hasattr or getattr first, and skips the use + GUARDED = 'absent, guarded' + #: guarded too, and the layer can't offer it honestly + UNSUPPORTED = 'unsupported' + + +#: stands for the running loop +RUNNING_LOOP = object() + + +@dataclass(frozen=True) +class Attribute: + """An attribute of an aiortc object or module, as a dependent uses it.""" + + owner: str + attribute: str + mode: Mode + expected: Expected + use: str + value: object = None + reason: str = '' + + +@dataclass(frozen=True) +class Dependency: + """A module of aiortc's own dependencies, which a dependent imports itself.""" + + module: str + use: str + reason: str + + +PIPECAT = 'pipecat:src/pipecat/transports/smallwebrtc/connection.py' +VISION_AGENTS = 'vision-agents:agents-core/vision_agents/core' +VISION_INWORLD = 'vision-agents:plugins/inworld/vision_agents/plugins/inworld/rtc_manager.py' +FASTRTC_TRACKS = 'fastrtc:backend/fastrtc/tracks.py' +ROBOFLOW_MANAGER = 'roboflow:stream_vision/streamvision/stream_manager/manager_app/webrtc.py' +ROBOFLOW_WORKER = 'roboflow:stream_vision/streamvision/webrtc_worker/webrtc.py' +GRADIO_BITRATE = 'gradio-pr:gradio/webrtc/bitrate.py' +NO_ENCODER = 'the engine encodes in its own threads, with no encoder object per sender; bitrates go through the codecs' + +NAMES = [ + Name('aiortc', None, f'{VISION_AGENTS}/avatars/avatar.py:4@902db86'), + Name('aiortc', 'AudioStreamTrack', 'vision-agents:plugins/local/vision_agents/plugins/local/tracks.py:20@902db86'), + Name('aiortc', 'MediaStreamError', f'{VISION_AGENTS}/utils/video_forwarder.py:8@902db86'), + Name('aiortc', 'MediaStreamTrack', f'{VISION_AGENTS}/edge/edge_transport.py:119@902db86'), + Name('aiortc', 'RTCBundlePolicy', f'{PIPECAT}:28@6f777a0'), + Name('aiortc', 'RTCBundlePolicy.MAX_BUNDLE', f'{PIPECAT}:336@6f777a0'), + Name('aiortc', 'RTCConfiguration', f'{VISION_INWORLD}:15@902db86'), + Name('aiortc', 'RTCDataChannel', f'{VISION_INWORLD}:15@902db86'), + Name('aiortc', 'RTCIceCandidate', 'fastrtc:backend/fastrtc/webrtc_connection_mixin.py:19@59ca4f7'), + Name('aiortc', 'RTCIceServer', f'{VISION_INWORLD}:15@902db86'), + Name('aiortc', 'RTCPeerConnection', f'{VISION_INWORLD}:15@902db86'), + Name( + 'aiortc', 'RTCRtpSender', 'vision-agents:plugins/openai/vision_agents/plugins/openai/rtc_manager.py:7@902db86' + ), + Name('aiortc', 'RTCSessionDescription', f'{VISION_INWORLD}:15@902db86'), + Name('aiortc', 'VideoStreamTrack', f'{VISION_AGENTS}/agents/agents.py:14@902db86'), + Name('aiortc', 'VideoStreamTrack.__init__', f'{ROBOFLOW_WORKER}:769@f79bf9f'), + Name('aiortc', '__version__', 'stream-py:getstream/video/rtc/__init__.py:55@8d7c51f'), + Name('aiortc', 'mediastreams', f'{VISION_AGENTS}/agents/agent_types.py:40@902db86'), + Name('aiortc', 'rtcdtlstransport', 'unitree:unitree_webrtc_connect/__init__.py:32@e0abc57'), + Name('aiortc.codecs', 'h264', f'{GRADIO_BITRATE}:21@b84a025'), + Name('aiortc.codecs', 'vpx', f'{GRADIO_BITRATE}:21@b84a025'), + Name('aiortc.contrib.media', 'AudioFrame', f'{FASTRTC_TRACKS}:32@59ca4f7'), + Name('aiortc.contrib.media', 'MediaPlayer', 'unitree:examples/go2/audio/internet_radio/stream_radio.py:6@e0abc57'), + Name('aiortc.contrib.media', 'MediaRecorder', 'streamlit-webrtc:pages/7_record.py:7@7484880'), + Name('aiortc.contrib.media', 'MediaRelay', 'vidgear:vidgear/gears/asyncio/webgear_rtc.py:80@3d37ad8'), + Name('aiortc.contrib.media', 'MediaStreamError', f'{ROBOFLOW_MANAGER}:18@f79bf9f'), + Name('aiortc.contrib.media', 'PlayerStreamTrack', f'{ROBOFLOW_WORKER}:24@f79bf9f'), + Name('aiortc.contrib.media', 'RelayStreamTrack', 'streamlit-webrtc:streamlit_webrtc/mix.py:13@7484880'), + Name('aiortc.contrib.media', 'VideoFrame', f'{FASTRTC_TRACKS}:32@59ca4f7'), + Name('aiortc.contrib.media', 'VideoFrame.from_ndarray', f'{FASTRTC_TRACKS}:132@59ca4f7'), + Name('aiortc.jitterbuffer', 'JitterBuffer', f'{PIPECAT}:35@6f777a0'), + Name('aiortc.mediastreams', None, f'{VISION_AGENTS}/agents/agent_types.py:5@902db86'), + Name('aiortc.mediastreams', 'AUDIO_PTIME', 'stream-py:getstream/video/rtc/audio_track.py:71@8d7c51f'), + Name( + 'aiortc.mediastreams', 'AudioStreamTrack', 'pipecat:src/pipecat/transports/smallwebrtc/transport.py:59@6f777a0' + ), + Name('aiortc.mediastreams', 'MediaStreamError', f'{VISION_AGENTS}/avatars/av_synchronizer.py:9@902db86'), + Name('aiortc.mediastreams', 'MediaStreamTrack', f'{VISION_AGENTS}/llm/llm.py:497@902db86'), + Name('aiortc.mediastreams', 'VIDEO_CLOCK_RATE', f'{FASTRTC_TRACKS}:33@59ca4f7'), + Name('aiortc.mediastreams', 'VIDEO_PTIME', 'roboflow:inference_sdk/webrtc/sources.py:21@f79bf9f'), + Name('aiortc.mediastreams', 'VIDEO_TIME_BASE', f'{FASTRTC_TRACKS}:33@59ca4f7'), + Name('aiortc.mediastreams', 'VideoStreamTrack', f'{VISION_AGENTS}/agents/agent_types.py:40@902db86'), + Name('aiortc.rtcdtlstransport', 'X509_DIGEST_ALGORITHMS', 'unitree:unitree_webrtc_connect/__init__.py:32@e0abc57'), + Name('aiortc.rtcicetransport', 'connection_kwargs', 'roboflow:inference_sdk/webrtc/turn_probe.py:16@f79bf9f'), + Name('aiortc.rtcicetransport', 'parse_stun_turn_uri', 'roboflow:inference_sdk/webrtc/turn_probe.py:16@f79bf9f'), + Name('aiortc.rtcrtpreceiver', 'RTCRtpReceiver', f'{ROBOFLOW_MANAGER}:19@f79bf9f'), + Name('aiortc.rtcrtpreceiver', 'RemoteStreamTrack', f'{PIPECAT}:36@6f777a0'), + Name('aiortc.rtcrtpsender', 'RTCRtpSender', 'vidgear:vidgear/gears/asyncio/webgear_rtc.py:82@3d37ad8'), + Name( + 'aiortc.rtcrtpsender', + 'RTCRtpSender.getCapabilities', + 'vidgear:vidgear/gears/asyncio/webgear_rtc.py:646@3d37ad8', + ), + Name('aiortc.rtcsctptransport', None, f'{PIPECAT}:27@6f777a0'), + Name('aiortc.rtcsctptransport', 'USERDATA_MAX_LENGTH', f'{PIPECAT}:79@6f777a0'), + Name('aiortc.rtp', 'RtpPacket', 'pipecat:tests/test_smallwebrtc_transport.py:430@6f777a0'), + Name('aiortc.sdp', 'candidate_from_sdp', 'gradio-pr:gradio/webrtc/connection.py:42@b84a025'), +] + +ATTRIBUTES = [ + Attribute('RTCPeerConnection', '_loop', Mode.WRITE, Expected.WORKS, f'{ROBOFLOW_WORKER}:217@f79bf9f', RUNNING_LOOP), + Attribute( + 'RTCPeerConnection', + 'emit', + Mode.CALL, + Expected.WORKS, + 'pipecat:tests/test_smallwebrtc_transport.py:131@6f777a0', + ), + Attribute('RTCPeerConnection', 'on', Mode.CALL, Expected.WORKS, f'{PIPECAT}:355@6f777a0'), + Attribute('RTCPeerConnection', 'remove_all_listeners', Mode.CALL, Expected.WORKS, f'{PIPECAT}:481@6f777a0'), + Attribute('RTCRtpTransceiver', 'direction', Mode.WRITE, Expected.WORKS, f'{PIPECAT}:508@6f777a0', 'recvonly'), + Attribute('RTCRtpReceiver', '_enabled', Mode.WRITE, Expected.WORKS, f'{PIPECAT}:156@6f777a0', value=False), + Attribute('RTCRtpReceiver', '_enabled', Mode.READ, Expected.WORKS, f'{PIPECAT}:219@6f777a0'), + Attribute( + 'RTCRtpReceiver', + '_RTCRtpReceiver__jitter_buffer', + Mode.WRITE, + Expected.NO_OP, + f'{PIPECAT}:158@6f777a0', + JitterBuffer(capacity=512, is_video=True), + reason="the engine's packet buffer grows to 2048 packets for large frames", + ), + Attribute( + 'RTCRtpReceiver', + '_RTCRtpReceiver__jitter_buffer', + Mode.READ, + Expected.GUARDED, + f'{ROBOFLOW_MANAGER}:212@f79bf9f', + ), + Attribute( + 'RTCRtpReceiver', + '_RTCRtpReceiver__decoder_queue', + Mode.READ, + Expected.GUARDED, + f'{ROBOFLOW_MANAGER}:202@f79bf9f', + ), + Attribute( + 'RTCRtpReceiver', + '_RTCRtpReceiver__stop_decoder', + Mode.READ, + Expected.GUARDED, + 'streamlit-webrtc:streamlit_webrtc/webrtc.py:372@7484880', + ), + Attribute('RemoteStreamTrack', '_queue', Mode.READ, Expected.WORKS, f'{PIPECAT}:188@6f777a0'), + Attribute('RemoteStreamTrack', '_queue.empty', Mode.CALL, Expected.WORKS, f'{PIPECAT}:194@6f777a0'), + Attribute('RemoteStreamTrack', '_queue.get_nowait', Mode.CALL, Expected.WORKS, f'{PIPECAT}:195@6f777a0'), + Attribute('RemoteStreamTrack', '_queue.task_done', Mode.CALL, Expected.WORKS, f'{PIPECAT}:196@6f777a0'), + Attribute('RemoteStreamTrack', '_queue.qsize', Mode.CALL, Expected.WORKS, f'{ROBOFLOW_MANAGER}:228@f79bf9f'), + Attribute('RelayStreamTrack', '_queue.qsize', Mode.CALL, Expected.WORKS, f'{ROBOFLOW_MANAGER}:193@f79bf9f'), + Attribute('PlayerStreamTrack', '_queue.qsize', Mode.CALL, Expected.WORKS, f'{ROBOFLOW_WORKER}:608@f79bf9f'), + Attribute('PlayerStreamTrack', '_queue.get_nowait', Mode.CALL, Expected.WORKS, f'{ROBOFLOW_WORKER}:609@f79bf9f'), + Attribute( + 'RTCRtpTransceiver', + '_codecs', + Mode.READ, + Expected.UNSUPPORTED, + f'{GRADIO_BITRATE}:119@b84a025', + reason=NO_ENCODER, + ), + Attribute( + 'RTCRtpSender', + '_RTCRtpSender__encoder', + Mode.READ, + Expected.UNSUPPORTED, + f'{GRADIO_BITRATE}:121@b84a025', + reason=NO_ENCODER, + ), + Attribute( + 'codecs.vpx', 'Vp8Encoder', Mode.READ, Expected.UNSUPPORTED, f'{GRADIO_BITRATE}:105@b84a025', reason=NO_ENCODER + ), + Attribute( + 'codecs.h264', + 'H264Encoder', + Mode.READ, + Expected.UNSUPPORTED, + f'{GRADIO_BITRATE}:106@b84a025', + reason=NO_ENCODER, + ), +] + +DEPENDENCIES = [ + Dependency('aioice', 'unitree:unitree_webrtc_connect/__init__.py:3@e0abc57', 'its patched connections go unused'), + Dependency('aioice', 'roboflow:inference_sdk/webrtc/turn_probe.py:14@f79bf9f', 'its TURN probe still runs'), + Dependency('aioice', f'{ROBOFLOW_WORKER}:14@f79bf9f', 'its mDNS lock goes unused'), + Dependency('cryptography', 'unitree:unitree_webrtc_connect/__init__.py:36@e0abc57', 'its digests go unused'), +] + + +def lookup(owner: object, dotted: str) -> object: + """The attribute at a dotted path, such as ``VideoFrame.from_ndarray``.""" + found = owner + for attribute in dotted.split('.'): + found = getattr(found, attribute) + return found + + +@contextlib.asynccontextmanager +async def connection() -> AsyncGenerator[RTCPeerConnection]: + pc = RTCPeerConnection(CONFIGURATION) + try: + yield pc + finally: + await pc.close() + + +@contextlib.asynccontextmanager +async def transceiver() -> AsyncGenerator[RTCRtpTransceiver]: + async with connection() as pc: + yield pc.addTransceiver('video') + + +@contextlib.asynccontextmanager +async def sender() -> AsyncGenerator[object]: + async with transceiver() as found: + yield found.sender + + +@contextlib.asynccontextmanager +async def receiver() -> AsyncGenerator[object]: + async with transceiver() as found: + yield found.receiver + + +@contextlib.asynccontextmanager +async def remote_track() -> AsyncGenerator[RemoteStreamTrack]: + generator = webrtc.VideoTrackGenerator() + track = RemoteStreamTrack._of(generator.track, None) + try: + yield track + finally: + track.stop() + + +@contextlib.asynccontextmanager +async def relayed_track() -> AsyncGenerator[object]: + async with remote_track() as track: + relayed = MediaRelay().subscribe(track) + try: + yield relayed + finally: + relayed.stop() + + +@contextlib.asynccontextmanager +async def player_track() -> AsyncGenerator[object]: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'media.webm' + write_media(path) + player = MediaPlayer(str(path)) + assert player.video is not None + try: + yield player.video + finally: + player.video.stop() + + +@contextlib.asynccontextmanager +async def vpx() -> AsyncGenerator[object]: + yield importlib.import_module('webrtc.aiortc.codecs.vpx') + + +@contextlib.asynccontextmanager +async def h264() -> AsyncGenerator[object]: + yield importlib.import_module('webrtc.aiortc.codecs.h264') + + +OWNERS: dict[str, Callable[[], contextlib.AbstractAsyncContextManager[object]]] = { + 'RTCPeerConnection': connection, + 'RTCRtpTransceiver': transceiver, + 'RTCRtpSender': sender, + 'RTCRtpReceiver': receiver, + 'RemoteStreamTrack': remote_track, + 'RelayStreamTrack': relayed_track, + 'PlayerStreamTrack': player_track, + 'codecs.vpx': vpx, + 'codecs.h264': h264, +} + + +def name_id(row: Name) -> str: + return f'{row.module}:{row.name}' + + +def attribute_id(row: Attribute) -> str: + return f'{row.owner}.{row.attribute}:{row.mode.value}' + + +@pytest.mark.parametrize('row', NAMES, ids=name_id) +def test_name(row: Name) -> None: + module = importlib.import_module(f'webrtc.{row.module}') + + if row.name is not None: + lookup(module, row.name) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('row', ATTRIBUTES, ids=attribute_id) +async def test_attribute(row: Attribute) -> None: + async with OWNERS[row.owner]() as owner: + if row.expected in {Expected.GUARDED, Expected.UNSUPPORTED}: + assert not hasattr(owner, row.attribute) + elif row.mode is Mode.WRITE: + value = asyncio.get_running_loop() if row.value is RUNNING_LOOP else row.value + setattr(owner, row.attribute, value) + assert getattr(owner, row.attribute) is value + elif row.mode is Mode.CALL: + assert callable(lookup(owner, row.attribute)) + else: + lookup(owner, row.attribute) + + +def test_rows_say_why() -> None: + """A no-op or unsupported row gives its reason.""" + unexplained = [ + row for row in ATTRIBUTES if row.expected in {Expected.NO_OP, Expected.UNSUPPORTED} and row.reason == '' + ] + + assert unexplained == [] + + +@isolated +def test_names_import_through_the_alias() -> None: + install_alias() + + for row in NAMES: + module = importlib.import_module(row.module) + assert module is importlib.import_module(f'webrtc.{row.module}') + if row.name is not None: + lookup(module, row.name) + + +@isolated +def test_layer_imports_none_of_the_dependencies() -> None: + package = importlib.import_module('webrtc.aiortc') + for module in pkgutil.walk_packages(package.__path__, 'webrtc.aiortc.'): + importlib.import_module(module.name) + + imported = [row.module for row in DEPENDENCIES if row.module in sys.modules] + assert imported == [] + + +def test_jitter_buffer_assembles_a_frame_of_many_packets() -> None: + """A 300-packet frame comes out whole once the next frame's first packet arrives.""" + buffer = JitterBuffer(capacity=512, is_video=True) + + frames: list[JitterFrame] = [] + for sequence_number in range(301): + timestamp = 0 if sequence_number < 300 else 3000 + packet = RtpPacket(payload_type=96, sequence_number=sequence_number, timestamp=timestamp) + # the depayloader sets it in a receiver + packet._data = b'x' # pyrefly: ignore[missing-attribute] + pli, frame = buffer.add(packet) + assert not pli + if frame is not None: + frames.append(frame) + + assert [len(frame.data) for frame in frames] == [300] + + +def test_stun_and_turn_urls_parse() -> None: + servers = [ + RTCIceServer('stun:stun.example.com'), + RTCIceServer('turns:turn.example.com:443', username='user', credential='secret'), + ] + + kwargs = connection_kwargs(servers) + + assert STUN_REGEX.fullmatch('stun:stun.example.com:3478') is not None + assert TURN_REGEX.fullmatch('turn:turn.example.com?transport=tcp') is not None + assert parse_stun_turn_uri('turn:turn.example.com') == { + 'scheme': 'turn', + 'host': 'turn.example.com', + 'port': 3478, + 'transport': 'udp', + } + assert kwargs == { + 'stun_server': ('stun.example.com', 3478), + 'turn_server': ('turn.example.com', 443), + 'turn_ssl': True, + 'turn_transport': 'tcp', + 'turn_username': 'user', + 'turn_password': 'secret', + } diff --git a/tests/aiortc/test_media.py b/tests/aiortc/test_media.py index 0979310..9f75dda 100644 --- a/tests/aiortc/test_media.py +++ b/tests/aiortc/test_media.py @@ -28,6 +28,7 @@ RTCOutboundRtpStreamStats, RTCPeerConnection, RTCRtpCodecCapability, + RTCRtpReceiver, RTCRtpSender, RTCSessionDescription, RTCStatsReport, @@ -187,6 +188,99 @@ async def test_unbuffered_relay_reads_the_latest_frame() -> None: await close(caller, callee) +async def received_alone(track: MediaStreamTrack) -> tuple[RTCPeerConnection, RTCPeerConnection, RemoteStreamTrack]: + """A caller sending the track, a callee and the track it receives.""" + caller, callee = new_pair() + tracks = received_tracks(callee) + caller.addTrack(track) + await negotiate(caller, callee) + return caller, callee, await next_track(tracks) + + +def only_receiver(pc: RTCPeerConnection) -> RTCRtpReceiver: + receivers = pc.getReceivers() + assert len(receivers) == 1 + return receivers[0] + + +@pytest.mark.asyncio +async def test_disabled_receiver_keeps_no_video() -> None: + caller, callee, track = await received_alone(VideoStreamTrack()) + try: + receiver = only_receiver(callee) + before = await recv_video(track) + + receiver._enabled = False + await asyncio.sleep(1) + receiver._enabled = True + after = await recv_video(track) + + assert before.pts is not None + assert after.pts is not None + assert after.pts - before.pts > 0.5 * 90000 + finally: + await close(caller, callee) + + +@pytest.mark.asyncio +async def test_disabled_receiver_keeps_no_audio() -> None: + caller, callee, track = await received_alone(AudioStreamTrack()) + try: + receiver = only_receiver(callee) + _ = await recv_audio(track) + + receiver._enabled = False + await asyncio.sleep(1) + receiver._enabled = True + _ = await recv_audio(track) + + # a second of audio is 50 frames + assert track._queue.qsize() < 10 + finally: + await close(caller, callee) + + +@pytest.mark.asyncio +async def test_disabling_the_receiver_keeps_a_waiting_read() -> None: + """A pending read gets the next frame, not the end of the track.""" + caller, callee, track = await received_alone(VideoStreamTrack()) + try: + receiver = only_receiver(callee) + reading = asyncio.ensure_future(track.recv()) + await asyncio.sleep(0) + + receiver._enabled = False + frame = await asyncio.wait_for(reading, TIMEOUT) + + assert isinstance(frame, av.VideoFrame) + assert track.readyState == 'live' + finally: + await close(caller, callee) + + +@pytest.mark.asyncio +async def test_queue_drains_the_frames_nobody_read() -> None: + caller, callee, track = await received_alone(VideoStreamTrack()) + try: + first = await recv_video(track) + await asyncio.sleep(0.5) + queue = track._queue + queued = queue.qsize() + + while not queue.empty(): + _ = queue.get_nowait() + queue.task_done() + latest = await recv_video(track) + + assert isinstance(queue, asyncio.Queue) + assert queued >= 5 + assert first.pts is not None + assert latest.pts is not None + assert latest.pts - first.pts > 0.3 * 90000 + finally: + await close(caller, callee) + + async def relayed_call(relay: MediaRelay, source: MediaStreamTrack, pcs: list[RTCPeerConnection]) -> RTCRtpSender: """Sends a proxy of the source on a new connection, adding the connections to pcs.""" caller, callee = new_pair()