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 datetime
36import logging
37import random
38import time
39from collections.abc import Callable
40from dataclasses import dataclass
41from typing import Any, Optional
44from . import clock
45from .codecs import frame_assembler, depayload, get_capabilities, is_rtx
46from .exceptions import InvalidStateError
47from .jitterbuffer import JitterBuffer
48from .mediastreams import MediaStreamError, MediaStreamTrack
49from .rate import RemoteBitrateEstimator
50from .rtcdtlstransport import RTCDtlsTransport
51from .rtcrtpparameters import (
52 RTCRtpCapabilities,
53 RTCRtpCodecParameters,
54 RTCRtpReceiveParameters,
55)
56from .rtp import (
57 RTCP_PSFB_APP,
58 RTCP_PSFB_PLI,
59 RTCP_RTPFB_NACK,
60 RTP_HISTORY_SIZE,
61 AnyRtcpPacket,
62 RtcpByePacket,
63 RtcpPsfbPacket,
64 RtcpReceiverInfo,
65 RtcpRrPacket,
66 RtcpRtpfbPacket,
67 RtcpSrPacket,
68 RtpPacket,
69 clamp_packets_lost,
70 pack_remb_fci,
71 unwrap_rtx,
72)
73from .stats import (
74 RTCInboundRtpStreamStats,
75 RTCRemoteOutboundRtpStreamStats,
76 RTCStatsReport,
77)
78from .utils import uint16_add, uint16_gt
80logger = logging.getLogger(__name__)
83class NackGenerator:
84 def __init__(self) -> None:
85 self.max_seq: Optional[int] = None
86 self.missing: set[int] = set()
88 def add(self, packet: RtpPacket) -> bool:
89 """
90 Mark a new packet as received, and deduce missing packets.
91 """
92 missed = False
94 if self.max_seq is None:
95 self.max_seq = packet.sequence_number
96 return missed
98 # mark missing packets
99 if uint16_gt(packet.sequence_number, self.max_seq):
100 seq = uint16_add(self.max_seq, 1)
101 while uint16_gt(packet.sequence_number, seq):
102 self.missing.add(seq)
103 missed = True
104 seq = uint16_add(seq, 1)
105 self.max_seq = packet.sequence_number
106 else:
107 self.missing.discard(packet.sequence_number)
109 # limit number of tracked packets
110 self.truncate()
112 return missed
114 def truncate(self) -> None:
115 """
116 Limit the number of missing packets we track.
118 Otherwise, the size of RTCP FB messages grows indefinitely.
119 """
120 if self.max_seq is not None:
121 min_seq = uint16_add(self.max_seq, -RTP_HISTORY_SIZE)
122 for seq in list(self.missing):
123 if uint16_gt(min_seq, seq):
124 self.missing.discard(seq)
127class StreamStatistics:
128 def __init__(self, clockrate: int) -> None:
129 self.base_seq: Optional[int] = None
130 self.max_seq: Optional[int] = None
131 self.cycles = 0
132 self.packets_received = 0
134 # jitter
135 self._clockrate = clockrate
136 self._jitter_q4 = 0
137 self._last_arrival: Optional[int] = None
138 self._last_timestamp: Optional[int] = None
140 # fraction lost
141 self._expected_prior = 0
142 self._received_prior = 0
144 def add(self, packet: RtpPacket) -> None:
145 in_order = self.max_seq is None or uint16_gt(
146 packet.sequence_number, self.max_seq
147 )
148 self.packets_received += 1
150 if self.base_seq is None:
151 self.base_seq = packet.sequence_number
153 if in_order:
154 arrival = int(time.time() * self._clockrate)
156 if self.max_seq is not None and packet.sequence_number < self.max_seq:
157 self.cycles += 1 << 16
158 self.max_seq = packet.sequence_number
160 if packet.timestamp != self._last_timestamp and self.packets_received > 1:
161 diff = abs(
162 (arrival - self._last_arrival)
163 - (packet.timestamp - self._last_timestamp)
164 )
165 self._jitter_q4 += diff - ((self._jitter_q4 + 8) >> 4)
167 self._last_arrival = arrival
168 self._last_timestamp = packet.timestamp
170 @property
171 def fraction_lost(self) -> int:
172 expected_interval = self.packets_expected - self._expected_prior
173 self._expected_prior = self.packets_expected
174 received_interval = self.packets_received - self._received_prior
175 self._received_prior = self.packets_received
176 lost_interval = expected_interval - received_interval
177 if expected_interval == 0 or lost_interval <= 0:
178 return 0
179 else:
180 return (lost_interval << 8) // expected_interval
182 @property
183 def jitter(self) -> int:
184 return self._jitter_q4 >> 4
186 @property
187 def packets_expected(self) -> int:
188 return self.cycles + self.max_seq - self.base_seq + 1
190 @property
191 def packets_lost(self) -> int:
192 return clamp_packets_lost(self.packets_expected - self.packets_received)
195class RemoteStreamTrack(MediaStreamTrack):
196 def __init__(self, kind: str, id: Optional[str] = None) -> None:
197 super().__init__()
198 self.kind = kind
199 if id is not None:
200 self._id = id
201 self._queue: asyncio.Queue = asyncio.Queue()
203 async def recv(self) -> Any:
204 """
205 Receive the next frame.
206 """
207 if self.readyState != "live":
208 raise MediaStreamError
210 frame = await self._queue.get()
211 if frame is None:
212 self.stop()
213 raise MediaStreamError
214 return frame
217class TimestampMapper:
218 def __init__(self) -> None:
219 self._last: Optional[int] = None
220 self._origin: Optional[int] = None
222 def map(self, timestamp: int) -> int:
223 if self._origin is None:
224 # first timestamp
225 self._origin = timestamp
226 elif timestamp < self._last:
227 # RTP timestamp wrapped
228 self._origin -= 1 << 32
230 self._last = timestamp
231 return timestamp - self._origin
234@dataclass
235class RTCRtpContributingSource:
236 """
237 The :class:`RTCRtpContributingSource` dictionary contains information about
238 a contributing source (CSRC).
239 """
241 timestamp: datetime.datetime
242 "The timestamp associated with this source."
243 source: int
244 "The CSRC identifier associated with this source."
247@dataclass
248class RTCRtpSynchronizationSource:
249 """
250 The :class:`RTCRtpSynchronizationSource` dictionary contains information about
251 a synchronization source (SSRC).
252 """
254 timestamp: datetime.datetime
255 "The timestamp associated with this source."
256 source: int
257 "The SSRC identifier associated with this source."
260class RTCRtpReceiver:
261 """
262 The :class:`RTCRtpReceiver` interface manages the reception and decoding
263 of data for a :class:`MediaStreamTrack`.
265 :param kind: The kind of media (`'audio'` or `'video'`).
266 :param transport: An :class:`RTCDtlsTransport`.
267 """
269 def __init__(self, kind: str, transport: RTCDtlsTransport) -> None:
270 if transport.state == "closed":
271 raise InvalidStateError
273 self._enabled = True
274 self.__active_ssrc: dict[int, datetime.datetime] = {}
275 self.__codecs: dict[int, RTCRtpCodecParameters] = {}
276 # Optional encoded-frame sinks (mic and webcam uplinks): when set, complete
277 # encoded frames are handed here instead of the Python decoder, so pcmflux
278 # and pixelflux are the only media decoders in the system.
279 self._encoded_audio_sink = None
280 self._encoded_video_sink = None
281 self.__kind = kind
282 if kind == "audio":
283 self.__jitter_buffer = JitterBuffer(capacity=16, prefetch=4)
284 self.__nack_generator = None
285 self.__remote_bitrate_estimator = None
286 else:
287 self.__jitter_buffer = JitterBuffer(capacity=128, is_video=True)
288 self.__nack_generator = NackGenerator()
289 self.__remote_bitrate_estimator = RemoteBitrateEstimator()
290 self._track: Optional[RemoteStreamTrack] = None
291 self.__rtcp_exited = asyncio.Event()
292 self.__rtcp_started = asyncio.Event()
293 self.__rtcp_task: Optional[asyncio.Future[None]] = None
294 self.__rtx_ssrc: dict[int, int] = {}
295 self.__started = False
296 self.__stats = RTCStatsReport()
297 self.__timestamp_mapper = TimestampMapper()
298 self.__transport = transport
300 # RTCP
301 self.__lsr: dict[int, int] = {}
302 self.__lsr_time: dict[int, float] = {}
303 self.__remote_streams: dict[int, StreamStatistics] = {}
304 self.__rtcp_ssrc: Optional[int] = None
306 # logging
307 self.__log_debug: Callable[..., None] = lambda *args: None
308 if logger.isEnabledFor(logging.DEBUG):
309 self.__log_debug = lambda msg, *args: logger.debug(
310 f"RTCRtpReceiver(%s) {msg}", self.__kind, *args
311 )
313 @property
314 def track(self) -> MediaStreamTrack:
315 """
316 The :class:`MediaStreamTrack` which is being handled by the receiver.
317 """
318 return self._track
320 @property
321 def transport(self) -> RTCDtlsTransport:
322 """
323 The :class:`RTCDtlsTransport` over which the media for the receiver's
324 track is received.
325 """
326 return self.__transport
328 @classmethod
329 def getCapabilities(cls, kind: str) -> Optional[RTCRtpCapabilities]:
330 """
331 Returns the most optimistic view of the system's capabilities for
332 receiving media of the given `kind`.
334 :rtype: :class:`RTCRtpCapabilities`
335 """
336 return get_capabilities(kind)
338 async def getStats(self) -> RTCStatsReport:
339 """
340 Returns statistics about the RTP receiver.
342 :rtype: :class:`RTCStatsReport`
343 """
344 for ssrc, stream in self.__remote_streams.items():
345 self.__stats.add(
346 RTCInboundRtpStreamStats(
347 # RTCStats
348 timestamp=clock.current_datetime(),
349 type="inbound-rtp",
350 id="inbound-rtp_" + str(id(self)),
351 # RTCStreamStats
352 ssrc=ssrc,
353 kind=self.__kind,
354 transportId=self.transport._stats_id,
355 # RTCReceivedRtpStreamStats
356 packetsReceived=stream.packets_received,
357 packetsLost=stream.packets_lost,
358 jitter=stream.jitter,
359 # RTPInboundRtpStreamStats
360 )
361 )
362 self.__stats.update(self.transport._get_stats())
364 return self.__stats
366 def getSynchronizationSources(self) -> list[RTCRtpSynchronizationSource]:
367 """
368 Returns a :class:`RTCRtpSynchronizationSource` for each unique SSRC identifier
369 received in the last 10 seconds.
370 """
371 cutoff = clock.current_datetime() - datetime.timedelta(seconds=10)
372 sources = []
373 for source, timestamp in self.__active_ssrc.items():
374 if timestamp >= cutoff:
375 sources.append(
376 RTCRtpSynchronizationSource(source=source, timestamp=timestamp)
377 )
378 return sources
380 async def receive(self, parameters: RTCRtpReceiveParameters) -> None:
381 """
382 Attempt to set the parameters controlling the receiving of media.
384 :param parameters: The :class:`RTCRtpParameters` for the receiver.
385 """
386 if not self.__started:
387 for codec in parameters.codecs:
388 self.__codecs[codec.payloadType] = codec
389 for encoding in parameters.encodings:
390 if encoding.rtx:
391 self.__rtx_ssrc[encoding.rtx.ssrc] = encoding.ssrc
393 # Received media is handed off encoded: audio to pcmflux, video to
394 # pixelflux, both of which decode off the GIL. A receiver with no
395 # encoded-frame sink has nothing to decode it in-process.
396 self.__transport._register_rtp_receiver(self, parameters)
397 self.__rtcp_task = asyncio.ensure_future(self._run_rtcp())
398 self.__started = True
400 async def request_keyframe(self) -> None:
401 """
402 Ask the sender for a keyframe (RTCP PLI) on every SSRC currently
403 delivering to this receiver; used by an encoded-frame sink whose
404 decoder lost its reference.
405 """
406 for ssrc in list(self.__active_ssrc.keys()):
407 try:
408 await self._send_rtcp_pli(ssrc)
409 except Exception:
410 pass
412 def setTransport(self, transport: RTCDtlsTransport) -> None:
413 self.__transport = transport
415 async def stop(self) -> None:
416 """
417 Irreversibly stop the receiver.
418 """
419 if self.__started:
420 self.__transport._unregister_rtp_receiver(self)
422 # shutdown RTCP task
423 await self.__rtcp_started.wait()
424 self.__rtcp_task.cancel()
425 await self.__rtcp_exited.wait()
427 def _handle_disconnect(self) -> None:
428 # End the media track (unused on Selkies' encoded-sink path) on disconnect.
429 if self._track is not None:
430 self._track.stop()
432 async def _handle_rtcp_packet(self, packet: AnyRtcpPacket) -> None:
433 self.__log_debug("< %s", packet)
435 if isinstance(packet, RtcpSrPacket):
436 self.__stats.add(
437 RTCRemoteOutboundRtpStreamStats(
438 # RTCStats
439 timestamp=clock.current_datetime(),
440 type="remote-outbound-rtp",
441 id=f"remote-outbound-rtp_{id(self)}",
442 # RTCStreamStats
443 ssrc=packet.ssrc,
444 kind=self.__kind,
445 transportId=self.transport._stats_id,
446 # RTCSentRtpStreamStats
447 packetsSent=packet.sender_info.packet_count,
448 bytesSent=packet.sender_info.octet_count,
449 # RTCRemoteOutboundRtpStreamStats
450 remoteTimestamp=clock.datetime_from_ntp(
451 packet.sender_info.ntp_timestamp
452 ),
453 )
454 )
455 self.__lsr[packet.ssrc] = (
456 (packet.sender_info.ntp_timestamp) >> 16
457 ) & 0xFFFFFFFF
458 self.__lsr_time[packet.ssrc] = time.time()
459 elif isinstance(packet, RtcpByePacket):
460 # Sender is done: end the media track (inert on the encoded-sink path).
461 if self._track is not None:
462 self._track.stop()
464 async def _handle_rtp_packet(self, packet: RtpPacket, arrival_time_ms: int) -> None:
465 """
466 Handle an incoming RTP packet.
467 """
468 self.__log_debug("< %s", packet)
470 # If the receiver is disabled, discard the packet.
471 if not self._enabled:
472 return
474 # answer transport-wide congestion control
475 if packet.extensions.transport_sequence_number is not None:
476 await self.transport._twcc_received(
477 packet.extensions.transport_sequence_number,
478 arrival_time_ms,
479 packet.ssrc,
480 self.__rtcp_ssrc,
481 )
483 # feed bitrate estimator
484 if self.__remote_bitrate_estimator is not None:
485 if packet.extensions.abs_send_time is not None:
486 remb = self.__remote_bitrate_estimator.add(
487 abs_send_time=packet.extensions.abs_send_time,
488 arrival_time_ms=arrival_time_ms,
489 payload_size=len(packet.payload) + packet.padding_size,
490 ssrc=packet.ssrc,
491 )
492 if self.__rtcp_ssrc is not None and remb is not None:
493 # send Receiver Estimated Maximum Bitrate feedback
494 rtcp_packet = RtcpPsfbPacket(
495 fmt=RTCP_PSFB_APP,
496 ssrc=self.__rtcp_ssrc,
497 media_ssrc=0,
498 fci=pack_remb_fci(*remb),
499 )
500 await self._send_rtcp(rtcp_packet)
502 # keep track of sources
503 self.__active_ssrc[packet.ssrc] = clock.current_datetime()
505 # check the codec is known
506 codec = self.__codecs.get(packet.payload_type)
507 if codec is None:
508 self.__log_debug(
509 "x RTP packet with unknown payload type %d", packet.payload_type
510 )
511 return
513 # feed RTCP statistics
514 if packet.ssrc not in self.__remote_streams:
515 self.__remote_streams[packet.ssrc] = StreamStatistics(codec.clockRate)
516 self.__remote_streams[packet.ssrc].add(packet)
518 # A FlexFEC repair stream numbers its own packets: fed to the NACK
519 # generator it would name holes the media stream never had, and nothing
520 # here recovers a frame from it.
521 if codec.mimeType.lower() == "video/flexfec-03":
522 return
524 # unwrap retransmission packet
525 if is_rtx(codec):
526 original_ssrc = self.__rtx_ssrc.get(packet.ssrc)
527 if original_ssrc is None:
528 self.__log_debug("x RTX packet from unknown SSRC %d", packet.ssrc)
529 return
531 apt = codec.parameters.get("apt")
532 if (
533 len(packet.payload) < 2
534 or not isinstance(apt, int)
535 or apt not in self.__codecs
536 ):
537 return
539 packet = unwrap_rtx(packet, payload_type=apt, ssrc=original_ssrc)
540 codec = self.__codecs[apt]
542 # send NACKs for any missing any packets
543 if self.__nack_generator is not None and self.__nack_generator.add(packet):
544 await self._send_rtcp_nack(
545 packet.ssrc, sorted(self.__nack_generator.missing)
546 )
548 # parse codec-specific information
549 try:
550 if packet.payload:
551 packet._data = depayload(codec, packet.payload) # type: ignore
552 else:
553 packet._data = b"" # type: ignore
554 packet._assemble = frame_assembler(codec) # type: ignore
555 except ValueError as exc:
556 self.__log_debug("x RTP payload parsing failed: %s", exc)
557 return
559 # try to re-assemble encoded frame
560 pli_flag, encoded_frame = self.__jitter_buffer.add(packet)
561 # check if the PLI should be sent
562 if pli_flag:
563 await self._send_rtcp_pli(packet.ssrc)
565 # if we have a complete encoded frame, hand it off
566 if encoded_frame is not None:
567 encoded_frame.timestamp = self.__timestamp_mapper.map(
568 encoded_frame.timestamp
569 )
570 if self._encoded_audio_sink is not None:
571 # Forward the encoded payload (bare Opus, or RED to de-frame downstream)
572 # rather than decoding it in Python.
573 self._encoded_audio_sink(codec, encoded_frame)
574 elif self._encoded_video_sink is not None:
575 # Same for video: the depacketized frame (Annex-B H.264 or H.265,
576 # VP8/VP9, an AV1 temporal unit) goes to the virtual webcam's decoder.
577 self._encoded_video_sink(codec, encoded_frame)
579 async def _run_rtcp(self) -> None:
580 self.__log_debug("- RTCP started")
581 self.__rtcp_started.set()
583 try:
584 while True:
585 # The interval between RTCP packets is varied randomly over the
586 # range [0.5, 1.5] times the calculated interval.
587 await asyncio.sleep(0.5 + random.random())
589 # RTCP RR
590 reports = []
591 for ssrc, stream in self.__remote_streams.items():
592 lsr = 0
593 dlsr = 0
594 if ssrc in self.__lsr:
595 lsr = self.__lsr[ssrc]
596 delay = time.time() - self.__lsr_time[ssrc]
597 if delay > 0 and delay < 65536:
598 dlsr = int(delay * 65536)
600 reports.append(
601 RtcpReceiverInfo(
602 ssrc=ssrc,
603 fraction_lost=stream.fraction_lost,
604 packets_lost=stream.packets_lost,
605 highest_sequence=stream.max_seq,
606 jitter=stream.jitter,
607 lsr=lsr,
608 dlsr=dlsr,
609 )
610 )
612 if self.__rtcp_ssrc is not None and reports:
613 packet = RtcpRrPacket(ssrc=self.__rtcp_ssrc, reports=reports)
614 await self._send_rtcp(packet)
616 except asyncio.CancelledError:
617 pass
619 self.__log_debug("- RTCP finished")
620 self.__rtcp_exited.set()
622 async def _send_rtcp(self, packet: AnyRtcpPacket) -> None:
623 self.__log_debug("> %s", packet)
624 try:
625 await self.transport._send_rtp(bytes(packet))
626 except ConnectionError:
627 pass
629 async def _send_rtcp_nack(self, media_ssrc: int, lost: list[int]) -> None:
630 """
631 Send an RTCP packet to report missing RTP packets.
632 """
633 if self.__rtcp_ssrc is not None:
634 packet = RtcpRtpfbPacket(
635 fmt=RTCP_RTPFB_NACK, ssrc=self.__rtcp_ssrc, media_ssrc=media_ssrc
636 )
637 packet.lost = lost
638 await self._send_rtcp(packet)
640 async def _send_rtcp_pli(self, media_ssrc: int) -> None:
641 """
642 Send an RTCP packet to report picture loss.
643 """
644 if self.__rtcp_ssrc is not None:
645 packet = RtcpPsfbPacket(
646 fmt=RTCP_PSFB_PLI, ssrc=self.__rtcp_ssrc, media_ssrc=media_ssrc
647 )
648 await self._send_rtcp(packet)
650 def _set_rtcp_ssrc(self, ssrc: int) -> None:
651 self.__rtcp_ssrc = ssrc