Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/webrtc/rtcrtptransceiver.py: 38%

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. 

33 

34import logging 

35from typing import Optional 

36 

37from .codecs import get_capabilities 

38from .rtcrtpparameters import ( 

39 RTCRtpCodecCapability, 

40 RTCRtpCodecParameters, 

41 RTCRtpHeaderExtensionParameters, 

42) 

43from .rtcrtpreceiver import RTCRtpReceiver 

44from .rtcrtpsender import RTCRtpSender 

45from .sdp import DIRECTIONS 

46 

47logger = logging.getLogger(__name__) 

48 

49 

50class RTCRtpTransceiver: 

51 """ 

52 The RTCRtpTransceiver interface describes a permanent pairing of an 

53 :class:`RTCRtpSender` and an :class:`RTCRtpReceiver`, along with some 

54 shared state. 

55 """ 

56 

57 def __init__( 

58 self, 

59 kind: str, 

60 receiver: RTCRtpReceiver, 

61 sender: RTCRtpSender, 

62 direction: str = "sendrecv", 

63 ): 

64 self.__currentDirection: Optional[str] = None 

65 self.__direction = direction 

66 self.__kind = kind 

67 self.__mid: Optional[str] = None 

68 self.__mline_index: Optional[int] = None 

69 self.__receiver = receiver 

70 self.__sender = sender 

71 self.__stopped = False 

72 

73 self._offerDirection: Optional[str] = None 

74 self._preferred_codecs: list[RTCRtpCodecCapability] = [] 

75 

76 # Owned and driven by RTCPeerConnection's BUNDLE bookkeeping. 

77 self._bundled = False 

78 self._codecs: list[RTCRtpCodecParameters] = [] 

79 self._headerExtensions: list[RTCRtpHeaderExtensionParameters] = [] 

80 

81 @property 

82 def currentDirection(self) -> Optional[str]: 

83 """ 

84 The currently negotiated direction of the transceiver. 

85 

86 One of `'sendrecv'`, `'sendonly'`, `'recvonly'`, `'inactive'`, or `None`. 

87 """ 

88 return self.__currentDirection 

89 

90 @property 

91 def direction(self) -> str: 

92 """ 

93 The preferred direction of the transceiver, which will be used in 

94 :meth:`RTCPeerConnection.createOffer` and 

95 :meth:`RTCPeerConnection.createAnswer`. 

96 

97 One of `'sendrecv'`, `'sendonly'`, `'recvonly'`, or `'inactive'`. 

98 """ 

99 return self.__direction 

100 

101 @direction.setter 

102 def direction(self, direction: str) -> None: 

103 assert direction in DIRECTIONS 

104 self.__direction = direction 

105 

106 @property 

107 def kind(self) -> str: 

108 return self.__kind 

109 

110 @property 

111 def mid(self) -> Optional[str]: 

112 return self.__mid 

113 

114 @property 

115 def receiver(self) -> RTCRtpReceiver: 

116 """ 

117 The :class:`RTCRtpReceiver` that handles receiving and decoding 

118 incoming media. 

119 """ 

120 return self.__receiver 

121 

122 @property 

123 def sender(self) -> RTCRtpSender: 

124 """ 

125 The :class:`RTCRtpSender` responsible for encoding and sending 

126 data to the remote peer. 

127 """ 

128 return self.__sender 

129 

130 @property 

131 def stopped(self) -> bool: 

132 return self.__stopped 

133 

134 def setCodecPreferences(self, codecs: list[RTCRtpCodecCapability]) -> None: 

135 """ 

136 Override the default codec preferences. 

137 

138 See :meth:`RTCRtpSender.getCapabilities` and 

139 :meth:`RTCRtpReceiver.getCapabilities` for the supported codecs. 

140 

141 :param codecs: A list of :class:`RTCRtpCodecCapability`, in decreasing order 

142 of preference. If empty, restores the default preferences. 

143 """ 

144 if not codecs: 

145 self._preferred_codecs = [] 

146 

147 capabilities = get_capabilities(self.kind).codecs 

148 unique: list[RTCRtpCodecCapability] = [] 

149 for codec in reversed(codecs): 

150 if codec not in capabilities: 

151 raise ValueError("Codec is not in capabilities") 

152 if codec not in unique: 

153 unique.insert(0, codec) 

154 self._preferred_codecs = unique 

155 

156 async def stop(self) -> None: 

157 """ 

158 Permanently stops the :class:`RTCRtpTransceiver`. 

159 """ 

160 await self.__receiver.stop() 

161 await self.__sender.stop() 

162 self.__stopped = True 

163 

164 def _setCurrentDirection(self, direction: str) -> None: 

165 self.__currentDirection = direction 

166 

167 if direction == "sendrecv": 

168 self.__sender._enabled = True 

169 self.__receiver._enabled = True 

170 elif direction == "sendonly": 

171 self.__sender._enabled = True 

172 self.__receiver._enabled = False 

173 elif direction == "recvonly": 

174 self.__sender._enabled = False 

175 self.__receiver._enabled = True 

176 elif direction == "inactive": 

177 self.__sender._enabled = False 

178 self.__receiver._enabled = False 

179 

180 def _set_mid(self, mid: str) -> None: 

181 self.__mid = mid 

182 

183 def _get_mline_index(self) -> Optional[int]: 

184 return self.__mline_index 

185 

186 def _set_mline_index(self, idx: int) -> None: 

187 self.__mline_index = idx