From 93637ceeec2f615eae1b417d07e3ebf932bb4196 Mon Sep 17 00:00:00 2001 From: "Ilya (Marshal)" Date: Fri, 16 Oct 2026 18:16:14 +0200 Subject: [PATCH] Fix generator frame drops, adapt relayed video and keep local sinks unadapted --- .../src/interfaces/rtc_video_track_source.cpp | 44 ++-- .../src/interfaces/rtc_video_track_source.h | 3 +- .../cpp/src/media/frame_adaptation.h | 43 ++++ .../cpp/src/media/local_video_sink.h | 60 ++++++ .../media/media_stream_track_processor.cpp | 2 + .../src/media/media_stream_track_processor.h | 3 +- .../cpp/src/media/relayed_video_track.cpp | 15 +- .../cpp/src/media/relayed_video_track.h | 6 +- .../cpp/src/media/track_generator.cpp | 15 +- python-webrtc/cpp/src/media/track_generator.h | 5 + python-webrtc/cpp/src/media/track_monitor.h | 4 +- tests/helpers.py | 2 + tests/test_media_e2e.py | 193 ++++++++++++++++-- 13 files changed, 342 insertions(+), 53 deletions(-) create mode 100644 python-webrtc/cpp/src/media/frame_adaptation.h create mode 100644 python-webrtc/cpp/src/media/local_video_sink.h diff --git a/python-webrtc/cpp/src/interfaces/rtc_video_track_source.cpp b/python-webrtc/cpp/src/interfaces/rtc_video_track_source.cpp index 3912623..560fb05 100644 --- a/python-webrtc/cpp/src/interfaces/rtc_video_track_source.cpp +++ b/python-webrtc/cpp/src/interfaces/rtc_video_track_source.cpp @@ -14,6 +14,9 @@ #include #include +#include "../media/frame_adaptation.h" +#include "../media/local_video_sink.h" + namespace python_webrtc { RTCVideoTrackSource::RTCVideoTrackSource(bool isScreencast, std::optional needsDenoising) @@ -100,29 +103,16 @@ namespace python_webrtc { } void RTCVideoTrackSource::Broadcast(const webrtc::VideoFrame &frame) { + _unadapted.OnFrame(frame); if (!_broadcaster.frame_wanted()) { return; } - int croppedWidth = 0; - int croppedHeight = 0; - int width = 0; - int height = 0; - // timed by arrival, as generator frames carry the application's timestamps - if (!_adapter.AdaptFrameResolution(frame.width(), frame.height(), webrtc::TimeNanos(), &croppedWidth, - &croppedHeight, &width, &height)) { + auto adapted = AdaptFrame(_adapter, frame); + if (adapted) { + _broadcaster.OnFrame(*adapted); + } else { _broadcaster.OnDiscardedFrame(); - return; } - if (width == frame.width() && height == frame.height()) { - _broadcaster.OnFrame(frame); - return; - } - webrtc::VideoFrame adapted = frame; - adapted.set_video_frame_buffer(frame.video_frame_buffer()->CropAndScale( - (frame.width() - croppedWidth) / 2, (frame.height() - croppedHeight) / 2, croppedWidth, croppedHeight, width, - height)); - adapted.clear_update_rect(); - _broadcaster.OnFrame(adapted); } void RTCVideoTrackSource::End() { @@ -142,13 +132,19 @@ namespace python_webrtc { void RTCVideoTrackSource::AddOrUpdateSink(webrtc::VideoSinkInterface *sink, const webrtc::VideoSinkWants &wants) { + auto *broadcaster = &_broadcaster; + if (wants.black_frames) { + broadcaster = &_blackBroadcaster; + } else if (LocalVideoSink::Is(sink)) { + broadcaster = &_unadapted; + } const std::scoped_lock lock(_sinksMutex); - auto [it, added] = _sinks.try_emplace(sink, wants.black_frames); - if (!added && it->second != wants.black_frames) { - (it->second ? _blackBroadcaster : _broadcaster).RemoveSink(sink); - it->second = wants.black_frames; + auto [it, added] = _sinks.try_emplace(sink, broadcaster); + if (!added && it->second != broadcaster) { + it->second->RemoveSink(sink); + it->second = broadcaster; } - (wants.black_frames ? _blackBroadcaster : _broadcaster).AddOrUpdateSink(sink, wants); + broadcaster->AddOrUpdateSink(sink, wants); _adapter.OnSinkWants(_broadcaster.wants()); } @@ -158,7 +154,7 @@ namespace python_webrtc { if (it == _sinks.end()) { return; } - (it->second ? _blackBroadcaster : _broadcaster).RemoveSink(sink); + it->second->RemoveSink(sink); _sinks.erase(it); _adapter.OnSinkWants(_broadcaster.wants()); } diff --git a/python-webrtc/cpp/src/interfaces/rtc_video_track_source.h b/python-webrtc/cpp/src/interfaces/rtc_video_track_source.h index 37f4eb6..09d74bc 100644 --- a/python-webrtc/cpp/src/interfaces/rtc_video_track_source.h +++ b/python-webrtc/cpp/src/interfaces/rtc_video_track_source.h @@ -84,10 +84,11 @@ namespace python_webrtc { const std::optional _needsDenoising; webrtc::VideoBroadcaster _broadcaster; webrtc::VideoAdapter _adapter; + webrtc::VideoBroadcaster _unadapted; // sinks of disabled or ended tracks: a black frame a second webrtc::VideoBroadcaster _blackBroadcaster; std::mutex _sinksMutex; - std::map *, bool> _sinks; + std::map *, webrtc::VideoBroadcaster *> _sinks; std::atomic _lastBlackMs{0}; std::atomic _width{0}; std::atomic _height{0}; diff --git a/python-webrtc/cpp/src/media/frame_adaptation.h b/python-webrtc/cpp/src/media/frame_adaptation.h new file mode 100644 index 0000000..be15fb8 --- /dev/null +++ b/python-webrtc/cpp/src/media/frame_adaptation.h @@ -0,0 +1,43 @@ +// +// 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. +// + +#ifndef PYTHON_WEBRTC_MEDIA_FRAME_ADAPTATION_H_ +#define PYTHON_WEBRTC_MEDIA_FRAME_ADAPTATION_H_ + +#include + +#include +#include +#include + +namespace python_webrtc { + + // The frame as the adapter's sinks want it, timed by its capture time: cropped and scaled, or none when dropped + inline std::optional AdaptFrame(webrtc::VideoAdapter &adapter, const webrtc::VideoFrame &frame) { + int croppedWidth = 0; + int croppedHeight = 0; + int width = 0; + int height = 0; + if (!adapter.AdaptFrameResolution(frame.width(), frame.height(), + frame.timestamp_us() * webrtc::kNumNanosecsPerMicrosec, &croppedWidth, + &croppedHeight, &width, &height)) { + return std::nullopt; + } + if (width == frame.width() && height == frame.height()) { + return frame; + } + webrtc::VideoFrame adapted = frame; + adapted.set_video_frame_buffer(frame.video_frame_buffer()->CropAndScale( + (frame.width() - croppedWidth) / 2, (frame.height() - croppedHeight) / 2, croppedWidth, croppedHeight, width, + height)); + adapted.clear_update_rect(); + return adapted; + } + +} // namespace python_webrtc + +#endif // PYTHON_WEBRTC_MEDIA_FRAME_ADAPTATION_H_ diff --git a/python-webrtc/cpp/src/media/local_video_sink.h b/python-webrtc/cpp/src/media/local_video_sink.h new file mode 100644 index 0000000..11d9a9f --- /dev/null +++ b/python-webrtc/cpp/src/media/local_video_sink.h @@ -0,0 +1,60 @@ +// +// 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. +// + +#ifndef PYTHON_WEBRTC_MEDIA_LOCAL_VIDEO_SINK_H_ +#define PYTHON_WEBRTC_MEDIA_LOCAL_VIDEO_SINK_H_ + +#include +#include + +#include +#include + +#include "../utils/libwebrtc_thread.h" + +namespace python_webrtc { + + // A sink of ours on a track (a processor, a monitor): it sees the frames of the track's source, never the ones a + // source adapts for an encoder. Registered while it exists, as sinks reach a source through libwebrtc's tracks. + class LocalVideoSink : public webrtc::VideoSinkInterface { + public: + LocalVideoSink() { + auto &sinks = Sinks(); + const std::scoped_lock lock(sinks.mutex); + sinks.sinks.insert(this); + } + + ~LocalVideoSink() override { + auto &sinks = Sinks(); + const std::scoped_lock lock(sinks.mutex); + sinks.sinks.erase(this); + } + + LocalVideoSink(const LocalVideoSink &) = delete; + LocalVideoSink &operator=(const LocalVideoSink &) = delete; + + static bool Is(const webrtc::VideoSinkInterface *sink) { + auto &sinks = Sinks(); + const std::scoped_lock lock(sinks.mutex); + return sinks.sinks.contains(sink); + } + + private: + struct Registry { + std::mutex mutex; + std::set *> sinks; + }; + + static Registry &Sinks() { + static ForkLocal registry; + return registry.Get(); + } + }; + +} // namespace python_webrtc + +#endif // PYTHON_WEBRTC_MEDIA_LOCAL_VIDEO_SINK_H_ diff --git a/python-webrtc/cpp/src/media/media_stream_track_processor.cpp b/python-webrtc/cpp/src/media/media_stream_track_processor.cpp index 612cf06..ce4f271 100644 --- a/python-webrtc/cpp/src/media/media_stream_track_processor.cpp +++ b/python-webrtc/cpp/src/media/media_stream_track_processor.cpp @@ -118,6 +118,8 @@ namespace python_webrtc { constexpr int64_t kRtpTicksPerMs = 90; timestampUs = _firstReceived->first + ((rtp - _firstReceived->second) * webrtc::kNumMicrosecsPerMillisec / kRtpTicksPerMs); + } else if (frame.presentation_timestamp()) { + timestampUs = frame.presentation_timestamp()->us(); } Push(VideoItem{.buffer = frame.video_frame_buffer(), .timestampUs = timestampUs, diff --git a/python-webrtc/cpp/src/media/media_stream_track_processor.h b/python-webrtc/cpp/src/media/media_stream_track_processor.h index e0a5b16..e79464e 100644 --- a/python-webrtc/cpp/src/media/media_stream_track_processor.h +++ b/python-webrtc/cpp/src/media/media_stream_track_processor.h @@ -27,6 +27,7 @@ #include "../interfaces/media_stream_track.h" #include "../utils/mailbox.h" #include "../utils/native_object.h" +#include "local_video_sink.h" #include "video_frame_buffer.h" namespace python_webrtc { @@ -35,7 +36,7 @@ namespace python_webrtc { class MediaStreamTrackProcessor : public NativeObject, public Emitter, public TrackEndObserver, - public webrtc::VideoSinkInterface, + public LocalVideoSink, public webrtc::AudioTrackSinkInterface { public: static constexpr const char *kName = "MediaStreamTrackProcessor"; diff --git a/python-webrtc/cpp/src/media/relayed_video_track.cpp b/python-webrtc/cpp/src/media/relayed_video_track.cpp index f6958b0..19ebab0 100644 --- a/python-webrtc/cpp/src/media/relayed_video_track.cpp +++ b/python-webrtc/cpp/src/media/relayed_video_track.cpp @@ -11,6 +11,8 @@ #include +#include "frame_adaptation.h" + namespace python_webrtc { RelayedVideoTrack::RelayedVideoTrack(webrtc::scoped_refptr track) @@ -27,7 +29,11 @@ namespace python_webrtc { if (!restamper) { restamper = std::make_unique(sink); } - _track->AddOrUpdateSink(restamper.get(), wants); + restamper->OnSinkWants(wants); + webrtc::VideoSinkWants delivered; + delivered.rotation_applied = wants.rotation_applied; + delivered.black_frames = wants.black_frames; + _track->AddOrUpdateSink(restamper.get(), delivered); } void RelayedVideoTrack::RemoveSink(webrtc::VideoSinkInterface *sink) { @@ -42,7 +48,12 @@ namespace python_webrtc { webrtc::VideoFrame restamped = frame; restamped.set_timestamp_us(_aligner.TranslateTimestamp(frame.timestamp_us(), webrtc::TimeMicros())); restamped.set_ntp_time_ms(0); - _sink->OnFrame(restamped); + auto adapted = AdaptFrame(_adapter, restamped); + if (adapted) { + _sink->OnFrame(*adapted); + } else { + _sink->OnDiscardedFrame(); + } } } // namespace python_webrtc diff --git a/python-webrtc/cpp/src/media/relayed_video_track.h b/python-webrtc/cpp/src/media/relayed_video_track.h index 96b77ae..7bda011 100644 --- a/python-webrtc/cpp/src/media/relayed_video_track.h +++ b/python-webrtc/cpp/src/media/relayed_video_track.h @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -22,7 +23,7 @@ namespace python_webrtc { // A track with a remote source as its sender sees it: frames restamped on the local clock without their remote NTP - // time, since encoders drop frames whose capture time goes back + // time, since encoders drop frames whose capture time goes back, and adapted to what the encoder wants class RelayedVideoTrack : public webrtc::VideoTrackInterface { public: explicit RelayedVideoTrack(webrtc::scoped_refptr track); @@ -66,6 +67,8 @@ namespace python_webrtc { public: explicit Restamper(webrtc::VideoSinkInterface *sink) : _sink(sink) {} + void OnSinkWants(const webrtc::VideoSinkWants &wants) { _adapter.OnSinkWants(wants); } + void OnFrame(const webrtc::VideoFrame &frame) override; void OnDiscardedFrame() override { _sink->OnDiscardedFrame(); } @@ -77,6 +80,7 @@ namespace python_webrtc { private: webrtc::VideoSinkInterface *const _sink; webrtc::TimestampAligner _aligner; + webrtc::VideoAdapter _adapter; }; const webrtc::scoped_refptr _track; diff --git a/python-webrtc/cpp/src/media/track_generator.cpp b/python-webrtc/cpp/src/media/track_generator.cpp index d690389..7c7dbe8 100644 --- a/python-webrtc/cpp/src/media/track_generator.cpp +++ b/python-webrtc/cpp/src/media/track_generator.cpp @@ -15,10 +15,10 @@ #include #include +#include #include #include #include -#include #include "../exceptions.h" #include "../utils/gil.h" @@ -93,11 +93,18 @@ namespace python_webrtc { if (!GetLive() || _muted) { return; } - // sinks see the timestamp of the application, encoders take the capture time from the NTP one + // sinks see the timestamp of the application, encoders a capture time on the system clock, spaced as it is + const std::scoped_lock lock(_videoMutex); + int64_t captureUs = _aligner.TranslateTimestamp(timestampUs, webrtc::TimeMicros()); + // encoders drop a frame whose capture millisecond isn't after the previous one's + if (captureUs / webrtc::kNumMicrosecsPerMillisec <= _lastCaptureMs) { + captureUs = (_lastCaptureMs + 1) * webrtc::kNumMicrosecsPerMillisec; + } + _lastCaptureMs = captureUs / webrtc::kNumMicrosecsPerMillisec; auto frame = webrtc::VideoFrame::Builder() .set_video_frame_buffer(buffer->ToWebrtc()) - .set_timestamp_us(timestampUs) - .set_ntp_time_ms(webrtc::Clock::GetRealTimeClock()->CurrentNtpInMilliseconds()) + .set_timestamp_us(captureUs) + .set_presentation_timestamp(webrtc::Timestamp::Micros(timestampUs)) .set_rotation(static_cast(rotation)) .build(); _videoSource->PushFrame(frame); diff --git a/python-webrtc/cpp/src/media/track_generator.h b/python-webrtc/cpp/src/media/track_generator.h index 43f2ee4..ad82d98 100644 --- a/python-webrtc/cpp/src/media/track_generator.h +++ b/python-webrtc/cpp/src/media/track_generator.h @@ -16,6 +16,7 @@ #include #include +#include #include @@ -65,6 +66,10 @@ namespace python_webrtc { std::shared_ptr _control = std::make_shared(); std::atomic _muted = false; + std::mutex _videoMutex; + webrtc::TimestampAligner _aligner; + int64_t _lastCaptureMs = 0; + // samples short of a 10 ms frame, sent with the next ones of the same format std::mutex _audioMutex; std::vector _pending; diff --git a/python-webrtc/cpp/src/media/track_monitor.h b/python-webrtc/cpp/src/media/track_monitor.h index 897e980..0db9b3d 100644 --- a/python-webrtc/cpp/src/media/track_monitor.h +++ b/python-webrtc/cpp/src/media/track_monitor.h @@ -19,10 +19,12 @@ #include #include +#include "local_video_sink.h" + namespace python_webrtc { // What a track carries, for its settings: the size and rate of its frames, the format of its audio - class TrackMonitor : public webrtc::VideoSinkInterface, public webrtc::AudioTrackSinkInterface { + class TrackMonitor : public LocalVideoSink, public webrtc::AudioTrackSinkInterface { public: struct Video { int width; diff --git a/tests/helpers.py b/tests/helpers.py index 0a62cc5..4a06791 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -356,6 +356,8 @@ async def write_video( timestamp += round(interval * 1_000_000) await asyncio.sleep(interval) + writer.release_lock() + @contextlib.asynccontextmanager async def writing(write: Callable[..., Awaitable[None]], *args: object, **kwargs: object) -> AsyncGenerator[None]: diff --git a/tests/test_media_e2e.py b/tests/test_media_e2e.py index 36cf778..21dfb86 100644 --- a/tests/test_media_e2e.py +++ b/tests/test_media_e2e.py @@ -14,9 +14,10 @@ import contextlib import dataclasses import gc +import itertools import math import weakref -from typing import cast +from typing import TYPE_CHECKING, cast import pytest @@ -31,11 +32,15 @@ stats_of_type, wait_for_event, wait_until, + wait_until_unmuted, write_video, writing, ) from tests.timing import seconds +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + TIMEOUT = seconds(20) #: no frame for this long means frames stopped; exact, as it defines the assertion QUIET = 0.5 @@ -184,20 +189,23 @@ async def test_a_remote_video_track_has_one_sender_a_connection( #: frames read to check their size FRAMES = 10 +#: a size too large for the bitrate the frames are sent at, and the frames of it +LARGE = (640, 480) +LARGE_I420 = bytes(640 * 480 * 3 // 2) -async def received_widths( +async def send_at_low_bitrate( caller: webrtc.RTCPeerConnection, callee: webrtc.RTCPeerConnection, + track: webrtc.MediaStreamTrack, *, preference: webrtc.RTCDegradationPreference, -) -> list[int]: - """The widths of frames received of 640x480 video sent at 100 kbps, too low a bitrate for its size.""" - generator = webrtc.VideoTrackGenerator() +) -> Awaitable[webrtc.Event]: + """Sends the track at 100 kbps, too low a bitrate for 640x480, and returns the callee's coming track event.""" init = webrtc.RTCRtpTransceiverInit( direction='sendonly', send_encodings=[webrtc.RTCRtpEncodingParameters(max_bitrate=100_000)] ) - sender = caller.add_transceiver(generator.track, init).sender + sender = caller.add_transceiver(track, init).sender parameters = sender.get_parameters() parameters.degradation_preference = preference await sender.set_parameters(parameters) @@ -205,19 +213,41 @@ async def received_widths( await connect(caller, callee, TIMEOUT) # libwebrtc drops frames too large for the bitrate only if it knows the bitrate before the first frame await wait_until(lambda: available_outgoing_bitrate(caller), 'a bandwidth estimate', TIMEOUT) - async with writing(write_video, generator, bytes(640 * 480 * 3 // 2), (640, 480)): - event = await received - assert isinstance(event, webrtc.RTCTrackEvent) - reader = webrtc.MediaStreamTrackProcessor( - webrtc.MediaStreamTrackProcessorInit(event.track, max_buffer_size=5) - ).readable.get_reader() - widths: list[int] = [] - while len(widths) < FRAMES: - frame = (await asyncio.wait_for(reader.read(), TIMEOUT)).value - assert isinstance(frame, webrtc.VideoFrame) - widths.append(frame.coded_width) - frame.close() - await reader.cancel() + return received + + +async def read_widths(track: webrtc.MediaStreamTrack, count: int) -> list[int]: + """The widths of the next frames of the track.""" + reader = webrtc.MediaStreamTrackProcessor( + webrtc.MediaStreamTrackProcessorInit(track, max_buffer_size=1) + ).readable.get_reader() + widths: list[int] = [] + while len(widths) < count: + frame = (await asyncio.wait_for(reader.read(), TIMEOUT)).value + assert isinstance(frame, webrtc.VideoFrame) + widths.append(frame.coded_width) + frame.close() + await reader.cancel() + return widths + + +async def received_track(received: Awaitable[webrtc.Event]) -> webrtc.MediaStreamTrack: + event = await received + assert isinstance(event, webrtc.RTCTrackEvent) + return event.track + + +async def received_widths( + caller: webrtc.RTCPeerConnection, + callee: webrtc.RTCPeerConnection, + *, + preference: webrtc.RTCDegradationPreference, +) -> list[int]: + """The widths of frames received of 640x480 video sent at a low bitrate.""" + generator = webrtc.VideoTrackGenerator() + received = await send_at_low_bitrate(caller, callee, generator.track, preference=preference) + async with writing(write_video, generator, LARGE_I420, LARGE): + widths = await read_widths(await received_track(received), FRAMES) generator.track.stop() return widths @@ -239,6 +269,131 @@ async def test_a_generator_keeps_its_size_under_maintain_resolution( assert widths == [640] * FRAMES +@pytest.mark.asyncio +async def test_a_sent_track_keeps_its_size_for_its_own_sinks( + caller: webrtc.RTCPeerConnection, callee: webrtc.RTCPeerConnection +) -> None: + """A processor of a track and its settings see the frames of its source, not the smaller ones the encoder takes.""" + generator = webrtc.VideoTrackGenerator() + received = await send_at_low_bitrate( + caller, callee, generator.track, preference=webrtc.RTCDegradationPreference.maintain_framerate + ) + async with writing(write_video, generator, LARGE_I420, LARGE): + assert all(width < 640 for width in await read_widths(await received_track(received), FRAMES)) + local = await read_widths(generator.track, FRAMES) + settings = generator.track.get_settings() + generator.track.stop() + + assert local == [640] * FRAMES + assert (settings.width, settings.height) == LARGE + + +@pytest.mark.asyncio +async def test_relayed_video_scales_down_for_the_encoder( + caller: webrtc.RTCPeerConnection, callee: webrtc.RTCPeerConnection, create_pc: CreatePC +) -> None: + """A remote track sent on at a low bitrate is sent in the smaller frames its encoder asks for.""" + relay, sink = create_pc(), create_pc() + generator = webrtc.VideoTrackGenerator() + async with writing(write_video, generator, LARGE_I420, LARGE): + remote = await connect_track(caller, callee, generator.track, timeout=TIMEOUT) + await wait_until_unmuted(remote, TIMEOUT) + # no frame reaches the relay's encoder before it knows its bitrate + received = await send_at_low_bitrate( + relay, sink, remote, preference=webrtc.RTCDegradationPreference.maintain_framerate + ) + async with writing(write_video, generator, LARGE_I420, LARGE): + widths = await read_widths(await received_track(received), FRAMES) + generator.track.stop() + + assert all(width < 640 for width in widths) + + +#: a frame interval of the producer, in microseconds, the unit of frame timestamps +FRAME_US = 33_333 +#: frames a producer writes at its pace first, a second of them, for the generator to learn its clock +PACED = 30 +#: frames a producer that fell behind writes at once +BURST = 4 + + +async def source_frames(pc: webrtc.RTCPeerConnection) -> int: + """The frames the encoders of the connection took from their sources.""" + report = await pc.get_stats() + sources = stats_of_type(report, 'media-source') + return sum(s.frames for s in sources if isinstance(s, webrtc.RTCVideoSourceStats) and s.frames is not None) + + +def frame_writer(generator: webrtc.VideoTrackGenerator) -> Callable[[], Awaitable[None]]: + """Writes the next frame of a producer to the generator, its timestamp a frame interval after the previous one.""" + writer = generator.writable.get_writer() + data = solid_i420(128, 128, 128) + timestamps = itertools.count(0, FRAME_US) + + async def write() -> None: + init = webrtc.VideoFrameBufferInit( + format='I420', coded_width=WIDTH, coded_height=HEIGHT, timestamp=next(timestamps) + ) + await writer.write(webrtc.VideoFrame(data, init)) + + return write + + +async def frames_taken(caller: webrtc.RTCPeerConnection, before: int, count: int) -> int: + """The frames the encoder took since it had taken ``before``, once it took ``count`` or a while passed.""" + + async def all_taken() -> bool: + return await source_frames(caller) >= before + count + + with contextlib.suppress(TimeoutError): + await wait_until(all_taken, 'the frames at the encoder', seconds(5)) + return await source_frames(caller) - before + + +@pytest.mark.asyncio +async def test_generator_frames_written_at_once_reach_the_encoder( + caller: webrtc.RTCPeerConnection, callee: webrtc.RTCPeerConnection +) -> None: + """Two frames written back to back, a frame interval apart in their timestamps, are both encoded.""" + generator = webrtc.VideoTrackGenerator() + write = frame_writer(generator) + await connect_track(caller, callee, generator.track, timeout=TIMEOUT) + await write() + await write() + taken = await frames_taken(caller, 0, 2) + generator.track.stop() + + assert taken == 2 + + +@pytest.mark.asyncio +async def test_late_generator_frames_written_at_once_reach_the_encoder( + caller: webrtc.RTCPeerConnection, callee: webrtc.RTCPeerConnection +) -> None: + """A producer that stalls, then writes the frames it owes at once, has none of them dropped as duplicates.""" + generator = webrtc.VideoTrackGenerator() + write = frame_writer(generator) + loop = asyncio.get_running_loop() + await connect_track(caller, callee, generator.track, timeout=TIMEOUT) + start = loop.time() + written = 0 + deadline = start + TIMEOUT + while written < PACED or await source_frames(caller) == 0: + assert loop.time() < deadline, 'no frame reached the encoder' + await write() + written += 1 + await asyncio.sleep(max(0.0, start + written * FRAME_US / 1e6 - loop.time())) + # it falls behind by the burst, then catches up + await asyncio.sleep(max(0.0, start + (written + BURST) * FRAME_US / 1e6 - loop.time())) + before = await source_frames(caller) + for _ in range(BURST): + await write() + taken = await frames_taken(caller, before, BURST) + generator.track.stop() + + assert taken == BURST + + #: the rids of three layers of 720p, lowest first, and their sizes LAYERS = {'q': (320, 180), 'h': (640, 360), 'f': (1280, 720)}