Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This Source Code Form is subject to the terms of the Mozilla Public
2# License, v. 2.0. If a copy of the MPL was not distributed with this
3# file, You can obtain one at https://mozilla.org/MPL/2.0/.
4#
5# This file incorporates work covered by the following copyright and
6# permission notice:
7#
8# Copyright (c) Jeremy Lainé.
9# All rights reserved.
10#
11# Redistribution and use in source and binary forms, with or without
12# modification, are permitted provided that the following conditions are met:
13#
14# * Redistributions of source code must retain the above copyright notice,
15# this list of conditions and the following disclaimer.
16# * Redistributions in binary form must reproduce the above copyright notice,
17# this list of conditions and the following disclaimer in the documentation
18# and/or other materials provided with the distribution.
19# * Neither the name of aiortc nor the names of its contributors may
20# be used to endorse or promote products derived from this software without
21# specific prior written permission.
22#
23# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
24# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
25# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
26# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
27# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
28# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
29# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
32# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
34import asyncio
35import logging
36import random
37import time
38import traceback
39import uuid
40from collections import deque
41from collections.abc import Callable
42from typing import Any, Optional, Union
45from . import clock, rtp
46from .pacer import CLASS_AUDIO, CLASS_VIDEO
47from .codecs import get_capabilities, get_encoder, is_rtx
48from .codecs.base import Encoder
49from .codecs.h264 import H264Encoder
50from .codecs.h264_skip import HEAD_BYTES, SkipPictures
51from .exceptions import InvalidStateError
52from .mediastreams import MediaStreamError, MediaStreamTrack
53from .rtcdtlstransport import RTCDtlsTransport
54from .rtcrtpparameters import (
55 RTCRtpCapabilities,
56 RTCRtpCodecParameters,
57 RTCRtpSendParameters,
58)
59from .rtp import (
60 RTCP_PSFB_APP,
61 RTCP_PSFB_FIR,
62 RTCP_PSFB_PLI,
63 RTCP_RTPFB_NACK,
64 RTCP_RTPFB_TWCC,
65 RtpHistory,
66 AnyRtcpPacket,
67 RtcpByePacket,
68 RtcpPsfbPacket,
69 RtcpRrPacket,
70 RtcpRtpfbPacket,
71 RtcpSdesPacket,
72 RtcpSenderInfo,
73 RtcpSourceInfo,
74 RtcpSrPacket,
75 RtcpXrPacket,
76 RtpPacket,
77 dependency_descriptor,
78 unpack_remb_fci,
79 wrap_rtx,
80 build_flexfec_03,
81)
82from .stats import (
83 RTCOutboundRtpStreamStats,
84 RTCRemoteInboundRtpStreamStats,
85 RTCStatsReport,
86)
87from .utils import random16, random32, uint16_add, uint32_add
88from pyee.asyncio import AsyncIOEventEmitter
90logger = logging.getLogger(__name__)
92RTT_ALPHA = 0.85
93# Receive-only reporters answered per sender report (libwebrtc answers at most 50 per XR).
94RRTR_REPORTERS_MAX = 50
95# How long a peer may send nothing at all before this sender stops: a browser sends
96# feedback for whatever it receives and checks consent on its path every few seconds
97# (2.5 s in libwebrtc, about 5 s in Firefox's stack) even while a still screen sends it
98# nothing, so a peer silent this long is gone (asleep, off the network), where ICE
99# consent would keep the stream going to it for half a minute.
100PEER_SILENCE_S = 10.0
101# A libwebrtc receiver with no frame to decode for three times the offer's rtx-time
102# (375 ms), and a packet in the last five seconds, asks for a key frame, and again
103# each 375 ms: a still screen sends nothing, so each still was answered with a key
104# frame and the cleanup after it, which ends in another still. A request this long
105# past the round trip after an acknowledged frame, with nothing lost since, is that
106# wait; one this soon after the last is a failed decode, which asks each rtx-time.
107STILL_REQUEST_S = 0.25
109# Where the encoder names each frame's reference and the peer reads the dependency
110# descriptor, a peer that does not own its display is served as a selective forwarding
111# unit serves a receiver on a narrow link (`_forward`): a delta frame finding
112# RESYNC_ROOM_FRAMES of the peer's frames in its pacer, or RESYNC_LAG_S of wait there, is
113# left out with the frames predicting from it, before any is numbered. The encoder is told
114# the run's first frame is lost once the peer has room, or sooner where waiting would cost
115# a key frame: at RESYNC_REACH_FRAMES, while it still holds the frame to predict from (eight
116# back), and in H.264 RESYNC_WRAP_LEAD frames ahead of a frame_num wrap (FRAME_NUM_WRAP
117# frames past a key frame, a frame never left out), unless the peer is far behind:
118# RESYNC_FAR_FRAMES or RESYNC_FAR_S, or a path that lost packets within RESYNC_LOSS_S while a
119# queue stood on it, which says it drops what overflows its buffer rather than queueing it.
120# A run the pacer cut (`_pacer_resync`) for such a peer is answered the same way, and one the
121# encoder has not predicted past in RESYNC_S costs a key frame, as the WebSockets page's decode
122# gate waits (LOST_RECOVERY_MS in lib/decode-gate.js).
123# Where the encoder is told which frames every peer was sent (`on_frame_out`, its last packet
124# on the wire) and holds (`on_frame_held`: transport-cc reported each of its packets received,
125# or the retransmission of one, within HELD_S, and the frame it predicts from is held), it pins
126# the newest anchor every peer holds, so a run of any depth is predicted past on its report,
127# and flags anchors predicting from a frame every peer was sent (`RTCEncodedFrame.anchor`):
128# such a frame goes to a peer without room as a key frame does, unless the peer is far behind,
129# and ends its run. A frame is numbered for FRAME_NUMBERS_MEMORY frames, as far back as a
130# pinned anchor may be. A peer that lost frames past repair (a NACK for packets gone from the
131# history, or its own PLI) is resynced from the newest frame it holds as a cut run is, where
132# anchors run (`_resync_held`); a stall of such requests past RESYNC_ANCHORED_S, from a decoder
133# that only takes a key frame, is sent one. A request from a peer whose link is dark, packets
134# going to it and none reported received for RESYNC_DARK_S, is answered by the one it makes once
135# it is back.
136RESYNC_ROOM_FRAMES = 3
137RESYNC_LAG_S = 0.1
138RESYNC_REACH_FRAMES = 5
139RESYNC_WRAP_LEAD = 2
140RESYNC_FAR_FRAMES = 12
141RESYNC_FAR_S = 1.0
142RESYNC_LOSS_S = 0.5
143RESYNC_S = 1.0
144# Where the encoder pins an anchor every peer holds (an anchor came since the codec's first
145# frame, its key frame being the first anchor), it predicts past a run of any depth on the
146# report, so a run waits RESYNC_ANCHORED_S for room.
147RESYNC_ANCHORED_S = 4.0
148RESYNC_DARK_S = 0.25
149# An anchor goes to a peer without room only while the queue standing in front of it is under
150# RESYNC_ANCHOR_S: on a link too narrow for the anchors alone they would never let it drain.
151RESYNC_ANCHOR_S = 0.25
152HELD_S = 8.0
153FRAME_NUMBERS_MEMORY = 4096
154# The shortest H.264 frame_num range, which every longer one is a multiple of.
155FRAME_NUM_WRAP = 16
157# Media packets per FlexFEC group at most; a group also closes with its frame.
158FEC_GROUP_PACKETS = 10
159# Smoothed loss past which every FlexFEC group takes its repairs, read on windows no queue
160# stood in for FEC_CLEAR_WINDOWS before them, so a queue's own overflow never counts.
161FEC_FULL_LOSS = 0.002
162FEC_CLEAR_WINDOWS = 3
165def random_sequence_number() -> int:
166 """
167 Generate a random RTP sequence number.
169 The sequence number is chosen in the lower half of the allowed range in
170 order to avoid wraparounds which break SRTP decryption.
172 See:
173 https://chromiumdash.appspot.com/commit/13b327b05fa3788b4daa9c3463e13282824cb320
174 """
175 return random16() % 32768
178#: The colour signal a stream was converted with, as the ITU-T H.273 codes the RTP
179#: colour-space header extension carries, for the two codecs whose bitstream cannot state it:
180#: BT.709 primaries, transfer, and matrix at limited range, with the BT.601 matrix for VP8,
181#: which is held to the only one a keyframe header's single colour-space bit can name. A
182#: receiver that reads the extension takes it over the bitstream, so H.264, H.265, and AV1 are
183#: left to their own headers rather than told here — theirs carry the range as well, which a
184#: 4:4:4 session signals as full and this table has no way to know.
185RTP_COLOR_SPACE = {
186 "video/vp8": (1, 1, 6, 1),
187 "video/vp9": (1, 1, 1, 1),
188}
191class RTCEncodedFrame:
192 def __init__(self, payloads: list[bytes], timestamp: int, audio_level: int,
193 keyframe: bool = False, timing: Optional[tuple] = None,
194 dependency: Optional[tuple] = None, anchor: bool = False, data: Any = None):
195 self.payloads = payloads
196 self.timestamp = timestamp
197 self.audio_level = audio_level
198 self.keyframe = keyframe
199 self.timing = timing
200 self.dependency = dependency
201 self.anchor = anchor
202 self.data = data
205def video_timing_legs(timing: Optional[tuple], now_ns: int, arrival_delta_ms: int) -> tuple:
206 """The video-timing extension of a frame packetized at `now_ns`
207 (CLOCK_MONOTONIC): the flags byte, timer-triggered, then six millisecond
208 legs from the capture. With `timing` as the capture library stamped it
209 (capture, encode start, encode end, in the same clock) the encode legs
210 are real and the packetization leg is the frame's whole age; without it,
211 the encode legs are unknown (0) and the packetization leg is the frame's
212 time in the sender, `arrival_delta_ms`. Pacer exit repeats packetization,
213 since the stamp is taken ahead of the pacer, and the two network legs
214 stay 0, as a sender that is not a middlebox leaves them.
215 """
216 if not timing or timing[0] <= 0:
217 return (0x01, 0, 0, arrival_delta_ms, arrival_delta_ms, 0, 0)
218 capture_ns, encode_start_ns, encode_end_ns = timing
219 leg = lambda instant_ns: min(0xFFFF, max(0, (instant_ns - capture_ns) // 1_000_000))
220 packetized = leg(now_ns)
221 return (0x01, leg(encode_start_ns), leg(encode_end_ns), packetized, packetized, 0, 0)
224class RTCRtpSender(AsyncIOEventEmitter):
225 """
226 The :class:`RTCRtpSender` interface provides the ability to control and
227 obtain details about how a particular :class:`MediaStreamTrack` is encoded
228 and sent to a remote peer.
230 :param trackOrKind: Either a :class:`MediaStreamTrack` instance or a
231 media kind (`'audio'` or `'video'`).
232 :param transport: An :class:`RTCDtlsTransport`.
233 """
235 def __init__(
236 self, trackOrKind: Union[MediaStreamTrack, str], transport: RTCDtlsTransport
237 ) -> None:
238 super().__init__()
239 if transport.state == "closed":
240 raise InvalidStateError
242 if isinstance(trackOrKind, MediaStreamTrack):
243 self.__kind = trackOrKind.kind
244 self.replaceTrack(trackOrKind)
245 else:
246 self.__kind = trackOrKind
247 self.replaceTrack(None)
248 self.__cname: Optional[str] = None
249 self._ssrc = random32()
250 self._rtx_ssrc = random32()
251 self._fec_ssrc = random32()
252 # Fresh UUID per sender: the msid stream identifier when the caller
253 # supplies no MediaStream grouping.
254 self._stream_id = str(uuid.uuid4())
255 self._enabled = True
256 self._peer_silent = False
257 # Told each frame once its last packet is on the wire, as
258 # `(capture_ns, payload bytes)`, where the owner wants it (stream stats).
259 self.on_frame_sent: Optional[Callable[[int, int], None]] = None
260 # Told each frame handed to the pacer with that wait, so the owner can count
261 # the frames the pacer drops instead, which call nothing.
262 self.on_frame_queued: Optional[Callable[[], None]] = None
263 self.__encoder: Optional[Encoder] = None
264 # The negotiated codecs and the one frames go out as; None drops them.
265 self.__codecs: list[RTCRtpCodecParameters] = []
266 self.__send_codec: Optional[RTCRtpCodecParameters] = None
267 # The codec a switch before the start named, taken out of the answer at the start.
268 self.__pending_codec: Optional[str] = None
269 self.__force_keyframe = False
270 self.__force_keyframe_used = False
271 # Last observed keyframe size (bytes) and whether it was a natural
272 # IDR: lets a late-attaching pacer bootstrap its IDR floor from the
273 # session-start keyframe instead of waiting for the next one.
274 self._keyframe_bytes: Optional[int] = None
275 self._keyframe_natural: bool = True
276 # Whether this peer is served selectively, as one not owning its display
277 # (`_forward`); told a run's first frame is lost and whether the run is within the
278 # encoder's reach, answering whether the encoder took it (None emits "lost_frame");
279 # and told each frame left out, which is neither queued nor sent.
280 self.selective: Optional[Callable[[], bool]] = None
281 self.on_resync: Optional[Callable[[int, bool], bool]] = None
282 self.on_frame_left_out: Optional[Callable[[], None]] = None
283 # Described frames handed to the pacer whose last packet has not left, as (tag of
284 # that packet, capture frame id), oldest first: a GOP reset cuts these
285 # (`_pacer_resync`). And the run of frames this peer is not sent: when it began
286 # (None for none), its first frame, how many it holds, whether the encoder took the
287 # report, the delta frames since the last key frame, and a count of pacer cuts, by
288 # which a frame being sent learns a cut took it.
289 self.__in_flight: deque[tuple[int, int]] = deque(maxlen=64)
290 # The skipped pictures a peer without the descriptor is sent for the H.264 frames it
291 # is not (`_stand_in`), and how many.
292 self.__skips: Optional[SkipPictures] = None
293 self.stand_ins = 0
294 self._resync_since: Optional[float] = None
295 self._resync_first: Optional[int] = None
296 self._resync_run = 0
297 self._resync_told = False
298 self._since_key = 0
299 self._resyncs = 0
300 # Told each described frame on the wire to the peer and each it holds, as `(frame id,
301 # key frame)`, and each key frame sent it; the described frames sent and not yet held,
302 # oldest first, as (frame id, frame predicted from, key frame, when sent,
303 # [(transport-wide, media sequence number)]); the frames held, oldest first, as far
304 # back as frames are numbered, since one may predict from a pinned anchor held long
305 # ago; whether a frame sent since the key frame was never held (`_confirm`); and the
306 # transport-wide sequence number each retransmitted packet last went out under.
307 self.on_frame_out: Optional[Callable[[int, bool], None]] = None
308 self.on_frame_held: Optional[Callable[[int, bool], None]] = None
309 self.on_key_sent: Optional[Callable[[], None]] = None
310 self._anchored = False
311 self.__unheld: deque[tuple] = deque(maxlen=1024)
312 self._stall_since: Optional[float] = None
313 self._stall_last = 0.0
314 self._held_at = 0.0
315 self._key_sent_at: Optional[float] = None
316 self.__held: dict[int, None] = {}
317 self._unheld_gap = False
318 self.__rtx_twcc: dict[int, int] = {}
319 self.__loop = asyncio.get_running_loop()
320 self.__mid: Optional[str] = None
321 self.__rtp_exited = asyncio.Event()
322 self.__rtp_header_extensions_map = rtp.HeaderExtensionsMap()
323 self.__rtp_started = asyncio.Event()
324 self.__rtp_task: Optional[asyncio.Future[None]] = None
325 self.__rtp_history = RtpHistory()
326 # The newest media sequence number sent when the pacer last abandoned a
327 # GOP: a NACK for it or anything before names a packet the keyframe replaces.
328 self.__last_sequence: Optional[int] = None
329 # Frames numbered on the wire for the dependency descriptor, and the number each
330 # capture frame id took, which a frame predicting from it is measured against.
331 self.__frame_number = 0
332 self.__frame_numbers: dict[int, int] = {}
333 # The transport-wide sequence number ending the newest frame handed over, and
334 # of the newest on the wire with when it left; when the peer last asked for a
335 # key frame; and when it lost a frame past repair, until a frame coded after
336 # that is on the wire (`_still_request`).
337 self._frame_handed: Optional[int] = None
338 self._frame_left: Optional[tuple[int, float]] = None
339 self._key_asked_at = 0.0
340 self._lost_at: Optional[float] = None
341 self.__rtcp_exited = asyncio.Event()
342 self.__rtcp_started = asyncio.Event()
343 self.__rtcp_task: Optional[asyncio.Future[None]] = None
344 self.__rtx_payload_type: Optional[int] = None
345 self.__rtx_sequence_number = random_sequence_number()
346 self.__fec_payload_type: Optional[int] = None
347 self.__fec_sequence_number = random_sequence_number()
348 # Repair packets per FlexFEC group, with interleaved masks: as many
349 # losses in a group as there are repairs are recovered. Follows the
350 # loss `steer_fec` is told about.
351 self.fec_repair_packets = 1
352 self._fec_loss = 0.0
353 # The loss of windows clear of a queue, and how many windows in a row
354 # have been (`steer_fec`).
355 self._fec_clear_loss = 0.0
356 self._fec_clear = 0
357 # Whether every group takes its repairs (`steer_fec`), and the repairs
358 # owed otherwise, in FEC_GROUP_PACKETS-ths (`_fec_repairs`).
359 self._fec_full = False
360 self._fec_credit = 0
361 self.__started = False
362 self.__stats = RTCStatsReport()
363 self.__transport = transport
365 # stats
366 self.__lsr: Optional[int] = None
367 self.__lsr_time: Optional[float] = None
368 # The RTP timestamp of the last frame sent, the CLOCK_MONOTONIC instant it
369 # stands for (its capture, where the frame says), and its clock rate: a
370 # sender report maps the RTP clock of the moment it is sent from these.
371 self.__rtp_timestamp = 0
372 self.__rtp_instant_ns = 0
373 self.__rtp_clock_rate = 0
374 self.__octet_count = 0
375 self.__packet_count = 0
376 self.__rtt: Optional[float] = None
377 # The last reference time each receive-only reporter sent (middle 32 bits of its
378 # NTP time) and the CLOCK_MONOTONIC instant it arrived, answered in every report.
379 self.__rrtrs: dict[int, tuple[int, int]] = {}
381 # logging
382 self.__log_debug: Callable[..., None] = lambda *args: None
383 if logger.isEnabledFor(logging.DEBUG):
384 self.__log_debug = lambda msg, *args: logger.debug(
385 f"RTCRtpSender(%s) {msg}", self.__kind, *args
386 )
388 @property
389 def kind(self) -> str:
390 return self.__kind
392 @property
393 def track(self) -> MediaStreamTrack:
394 """
395 The :class:`MediaStreamTrack` which is being handled by the sender.
396 """
397 return self.__track
399 @property
400 def transport(self) -> RTCDtlsTransport:
401 """
402 The :class:`RTCDtlsTransport` over which media data for the track is
403 transmitted.
404 """
405 return self.__transport
407 @classmethod
408 def getCapabilities(cls, kind: str) -> RTCRtpCapabilities:
409 """
410 Returns the most optimistic view of the system's capabilities for
411 sending media of the given `kind`.
413 :rtype: :class:`RTCRtpCapabilities`
414 """
415 return get_capabilities(kind)
417 async def getStats(self) -> RTCStatsReport:
418 """
419 Returns statistics about the RTP sender.
421 :rtype: :class:`RTCStatsReport`
422 """
423 self.__stats.add(
424 RTCOutboundRtpStreamStats(
425 # RTCStats
426 timestamp=clock.current_datetime(),
427 type="outbound-rtp",
428 id="outbound-rtp_" + str(id(self)),
429 # RTCStreamStats
430 ssrc=self._ssrc,
431 kind=self.__kind,
432 transportId=self.transport._stats_id,
433 # RTCSentRtpStreamStats
434 packetsSent=self.__packet_count,
435 bytesSent=self.__octet_count,
436 # RTCOutboundRtpStreamStats
437 trackId=str(id(self.track)),
438 )
439 )
440 self.__stats.update(self.transport._get_stats())
442 return self.__stats
444 def replaceTrack(self, track: Optional[MediaStreamTrack]) -> None:
445 self.__track = track
446 if track is not None:
447 self._track_id = track.id
448 else:
449 self._track_id = str(uuid.uuid4())
451 def setTransport(self, transport: RTCDtlsTransport) -> None:
452 self.__transport = transport
454 async def send(self, parameters: RTCRtpSendParameters) -> None:
455 """
456 Attempt to set the parameters controlling the sending of media.
458 :param parameters: The :class:`RTCRtpSendParameters` for the sender.
459 """
460 if not self.__started:
461 self.__cname = parameters.rtcp.cname
462 self.__mid = parameters.muxId
464 # make note of the RTP header extension IDs
465 self.__transport._register_rtp_sender(self, parameters)
466 self.__rtp_header_extensions_map.configure(parameters)
468 # Send with the first codec that actually has an encoder: auxiliary
469 # entries (rtx, flexfec) can top the negotiated list when the media
470 # codec was filtered out, and starting RTP on them kills the sender.
471 send_codec = None
472 for codec in parameters.codecs:
473 if is_rtx(codec) or codec.mimeType.lower() == "video/flexfec-03":
474 continue
475 send_codec = codec
476 break
477 if send_codec is None:
478 raise InvalidStateError("No sendable media codec was negotiated")
479 self.__codecs = list(parameters.codecs)
480 if self.__pending_codec is not None:
481 send_codec = self.negotiated_codec(self.__pending_codec) or send_codec
482 self.__send_codec = send_codec
483 self.__rtx_payload_type = self._rtx_payload_type_for(send_codec)
485 # make note of the FlexFEC payload type (negotiated => protect video)
486 for codec in parameters.codecs:
487 if codec.mimeType.lower() == "video/flexfec-03":
488 self.__fec_payload_type = codec.payloadType
489 break
491 self.__rtp_task = asyncio.ensure_future(self._run_rtp())
492 self.__rtcp_task = asyncio.ensure_future(self._run_rtcp())
493 self.__started = True
495 def _rtx_payload_type_for(self, codec: RTCRtpCodecParameters) -> Optional[int]:
496 for candidate in self.__codecs:
497 if is_rtx(candidate) and candidate.parameters["apt"] == codec.payloadType:
498 return candidate.payloadType
499 return None
501 def negotiated_codec(self, mime_type: str) -> Optional[RTCRtpCodecParameters]:
502 """The negotiated media codec of `mime_type`, or None when the peer did not take it."""
503 wanted = mime_type.lower()
504 for codec in self.__codecs:
505 if codec.mimeType.lower() == wanted:
506 return codec
507 return None
509 def switch_codec(self, mime_type: str) -> bool:
510 """Send the track's frames as the negotiated codec of `mime_type` from
511 now on: its payload type, its RTX type, and its own packer. A codec the
512 peer never took drops the frames instead, until a switch names one it
513 did; before the sender starts, the switch names the codec it starts on,
514 out of the answer."""
515 if not self.__started:
516 self.__pending_codec = mime_type
517 return True
518 codec = self.negotiated_codec(mime_type)
519 self.__send_codec = codec
520 self.__encoder = None
521 self._anchored = False
522 self.__rtx_payload_type = self._rtx_payload_type_for(codec) if codec else None
523 return codec is not None
525 async def stop(self) -> None:
526 """
527 Irreversibly stop the sender.
528 """
529 if self.__started:
530 self.__transport._unregister_rtp_sender(self)
532 # shutdown RTP and RTCP tasks
533 await asyncio.gather(self.__rtp_started.wait(), self.__rtcp_started.wait())
534 self.__rtp_task.cancel()
535 self.__rtcp_task.cancel()
536 await asyncio.gather(self.__rtp_exited.wait(), self.__rtcp_exited.wait())
538 async def _handle_rtcp_packet(self, packet: AnyRtcpPacket) -> None:
539 if isinstance(packet, (RtcpRrPacket, RtcpSrPacket)):
540 for report in filter(lambda x: x.ssrc == self._ssrc, packet.reports):
541 # estimate round-trip time
542 if self.__lsr == report.lsr and report.dlsr:
543 rtt = time.time() - self.__lsr_time - (report.dlsr / 65536)
544 if self.__rtt is None:
545 self.__rtt = rtt
546 else:
547 self.__rtt = RTT_ALPHA * self.__rtt + (1 - RTT_ALPHA) * rtt
549 self.__stats.add(
550 RTCRemoteInboundRtpStreamStats(
551 # RTCStats
552 timestamp=clock.current_datetime(),
553 type="remote-inbound-rtp",
554 id="remote-inbound-rtp_" + str(id(self)),
555 # RTCStreamStats
556 ssrc=packet.ssrc,
557 kind=self.__kind,
558 transportId=self.transport._stats_id,
559 # RTCReceivedRtpStreamStats
560 packetsReceived=self.__packet_count - report.packets_lost,
561 packetsLost=report.packets_lost,
562 jitter=report.jitter,
563 # RTCRemoteInboundRtpStreamStats
564 roundTripTime=self.__rtt,
565 fractionLost=report.fraction_lost,
566 )
567 )
568 elif isinstance(packet, RtcpRtpfbPacket) and packet.fmt == RTCP_RTPFB_NACK:
569 gone = self._past_repair(packet.lost)
570 if gone:
571 self.__rtp_history.abandon()
572 for seq in packet.lost:
573 if self.__rtp_history.abandoned(seq):
574 continue
575 sent, frame, times = self.__rtp_history.nacked(seq)
576 if sent is None:
577 # A list runs oldest first, so one let go says nothing about the rest.
578 gone = True
579 continue
580 # A NACK after the first, sent once the retransmission and any FlexFEC
581 # repair of the packet had a round trip to reach the peer, says neither
582 # did: the frame is lost to it, and the encoder is told once so the frames
583 # after it stop predicting from it.
584 lost = (times > 1 and frame is not None
585 and self.__rtp_history.unrepaired(seq, time.time(), self.__rtt or 0.0))
586 if not await self._retransmit(sent):
587 break
588 # A first repair goes out twice, back to back: lost with the original, it would
589 # cost the frame a second NACK a round trip later and a prediction past it.
590 if times == 1 and not await self._retransmit(sent):
591 break
592 self.__rtp_history.repaired(seq, time.time() + self.transport._send_delay())
593 if lost and self.__rtp_history.newly_lost(frame, time.time()):
594 self.emit("lost_frame", frame)
595 self._lost_at = time.monotonic()
596 if gone and not self._resync_held():
597 # Gone from the history, or past repair, where no resync answers it: only a key
598 # frame brings the peer back.
599 self._emit_pli_event()
600 elif isinstance(packet, RtcpXrPacket) and packet.rrtr is not None:
601 self.__rrtrs.pop(packet.ssrc, None)
602 if len(self.__rrtrs) >= RRTR_REPORTERS_MAX:
603 del self.__rrtrs[next(iter(self.__rrtrs))]
604 self.__rrtrs[packet.ssrc] = ((packet.rrtr >> 16) & 0xFFFFFFFF, time.monotonic_ns())
605 elif isinstance(packet, RtcpRtpfbPacket) and packet.fmt == RTCP_RTPFB_TWCC:
606 self.transport._twcc_process_feedback(packet.fci)
607 if self.__unheld:
608 self._confirm()
609 elif isinstance(packet, RtcpPsfbPacket) and packet.fmt in (RTCP_PSFB_PLI, RTCP_PSFB_FIR):
610 # A Full Intra Request (RFC 5104) asks what a PLI does: a key frame.
611 if not self._still_request() and not self._resync_held():
612 self._send_keyframe()
613 self._emit_pli_event()
614 elif isinstance(packet, RtcpPsfbPacket) and packet.fmt == RTCP_PSFB_APP:
615 try:
616 bitrate, ssrcs = unpack_remb_fci(packet.fci)
617 if self._ssrc in ssrcs:
618 self.__log_debug(
619 "- receiver estimated maximum bitrate %d bps", bitrate
620 )
621 if self.__encoder and hasattr(self.__encoder, "target_bitrate"):
622 self.__encoder.target_bitrate = bitrate
623 except ValueError:
624 pass
626 def _still_request(self) -> bool:
627 """Whether a key-frame request is a receiver's wait on a still screen
628 (STILL_REQUEST_S), which a key frame would only restart: the newest frame is
629 on the wire and acknowledged, it left STILL_REQUEST_S and the round trip ago,
630 no frame lost past repair awaits one predicting past it, and the request
631 before came STILL_REQUEST_S or more earlier."""
632 now = time.monotonic()
633 before, self._key_asked_at = self._key_asked_at, now
634 left = self._frame_left
635 return (left is not None and left[0] == self._frame_handed and self._lost_at is None
636 and now - left[1] >= STILL_REQUEST_S + (self.__rtt or 0.0)
637 and now - before >= STILL_REQUEST_S
638 and self.transport._twcc_acked(left[0]))
640 def _frame_on_wire(self, seq: int, keyframe: bool, timing: Optional[tuple],
641 captured: int, size: int, frame_id: Optional[int] = None) -> None:
642 """A frame's last packet left for the wire. One whose encode began after the
643 peer last lost a frame predicts past it, as a key frame predicts from nothing."""
644 if frame_id is not None and self.on_frame_out is not None:
645 self.on_frame_out(frame_id, keyframe)
646 self._frame_left = (seq, time.monotonic())
647 while self.__in_flight and ((seq - self.__in_flight[0][0]) & 0xFFFF) < 0x8000:
648 self.__in_flight.popleft()
649 encoded = timing[1] / 1e9 if timing and len(timing) > 1 and timing[1] > 0 else 0.0
650 if self._lost_at is not None and (keyframe or encoded > self._lost_at):
651 self._lost_at = None
652 if self.on_frame_sent is not None:
653 self.on_frame_sent(captured, size)
655 def _emit_pli_event(self):
656 """
657 Emit a "pli" event to notify the application layer, which is
658 responsible for instructing the encoder to generate that keyframe
659 """
660 self.emit("pli")
662 def _pacer_resync(self, first_tag: Optional[int]) -> bool:
663 """A pacer GOP reset (`RtpPacer._reset_gop`), told the oldest packet it dropped.
665 Where the encoder names each frame's reference, the peer reads the
666 dependency descriptor and does not own its display (`selective`), the
667 frames whose last packet had not left are lost to the peer: they leave
668 the frames it was sent, so `_forward` holds back whatever predicts from
669 them, NACKs for the dropped packets go unanswered, and the run is
670 answered as a run left out is. False where this sender cannot, which
671 leaves the pacer its key frame, and for the display's owner and a peer
672 without the descriptor: the cut leaves a gap in its sequence numbers
673 that no retransmission fills, which its receiver closes on a key frame
674 (a burst of loss on a software encoder's stream otherwise took one to
675 two seconds to close).
676 """
677 if (first_tag is None or not self.__in_flight or self.selective is None
678 or not self.selective()
679 or not self.__rtp_header_extensions_map.has_dependency_descriptor()):
680 return False
681 lost = [fid for tag, fid in self.__in_flight if ((tag - first_tag) & 0xFFFF) < 0x8000]
682 if not lost:
683 return False
684 for fid in lost:
685 self.__frame_numbers.pop(fid, None)
686 self.__in_flight.clear()
687 self.__rtp_history.abandon()
688 self._resyncs += 1
689 if self._resync_since is None:
690 self._open_run(lost[0])
691 self._resync_run = len(lost)
692 else:
693 self._resync_run += len(lost)
694 self._repair()
695 return True
697 def _confirm(self) -> None:
698 """Tell `on_frame_held` of each frame sent the peer now holds: every packet of it
699 reported received, or the retransmission of one, and the frame it predicts from
700 held. A frame not held within HELD_S never will be. A peer without the dependency
701 descriptor decodes nothing past a frame it lacks until a key frame, so it holds a
702 frame only once it holds every frame sent before it."""
703 acked = self.transport.twcc_arrived
704 now = time.monotonic()
705 rtx = self.__rtx_twcc
706 in_order = not self.__rtp_header_extensions_map.has_dependency_descriptor()
707 kept = []
708 for entry in self.__unheld:
709 frame_id, reference, keyframe, sent_at, packets = entry
710 if now - sent_at > HELD_S:
711 self._unheld_gap = self._unheld_gap or in_order
712 continue
713 if (not (in_order and (kept or self._unheld_gap))
714 and all(acked(twcc) or (media in rtx and acked(rtx[media])) for twcc, media in packets)
715 and (keyframe or reference in self.__held)):
716 self.__held.pop(frame_id, None)
717 self.__held[frame_id] = None
718 self._held_at = now
719 if len(self.__held) > FRAME_NUMBERS_MEMORY:
720 del self.__held[next(iter(self.__held))]
721 if self.on_frame_held is not None:
722 self.on_frame_held(frame_id, keyframe)
723 else:
724 kept.append(entry)
725 self.__unheld.clear()
726 self.__unheld.extend(kept)
727 if len(rtx) > 1024:
728 for media in list(rtx)[:512]:
729 del rtx[media]
731 def _past_repair(self, lost: list) -> bool:
732 """Whether a NACK from a peer that does not own its display, and reads the dependency
733 descriptor, names more than its pacer queue holds, a first repair going out twice: the
734 burst a link that went dark asks for once it is back, whose retransmissions would
735 overflow the queue and cost every peer a key frame, and hold every newer frame behind
736 them. It is answered as a loss past repair."""
737 if (self.__kind != "video" or self.selective is None or not self.selective()
738 or not self.__rtp_header_extensions_map.has_dependency_descriptor()):
739 return False
740 room = self.transport.video_room()
741 held = (self.__rtp_history.get(seq) for seq in lost)
742 return room is not None and 2 * sum(len(p.payload) for p in held if p is not None) > room
744 def _resync_held(self) -> bool:
745 """Answer a peer that lost frames past repair as a run the pacer cut is, where it does
746 not own its display and anchors run: the frames sent it after the newest it holds leave
747 the frames it was sent, and the encoder is told the first is lost (`_repair`), so a frame
748 predicting from one it holds comes next. A request while that run is open, or while the
749 key frame it was sent is on its way, is answered by them. False where it cannot, which
750 asks for a key frame: past RESYNC_ANCHORED_S of such requests one after another while
751 it holds frames it is sent (a decoder that only takes a key frame; one holding nothing
752 new between two requests is starved, as through an outage, and starts the count again),
753 where it holds nothing and no key frame is due, and for a peer without the dependency
754 descriptor, which decodes nothing past the packets it lacks until a key frame. A frame
755 held only after a resync let it go is passed over, and a peer whose link is dark
756 (RESYNC_DARK_S) is answered by its request once it is back: neither a resync nor a key
757 frame reaches it now. One sent nothing since its last report, as while it waits for a key
758 frame, is not dark."""
759 now = time.monotonic()
760 if (self.selective is None or not self.selective() or not self._anchored
761 or not self.__rtp_header_extensions_map.has_dependency_descriptor()):
762 return False
763 if 0 < self.transport._twcc_received_at < now - RESYNC_DARK_S < self.transport._twcc_sent_at:
764 return True
765 if (self._stall_since is None or now - self._stall_last > RESYNC_ANCHORED_S
766 or self._held_at < self._stall_last):
767 self._stall_since = now
768 self._stall_last = now
769 if now - self._stall_since > RESYNC_ANCHORED_S:
770 self._stall_since = None
771 return False
772 if self._resync_since is not None:
773 return True
774 if not self.__held:
775 return (self._key_sent_at is not None and now - self._key_sent_at
776 < max(RESYNC_S, 2 * (self._video_backlog()[1] + (self.__rtt or 0.0))))
777 number = next((n for n in map(self.__frame_numbers.get, reversed(self.__held)) if n is not None), None)
778 if number is None:
779 return False
780 after = [fid for fid, n in self.__frame_numbers.items() if 0 < ((n - number) & 0xFFFF) < 0x8000]
781 if not after:
782 return True
783 for fid in after:
784 self.__frame_numbers.pop(fid, None)
785 # The encoder is told the newest of them predicting from none of the others: it predicted
786 # past the frames before that one already, and a report naming one of those is ignored.
787 refs = {entry[0]: entry[1] for entry in self.__unheld}
788 later = set(after)
789 self._open_run(next((fid for fid in reversed(after) if fid in refs and refs[fid] not in later),
790 after[0]))
791 self._resync_run = len(after)
792 self._repair()
793 return True
795 def _video_backlog(self) -> tuple[int, float, float]:
796 """The peer's frames in its pacer, how long a packet waits to reach it, and the
797 seconds since its path last lost one (`RTCDtlsTransport.video_backlog`)."""
798 backlog = getattr(self.transport, "video_backlog", None)
799 return backlog() if backlog is not None else (0, 0.0, float("inf"))
801 def _forward(self, frame_id: int, reference: Optional[int], keyframe: bool,
802 anchor: bool = False) -> bool:
803 """Whether a frame naming the frame it predicts from goes to this peer; False
804 for one left out, or held back for predicting from one (RESYNC_ROOM_FRAMES).
805 The first frame predicting past an open run, which decodes on the peer, ends it,
806 and an `anchor` goes without room unless the peer is far behind."""
807 if keyframe:
808 self._since_key = 0
809 self._close_run()
810 return True
811 self._since_key += 1
812 self._anchored = self._anchored or anchor
813 sent = reference is None or reference in self.__frame_numbers
814 # In H.264 a frame where frame_num may wrap is never left out.
815 kept = self._numbered() and not self._since_key % FRAME_NUM_WRAP
816 if self._resync_since is not None:
817 if sent and (not anchor or kept or self._anchor_room()):
818 self._close_run()
819 return True
820 if sent:
821 # An anchor left out predicts past the run: the encoder is told of it.
822 self._resync_first = frame_id
823 self._resync_told = False
824 self._resync_run += 1
825 self._repair()
826 return False
827 if not sent or self.selective is None or kept:
828 return True
829 if anchor and self._anchor_room():
830 return True
831 frames, wait, _ = self._video_backlog()
832 if (frames < RESYNC_ROOM_FRAMES and wait < RESYNC_LAG_S) or not self.selective():
833 return True
834 self._open_run(frame_id)
835 self._repair()
836 return False
838 def _anchor_room(self) -> bool:
839 """Whether this peer is sent an anchor without room: not far behind, and the queue
840 in front of it under RESYNC_ANCHOR_S."""
841 return not self._far() and self._video_backlog()[1] < RESYNC_ANCHOR_S
843 def _far(self) -> bool:
844 """Whether this peer is too far behind to be sent a frame without room: its
845 pacer holds RESYNC_FAR_FRAMES or RESYNC_FAR_S, or its path lost packets within
846 RESYNC_LOSS_S while a queue stood on it."""
847 frames, wait, lost = self._video_backlog()
848 return (frames >= RESYNC_FAR_FRAMES or wait >= RESYNC_FAR_S
849 or (lost < RESYNC_LOSS_S and wait >= RESYNC_LAG_S))
851 def _numbered(self) -> bool:
852 """Whether the stream is H.264, whose frame_num wraps (FRAME_NUM_WRAP)."""
853 codec = self.__send_codec
854 return codec is not None and codec.mimeType.lower() == "video/h264"
856 def _numbers_frames(self) -> bool:
857 """Whether frames naming their reference are numbered for this peer and left out
858 where it has no room: one reading the dependency descriptor, or an H.264 one
859 without it (Firefox answers so), which finds a frame's reference by sequence
860 number, unbroken by a frame left out before it is packed; its frames carry no
861 descriptor."""
862 return self.__rtp_header_extensions_map.has_dependency_descriptor() or self._numbered()
864 def _learn_parameter_sets(self, enc_frame: RTCEncodedFrame) -> None:
865 """Keep the SPS and PPS a key frame sent this peer carries, which its stand-ins'
866 slice headers are read against (`_stand_in`)."""
867 if enc_frame.data is None:
868 return
869 if self.__skips is None:
870 self.__skips = SkipPictures()
871 self.__skips.learn([bytes(nal) for nal in H264Encoder._split_bitstream(memoryview(enc_frame.data))])
873 def _stand_in(self, enc_frame: RTCEncodedFrame) -> Optional[RTCEncodedFrame]:
874 """A picture of skipped macroblocks in place of an H.264 frame left out for a peer
875 without the dependency descriptor (`h264_skip`), under the frame's own frame_num and
876 reference marking: its receiver hands the decoder every frame it is sent by sequence
877 number, and a decoder that refuses a frame_num gap (OpenH264) decodes on, repeating
878 its picture until the frame predicting past the run (`_repair`). Not numbered, so it
879 is neither a frame this peer holds nor one a later frame may predict from. None where
880 the frame cannot stand in, which leaves it out as before."""
881 if self.__skips is None or enc_frame.data is None or not self._numbered():
882 return None
883 built = self.__skips.stand_in(
884 [bytes(nal[:HEAD_BYTES]) for nal in H264Encoder._split_bitstream(memoryview(enc_frame.data))])
885 if built is None:
886 return None
887 self.stand_ins += 1
888 return RTCEncodedFrame(H264Encoder._packetize(built), enc_frame.timestamp, None, False, enc_frame.timing)
890 def _open_run(self, frame_id: int) -> None:
891 self._resync_since = time.monotonic()
892 self._resync_first = frame_id
893 self._resync_run = 1
894 self._resync_told = False
896 def _close_run(self) -> None:
897 self._resync_since = None
898 self._resync_first = None
899 self._resync_run = 0
900 self._resync_told = False
902 def _repair(self) -> None:
903 """A frame of the open run was held back: tell the encoder of the run's first
904 frame once the peer has room, or sooner where waiting would cost a key frame
905 (RESYNC_REACH_FRAMES); past RESYNC_S (RESYNC_ANCHORED_S where anchors run), ask
906 for a key frame instead."""
907 if time.monotonic() - self._resync_since > (RESYNC_ANCHORED_S if self._anchored else RESYNC_S):
908 logger.info("RTCRtpSender(%s) frame %s was never predicted past; asking for a key frame",
909 self.__kind, self._resync_first)
910 self._close_run()
911 self._emit_pli_event()
912 return
913 if self._resync_told:
914 return
915 frames, wait, _ = self._video_backlog()
916 due = (self._resync_run >= RESYNC_REACH_FRAMES
917 or (self._numbered() and -self._since_key % FRAME_NUM_WRAP <= RESYNC_WRAP_LEAD))
918 if (frames < RESYNC_ROOM_FRAMES and wait < RESYNC_LAG_S) or (due and not self._far()):
919 # Where anchors run, a run of any depth is predicted past from the one pinned.
920 reach = self._resync_run <= RESYNC_REACH_FRAMES or self._anchored
921 self._lost_at = time.monotonic()
922 if self.on_resync is not None:
923 self._resync_told = self.on_resync(self._resync_first, reach)
924 else:
925 self.emit("lost_frame", self._resync_first)
926 self._resync_told = True
928 def _undecodable(self, frame_id: int, reference: Optional[int]) -> None:
929 """A frame predicting from one this peer was never sent (a peer paused across
930 it): nothing but a key frame decodes here, and one is asked for."""
931 logger.info("RTCRtpSender(%s) frame %s predicts from %s, which this peer "
932 "was not sent; asking for a key frame", self.__kind, frame_id, reference)
933 self._emit_pli_event()
935 def _peer_gone(self) -> bool:
936 """Whether the peer has sent nothing at all for PEER_SILENCE_S, so frames are not
937 sent to a client that left without a goodbye. Its first packet after the silence
938 asks for a key frame: the frames skipped meanwhile broke the prediction chain, and
939 `_describe` already leaves out the ones predicting from them."""
940 silent = time.monotonic() - self.transport._peer_heard_at() > PEER_SILENCE_S
941 if silent != self._peer_silent:
942 self._peer_silent = silent
943 logger.info("%s sender %s: the peer %s", self.__kind, self._ssrc,
944 "went silent; sending stopped" if silent else "is back; sending again")
945 if not silent:
946 self._emit_pli_event()
947 return silent
949 def _fec_repairs(self, packets: int) -> int:
950 """How many repairs a FlexFEC group of `packets` media packets gets:
951 `fec_repair_packets`, never more than it has packets, where `steer_fec`
952 wants every group repaired. Otherwise `fec_repair_packets` per
953 FEC_GROUP_PACKETS media packets, what a group falls short of carried to
954 the next: a group closes at every frame, so a low-rate stream's frames
955 of one or two packets would each take a repair and double what the
956 path carries."""
957 if self._fec_full:
958 self._fec_credit = 0
959 return min(self.fec_repair_packets, packets)
960 self._fec_credit += packets * self.fec_repair_packets
961 repairs = min(packets, self._fec_credit // FEC_GROUP_PACKETS)
962 self._fec_credit -= repairs * FEC_GROUP_PACKETS
963 return repairs
965 def steer_fec(self, loss_fraction: float, queued: bool = False) -> None:
966 """Set the FlexFEC repair density from a measured loss fraction: one
967 repair per group under 2% loss, two under 8%, three above, read on a
968 smoothed loss so a single small window neither adds nor drops a
969 repair on its own. Every group takes them while the path loses
970 packets with no queue standing (`queued`), a loss a repair recovers
971 without a retransmission's round trip: the loss read on windows
972 FEC_CLEAR_WINDOWS past the last queue, since a queue's overflow shows
973 in the windows it stood in and the ones just after. With nothing lost
974 that way, or while a queue stands, whose overflow a repair per frame
975 would only deepen, the repairs follow the media packets
976 (`_fec_repairs`)."""
977 self._fec_loss += (loss_fraction - self._fec_loss) * 0.3
978 self._fec_clear = 0 if queued else self._fec_clear + 1
979 if self._fec_clear > FEC_CLEAR_WINDOWS:
980 self._fec_clear_loss += (loss_fraction - self._fec_clear_loss) * 0.3
981 self.fec_repair_packets = 1 + (self._fec_loss > 0.02) + (self._fec_loss > 0.08)
982 self._fec_full = self._fec_clear_loss > FEC_FULL_LOSS and not queued
984 async def _next_encoded_frame(self) -> Optional[RTCEncodedFrame]:
985 data = await self.__track.recv()
987 # If the sender is disabled, drop the frame instead of packing it.
988 # We still want to read from the track in order to avoid frames
989 # accumulating in memory.
990 if not self._enabled or self.__send_codec is None or self._peer_gone():
991 return None
992 # A frame the capture coded before a codec switch is not this codec's to pack.
993 codec = getattr(data, "codec", None)
994 if codec is not None and codec.lower() != self.__send_codec.mimeType.lower():
995 return None
997 if self.__encoder is None:
998 self.__encoder = get_encoder(self.__send_codec)
1000 # Tracks serve frames pixelflux/pcmflux have already encoded; the sender
1001 # only packs them into RTP payloads. Keyframes are requested out of band
1002 # (the capture side produces the IDR), so no encode runs here.
1003 self.__force_keyframe_used = False
1004 try:
1005 payloads, timestamp, keyframe = self.__encoder.pack(data)
1006 except ValueError as e:
1007 # A frame its packer cannot read is dropped, not the sender with it.
1008 logger.warning("RTCRtpSender(%s) dropped a frame it cannot pack: %s", self.__kind, e)
1009 if self.__kind == "video":
1010 self._emit_pli_event()
1011 return None
1013 # If the packer did not return any payloads, return `None`.
1014 if not payloads:
1015 return None
1017 return RTCEncodedFrame(payloads, timestamp, None, data.keyframe, data.timing, data.dependency,
1018 getattr(data, "anchor", False), data.data)
1020 def _describe(self, frame_id: int, reference: Optional[int], keyframe: bool) -> Optional[tuple]:
1021 """Number a frame on the wire and measure how far back it predicts, for its
1022 dependency descriptor: (number, frames back), the latter None for a frame that
1023 predicts from nothing. None for a frame predicting from one this sender never sent
1024 (a peer paused across it): undecodable here, so the caller leaves it out."""
1025 if keyframe:
1026 self.__frame_numbers.clear()
1027 fdiff = None
1028 if reference is not None:
1029 known = self.__frame_numbers.get(reference)
1030 if known is None:
1031 return None
1032 fdiff = (self.__frame_number - known) & 0xFFFF
1033 if not 1 <= fdiff <= 4096:
1034 return None
1035 number = self.__frame_number
1036 self.__frame_number = (number + 1) & 0xFFFF
1037 self.__frame_numbers[frame_id] = number
1038 if len(self.__frame_numbers) > FRAME_NUMBERS_MEMORY:
1039 del self.__frame_numbers[next(iter(self.__frame_numbers))]
1040 return number, fdiff
1042 async def _retransmit(self, packet: RtpPacket) -> bool:
1043 """Retransmit an RTP packet reported lost; False when the pacer dropped it."""
1044 media_seq = packet.sequence_number
1045 if self.__rtx_payload_type is not None:
1046 packet = wrap_rtx(
1047 packet,
1048 payload_type=self.__rtx_payload_type,
1049 sequence_number=self.__rtx_sequence_number,
1050 ssrc=self._rtx_ssrc,
1051 )
1052 self.__rtx_sequence_number = uint16_add(self.__rtx_sequence_number, 1)
1054 # A retransmission is a new packet on the wire: give it its own
1055 # transport-wide sequence number.
1056 packet.extensions.transport_sequence_number = self.transport._twcc_next(
1057 len(packet.payload)
1058 )
1059 if self.__unheld:
1060 self.__rtx_twcc[media_seq] = packet.extensions.transport_sequence_number
1061 self.__log_debug("> %s", packet)
1062 return await self._send(packet.serialize(self.__rtp_header_extensions_map),
1063 packet.extensions.transport_sequence_number)
1065 async def _send(self, packet_bytes: bytes, twcc_seq: Optional[int] = None) -> bool:
1066 """Hand a packet to the transport. A video packet the pacer refuses was
1067 abandoned with its GOP, and every packet the history holds with it: NACKs
1068 for them go unanswered, since the keyframe the pacer asked for is their
1069 repair and a late retransmission would only land behind it."""
1070 sent = await self.transport._send_rtp(
1071 packet_bytes, rtc_class=CLASS_AUDIO if self.__kind == "audio" else CLASS_VIDEO,
1072 twcc_seq=twcc_seq)
1073 if not sent:
1074 self.__rtp_history.abandon()
1075 return sent
1077 def _send_keyframe(self) -> None:
1078 """
1079 Request the next frame to be a keyframe.
1080 """
1081 self.__force_keyframe = True
1083 def request_keyframe(self) -> None:
1084 """Public alias used by the transport pacer's GOP-reset recovery hook."""
1085 self._send_keyframe()
1087 async def _run_rtp(self) -> None:
1088 self.__log_debug("- RTP started")
1089 self.__rtp_started.set()
1090 if self.__kind == "video" and hasattr(self.transport, "set_video_resync"):
1091 self.transport.set_video_resync(self._pacer_resync)
1093 sequence_number = random_sequence_number()
1094 timestamp_origin = random32()
1095 # Timer-triggered video-timing diagnostics (~5 flagged frames/s, like libwebrtc).
1096 last_video_timing = 0.0
1097 last_abs_capture_ns = 0
1098 # FlexFEC group: serialized media packets awaiting their XOR repair
1099 # packets (flushed per frame, or every FEC_GROUP_PACKETS within a large frame).
1100 fec_group: list[bytes] = []
1101 fec_first_seq = 0
1102 try:
1103 while True:
1104 if not self.__track:
1105 await asyncio.sleep(0.02)
1106 continue
1108 # Fetch the next encoded frame. This can be `None` if the sender
1109 # is disabled, in which case we just continue the loop.
1110 enc_frame = await self._next_encoded_frame()
1111 if enc_frame is None:
1112 # A frame this peer is not sent leaves its picture behind the screen.
1113 self._frame_left = None
1114 continue
1115 codec = self.__send_codec
1116 frame_time = time.time()
1117 instant_ns = (enc_frame.timing[0] if enc_frame.timing and enc_frame.timing[0] > 0
1118 else time.monotonic_ns())
1120 if self.__kind == "video" and (
1121 self.__force_keyframe_used or enc_frame.keyframe
1122 ):
1123 # Report keyframe size to the pacer: feeds its IDR-aware
1124 # queue budget and resurrects video after a GOP reset.
1125 # Forced (recovery) keyframes resurrect but must not
1126 # shrink the IDR floor. Every JPEG frame is one, so every
1127 # one feeds the floor. Remember for late attach.
1128 natural = not self.__force_keyframe_used
1129 size = sum(len(p_) for p_ in enc_frame.payloads)
1130 self._keyframe_bytes = size
1131 self._keyframe_natural = natural
1132 self.transport.note_video_keyframe(size, natural=natural)
1134 timestamp = uint32_add(timestamp_origin, enc_frame.timestamp)
1135 described = None
1136 standing_in = False
1137 descriptor = self.__rtp_header_extensions_map.has_dependency_descriptor()
1138 if enc_frame.dependency is not None and self._numbers_frames():
1139 if not self._forward(*enc_frame.dependency, enc_frame.keyframe, enc_frame.anchor):
1140 self._frame_left = None
1141 if self.on_frame_left_out is not None:
1142 self.on_frame_left_out()
1143 stand_in = None if descriptor else self._stand_in(enc_frame)
1144 if stand_in is None:
1145 continue
1146 enc_frame, standing_in = stand_in, True
1147 else:
1148 described = self._describe(*enc_frame.dependency, enc_frame.keyframe)
1149 if described is None:
1150 self._frame_left = None
1151 self._undecodable(*enc_frame.dependency)
1152 continue
1153 if not descriptor and enc_frame.keyframe and self._numbered():
1154 self._learn_parameter_sets(enc_frame)
1156 # abs-capture-time rides the first packet of a key frame and of a
1157 # frame a second after the last one, as libwebrtc paces it: the RTP
1158 # clock follows the capture, so a receiver extrapolates between them.
1159 abs_capture_time = None
1160 if self.__kind == "video" and (
1161 enc_frame.keyframe or instant_ns - last_abs_capture_ns >= 1_000_000_000
1162 ):
1163 last_abs_capture_ns = instant_ns
1164 abs_capture_time = clock.ntp_from_monotonic_ns(instant_ns)
1166 # Every datagram of the frame is built before the first is sent:
1167 # building and sending in turn costs a third more per packet, as each
1168 # evicts what the other just warmed, and the frame is decodable only
1169 # once its last packet is in. Entries: bytes, transport-wide sequence
1170 # number, media sequence number (None for FlexFEC), payload length.
1171 outgoing: list[tuple[bytes, Optional[int], Optional[int], int]] = []
1172 # The first sequence number and count of each FlexFEC group it repairs.
1173 protected: list[tuple[int, int]] = []
1174 for i, payload in enumerate(enc_frame.payloads):
1175 packet = RtpPacket(
1176 payload_type=codec.payloadType,
1177 sequence_number=sequence_number,
1178 timestamp=timestamp,
1179 )
1180 packet.ssrc = self._ssrc
1181 packet.payload = payload
1182 packet.marker = (i == len(enc_frame.payloads) - 1) and 1 or 0
1184 # set header extensions
1185 packet.extensions.abs_send_time = (
1186 clock.current_ntp_time() >> 14
1187 ) & 0x00FFFFFF
1188 packet.extensions.mid = self.__mid
1189 packet.extensions.transport_sequence_number = (
1190 self.transport._twcc_next(len(payload))
1191 )
1192 if enc_frame.audio_level is not None:
1193 packet.extensions.audio_level = (False, -enc_frame.audio_level)
1194 # https://webrtc.googlesource.com/src/+/main/docs/native-code/rtp-hdrext/playout-delay/README.md
1195 # set min and max to 0 to hint the receiver to render frames as soon as possible
1196 packet.extensions.playout_delay = (0, 0)
1197 # The colour signal rides every packet of a key frame, as libwebrtc sends
1198 # it; the receiver keeps it for the frames that follow.
1199 color_space = RTP_COLOR_SPACE.get(codec.mimeType.lower())
1200 if enc_frame.keyframe and color_space is not None:
1201 packet.extensions.color_space = color_space
1202 # video-timing rides the LAST packet of a frame, about five
1203 # times a second like libwebrtc's timer-triggered frames.
1204 if (
1205 packet.marker
1206 and self.__kind == "video"
1207 and frame_time - last_video_timing >= 0.2
1208 ):
1209 last_video_timing = frame_time
1210 arrival_ms = min(0xFFFF, max(0, int((time.time() - frame_time) * 1000)))
1211 packet.extensions.video_timing = video_timing_legs(
1212 enc_frame.timing, time.monotonic_ns(), arrival_ms)
1213 if described is not None and descriptor:
1214 packet.extensions.dependency_descriptor = dependency_descriptor(
1215 i == 0, bool(packet.marker), described[0], described[1], enc_frame.keyframe)
1216 if i == 0 and abs_capture_time is not None:
1217 packet.extensions.abs_capture_time = abs_capture_time
1218 self.__log_debug("> %s", packet)
1219 self.__rtp_history.add(
1220 packet, frame_time, enc_frame.dependency[0] if described is not None and descriptor else None)
1221 packet_bytes = packet.serialize(self.__rtp_header_extensions_map)
1222 outgoing.append((packet_bytes, packet.extensions.transport_sequence_number,
1223 packet.sequence_number, len(payload)))
1224 sequence_number = uint16_add(sequence_number, 1)
1226 if self.__fec_payload_type is not None:
1227 if not fec_group:
1228 fec_first_seq = packet.sequence_number
1229 fec_group.append(
1230 self.__rtp_header_extensions_map.for_fec(packet_bytes)
1231 )
1232 if packet.marker or len(fec_group) == FEC_GROUP_PACKETS:
1233 repairs = self._fec_repairs(len(fec_group))
1234 for repair in range(repairs):
1235 fec_bytes = build_flexfec_03(
1236 fec_group,
1237 fec_first_seq,
1238 self._ssrc,
1239 self.__fec_payload_type,
1240 self.__fec_sequence_number,
1241 packet.timestamp,
1242 self._fec_ssrc,
1243 range(repair, len(fec_group), repairs),
1244 )
1245 self.__fec_sequence_number = uint16_add(
1246 self.__fec_sequence_number, 1
1247 )
1248 outgoing.append((fec_bytes, None, None, 0))
1249 if repairs:
1250 protected.append((fec_first_seq, len(fec_group)))
1251 fec_group = []
1253 last = next((seq for _, seq, media, _ in reversed(outgoing)
1254 if media is not None and seq is not None), None)
1255 # A stand-in is no frame of the stream's: the pacer counts no frame for it, and
1256 # the peer's frame was told left out already.
1257 if last is not None and self.__kind == "video" and not standing_in:
1258 self._frame_handed = last
1259 captured = enc_frame.timing[0] if enc_frame.timing else 0
1260 self.transport.frame_end(last, self._frame_on_wire, last, enc_frame.keyframe,
1261 enc_frame.timing, captured,
1262 sum(len(p_) for p_ in enc_frame.payloads),
1263 enc_frame.dependency[0] if described is not None else None)
1264 if described is not None:
1265 self.__in_flight.append((last, enc_frame.dependency[0]))
1266 if enc_frame.keyframe:
1267 self._key_sent_at = time.monotonic()
1268 self.__unheld.clear()
1269 self.__held.clear()
1270 self._unheld_gap = False
1271 if self.on_key_sent is not None:
1272 self.on_key_sent()
1273 self.__unheld.append((
1274 *enc_frame.dependency, enc_frame.keyframe, time.monotonic(),
1275 [(seq, media) for _, seq, media, _ in outgoing if media is not None]))
1276 if self.on_frame_queued is not None:
1277 self.on_frame_queued()
1278 resyncs = self._resyncs
1279 for index, (packet_bytes, twcc_seq, media_seq, size) in enumerate(outgoing):
1280 if self._resyncs != resyncs:
1281 # A pacer reset cut this frame (`_pacer_resync`): the rest of it
1282 # would only queue in front of the frame that resyncs the peer,
1283 # and goes as the packets the pacer drops do.
1284 for _, seq, _, _ in outgoing[index:]:
1285 if seq is not None:
1286 self.transport._twcc_dropped(seq)
1287 if last is not None:
1288 self.transport.forget_frame_end(last)
1289 break
1290 if media_seq is not None:
1291 self.__last_sequence = media_seq
1292 await self._send(packet_bytes, twcc_seq)
1293 if media_seq is not None:
1294 self.__octet_count += size
1295 self.__packet_count += 1
1296 if protected:
1297 repaired = time.time() + self.transport._send_delay()
1298 for first, count in protected:
1299 for offset in range(count):
1300 self.__rtp_history.repaired(uint16_add(first, offset), repaired)
1301 self.__rtp_timestamp = timestamp
1302 self.__rtp_instant_ns = instant_ns
1303 self.__rtp_clock_rate = codec.clockRate
1304 except (asyncio.CancelledError, ConnectionError, MediaStreamError):
1305 pass
1306 except Exception:
1307 # we *need* to set __rtp_exited, otherwise RTCRtpSender.stop() will hang,
1308 # so issue a warning if we hit an unexpected exception
1309 self.__log_warning(traceback.format_exc())
1311 # stop track
1312 if self.__track:
1313 self.__track.stop()
1314 self.__track = None
1316 # release encoder
1317 self.__encoder = None
1319 self.__log_debug("- RTP finished")
1320 self.__rtp_exited.set()
1322 async def _run_rtcp(self) -> None:
1323 self.__log_debug("- RTCP started")
1324 self.__rtcp_started.set()
1326 try:
1327 while True:
1328 # The interval between RTCP packets is varied randomly over the
1329 # range [0.5, 1.5] times the calculated interval.
1330 await asyncio.sleep(0.5 + random.random())
1332 # RTCP SR
1333 ntp_timestamp, rtp_timestamp = self._sender_clock(time.monotonic_ns())
1334 packets: list[AnyRtcpPacket] = [
1335 RtcpSrPacket(
1336 ssrc=self._ssrc,
1337 sender_info=RtcpSenderInfo(
1338 ntp_timestamp=ntp_timestamp,
1339 rtp_timestamp=rtp_timestamp,
1340 packet_count=self.__packet_count & 0xFFFFFFFF,
1341 octet_count=self.__octet_count & 0xFFFFFFFF,
1342 ),
1343 )
1344 ]
1345 self.__lsr = (ntp_timestamp >> 16) & 0xFFFFFFFF
1346 self.__lsr_time = time.time()
1348 # RTCP SDES
1349 if self.__cname is not None:
1350 packets.append(
1351 RtcpSdesPacket(
1352 chunks=[
1353 RtcpSourceInfo(
1354 ssrc=self._ssrc,
1355 items=[(1, self.__cname.encode("utf8"))],
1356 )
1357 ]
1358 )
1359 )
1361 # RTCP XR DLRR, after the SDES as libwebrtc sends it
1362 if self.__rrtrs:
1363 packets.append(self._dlrr_report(time.monotonic_ns()))
1365 await self._send_rtcp(packets)
1366 except asyncio.CancelledError:
1367 pass
1369 # RTCP BYE
1370 packet = RtcpByePacket(sources=[self._ssrc])
1371 await self._send_rtcp([packet])
1373 self.__log_debug("- RTCP finished")
1374 self.__rtcp_exited.set()
1376 def _sender_clock(self, now_ns: int) -> tuple[int, int]:
1377 """The NTP timestamp of `now_ns` (CLOCK_MONOTONIC) and the RTP timestamp the
1378 media clock reads then, carried forward from the last frame sent (RFC 3550
1379 6.4.1): a report pairs the instant it is sent with the RTP clock at that
1380 instant, not with the last packet's timestamp, which stands for its capture.
1381 Zero before any frame, which a receiver takes for no mapping."""
1382 if not self.__rtp_instant_ns:
1383 return 0, 0
1384 elapsed = (now_ns - self.__rtp_instant_ns) * self.__rtp_clock_rate // 1_000_000_000
1385 return clock.ntp_from_monotonic_ns(now_ns), (self.__rtp_timestamp + elapsed) & 0xFFFFFFFF
1387 def _dlrr_report(self, now_ns: int) -> RtcpXrPacket:
1388 """The DLRR answering each reporter's last reference time, sent beside every sender
1389 report rather than once per reference time: libwebrtc drops the round trip it
1390 reports for a stream at a sender report arriving without one."""
1391 return RtcpXrPacket(
1392 ssrc=self._ssrc,
1393 dlrr=[
1394 (ssrc, lrr, (((now_ns - arrived_ns) << 16) // 1_000_000_000) & 0xFFFFFFFF)
1395 for ssrc, (lrr, arrived_ns) in self.__rrtrs.items()
1396 ],
1397 )
1399 async def _send_rtcp(self, packets: list[AnyRtcpPacket]) -> None:
1400 payload = b""
1401 for packet in packets:
1402 self.__log_debug("> %s", packet)
1403 payload += bytes(packet)
1405 try:
1406 await self.transport._send_rtp(payload)
1407 except ConnectionError:
1408 pass
1410 def __log_warning(self, msg: str, *args: object) -> None:
1411 logger.warning(f"RTCRtpsender(%s) {msg}", self.__kind, *args)