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 copy
36import logging
37import uuid
38from typing import Optional, Union
40from pyee.asyncio import AsyncIOEventEmitter
42from . import clock, rtp, sdp
43from .codecs import CODECS, HEADER_EXTENSIONS, is_red, is_rtx, red_block_payload_type
44from .events import RTCTrackEvent
45from .exceptions import (
46 InternalError,
47 InvalidAccessError,
48 InvalidStateError,
49 OperationError,
50)
51from .mediastreams import MediaStreamTrack
52from .rtcconfiguration import RTCBundlePolicy, RTCConfiguration
53from .rtcdatachannel import RTCDataChannel, RTCDataChannelParameters
54from .rtcdtlstransport import RTCCertificate, RTCDtlsParameters, RTCDtlsTransport
55from .rtcicetransport import (
56 RTCIceCandidate,
57 RTCIceGatherer,
58 RTCIceParameters,
59 RTCIceTransport,
60)
61from .rtcrtpparameters import (
62 RTCRtpCodecCapability,
63 RTCRtpCodecParameters,
64 RTCRtpDecodingParameters,
65 RTCRtpHeaderExtensionParameters,
66 RTCRtpParameters,
67 RTCRtpReceiveParameters,
68 RTCRtpRtxParameters,
69 RTCRtpSendParameters,
70)
71from .rtcrtpreceiver import RemoteStreamTrack, RTCRtpReceiver
72from .rtcrtpsender import RTCRtpSender
73from .rtcrtptransceiver import RTCRtpTransceiver
74from .rtcsctptransport import RTCSctpCapabilities, RTCSctpTransport
75from .rtcsessiondescription import RTCSessionDescription
76from .stats import RTCStatsReport
78DISCARD_HOST = "0.0.0.0"
79DISCARD_PORT = 9
80MEDIA_KINDS = ["audio", "video"]
82logger = logging.getLogger(__name__)
85def filter_preferred_codecs(
86 codecs: list[RTCRtpCodecParameters], preferred: list[RTCRtpCodecCapability]
87) -> list[RTCRtpCodecParameters]:
88 if not preferred:
89 return codecs
91 rtx_codecs = list(filter(is_rtx, codecs))
92 rtx_enabled = next(filter(is_rtx, preferred), None) is not None
94 filtered = []
95 for pref in filter(lambda x: not is_rtx(x), preferred):
96 for codec in codecs:
97 if (
98 codec.mimeType.lower() == pref.mimeType.lower()
99 and codec.parameters == pref.parameters
100 ):
101 filtered.append(codec)
103 # add corresponding RTX
104 if rtx_enabled:
105 for rtx in rtx_codecs:
106 if rtx.parameters["apt"] == codec.payloadType:
107 filtered.append(rtx)
108 break
110 break
112 # Drop audio/red whose inner opus payload type was not negotiated, so we
113 # never keep a redundancy codec without the codec it wraps (mirrors rtx/apt).
114 present_pts = {c.payloadType for c in filtered if not is_red(c)}
115 filtered = [
116 c
117 for c in filtered
118 if not is_red(c) or red_block_payload_type(c.parameters) in present_pts
119 ]
121 return filtered
124def receive_only_payload_types() -> dict[str, dict[int, int]]:
125 """The payload types a section only the browser encodes for (the webcam, the microphone)
126 takes in place of the shared ones, by kind: payload types no codec of either kind holds,
127 the dynamic range first for video and the unassigned one (35-63) from the top for audio,
128 static ones (G.722, PCMU, PCMA) too. Such a section's codecs differ in parameters from the
129 display's and the playback's (profiles, sps-pps-idr-in-keyframe, the client's stereo, the
130 playback's ptime, which Chromium folds into every codec of its section), and one payload
131 type with two sets of parameters in a BUNDLE group is a collision Chromium logs (and may
132 one day refuse)."""
133 used = {c.payloadType for kind in CODECS for c in CODECS[kind]}
134 mapping: dict[str, dict[int, int]] = {"video": {}, "audio": {}}
135 free_video = [pt for pt in (*rtp.DYNAMIC_PAYLOAD_TYPES, *rtp.UNASSIGNED_PAYLOAD_TYPES) if pt not in used]
136 for c in CODECS["video"]:
137 if c.payloadType not in mapping["video"]:
138 mapping["video"][c.payloadType] = free_video.pop(0)
139 taken = used | set(mapping["video"].values())
140 free_audio = [pt for pt in reversed(rtp.UNASSIGNED_PAYLOAD_TYPES) if pt not in taken]
141 for c in CODECS["audio"]:
142 if c.payloadType not in mapping["audio"]:
143 mapping["audio"][c.payloadType] = free_audio.pop(0)
144 return mapping
147def with_payload_types(
148 codec: RTCRtpCodecParameters, mapping: dict[int, int]
149) -> RTCRtpCodecParameters:
150 """`codec` under the payload types `mapping` gives, the one an RTX or RED codec names
151 with it."""
152 codec = copy.deepcopy(codec)
153 codec.payloadType = mapping.get(codec.payloadType, codec.payloadType)
154 if is_rtx(codec) and isinstance(codec.parameters.get("apt"), int):
155 codec.parameters["apt"] = mapping.get(codec.parameters["apt"], codec.parameters["apt"])
156 elif is_red(codec):
157 codec.parameters = {
158 "/".join(str(mapping.get(int(pt), int(pt))) if pt.isdigit() else pt
159 for pt in str(key).split("/")): value
160 for key, value in codec.parameters.items()
161 }
162 return codec
165def find_common_codecs(
166 local_codecs: list[RTCRtpCodecParameters],
167 remote_codecs: list[RTCRtpCodecParameters],
168) -> list[RTCRtpCodecParameters]:
169 common = []
170 common_base: dict[int, RTCRtpCodecParameters] = {}
171 for c in remote_codecs:
172 # for RTX, check we accepted the base codec
173 if is_rtx(c):
174 apt = c.parameters.get("apt")
175 if isinstance(apt, int) and apt in common_base:
176 base = common_base[apt]
177 if c.clockRate == base.clockRate:
178 common.append(copy.deepcopy(c))
179 continue
181 # handle other codecs
182 for codec in local_codecs:
183 if is_codec_compatible(codec, c):
184 codec = copy.deepcopy(codec)
185 if c.payloadType in rtp.DYNAMIC_PAYLOAD_TYPES or c.payloadType in rtp.UNASSIGNED_PAYLOAD_TYPES:
186 codec.payloadType = c.payloadType
187 if is_red(codec):
188 # The payload type of the codec it wraps, as negotiated.
189 codec.parameters = dict(c.parameters)
190 codec.rtcpFeedback = list(
191 filter(lambda x: x in c.rtcpFeedback, codec.rtcpFeedback)
192 )
193 common.append(codec)
194 common_base[codec.payloadType] = codec
195 break
196 return common
199def find_common_header_extensions(
200 local_extensions: list[RTCRtpHeaderExtensionParameters],
201 remote_extensions: list[RTCRtpHeaderExtensionParameters],
202) -> list[RTCRtpHeaderExtensionParameters]:
203 common = []
204 for rx in remote_extensions:
205 for lx in local_extensions:
206 if lx.uri == rx.uri:
207 common.append(rx)
208 return common
211def is_codec_compatible(a: RTCRtpCodecParameters, b: RTCRtpCodecParameters) -> bool:
212 if a.mimeType.lower() != b.mimeType.lower() or a.clockRate != b.clockRate:
213 return False
215 if a.mimeType.lower() == "video/h264":
217 def packetization(c: RTCRtpCodecParameters) -> int:
218 return int(c.parameters.get("packetization-mode", "0"))
220 def profile(c: RTCRtpCodecParameters) -> sdp.H264Profile:
221 # for backwards compatibility with older versions of WebRTC,
222 # consider the absence of a profile-level-id parameter to mean
223 # "constrained baseline level 3.1"
224 return sdp.parse_h264_profile_level_id(
225 str(c.parameters.get("profile-level-id", "42E01F"))
226 )[0]
228 if packetization(a) != packetization(b):
229 return False
230 try:
231 return profile(a) == profile(b)
232 except ValueError:
233 # Profiles the parser doesn't model (e.g. High 4:4:4 f4001f used by
234 # full-color streams): the sender payloads the encoder's bitstream
235 # as-is, so byte-equal raw ids are compatible.
236 def raw_id(c: RTCRtpCodecParameters) -> str:
237 return str(c.parameters.get("profile-level-id", "42E01F")).lower()
239 return raw_id(a) == raw_id(b)
241 return True
244def add_transport_description(
245 media: sdp.MediaDescription, dtlsTransport: RTCDtlsTransport
246) -> None:
247 # ice
248 iceTransport = dtlsTransport.transport
249 iceGatherer = iceTransport.iceGatherer
250 media.ice_candidates = iceGatherer.getLocalCandidates()
251 media.ice_candidates_complete = iceGatherer.state == "completed"
252 media.ice = iceGatherer.getLocalParameters()
253 if media.ice_candidates:
254 media.host = media.ice_candidates[0].ip
255 media.port = media.ice_candidates[0].port
256 else:
257 media.host = DISCARD_HOST
258 media.port = DISCARD_PORT
260 # dtls
261 if media.dtls is None:
262 media.dtls = dtlsTransport.getLocalParameters()
263 else:
264 media.dtls.fingerprints = dtlsTransport.getLocalParameters().fingerprints
267async def add_remote_candidates(
268 iceTransport: RTCIceTransport, media: sdp.MediaDescription
269) -> None:
270 coros = map(iceTransport.addRemoteCandidate, media.ice_candidates)
271 await asyncio.gather(*coros)
273 if media.ice_candidates_complete:
274 await iceTransport.addRemoteCandidate(None)
277def allocate_mid(mids: set[str]) -> str:
278 """
279 Allocate a MID which has not been used yet.
280 """
281 i = 0
282 while True:
283 mid = str(i)
284 if mid not in mids:
285 mids.add(mid)
286 return mid
287 i += 1
290def create_media_description_for_sctp(
291 sctp: RTCSctpTransport, legacy: bool, mid: str
292) -> sdp.MediaDescription:
293 if legacy:
294 media = sdp.MediaDescription(
295 kind="application", port=DISCARD_PORT, profile="DTLS/SCTP", fmt=[sctp.port]
296 )
297 media.sctpmap[sctp.port] = f"webrtc-datachannel {sctp._outbound_streams_count}"
298 else:
299 media = sdp.MediaDescription(
300 kind="application",
301 port=DISCARD_PORT,
302 profile="UDP/DTLS/SCTP",
303 fmt=["webrtc-datachannel"],
304 )
305 media.sctp_port = sctp.port
307 media.rtp.muxId = mid
308 media.sctpCapabilities = sctp.getCapabilities()
309 add_transport_description(media, sctp.transport)
311 return media
314def create_media_description_for_transceiver(
315 transceiver: RTCRtpTransceiver, cname: str, direction: str, mid: str
316) -> sdp.MediaDescription:
317 media = sdp.MediaDescription(
318 kind=transceiver.kind,
319 port=DISCARD_PORT,
320 profile="UDP/TLS/RTP/SAVPF",
321 fmt=[c.payloadType for c in transceiver._codecs],
322 )
323 media.direction = direction
324 media.msid = f"{transceiver.sender._stream_id} {transceiver.sender._track_id}"
326 media.rtp = RTCRtpParameters(
327 codecs=transceiver._codecs,
328 headerExtensions=transceiver._headerExtensions,
329 muxId=mid,
330 )
331 media.rtcp_host = DISCARD_HOST
332 media.rtcp_port = DISCARD_PORT
333 media.rtcp_mux = True
334 media.ssrc = [sdp.SsrcDescription(ssrc=transceiver.sender._ssrc, cname=cname)]
336 # if RTX is enabled, add corresponding SSRC
337 if next(filter(is_rtx, media.rtp.codecs), None):
338 media.ssrc.append(
339 sdp.SsrcDescription(ssrc=transceiver.sender._rtx_ssrc, cname=cname)
340 )
341 media.ssrc_group.append(
342 sdp.GroupDescription(
343 semantic="FID",
344 items=[transceiver.sender._ssrc, transceiver.sender._rtx_ssrc],
345 )
346 )
348 # if FlexFEC is on the table, announce the repair stream (FEC-FR group) and
349 # declare our send resolution envelope (RFC 6236) for the video line.
350 if next(
351 (c for c in media.rtp.codecs if c.mimeType.lower() == "video/flexfec-03"), None
352 ):
353 media.ssrc.append(
354 sdp.SsrcDescription(ssrc=transceiver.sender._fec_ssrc, cname=cname)
355 )
356 media.ssrc_group.append(
357 sdp.GroupDescription(
358 semantic="FEC-FR",
359 items=[transceiver.sender._ssrc, transceiver.sender._fec_ssrc],
360 )
361 )
362 if transceiver.kind == "video":
363 media.imageattr = ["* send [x=[16:8192],y=[16:8192]] recv *"]
365 add_transport_description(media, transceiver.receiver.transport)
367 return media
370def and_direction(a: str, b: str) -> str:
371 return sdp.DIRECTIONS[sdp.DIRECTIONS.index(a) & sdp.DIRECTIONS.index(b)]
374def or_direction(a: str, b: str) -> str:
375 return sdp.DIRECTIONS[sdp.DIRECTIONS.index(a) | sdp.DIRECTIONS.index(b)]
378def reverse_direction(direction: str) -> str:
379 if direction == "sendonly":
380 return "recvonly"
381 elif direction == "recvonly":
382 return "sendonly"
383 return direction
386def wrap_session_description(
387 session_description: Optional[sdp.SessionDescription],
388) -> Optional[RTCSessionDescription]:
389 if session_description is not None:
390 return RTCSessionDescription(
391 sdp=str(session_description), type=session_description.type
392 )
393 return None
396class RTCPeerConnection(AsyncIOEventEmitter):
397 """
398 The :class:`RTCPeerConnection` interface represents a WebRTC connection
399 between the local computer and a remote peer.
401 :param configuration: An optional :class:`RTCConfiguration`.
402 """
404 def __init__(self, configuration: Optional[RTCConfiguration] = None) -> None:
405 super().__init__()
406 self.__certificates = [RTCCertificate.generateCertificate()]
407 self.__cname = f"{uuid.uuid4()}"
408 self.__configuration = configuration or RTCConfiguration()
409 self.__dtlsTransports: set[RTCDtlsTransport] = set()
410 self.__iceTransports: set[RTCIceTransport] = set()
411 self.__remoteDtls: dict[
412 Union[RTCRtpTransceiver, RTCSctpTransport], RTCDtlsParameters
413 ] = {}
414 self.__remoteIce: dict[
415 Union[RTCRtpTransceiver, RTCSctpTransport], RTCIceParameters
416 ] = {}
417 self.__seenMids: set[str] = set()
418 self.__sctp: Optional[RTCSctpTransport] = None
419 self.__sctp_mline_index: Optional[int] = None
420 # Offer the standards SCTP SDP (UDP/DTLS/SCTP + a=sctp-port +
421 # a=max-message-size): Chromium ignores a=max-message-size on the legacy
422 # DTLS/SCTP form and caps data channel messages at 64 KiB. Answers still
423 # adapt to whichever form the peer offered.
424 self._sctpLegacySdp = False
425 self.__sctpRemotePort: Optional[int] = None
426 self.__sctpRemoteCaps: Optional[RTCSctpCapabilities] = None
427 self.__stream_id = str(uuid.uuid4())
428 self.__transceivers: list[RTCRtpTransceiver] = []
430 self.__closeTask: Optional[asyncio.Task] = None
431 self.__connectionState = "new"
432 self.__iceConnectionState = "new"
433 self.__iceGatheringState = "new"
434 self.__isClosed: Optional[asyncio.Future[bool]] = None
435 self.__signalingState = "stable"
437 self.__currentLocalDescription: Optional[sdp.SessionDescription] = None
438 self.__currentRemoteDescription: Optional[sdp.SessionDescription] = None
439 self.__pendingLocalDescription: Optional[sdp.SessionDescription] = None
440 self.__pendingRemoteDescription: Optional[sdp.SessionDescription] = None
442 @property
443 def connectionState(self) -> str:
444 """
445 The current connection state.
447 Possible values: `"connected"`, `"connecting"`, `"closed"`, `"failed"`, `"new`".
449 When the state changes, the `"connectionstatechange"` event is fired.
450 """
451 return self.__connectionState
453 @property
454 def iceConnectionState(self) -> str:
455 """
456 The current ICE connection state.
458 Possible values: `"checking"`, `"completed"`, `"closed"`, `"failed"`, `"new`".
460 When the state changes, the `"iceconnectionstatechange"` event is fired.
461 """
462 return self.__iceConnectionState
464 @property
465 def iceGatheringState(self) -> str:
466 """
467 The current ICE gathering state.
469 Possible values: `"complete"`, `"gathering"`, `"new`".
471 When the state changes, the `"icegatheringstatechange"` event is fired.
472 """
473 return self.__iceGatheringState
475 @property
476 def localDescription(self) -> RTCSessionDescription:
477 """
478 An :class:`RTCSessionDescription` describing the session for
479 the local end of the connection.
480 """
481 return wrap_session_description(self.__localDescription())
483 @property
484 def remoteDescription(self) -> RTCSessionDescription:
485 """
486 An :class:`RTCSessionDescription` describing the session for
487 the remote end of the connection.
488 """
489 return wrap_session_description(self.__remoteDescription())
491 @property
492 def sctp(self) -> Optional[RTCSctpTransport]:
493 """
494 An :class:`RTCSctpTransport` describing the SCTP transport being used
495 for datachannels or `None`.
496 """
497 return self.__sctp
499 @property
500 def signalingState(self) -> str:
501 """
502 The current signaling state.
504 Possible values: `"closed"`, `"have-local-offer"`, `"have-remote-offer`",
505 `"stable"`.
507 When the state changes, the `"signalingstatechange"` event is fired.
508 """
509 return self.__signalingState
511 async def addIceCandidate(self, candidate: Optional[RTCIceCandidate]) -> None:
512 """
513 Add a new :class:`RTCIceCandidate` received from the remote peer.
515 The specified candidate must have a value for either `sdpMid` or
516 `sdpMLineIndex`.
518 :param candidate: The new remote candidate or `None` to signal
519 end-of-candidates.
520 """
521 if (
522 candidate is not None
523 and candidate.sdpMid is None
524 and candidate.sdpMLineIndex is None
525 ):
526 raise ValueError("Candidate must have either sdpMid or sdpMLineIndex")
528 for transceiver in self.__transceivers:
529 if (
530 candidate is None
531 or (
532 candidate.sdpMid == transceiver.mid
533 or candidate.sdpMLineIndex == transceiver._get_mline_index()
534 )
535 and not transceiver._bundled
536 ):
537 iceTransport = transceiver.receiver.transport.transport
538 await iceTransport.addRemoteCandidate(candidate)
540 if self.__sctp and (
541 candidate is None
542 or (
543 candidate.sdpMid == self.__sctp.mid
544 or candidate.sdpMLineIndex == self.__sctp_mline_index
545 )
546 and not self.__sctp._bundled
547 ):
548 iceTransport = self.__sctp.transport.transport
549 await iceTransport.addRemoteCandidate(candidate)
551 # Update the remote description if it exists.
552 # The spec forbids adding candidates prior to setRemoteDescription
553 # but this used to work in previous versions.
554 if remote_description := self.__remoteDescription():
555 media = remote_description.media
556 for sdp_m_line_index in range(0, len(media)):
557 if candidate is None:
558 media[sdp_m_line_index].ice_candidates_complete = True
559 elif (
560 candidate.sdpMLineIndex == sdp_m_line_index
561 or candidate.sdpMid == media[sdp_m_line_index].rtp.muxId
562 ):
563 media[sdp_m_line_index].ice_candidates.append(candidate)
564 else:
565 logger.warning(
566 "RTCPeerConnection addIceCandidate called without remote description"
567 )
569 def addTrack(self, track: MediaStreamTrack) -> RTCRtpSender:
570 """
571 Add a :class:`MediaStreamTrack` to the set of media tracks which
572 will be transmitted to the remote peer.
573 """
574 # check state is valid
575 self.__assertNotClosed()
576 if track.kind not in ["audio", "video"]:
577 raise InternalError(f'Invalid track kind "{track.kind}"')
579 # don't add track twice
580 self.__assertTrackHasNoSender(track)
582 for transceiver in self.__transceivers:
583 if transceiver.kind == track.kind:
584 if transceiver.sender.track is None:
585 transceiver.sender.replaceTrack(track)
586 transceiver.direction = or_direction(
587 transceiver.direction, "sendonly"
588 )
589 return transceiver.sender
591 transceiver = self.__createTransceiver(
592 direction="sendrecv", kind=track.kind, sender_track=track
593 )
594 return transceiver.sender
596 def addTransceiver(
597 self, trackOrKind: Union[str, MediaStreamTrack], direction: str = "sendrecv"
598 ) -> RTCRtpTransceiver:
599 """
600 Add a new :class:`RTCRtpTransceiver`.
601 """
602 self.__assertNotClosed()
604 # determine track or kind
605 if isinstance(trackOrKind, MediaStreamTrack):
606 kind = trackOrKind.kind
607 track = trackOrKind
608 else:
609 kind = trackOrKind
610 track = None
611 if kind not in ["audio", "video"]:
612 raise InternalError(f'Invalid track kind "{kind}"')
614 # check direction
615 if direction not in sdp.DIRECTIONS:
616 raise InternalError(f'Invalid direction "{direction}"')
618 # don't add track twice
619 if track:
620 self.__assertTrackHasNoSender(track)
622 return self.__createTransceiver(
623 direction=direction, kind=kind, sender_track=track
624 )
626 async def close(self) -> None:
627 """
628 Terminate the ICE agent, ending ICE processing and streams.
629 """
630 if self.__isClosed:
631 await self.__isClosed
632 return
633 self.__isClosed = asyncio.Future()
634 self.__setSignalingState("closed")
636 # stop senders / receivers
637 for transceiver in self.__transceivers:
638 await transceiver.stop()
639 if self.__sctp:
640 await self.__sctp.stop()
642 # stop transports
643 for transceiver in self.__transceivers:
644 await transceiver.receiver.transport.stop()
645 await transceiver.receiver.transport.transport.stop()
646 if self.__sctp:
647 await self.__sctp.transport.stop()
648 await self.__sctp.transport.transport.stop()
650 # update states
651 self.__updateIceGatheringState()
652 self.__updateIceConnectionState()
653 self.__updateConnectionState()
655 # no more events will be emitted, so remove all event listeners
656 # to facilitate garbage collection.
657 self.remove_all_listeners()
659 self.__isClosed.set_result(True)
661 async def createAnswer(self) -> RTCSessionDescription:
662 """
663 Create an SDP answer to an offer received from a remote peer during
664 the offer/answer negotiation of a WebRTC connection.
666 :rtype: :class:`RTCSessionDescription`
667 """
668 # check state is valid
669 self.__assertNotClosed()
670 if self.signalingState not in ["have-remote-offer", "have-local-pranswer"]:
671 raise InvalidStateError(
672 f'Cannot create answer in signaling state "{self.signalingState}"'
673 )
675 # create description
676 ntp_seconds = clock.current_ntp_time() >> 32
677 description = sdp.SessionDescription()
678 description.origin = f"- {ntp_seconds} {ntp_seconds} IN IP4 0.0.0.0"
679 description.msid_semantic.append(
680 sdp.GroupDescription(semantic="WMS", items=["*"])
681 )
682 description.type = "answer"
684 for remote_m in self.__remoteDescription().media:
685 if remote_m.kind in ["audio", "video"]:
686 transceiver = self.__getTransceiverByMid(remote_m.rtp.muxId)
687 media = create_media_description_for_transceiver(
688 transceiver,
689 cname=self.__cname,
690 direction=and_direction(
691 transceiver.direction, transceiver._offerDirection
692 ),
693 mid=transceiver.mid,
694 )
695 dtlsTransport = transceiver.receiver.transport
696 else:
697 media = create_media_description_for_sctp(
698 self.__sctp, legacy=self._sctpLegacySdp, mid=self.__sctp.mid
699 )
700 dtlsTransport = self.__sctp.transport
702 # determine DTLS role, or preserve the currently configured role
703 if dtlsTransport._role == "auto":
704 media.dtls.role = "client"
705 else:
706 media.dtls.role = dtlsTransport._role
708 description.media.append(media)
710 bundle = sdp.GroupDescription(semantic="BUNDLE", items=[])
711 for media in description.media:
712 bundle.items.append(media.rtp.muxId)
713 description.group.append(bundle)
715 return wrap_session_description(description)
717 def createDataChannel(
718 self,
719 label: str,
720 maxPacketLifeTime: Optional[int] = None,
721 maxRetransmits: Optional[int] = None,
722 ordered: bool = True,
723 protocol: str = "",
724 negotiated: bool = False,
725 id: Optional[int] = None,
726 ) -> RTCDataChannel:
727 """
728 Create a data channel with the given label.
730 :rtype: :class:`RTCDataChannel`
731 """
732 if maxPacketLifeTime is not None and maxRetransmits is not None:
733 raise ValueError("Cannot specify both maxPacketLifeTime and maxRetransmits")
735 if not self.__sctp:
736 self.__createSctpTransport()
738 parameters = RTCDataChannelParameters(
739 id=id,
740 label=label,
741 maxPacketLifeTime=maxPacketLifeTime,
742 maxRetransmits=maxRetransmits,
743 negotiated=negotiated,
744 ordered=ordered,
745 protocol=protocol,
746 )
747 return RTCDataChannel(self.__sctp, parameters)
749 async def createOffer(self) -> RTCSessionDescription:
750 """
751 Create an SDP offer for the purpose of starting a new WebRTC
752 connection to a remote peer.
754 :rtype: :class:`RTCSessionDescription`
755 """
756 # check state is valid
757 self.__assertNotClosed()
759 # offer codecs
760 receive_only = receive_only_payload_types()
761 for transceiver in self.__transceivers:
762 transceiver._codecs = filter_preferred_codecs(
763 CODECS[transceiver.kind][:], transceiver._preferred_codecs
764 )
765 if transceiver.direction == "recvonly":
766 transceiver._codecs = [
767 with_payload_types(c, receive_only[transceiver.kind])
768 for c in transceiver._codecs
769 ]
770 transceiver._headerExtensions = HEADER_EXTENSIONS[transceiver.kind][:]
772 mids = self.__seenMids.copy()
774 # create description
775 ntp_seconds = clock.current_ntp_time() >> 32
776 description = sdp.SessionDescription()
777 description.origin = f"- {ntp_seconds} {ntp_seconds} IN IP4 0.0.0.0"
778 description.msid_semantic.append(
779 sdp.GroupDescription(semantic="WMS", items=["*"])
780 )
781 description.type = "offer"
783 def get_media(
784 description: sdp.SessionDescription,
785 ) -> list[sdp.MediaDescription]:
786 return description.media if description else []
788 def get_media_section(
789 media: list[sdp.MediaDescription], i: int
790 ) -> Optional[sdp.MediaDescription]:
791 return media[i] if i < len(media) else None
793 # handle existing transceivers / sctp
794 local_media = get_media(self.__localDescription())
795 remote_media = get_media(self.__remoteDescription())
796 for i in range(max(len(local_media), len(remote_media))):
797 local_m = get_media_section(local_media, i)
798 remote_m = get_media_section(remote_media, i)
799 media_kind = local_m.kind if local_m else remote_m.kind
800 mid = local_m.rtp.muxId if local_m else remote_m.rtp.muxId
801 if media_kind in ["audio", "video"]:
802 transceiver = self.__getTransceiverByMid(mid)
803 transceiver._set_mline_index(i)
804 description.media.append(
805 create_media_description_for_transceiver(
806 transceiver,
807 cname=self.__cname,
808 direction=transceiver.direction,
809 mid=mid,
810 )
811 )
812 elif media_kind == "application":
813 self.__sctp_mline_index = i
814 description.media.append(
815 create_media_description_for_sctp(
816 self.__sctp, legacy=self._sctpLegacySdp, mid=mid
817 )
818 )
820 # handle new transceivers / sctp
821 def next_mline_index() -> int:
822 return len(description.media)
824 def add_datachannel_section() -> None:
825 self.__sctp_mline_index = next_mline_index()
826 description.media.append(
827 create_media_description_for_sctp(
828 self.__sctp, legacy=self._sctpLegacySdp, mid=allocate_mid(mids)
829 )
830 )
832 # JSEP specifies data channels should only be added after audio/video
833 # m-sections, but the `alwaysNegotiateDataChannels` option overrides that:
834 # https://w3c.github.io/webrtc-extensions/#always-negotiating-datachannels
835 if not self.__sctp and self.__configuration.alwaysNegotiateDataChannels:
836 self.__createSctpTransport()
837 add_datachannel_section()
839 for transceiver in filter(
840 lambda x: x.mid is None and not x.stopped, self.__transceivers
841 ):
842 transceiver._set_mline_index(next_mline_index())
843 description.media.append(
844 create_media_description_for_transceiver(
845 transceiver,
846 cname=self.__cname,
847 direction=transceiver.direction,
848 mid=allocate_mid(mids),
849 )
850 )
851 if (
852 self.__sctp
853 and self.__sctp.mid is None
854 and not self.__configuration.alwaysNegotiateDataChannels
855 ):
856 add_datachannel_section()
858 bundle = sdp.GroupDescription(semantic="BUNDLE", items=[])
859 for media in description.media:
860 bundle.items.append(media.rtp.muxId)
861 description.group.append(bundle)
863 return wrap_session_description(description)
865 def getReceivers(self) -> list[RTCRtpReceiver]:
866 """
867 Returns the list of :class:`RTCRtpReceiver` objects that are currently
868 attached to the connection.
869 """
870 return list(map(lambda x: x.receiver, self.__transceivers))
872 def getSenders(self) -> list[RTCRtpSender]:
873 """
874 Returns the list of :class:`RTCRtpSender` objects that are currently
875 attached to the connection.
876 """
877 return list(map(lambda x: x.sender, self.__transceivers))
879 async def getStats(self) -> RTCStatsReport:
880 """
881 Returns statistics for the connection.
883 :rtype: :class:`RTCStatsReport`
884 """
885 merged = RTCStatsReport()
886 coros = [x.getStats() for x in self.getSenders()] + [
887 x.getStats() for x in self.getReceivers()
888 ]
889 for report in await asyncio.gather(*coros):
890 merged.update(report)
891 return merged
893 def getTransceivers(self) -> list[RTCRtpTransceiver]:
894 """
895 Returns the list of :class:`RTCRtpTransceiver` objects that are currently
896 attached to the connection.
897 """
898 return list(self.__transceivers)
900 async def setLocalDescription(
901 self, sessionDescription: Optional[RTCSessionDescription] = None
902 ) -> None:
903 """
904 Change the local description associated with the connection.
906 :param sessionDescription: An :class:`RTCSessionDescription` generated
907 by :meth:`createOffer` or :meth:`createAnswer()`
908 or `None` to implicitly create an offer or create
909 an answer, as needed.
910 """
911 # check state is valid
912 self.__assertNotClosed()
914 if sessionDescription is None:
915 # https://w3c.github.io/webrtc-pc/#dom-peerconnection-setlocaldescription
916 # If left out, then setLocalDescription will implicitly create an offer
917 # or create an answer, as needed.
918 if self.signalingState == "have-remote-offer":
919 sessionDescription = await self.createAnswer()
920 else:
921 sessionDescription = await self.createOffer()
922 self.__log_debug(
923 "setLocalDescription(%s, implicit)\n%s",
924 sessionDescription.type,
925 sessionDescription.sdp,
926 )
927 else:
928 self.__log_debug(
929 "setLocalDescription(%s)\n%s",
930 sessionDescription.type,
931 sessionDescription.sdp,
932 )
934 # parse and validate description
935 description = sdp.SessionDescription.parse(sessionDescription.sdp)
936 description.type = sessionDescription.type
937 self.__validate_description(description, is_local=True)
939 # update signaling state
940 if description.type == "offer":
941 self.__setSignalingState("have-local-offer")
942 elif description.type == "answer":
943 self.__setSignalingState("stable")
945 # assign MID
946 for i, media in enumerate(description.media):
947 mid = media.rtp.muxId
948 self.__seenMids.add(mid)
949 if media.kind in ["audio", "video"]:
950 transceiver = self.__getTransceiverByMLineIndex(i)
951 transceiver._set_mid(mid)
952 elif media.kind == "application":
953 self.__sctp.mid = mid
955 # set ICE role; an ICE-lite agent is controlled whichever side offers
956 if description.type == "offer":
957 for iceTransport in self.__iceTransports:
958 if not iceTransport._role_set:
959 iceTransport._connection.ice_controlling = (
960 not iceTransport._connection.ice_lite
961 )
962 iceTransport._role_set = True
964 # set DTLS role
965 if description.type == "answer":
966 for i, media in enumerate(description.media):
967 if media.kind in ["audio", "video"]:
968 transceiver = self.__getTransceiverByMLineIndex(i)
969 transceiver.receiver.transport._set_role(media.dtls.role)
970 elif media.kind == "application":
971 self.__sctp.transport._set_role(media.dtls.role)
973 # configure direction
974 for t in self.__transceivers:
975 if description.type in ["answer", "pranswer"]:
976 t._setCurrentDirection(and_direction(t.direction, t._offerDirection))
978 # gather candidates
979 await self.__gather()
980 for i, media in enumerate(description.media):
981 if media.kind in ["audio", "video"]:
982 transceiver = self.__getTransceiverByMLineIndex(i)
983 add_transport_description(media, transceiver.receiver.transport)
984 elif media.kind == "application":
985 add_transport_description(media, self.__sctp.transport)
987 # connect
988 asyncio.ensure_future(self.__connect())
990 # replace description
991 if description.type == "answer":
992 self.__currentLocalDescription = description
993 self.__pendingLocalDescription = None
994 else:
995 self.__pendingLocalDescription = description
997 async def setRemoteDescription(
998 self, sessionDescription: RTCSessionDescription
999 ) -> None:
1000 """
1001 Changes the remote description associated with the connection.
1003 :param sessionDescription: An :class:`RTCSessionDescription` created from
1004 information received over the signaling channel.
1005 """
1006 self.__log_debug(
1007 "setRemoteDescription(%s)\n%s",
1008 sessionDescription.type,
1009 sessionDescription.sdp,
1010 )
1012 # parse and validate description
1013 description = sdp.SessionDescription.parse(sessionDescription.sdp)
1014 description.type = sessionDescription.type
1015 self.__validate_description(description, is_local=False)
1017 # apply description
1018 iceCandidates: dict[RTCIceTransport, sdp.MediaDescription] = {}
1019 trackEvents = []
1020 for i, media in enumerate(description.media):
1021 dtlsTransport: Optional[RTCDtlsTransport] = None
1022 self.__seenMids.add(media.rtp.muxId)
1023 if media.port == 0:
1024 # The peer rejected this m-line (RFC 3264: port zero); nothing
1025 # negotiates. Drop its never-started transports from the state
1026 # aggregation so the connection doesn't wait on them forever —
1027 # close() still reaches them through the transceiver list.
1028 if media.kind in ["audio", "video"]:
1029 for transceiver in self.__transceivers:
1030 if transceiver.kind == media.kind and transceiver.mid in [
1031 None,
1032 media.rtp.muxId,
1033 ]:
1034 if transceiver.mid is None:
1035 transceiver._set_mid(media.rtp.muxId)
1036 transceiver._set_mline_index(i)
1037 if description.type in ["answer", "pranswer"]:
1038 transceiver._setCurrentDirection("inactive")
1039 rejectedDtls = transceiver.receiver.transport
1040 if not any(
1041 t is not transceiver
1042 and t.receiver.transport is rejectedDtls
1043 for t in self.__transceivers
1044 ):
1045 self.__dtlsTransports.discard(rejectedDtls)
1046 self.__iceTransports.discard(rejectedDtls.transport)
1047 break
1048 elif media.kind == "application" and self.__sctp:
1049 self.__dtlsTransports.discard(self.__sctp.transport)
1050 self.__iceTransports.discard(self.__sctp.transport.transport)
1051 continue
1052 if media.kind in ["audio", "video"]:
1053 # find transceiver
1054 transceiver = None
1055 for t in self.__transceivers:
1056 if t.kind == media.kind and t.mid in [None, media.rtp.muxId]:
1057 transceiver = t
1058 break
1059 if transceiver is None:
1060 transceiver = self.__createTransceiver(
1061 direction="recvonly", kind=media.kind
1062 )
1063 if transceiver.mid is None:
1064 transceiver._set_mid(media.rtp.muxId)
1065 transceiver._set_mline_index(i)
1067 # negotiate codecs
1068 common = filter_preferred_codecs(
1069 find_common_codecs(CODECS[media.kind], media.rtp.codecs),
1070 transceiver._preferred_codecs,
1071 )
1073 if not len(common):
1074 raise OperationError(
1075 "Failed to set remote {} description send parameters".format(
1076 media.kind
1077 )
1078 )
1080 transceiver._codecs = common
1081 transceiver._headerExtensions = find_common_header_extensions(
1082 HEADER_EXTENSIONS[media.kind], media.rtp.headerExtensions
1083 )
1085 # configure direction
1086 direction = reverse_direction(media.direction)
1087 if description.type in ["answer", "pranswer"]:
1088 transceiver._setCurrentDirection(direction)
1089 else:
1090 transceiver._offerDirection = direction
1092 # create remote stream track
1093 if (
1094 direction in ["recvonly", "sendrecv"]
1095 and not transceiver.receiver.track
1096 ):
1097 transceiver.receiver._track = RemoteStreamTrack(
1098 kind=media.kind, id=description.webrtc_track_id(media)
1099 )
1100 trackEvents.append(
1101 RTCTrackEvent(
1102 receiver=transceiver.receiver,
1103 track=transceiver.receiver.track,
1104 transceiver=transceiver,
1105 )
1106 )
1108 # memorise transport parameters
1109 dtlsTransport = transceiver.receiver.transport
1110 self.__remoteDtls[transceiver] = media.dtls
1111 self.__remoteIce[transceiver] = media.ice
1113 elif media.kind == "application":
1114 if not self.__sctp:
1115 self.__createSctpTransport()
1116 if self.__sctp.mid is None:
1117 self.__sctp.mid = media.rtp.muxId
1118 self.__sctp_mline_index = i
1120 # configure sctp
1121 if media.profile == "DTLS/SCTP":
1122 self._sctpLegacySdp = True
1123 self.__sctpRemotePort = int(media.fmt[0])
1124 else:
1125 self._sctpLegacySdp = False
1126 self.__sctpRemotePort = media.sctp_port
1127 self.__sctpRemoteCaps = media.sctpCapabilities
1129 # memorise transport parameters
1130 dtlsTransport = self.__sctp.transport
1131 self.__remoteDtls[self.__sctp] = media.dtls
1132 self.__remoteIce[self.__sctp] = media.ice
1134 if dtlsTransport is not None:
1135 # add ICE candidates
1136 iceTransport = dtlsTransport.transport
1137 iceCandidates[iceTransport] = media
1139 # set ICE role
1140 if description.type == "offer" and not iceTransport._role_set:
1141 iceTransport._connection.ice_controlling = (
1142 media.ice.iceLite and not iceTransport._connection.ice_lite
1143 )
1144 iceTransport._role_set = True
1146 # set DTLS role
1147 if description.type == "offer" and media.dtls.role == "client":
1148 dtlsTransport._set_role(role="server")
1149 if description.type == "answer":
1150 dtlsTransport._set_role(
1151 role="server" if media.dtls.role == "client" else "client"
1152 )
1154 # remove bundled transports
1155 bundle = next((x for x in description.group if x.semantic == "BUNDLE"), None)
1156 if bundle and bundle.items:
1157 # find main media stream
1158 primaryMid = bundle.items[0]
1159 primaryTransport = None
1160 for transceiver in self.__transceivers:
1161 if transceiver.mid == primaryMid:
1162 primaryTransport = transceiver.receiver.transport
1163 break
1164 if self.__sctp and self.__sctp.mid == primaryMid:
1165 primaryTransport = self.__sctp.transport
1167 # replace transport for bundled media; a slave that already shares
1168 # the primary transport (same-kind transceivers are created on one
1169 # transport) must not retire it as an "old" transport — e.g. a
1170 # Firefox answer that rejected the first m-line re-anchors the
1171 # bundle on such a shared transport.
1172 oldTransports = set()
1173 slaveMids = bundle.items[1:]
1174 for transceiver in self.__transceivers:
1175 if transceiver.mid in slaveMids and not transceiver._bundled:
1176 if transceiver.receiver.transport is not primaryTransport:
1177 oldTransports.add(transceiver.receiver.transport)
1178 transceiver.receiver.setTransport(primaryTransport)
1179 transceiver.sender.setTransport(primaryTransport)
1180 transceiver._bundled = True
1181 if (
1182 self.__sctp
1183 and self.__sctp.mid in slaveMids
1184 and not self.__sctp._bundled
1185 ):
1186 if self.__sctp.transport is not primaryTransport:
1187 oldTransports.add(self.__sctp.transport)
1188 self.__sctp.setTransport(primaryTransport)
1189 self.__sctp._bundled = True
1191 # stop and discard old ICE transports
1192 for dtlsTransport in oldTransports:
1193 await dtlsTransport.stop()
1194 await dtlsTransport.transport.stop()
1195 self.__dtlsTransports.discard(dtlsTransport)
1196 self.__iceTransports.discard(dtlsTransport.transport)
1197 iceCandidates.pop(dtlsTransport.transport, None)
1198 self.__updateIceGatheringState()
1199 self.__updateIceConnectionState()
1200 self.__updateConnectionState()
1202 # add remote candidates
1203 coros = [
1204 add_remote_candidates(iceTransport, media)
1205 for iceTransport, media in iceCandidates.items()
1206 ]
1207 await asyncio.gather(*coros)
1209 # Emits the bare track rather than a wrapper event object; the rtc.py
1210 # listeners rely on receiving the track object itself.
1211 for event in trackEvents:
1212 self.emit("track", event.track)
1214 # connect
1215 asyncio.ensure_future(self.__connect())
1217 # update signaling state
1218 if description.type == "offer":
1219 self.__setSignalingState("have-remote-offer")
1220 elif description.type == "answer":
1221 self.__setSignalingState("stable")
1223 # replace description
1224 if description.type == "answer":
1225 self.__currentRemoteDescription = description
1226 self.__pendingRemoteDescription = None
1227 else:
1228 self.__pendingRemoteDescription = description
1230 async def __connect(self) -> None:
1231 for transceiver in self.__transceivers:
1232 dtlsTransport = transceiver.receiver.transport
1233 iceTransport = dtlsTransport.transport
1234 if (
1235 iceTransport.iceGatherer.getLocalCandidates()
1236 and transceiver in self.__remoteIce
1237 ):
1238 await iceTransport.start(self.__remoteIce[transceiver])
1239 if dtlsTransport.state == "new":
1240 await dtlsTransport.start(self.__remoteDtls[transceiver])
1241 if dtlsTransport.state == "connected":
1242 if transceiver.currentDirection in ["sendonly", "sendrecv"]:
1243 await transceiver.sender.send(self.__localRtp(transceiver))
1244 if transceiver.currentDirection in ["recvonly", "sendrecv"]:
1245 await transceiver.receiver.receive(
1246 self.__remoteRtp(transceiver)
1247 )
1248 if self.__sctp:
1249 dtlsTransport = self.__sctp.transport
1250 iceTransport = dtlsTransport.transport
1251 if (
1252 iceTransport.iceGatherer.getLocalCandidates()
1253 and self.__sctp in self.__remoteIce
1254 ):
1255 await iceTransport.start(self.__remoteIce[self.__sctp])
1256 if dtlsTransport.state == "new":
1257 await dtlsTransport.start(self.__remoteDtls[self.__sctp])
1258 if dtlsTransport.state == "connected":
1259 await self.__sctp.start(
1260 self.__sctpRemoteCaps, self.__sctpRemotePort
1261 )
1263 async def __gather(self) -> None:
1264 coros = map(lambda t: t.iceGatherer.gather(), self.__iceTransports)
1265 await asyncio.gather(*coros)
1267 def __assertNotClosed(self) -> None:
1268 if self.__isClosed:
1269 raise InvalidStateError("RTCPeerConnection is closed")
1271 def __assertTrackHasNoSender(self, track: MediaStreamTrack) -> None:
1272 for sender in self.getSenders():
1273 if sender.track == track:
1274 raise InvalidAccessError("Track already has a sender")
1276 def __createDtlsTransport(self) -> RTCDtlsTransport:
1277 # create ICE transport
1278 if len(self.__transceivers) > 0 or self.__sctp:
1279 if len(self.__transceivers) > 0:
1280 parameters = self.__transceivers[
1281 0
1282 ].receiver.transport.transport.iceGatherer.getLocalParameters()
1283 else:
1284 parameters = (
1285 self.__sctp.transport.transport.iceGatherer.getLocalParameters()
1286 )
1287 iceGatherer = RTCIceGatherer(
1288 iceServers=self.__configuration.iceServers,
1289 local_username=parameters.usernameFragment,
1290 local_password=parameters.password,
1291 ice_host_public_ips=self.__configuration.iceHostPublicIps,
1292 ice_port_range=self.__configuration.icePortRange,
1293 ice_udp_mux=self.__configuration.iceUdpMux,
1294 ice_tcp_mux=self.__configuration.iceTcpMux,
1295 ice_lite=self.__configuration.iceLite,
1296 )
1297 else:
1298 iceGatherer = RTCIceGatherer(
1299 iceServers=self.__configuration.iceServers,
1300 ice_host_public_ips=self.__configuration.iceHostPublicIps,
1301 ice_port_range=self.__configuration.icePortRange,
1302 ice_udp_mux=self.__configuration.iceUdpMux,
1303 ice_tcp_mux=self.__configuration.iceTcpMux,
1304 ice_lite=self.__configuration.iceLite,
1305 )
1307 iceGatherer.on("statechange", self.__updateIceGatheringState)
1308 iceTransport = RTCIceTransport(iceGatherer)
1309 iceTransport.on("statechange", self.__updateIceConnectionState)
1310 iceTransport.on("statechange", self.__updateConnectionState)
1311 self.__iceTransports.add(iceTransport)
1313 # create DTLS transport
1314 dtlsTransport = RTCDtlsTransport(iceTransport, self.__certificates)
1315 dtlsTransport.on("statechange", self.__updateConnectionState)
1316 self.__dtlsTransports.add(dtlsTransport)
1318 # update states
1319 self.__updateIceGatheringState()
1320 self.__updateIceConnectionState()
1321 self.__updateConnectionState()
1323 return dtlsTransport
1325 def __createSctpTransport(self) -> None:
1326 dtlsTransport = None
1327 bundled = (
1328 self.__configuration.bundlePolicy == RTCBundlePolicy.MAX_BUNDLE
1329 and len(self.__transceivers) > 0
1330 )
1331 if bundled:
1332 dtlsTransport = self.__transceivers[0].receiver.transport
1333 else:
1334 dtlsTransport = self.__createDtlsTransport()
1335 self.__sctp = RTCSctpTransport(dtlsTransport)
1336 self.__sctp._bundled = bundled
1337 self.__sctp.mid = None
1339 @self.__sctp.on("datachannel")
1340 def on_datachannel(channel: RTCDataChannel) -> None:
1341 self.emit("datachannel", channel)
1343 def __createTransceiver(
1344 self, direction: str, kind: str, sender_track: Optional[MediaStreamTrack] = None
1345 ) -> RTCRtpTransceiver:
1346 dtlsTransport = None
1347 bundled = False
1348 if self.__configuration.bundlePolicy == RTCBundlePolicy.MAX_BUNDLE:
1349 if len(self.__transceivers) > 0:
1350 dtlsTransport = self.__transceivers[0].receiver.transport
1351 bundled = True
1352 elif self.__sctp:
1353 dtlsTransport = self.__sctp.transport
1354 bundled = True
1355 elif self.__configuration.bundlePolicy == RTCBundlePolicy.BALANCED:
1356 transceiver = next(
1357 filter(lambda t: t.kind == kind, self.__transceivers), None
1358 )
1359 if transceiver:
1360 dtlsTransport = transceiver.receiver.transport
1361 bundled = True
1363 if not dtlsTransport:
1364 dtlsTransport = self.__createDtlsTransport()
1366 transceiver = RTCRtpTransceiver(
1367 direction=direction,
1368 kind=kind,
1369 sender=RTCRtpSender(sender_track or kind, dtlsTransport),
1370 receiver=RTCRtpReceiver(kind, dtlsTransport),
1371 )
1372 transceiver.receiver._set_rtcp_ssrc(transceiver.sender._ssrc)
1373 transceiver.sender._stream_id = self.__stream_id
1374 transceiver._bundled = bundled
1375 self.__transceivers.append(transceiver)
1376 return transceiver
1378 def __getTransceiverByMid(self, mid: str) -> Optional[RTCRtpTransceiver]:
1379 return next(filter(lambda x: x.mid == mid, self.__transceivers), None)
1381 def __getTransceiverByMLineIndex(self, index: int) -> Optional[RTCRtpTransceiver]:
1382 return next(
1383 filter(lambda x: x._get_mline_index() == index, self.__transceivers), None
1384 )
1386 def __localDescription(self) -> Optional[sdp.SessionDescription]:
1387 return self.__pendingLocalDescription or self.__currentLocalDescription
1389 def __localRtp(self, transceiver: RTCRtpTransceiver) -> RTCRtpSendParameters:
1390 rtp = RTCRtpSendParameters(
1391 codecs=transceiver._codecs,
1392 headerExtensions=transceiver._headerExtensions,
1393 muxId=transceiver.mid,
1394 )
1395 rtp.rtcp.cname = self.__cname
1396 rtp.rtcp.ssrc = transceiver.sender._ssrc
1397 rtp.rtcp.mux = True
1398 return rtp
1400 def __log_debug(self, msg: str, *args: object) -> None:
1401 logger.debug(f"RTCPeerConnection() {msg}", *args)
1403 def __remoteDescription(self) -> Optional[sdp.SessionDescription]:
1404 return self.__pendingRemoteDescription or self.__currentRemoteDescription
1406 def __remoteRtp(self, transceiver: RTCRtpTransceiver) -> RTCRtpReceiveParameters:
1407 media = self.__remoteDescription().media[transceiver._get_mline_index()]
1409 receiveParameters = RTCRtpReceiveParameters(
1410 codecs=transceiver._codecs,
1411 headerExtensions=transceiver._headerExtensions,
1412 muxId=media.rtp.muxId,
1413 rtcp=media.rtp.rtcp,
1414 )
1415 if len(media.ssrc):
1416 # The retransmission stream is the FID group's second member; a
1417 # remote that also sends FEC lists a third SSRC, so the count of
1418 # SSRCs alone names nothing.
1419 rtx_ssrc = next((group.items[1] for group in media.ssrc_group
1420 if group.semantic == "FID" and len(group.items) == 2
1421 and group.items[0] == media.ssrc[0].ssrc),
1422 media.ssrc[1].ssrc if len(media.ssrc) == 2 else None)
1423 encodings: dict[int, RTCRtpDecodingParameters] = {}
1424 for codec in transceiver._codecs:
1425 if is_rtx(codec):
1426 apt = codec.parameters.get("apt")
1427 if isinstance(apt, int) and apt in encodings and rtx_ssrc is not None:
1428 encodings[apt].rtx = RTCRtpRtxParameters(ssrc=rtx_ssrc)
1429 continue
1431 encodings[codec.payloadType] = RTCRtpDecodingParameters(
1432 ssrc=media.ssrc[0].ssrc, payloadType=codec.payloadType
1433 )
1434 receiveParameters.encodings = list(encodings.values())
1435 return receiveParameters
1437 def __setSignalingState(self, state: str) -> None:
1438 self.__signalingState = state
1439 self.emit("signalingstatechange")
1441 def __updateConnectionState(self) -> None:
1442 # compute new state
1443 # NOTE: we do not have a "disconnected" state
1444 dtlsStates = set(map(lambda x: x.state, self.__dtlsTransports))
1445 iceStates = set(map(lambda x: x.state, self.__iceTransports))
1446 if self.__isClosed:
1447 state = "closed"
1448 elif "failed" in iceStates or "failed" in dtlsStates:
1449 state = "failed"
1450 elif not iceStates.difference(["new", "closed"]) and not dtlsStates.difference(
1451 ["new", "closed"]
1452 ):
1453 state = "new"
1454 elif "checking" in iceStates or "connecting" in dtlsStates:
1455 state = "connecting"
1456 elif "new" in dtlsStates:
1457 # this avoids a spurious connecting -> connected -> connecting
1458 # transition after ICE connects but before DTLS starts
1459 state = "connecting"
1460 else:
1461 state = "connected"
1463 # update state
1464 if state != self.__connectionState:
1465 self.__log_debug("connectionState %s -> %s", self.__connectionState, state)
1466 self.__connectionState = state
1467 self.emit("connectionstatechange")
1469 # if all DTLS connections are closed, initiate a shutdown
1470 if (
1471 not self.__isClosed
1472 and self.__closeTask is None
1473 and dtlsStates == set(["closed"])
1474 ):
1475 self.__closeTask = asyncio.ensure_future(self.close())
1477 def __updateIceConnectionState(self) -> None:
1478 # compute new state
1479 # NOTE: we do not have "connected" or "disconnected" states
1480 states = set(map(lambda x: x.state, self.__iceTransports))
1481 if self.__isClosed:
1482 state = "closed"
1483 elif "failed" in states:
1484 state = "failed"
1485 elif states == set(["completed"]):
1486 state = "completed"
1487 elif "checking" in states:
1488 state = "checking"
1489 else:
1490 state = "new"
1492 # update state
1493 if state != self.__iceConnectionState:
1494 self.__log_debug(
1495 "iceConnectionState %s -> %s", self.__iceConnectionState, state
1496 )
1497 self.__iceConnectionState = state
1498 self.emit("iceconnectionstatechange")
1500 def __updateIceGatheringState(self) -> None:
1501 # compute new state
1502 states = set(map(lambda x: x.iceGatherer.state, self.__iceTransports))
1503 if states == set(["completed"]):
1504 state = "complete"
1505 elif "gathering" in states:
1506 state = "gathering"
1507 else:
1508 state = "new"
1510 # update state
1511 if state != self.__iceGatheringState:
1512 self.__log_debug(
1513 "iceGatheringState %s -> %s", self.__iceGatheringState, state
1514 )
1515 self.__iceGatheringState = state
1516 self.emit("icegatheringstatechange")
1518 def __validate_description(
1519 self, description: sdp.SessionDescription, is_local: bool
1520 ) -> None:
1521 # check description is compatible with signaling state
1522 if is_local:
1523 if description.type == "offer":
1524 if self.signalingState not in ["stable", "have-local-offer"]:
1525 raise InvalidStateError(
1526 "Cannot handle offer in signaling state "
1527 f'"{self.signalingState}"'
1528 )
1529 elif description.type == "answer":
1530 if self.signalingState not in [
1531 "have-remote-offer",
1532 "have-local-pranswer",
1533 ]:
1534 raise InvalidStateError(
1535 "Cannot handle answer in signaling state "
1536 f'"{self.signalingState}"'
1537 )
1538 else:
1539 if description.type == "offer":
1540 if self.signalingState not in ["stable", "have-remote-offer"]:
1541 raise InvalidStateError(
1542 "Cannot handle offer in signaling state "
1543 f'"{self.signalingState}"'
1544 )
1545 elif description.type == "answer":
1546 if self.signalingState not in [
1547 "have-local-offer",
1548 "have-remote-pranswer",
1549 ]:
1550 raise InvalidStateError(
1551 "Cannot handle answer in signaling state "
1552 f'"{self.signalingState}"'
1553 )
1555 for media in description.media:
1556 # A rejected m-line (RFC 3264: port zero) negotiates nothing and
1557 # carries no ICE/DTLS parameters — e.g. Firefox answering a video
1558 # section it has no codec for — so none of these checks apply.
1559 if media.port == 0:
1560 continue
1562 # check ICE credentials were provided
1563 if not media.ice.usernameFragment or not media.ice.password:
1564 raise ValueError("ICE username fragment or password is missing")
1566 # check DTLS role is allowed
1567 if description.type in ["answer", "pranswer"] and media.dtls.role not in [
1568 "client",
1569 "server",
1570 ]:
1571 raise ValueError(
1572 "DTLS setup attribute must be 'active' or 'passive' for an answer"
1573 )
1575 # check RTCP mux is used
1576 if media.kind in ["audio", "video"] and not media.rtcp_mux:
1577 raise ValueError("RTCP mux is not enabled")
1579 # check the number of media section matches
1580 if description.type in ["answer", "pranswer"]:
1581 offer = (
1582 self.__remoteDescription() if is_local else self.__localDescription()
1583 )
1584 offer_media = [(media.kind, media.rtp.muxId) for media in offer.media]
1585 answer_media = [
1586 (media.kind, media.rtp.muxId) for media in description.media
1587 ]
1588 if answer_media != offer_media:
1589 raise ValueError("Media sections in answer do not match offer")