lagarde.py

   1 #!/usr/bin/python3
   2
   3 import argparse
   4 import asyncio
   5 import json
   6 import logging
   7 import ssl
   8 from typing import Optional, List
   9
  10 import gi
  11 import websockets
  12
  13 gi.require_version('Gst', '1.0')
  14 from gi.repository import Gst
  15
  16 gi.require_version('GstWebRTC', '1.0')
  17 from gi.repository import GstWebRTC
  18
  19 gi.require_version('GstSdp', '1.0')
  20 from gi.repository import GstSdp
  21
  22 log = logging.getLogger(__name__)
  23
  24
  25 class Lagarde:
  26     def __init__(self):
  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
  34         self.pipe = None
  35         self.webrtcbin = None
  36
  37     def on_negotiation_needed(self, element):
  38         log.debug('on_negotiation_needed')
  39
  40     def on_ice_candidate(self, element, mlineindex, candidate):
  41         log.debug('on_ice_candidate')
  42         self.generated_ice_candidates.append((mlineindex, candidate))
  43
  44     def webrtcbin_pad_added(self, element, pad):
  45         log.debug('webrtcbin_pad_added')
  46         if pad.direction != Gst.PadDirection.SRC:
  47             return
  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)
  53
  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')
  58             return
  59
  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
  64             name = s.get_name()
  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')
  70                 self.pipe.add(q)
  71                 self.pipe.add(conv)
  72                 self.pipe.add(sink)
  73                 self.pipe.sync_children_states()
  74                 pad.link(q.get_static_pad('sink'))
  75                 q.link(conv)
  76                 conv.link(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')
  82                 self.pipe.add(q)
  83                 self.pipe.add(conv)
  84                 self.pipe.add(resample)
  85                 self.pipe.add(sink)
  86                 self.pipe.sync_children_states()
  87                 pad.link(q.get_static_pad('sink'))
  88                 q.link(conv)
  89                 conv.link(resample)
  90                 resample.link(sink)
  91
  92     async def listen_to_gstreamer_bus(self):
  93         Gst.init(None)
  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)
 102         try:
 103             while True:
 104                 if bus.have_pending():
 105                     msg = bus.pop()
 106                     if msg.type == Gst.MessageType.ERROR:
 107                         log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
 108                         return
 109                     elif msg.type == Gst.MessageType.EOS:  # end of stream
 110                         log.info(f'Gstreamer message bus reports end of stream')
 111                         return
 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
 123
 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,
 139                         'Type': "gotAnswer",
 140                         'Value': json.dumps({
 141                             'type': 'answer',
 142                             'sdp': sdp_answer
 143                         })
 144                     })
 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)
 150
 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'])
 155
 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],
 163                     })
 164                     icemsg = json.dumps({
 165                         'SessionID': self.session_id,
 166                         'Type': 'addCalleeIceCandidate',
 167                         'Value': icemsg_value,
 168                     })
 169                     log.info(f'Send ICE candidate: {icemsg_value}')
 170                     await self.websocket.send(icemsg)
 171
 172                 else:
 173                     await asyncio.sleep(0.1)
 174         finally:
 175             self.pipe.set_state(Gst.State.NULL)
 176
 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}')
 190                 self.sdp_offer = 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
 198                 return
 199             else:
 200                 log.error(f'Unknown message type {msg_type}')
 201
 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
 206         try:
 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)
 213                 for d in done:
 214                     d.result()
 215                 for p in pending:
 216                     p.cancel()
 217         except OSError as e:
 218             print(e)
 219
 220
 221 def main():
 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()
 227     lagarde = Lagarde()
 228     asyncio.run(lagarde.run(args.uri), debug=True)
 229
 230
 231 if __name__ == '__main__':
 232     main()