8 from typing import Optional, List
13 gi.require_version('Gst', '1.0')
14 from gi.repository import Gst
16 gi.require_version('GstWebRTC', '1.0')
17 from gi.repository import GstWebRTC
19 gi.require_version('GstSdp', '1.0')
20 from gi.repository import GstSdp
22 log = logging.getLogger(__name__)
27 self.sdp_offer: Optional[str] = None
28 self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
29 self.session_id = None
30 self.received_ice_candidates = []
31 self.generated_ice_candidates = []
32 self.user_fragments: Optional[List] = None
33 self.mids: Optional[List] = None
37 def on_negotiation_needed(self, element):
38 log.debug('on_negotiation_needed')
40 def on_ice_candidate(self, element, mlineindex, candidate):
41 log.debug('on_ice_candidate')
42 self.generated_ice_candidates.append((mlineindex, candidate))
44 def webrtcbin_pad_added(self, element, pad):
45 log.debug('webrtcbin_pad_added')
46 if pad.direction != Gst.PadDirection.SRC:
48 decodebin = Gst.ElementFactory.make('decodebin')
49 decodebin.connect('pad-added', self.decodebin_pad_added)
50 self.pipe.add(decodebin)
51 decodebin.sync_state_with_parent()
52 self.webrtcbin.link(decodebin)
54 def decodebin_pad_added(self, element, pad):
55 log.debug('decodebin_pad_added')
56 if not pad.has_current_caps():
57 log.debug(pad, 'has no caps, ignoring')
60 caps = pad.get_current_caps()
61 padsize = caps.get_size()
62 for i in range(padsize):
63 s = caps.get_structure(i) # Gst.Structure
65 if name.startswith('video'):
66 q = Gst.ElementFactory.make('queue')
67 conv = Gst.ElementFactory.make('videoconvert')
68 # sink = Gst.ElementFactory.make('autovideosink') # needs XDG_RUNTIME_DIR
69 sink = Gst.ElementFactory.make('xvimagesink')
73 self.pipe.sync_children_states()
74 pad.link(q.get_static_pad('sink'))
77 elif name.startswith('audio'):
78 q = Gst.ElementFactory.make('queue')
79 conv = Gst.ElementFactory.make('audioconvert')
80 resample = Gst.ElementFactory.make('audioresample')
81 sink = Gst.ElementFactory.make('autoaudiosink')
84 self.pipe.add(resample)
86 self.pipe.sync_children_states()
87 pad.link(q.get_static_pad('sink'))
92 async def listen_to_gstreamer_bus(self):
94 self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
95 self.pipe = Gst.Pipeline.new("pipeline")
96 Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
97 bus = Gst.Pipeline.get_bus(self.pipe)
98 self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed)
99 self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate)
100 self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added)
101 self.pipe.set_state(Gst.State.PLAYING)
104 if bus.have_pending():
106 if msg.type == Gst.MessageType.ERROR:
107 log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
109 elif msg.type == Gst.MessageType.EOS: # end of stream
110 log.info(f'Gstreamer message bus reports end of stream')
112 elif self.sdp_offer is not None:
113 res, sm = GstSdp.SDPMessage.new()
114 assert res == GstSdp.SDPResult.OK
115 GstSdp.sdp_message_parse_buffer(bytes(self.sdp_offer.encode()), sm)
116 # the three lines above can also be done this way in new versions of GStreamer:
117 # sm = GstSdp.SDPMessage.new_from_text(sdp_offer)
118 rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
119 gst_promise = Gst.Promise.new()
120 self.webrtcbin.emit('set-remote-description', rd, gst_promise)
121 await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
122 self.sdp_offer = None
124 gst_promise = Gst.Promise.new()
125 self.webrtcbin.emit('create-answer', None, gst_promise)
126 result = await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
127 assert result == Gst.PromiseResult.REPLIED
128 reply = gst_promise.get_reply()
129 answer = reply.get_value('answer')
130 sdp_message = answer.sdp
131 self.user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag')
132 for i in range(sdp_message.medias_len())]
133 self.mids = [sdp_message.get_media(i).get_attribute_val('mid')
134 for i in range(sdp_message.medias_len())]
135 sdp_answer = sdp_message.as_text()
136 log.info(f'Send SDP answer:\n{sdp_answer}')
137 sdp_answer_msg = json.dumps({
138 'SessionID': self.session_id,
140 'Value': json.dumps({
145 gst_promise = Gst.Promise.new()
146 self.webrtcbin.emit('set-local-description', answer, gst_promise)
147 await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
148 gst_promise.get_reply()
149 await self.websocket.send(sdp_answer_msg)
151 elif len(self.received_ice_candidates) > 0:
152 ic = self.received_ice_candidates.pop(0)
153 if ic['candidate'] != '':
154 self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
156 elif len(self.generated_ice_candidates) > 0:
157 mlineindex, candidate = self.generated_ice_candidates.pop(0)
158 icemsg_value = json.dumps({
159 "candidate": candidate,
160 "sdpMid": self.mids[mlineindex],
161 "sdpMLineIndex": mlineindex,
162 "usernameFragment": self.user_fragments[mlineindex],
164 icemsg = json.dumps({
165 'SessionID': self.session_id,
166 'Type': 'addCalleeIceCandidate',
167 'Value': icemsg_value,
169 log.info(f'Send ICE candidate: {icemsg_value}')
170 await self.websocket.send(icemsg)
173 await asyncio.sleep(0.1)
175 self.pipe.set_state(Gst.State.NULL)
177 async def talk_to_websocket(self, uri):
178 async for msg in self.websocket:
179 msg_json = json.loads(msg)
180 msg_type = msg_json['Type']
181 msg_value = msg_json['Value']
182 assert self.session_id is None or self.session_id == msg_json['SessionID']
183 if msg_type == 'newSession':
184 self.session_id = msg_json['SessionID']
185 log.info(f"New session {self.session_id}")
186 elif msg_type == 'gotOffer':
187 value_json = json.loads(msg_value)
188 sdp = value_json['sdp']
189 log.info(f'Got SDP offer:\n{sdp}')
191 elif msg_type == 'addCallerIceCandidate':
192 value_json = json.loads(msg_value)
193 log.info(f'Got ICE candidate: {value_json}')
194 self.received_ice_candidates.append(value_json)
195 elif msg_type == 'roomClosed':
196 log.info(f'Oh noes, the room went away (session {self.session_id})!')
197 self.session_id = None
200 log.error(f'Unknown message type {msg_type}')
202 async def run(self, uri):
203 ssl_context = ssl.SSLContext()
204 ssl_context.check_hostname = False
205 ssl_context.verify_mode = ssl.CERT_NONE
207 async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
208 talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
209 listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
210 done, pending = await asyncio.wait(
211 [talk_to_websocket_task, listen_to_gstreamer_bus_task],
212 return_when=asyncio.FIRST_COMPLETED)
222 logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s')
223 parser = argparse.ArgumentParser()
224 parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
225 help='Signalling server URI')
226 args = parser.parse_args()
228 asyncio.run(lagarde.run(args.uri), debug=True)
231 if __name__ == '__main__':