#!/usr/bin/python3 import argparse import asyncio import json import logging import ssl from typing import Optional, List import gi import websockets gi.require_version('Gst', '1.0') from gi.repository import Gst gi.require_version('GstWebRTC', '1.0') from gi.repository import GstWebRTC gi.require_version('GstSdp', '1.0') from gi.repository import GstSdp log = logging.getLogger(__name__) class Lagarde: def __init__(self): self.sdp_offer: Optional[str] = None self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None self.session_id = None self.received_ice_candidates = [] self.generated_ice_candidates = [] self.user_fragments: Optional[List] = None self.mids: Optional[List] = None self.pipe = None self.webrtcbin = None def on_negotiation_needed(self, element): log.info('on_negotiation_needed') def on_ice_candidate(self, element, mlineindex, candidate): log.info('on_ice_candidate') self.generated_ice_candidates.append((mlineindex, candidate)) def webrtcbin_pad_added(self, element, pad): log.info('webrtcbin_pad_added') if pad.direction != Gst.PadDirection.SRC: return decodebin = Gst.ElementFactory.make('decodebin') decodebin.connect('pad-added', self.decodebin_pad_added) self.pipe.add(decodebin) decodebin.sync_state_with_parent() self.webrtcbin.link(decodebin) def decodebin_pad_added(self, element, pad): log.info('decodebin_pad_added') if not pad.has_current_caps(): log.info(pad, 'has no caps, ignoring') return caps = pad.get_current_caps() # assert (len(caps)) # we have a Gst.Caps object and it has no length # s = caps[0] # also, it's not a list padsize = caps.get_size() assert(padsize > 0) for i in range(padsize): # pythonic?! s = caps.get_structure(i) # Gst.Structure name = s.get_name() if name.startswith('video'): q = Gst.ElementFactory.make('queue') conv = Gst.ElementFactory.make('videoconvert') # sink = Gst.ElementFactory.make('autovideosink') # needs XDG_RUNTIME_DIR sink = Gst.ElementFactory.make('xvimagesink') self.pipe.add(q) self.pipe.add(conv) self.pipe.add(sink) self.pipe.sync_children_states() pad.link(q.get_static_pad('sink')) q.link(conv) conv.link(sink) elif name.startswith('audio'): q = Gst.ElementFactory.make('queue') conv = Gst.ElementFactory.make('audioconvert') resample = Gst.ElementFactory.make('audioresample') sink = Gst.ElementFactory.make('autoaudiosink') self.pipe.add(q) self.pipe.add(conv) self.pipe.add(resample) self.pipe.add(sink) self.pipe.sync_children_states() pad.link(q.get_static_pad('sink')) q.link(conv) conv.link(resample) resample.link(sink) async def listen_to_gstreamer_bus(self): Gst.init(None) self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace') self.pipe = Gst.Pipeline.new("pipeline") Gst.Bin.do_add_element(self.pipe, self.webrtcbin) bus = Gst.Pipeline.get_bus(self.pipe) self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed) self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate) self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added) self.pipe.set_state(Gst.State.PLAYING) try: while True: if bus.have_pending(): msg = bus.pop() # Gst.Message, has to be unref'ed. if msg.type != Gst.MessageType.STATE_CHANGED: # log.info(f'Receive Gst.Message: {msg.type}, {msg.seqnum}, {msg.get_structure()}') # log.info(f'{webrtcbin.props.signaling_state} {webrtcbin.props.ice_gathering_state} {webrtcbin.props.ice_connection_state}') # Gst.Message.unref(msg) pass elif self.sdp_offer is not None: res, sm = GstSdp.SDPMessage.new() assert res == GstSdp.SDPResult.OK GstSdp.sdp_message_parse_buffer(bytes(self.sdp_offer.encode()), sm) # the three lines above can also be done this way in new versions of GStreamer: # sm = GstSdp.SDPMessage.new_from_text(sdp_offer) rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm) gst_promise = Gst.Promise.new() self.webrtcbin.emit('set-remote-description', rd, gst_promise) gst_promise.wait() self.sdp_offer = None log.info('create-answer') gst_promise = Gst.Promise.new() self.webrtcbin.emit('create-answer', None, gst_promise) result = gst_promise.wait() assert result == Gst.PromiseResult.REPLIED reply = gst_promise.get_reply() answer = reply.get_value('answer') sdp_message = answer.sdp self.user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag') for i in range(sdp_message.medias_len())] self.mids = [sdp_message.get_media(i).get_attribute_val('mid') for i in range(sdp_message.medias_len())] sdp_answer = sdp_message.as_text() log.info(sdp_answer) sdp_answer_msg = json.dumps({ 'SessionID': self.session_id, 'Type': "gotAnswer", 'Value': json.dumps({ 'type': 'answer', 'sdp': sdp_answer }) }) gst_promise = Gst.Promise.new() self.webrtcbin.emit('set-local-description', answer, gst_promise) gst_promise.wait() gst_promise.get_reply() await self.websocket.send(sdp_answer_msg) elif len(self.received_ice_candidates) > 0: ic = self.received_ice_candidates.pop(0) if ic['candidate'] != '': self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate']) elif len(self.generated_ice_candidates) > 0: mlineindex, candidate = self.generated_ice_candidates.pop(0) icemsg = json.dumps({ 'SessionID': self.session_id, 'Type': 'addCalleeIceCandidate', 'Value': json.dumps({ "candidate": candidate, "sdpMid": self.mids[mlineindex], "sdpMLineIndex": mlineindex, "usernameFragment": self.user_fragments[mlineindex], }) }) log.info(f'send_ice_candidate_message with {icemsg}') await self.websocket.send(icemsg) else: await asyncio.sleep(0.1) finally: self.pipe.set_state(Gst.State.NULL) async def talk_to_websocket(self, uri): ssl_context = ssl.SSLContext() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE async with websockets.connect(uri, ssl=ssl_context) as self.websocket: async for msg in self.websocket: msg_json = json.loads(msg) msg_type = msg_json['Type'] msg_value = msg_json['Value'] self.session_id = msg_json['SessionID'] log.info(f"receive for session {self.session_id} type {msg_type}") if msg_type == 'newSession': pass elif msg_type == 'gotOffer': value_json = json.loads(msg_value) sdp = value_json['sdp'] log.info(f'SDP: {sdp}') self.sdp_offer = sdp elif msg_type == 'addCallerIceCandidate': value_json = json.loads(msg_value) log.info(f'ICE: {value_json}') self.received_ice_candidates.append(value_json) elif msg_type == 'roomClosed': log.info('Oh noes, the room went away!') # and here we should clean up else: log.error(f'Unknown message type {msg_type}') async def run(self, uri): talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri)) listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus()) done, pending = await asyncio.wait( [talk_to_websocket_task, listen_to_gstreamer_bus_task], return_when=asyncio.FIRST_COMPLETED) for d in done: d.result() for p in pending: p.cancel() def main(): logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s') parser = argparse.ArgumentParser() parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug', help='Signalling server URI') args = parser.parse_args() lagarde = Lagarde() asyncio.run(lagarde.run(args.uri), debug=True) if __name__ == '__main__': main()