|
import logging
|
|
from typing import List, Optional
|
|
|
|
from .codecs import get_capabilities
|
|
from .rtcdtlstransport import RTCDtlsTransport
|
|
from .rtcrtpparameters import (
|
|
RTCRtpCodecCapability,
|
|
RTCRtpCodecParameters,
|
|
RTCRtpHeaderExtensionParameters,
|
|
)
|
|
from .rtcrtpreceiver import RTCRtpReceiver
|
|
from .rtcrtpsender import RTCRtpSender
|
|
from .sdp import DIRECTIONS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RTCRtpTransceiver:
|
|
"""
|
|
The RTCRtpTransceiver interface describes a permanent pairing of an
|
|
:class:`RTCRtpSender` and an :class:`RTCRtpReceiver`, along with some
|
|
shared state.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
kind: str,
|
|
receiver: RTCRtpReceiver,
|
|
sender: RTCRtpSender,
|
|
direction: str = "sendrecv",
|
|
):
|
|
self.__currentDirection: Optional[str] = None
|
|
self.__direction = direction
|
|
self.__kind = kind
|
|
self.__mid: Optional[str] = None
|
|
self.__mline_index: Optional[int] = None
|
|
self.__receiver = receiver
|
|
self.__sender = sender
|
|
self.__stopped = False
|
|
|
|
self._offerDirection: Optional[str] = None
|
|
self._preferred_codecs: List[RTCRtpCodecCapability] = []
|
|
self._transport: RTCDtlsTransport = None
|
|
|
|
|
|
self._bundled = False
|
|
self._codecs: List[RTCRtpCodecParameters] = []
|
|
self._headerExtensions: List[RTCRtpHeaderExtensionParameters] = []
|
|
|
|
@property
|
|
def currentDirection(self) -> Optional[str]:
|
|
"""
|
|
The currently negotiated direction of the transceiver.
|
|
|
|
One of `'sendrecv'`, `'sendonly'`, `'recvonly'`, `'inactive'` or `None`.
|
|
"""
|
|
return self.__currentDirection
|
|
|
|
@property
|
|
def direction(self) -> str:
|
|
"""
|
|
The preferred direction of the transceiver, which will be used in
|
|
:meth:`RTCPeerConnection.createOffer` and
|
|
:meth:`RTCPeerConnection.createAnswer`.
|
|
|
|
One of `'sendrecv'`, `'sendonly'`, `'recvonly'` or `'inactive'`.
|
|
"""
|
|
return self.__direction
|
|
|
|
@direction.setter
|
|
def direction(self, direction: str) -> None:
|
|
assert direction in DIRECTIONS
|
|
self.__direction = direction
|
|
|
|
@property
|
|
def kind(self) -> str:
|
|
return self.__kind
|
|
|
|
@property
|
|
def mid(self) -> Optional[str]:
|
|
return self.__mid
|
|
|
|
@property
|
|
def receiver(self) -> RTCRtpReceiver:
|
|
"""
|
|
The :class:`RTCRtpReceiver` that handles receiving and decoding
|
|
incoming media.
|
|
"""
|
|
return self.__receiver
|
|
|
|
@property
|
|
def sender(self) -> RTCRtpSender:
|
|
"""
|
|
The :class:`RTCRtpSender` responsible for encoding and sending
|
|
data to the remote peer.
|
|
"""
|
|
return self.__sender
|
|
|
|
@property
|
|
def stopped(self) -> bool:
|
|
return self.__stopped
|
|
|
|
def setCodecPreferences(self, codecs: List[RTCRtpCodecCapability]) -> None:
|
|
"""
|
|
Override the default codec preferences.
|
|
|
|
See :meth:`RTCRtpSender.getCapabilities` and
|
|
:meth:`RTCRtpReceiver.getCapabilities` for the supported codecs.
|
|
|
|
:param codecs: A list of :class:`RTCRtpCodecCapability`, in decreasing order
|
|
of preference. If empty, restores the default preferences.
|
|
"""
|
|
if not codecs:
|
|
self._preferred_codecs = []
|
|
|
|
capabilities = get_capabilities(self.kind).codecs
|
|
unique: List[RTCRtpCodecCapability] = []
|
|
for codec in reversed(codecs):
|
|
if codec not in capabilities:
|
|
raise ValueError("Codec is not in capabilities")
|
|
if codec not in unique:
|
|
unique.insert(0, codec)
|
|
self._preferred_codecs = unique
|
|
|
|
async def stop(self) -> None:
|
|
"""
|
|
Permanently stops the :class:`RTCRtpTransceiver`.
|
|
"""
|
|
await self.__receiver.stop()
|
|
await self.__sender.stop()
|
|
self.__stopped = True
|
|
|
|
def _setCurrentDirection(self, direction: str) -> None:
|
|
self.__currentDirection = direction
|
|
|
|
if direction == "sendrecv":
|
|
self.__sender._enabled = True
|
|
self.__receiver._enabled = True
|
|
elif direction == "sendonly":
|
|
self.__sender._enabled = True
|
|
self.__receiver._enabled = False
|
|
elif direction == "recvonly":
|
|
self.__sender._enabled = False
|
|
self.__receiver._enabled = True
|
|
elif direction == "inactive":
|
|
self.__sender._enabled = False
|
|
self.__receiver._enabled = False
|
|
|
|
def _set_mid(self, mid: str) -> None:
|
|
self.__mid = mid
|
|
|
|
def _get_mline_index(self) -> Optional[int]:
|
|
return self.__mline_index
|
|
|
|
def _set_mline_index(self, idx: int) -> None:
|
|
self.__mline_index = idx
|
|
|
