lagarde.py

   1 #!/usr/bin/python3
   2
   3 import argparse
   4 import asyncio
   5 import datetime
   6 import json
   7 import logging
   8 import pathlib
   9 import ssl
  10 import sys
  11 from typing import Optional, List
  12
  13 import websockets
  14
  15 import gi
  16
  17 gi.require_version('Gst', '1.0')
  18 from gi.repository import Gst
  19
  20 gi.require_version('GstWebRTC', '1.0')
  21 from gi.repository import GstWebRTC
  22
  23 gi.require_version('GstSdp', '1.0')
  24 from gi.repository import GstSdp
  25
  26 log = logging.getLogger(__name__)
  27
  28
  29 class Lagarde:
  30     def __init__(self):
  31         self.sdp_offer: Optional[str] = None
  32         self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
  33         self.session_id = None
  34         self.received_ice_candidates = []
  35         self.generated_ice_candidates = []
  36         self.user_fragments: Optional[List] = None
  37         self.mids: Optional[List] = None
  38         self.pipe = None
  39         self.webrtcbin = None
  40
  41     def on_negotiation_needed(self, element):
  42         log.info('on_negotiation_needed')
  43
  44     def on_ice_candidate(self, element, mlineindex, candidate):
  45         log.info('on_ice_candidate')
  46         self.generated_ice_candidates.append((mlineindex, candidate))
  47
  48     def webrtcbin_pad_added(self, element, pad):
  49         log.info('webrtcbin_pad_added')
  50         if pad.direction != Gst.PadDirection.SRC:
  51             return
  52         decodebin = Gst.ElementFactory.make('decodebin')
  53         decodebin.connect('pad-added', self.decodebin_pad_added)
  54         self.pipe.add(decodebin)
  55         decodebin.sync_state_with_parent()
  56         self.webrtcbin.link(decodebin)
  57
  58     def decodebin_pad_added(self, element, pad):
  59         log.info('decodebin_pad_added')
  60         if not pad.has_current_caps():
  61             log.info(pad, 'has no caps, ignoring')
  62             return
  63
  64         caps = pad.get_current_caps()
  65         assert (len(caps))
  66         s = caps[0]
  67         name = s.get_name()
  68         if name.startswith('video'):
  69             q = Gst.ElementFactory.make('queue')
  70             conv = Gst.ElementFactory.make('videoconvert')
  71             sink = Gst.ElementFactory.make('autovideosink')
  72             self.pipe.add(q, conv, 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, conv, resample, sink)
  83             self.pipe.sync_children_states()
  84             pad.link(q.get_static_pad('sink'))
  85             q.link(conv)
  86             conv.link(resample)
  87             resample.link(sink)
  88
  89     async def listen_to_gstreamer_bus(self):
  90         Gst.init(None)
  91         self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
  92         self.pipe = Gst.Pipeline.new("pipeline")
  93         Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
  94         bus = Gst.Pipeline.get_bus(self.pipe)
  95         self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed)
  96         self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate)
  97         self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added)
  98         self.pipe.set_state(Gst.State.PLAYING)
  99         try:
 100             while True:
 101                 if bus.have_pending():
 102                     msg = bus.pop()  # Gst.Message, has to be unref'ed.
 103                     if msg.type != Gst.MessageType.STATE_CHANGED:
 104                         # log.info(f'Receive Gst.Message: {msg.type}, {msg.seqnum}, {msg.get_structure()}')
 105                         # log.info(f'{webrtcbin.props.signaling_state} {webrtcbin.props.ice_gathering_state} {webrtcbin.props.ice_connection_state}')
 106                         # Gst.Message.unref(msg)
 107                         pass
 108                 elif self.sdp_offer is not None:
 109                     res, sm = GstSdp.SDPMessage.new()
 110                     assert res == GstSdp.SDPResult.OK
 111                     GstSdp.sdp_message_parse_buffer(bytes(self.sdp_offer.encode()), sm)
 112                     # the three lines above can also be done this way in new versions of GStreamer:
 113                     # sm = GstSdp.SDPMessage.new_from_text(sdp_offer)
 114                     rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
 115                     gst_promise = Gst.Promise.new()
 116                     self.webrtcbin.emit('set-remote-description', rd, gst_promise)
 117                     gst_promise.wait()
 118                     self.sdp_offer = None
 119
 120                     log.info('create-answer')
 121                     gst_promise = Gst.Promise.new()
 122                     self.webrtcbin.emit('create-answer', None, gst_promise)
 123                     result = gst_promise.wait()
 124                     assert result == Gst.PromiseResult.REPLIED
 125                     reply = gst_promise.get_reply()
 126                     answer = reply.get_value('answer')
 127                     sdp_message = answer.sdp
 128                     self.user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag')
 129                                           for i in range(sdp_message.medias_len())]
 130                     self.mids = [sdp_message.get_media(i).get_attribute_val('mid')
 131                                           for i in range(sdp_message.medias_len())]
 132                     sdp_answer = sdp_message.as_text()
 133                     log.info(sdp_answer)
 134                     sdp_answer_msg = json.dumps({
 135                         'SessionID': self.session_id,
 136                         'Type': "gotAnswer",
 137                         'Value': json.dumps({
 138                             'type': 'answer',
 139                             'sdp': sdp_answer
 140                         })
 141                     })
 142                     gst_promise = Gst.Promise.new()
 143                     self.webrtcbin.emit('set-local-description', answer, gst_promise)
 144                     gst_promise.wait()
 145                     gst_promise.get_reply()
 146                     await self.websocket.send(sdp_answer_msg)
 147
 148                 elif len(self.received_ice_candidates) > 0:
 149                     ic = self.received_ice_candidates.pop(0)
 150                     if ic['candidate'] != '':
 151                         self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
 152
 153                 elif len(self.generated_ice_candidates) > 0:
 154                     mlineindex, candidate = self.generated_ice_candidates.pop(0)
 155                     icemsg = json.dumps({
 156                         'SessionID': self.session_id,
 157                         'Type': 'addCalleeIceCandidate',
 158                         'Value': json.dumps({
 159                             "candidate": candidate,
 160                             "sdpMid": self.mids[mlineindex],
 161                             "sdpMLineIndex": mlineindex,
 162                             "usernameFragment": self.user_fragments[mlineindex],
 163                         })
 164                     })
 165                     log.info(f'send_ice_candidate_message with {icemsg}')
 166                     await self.websocket.send(icemsg)
 167
 168                 else:
 169                     await asyncio.sleep(0.1)
 170         finally:
 171             self.pipe.set_state(Gst.State.NULL)
 172
 173     async def talk_to_websocket(self, uri):
 174         ssl_context = ssl.SSLContext()
 175         ssl_context.check_hostname = False
 176         ssl_context.verify_mode = ssl.CERT_NONE
 177         async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
 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                 self.session_id = msg_json['SessionID']
 183                 log.info(f"receive for session {self.session_id} type {msg_type}")
 184                 if msg_type == 'newSession':
 185                     pass
 186                 elif msg_type == 'gotOffer':
 187                     value_json = json.loads(msg_value)
 188                     sdp = value_json['sdp']
 189                     log.info(f'SDP: {sdp}')
 190                     self.sdp_offer = sdp
 191                 elif msg_type == 'addCallerIceCandidate':
 192                     value_json = json.loads(msg_value)
 193                     log.info(f'ICE: {value_json}')
 194                     self.received_ice_candidates.append(value_json)
 195                 elif msg_type == 'roomClosed':
 196                     log.info('Oh noes, the room went away!')
 197                     # and here we should clean up
 198                 else:
 199                     log.error(f'Unknown message type {msg_type}')
 200
 201     async def run(self, uri):
 202         talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
 203         listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
 204         done, pending = await asyncio.wait(
 205             [talk_to_websocket_task, listen_to_gstreamer_bus_task],
 206             return_when=asyncio.FIRST_COMPLETED)
 207         for d in done:
 208             d.result()
 209         for p in pending:
 210             p.cancel()
 211
 212
 213 def main():
 214     logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s')
 215     parser = argparse.ArgumentParser()
 216     parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
 217                         help='Signalling server URI')
 218     args = parser.parse_args()
 219     lagarde = Lagarde()
 220     asyncio.run(lagarde.run(args.uri), debug=True)
 221
 222
 223 if __name__ == '__main__':
 224     main()