lagarde.py

   1 #!/usr/bin/python3
   2 import argparse
   3 import asyncio
   4 import json
   5 import logging
   6 import ssl
   7 import queue
   8
   9 import gi
  10 import websockets
  11
  12 gi.require_version('Gst', '1.0')
  13 from gi.repository import Gst
  14
  15 gi.require_version('GstWebRTC', '1.0')
  16 from gi.repository import GstWebRTC
  17
  18 gi.require_version('GstSdp', '1.0')
  19 from gi.repository import GstSdp
  20
  21 log = logging.getLogger(__name__)
  22
  23
  24 class Events:
  25     def __init__(self):
  26         self.sdp_offer = queue.Queue()
  27         self.sdp_answer = queue.Queue()
  28         self.generated_ice_candidates = queue.Queue()
  29         self.received_ice_candidates = queue.Queue()
  30         self.sdp_info = queue.Queue()  # (sdp_mids, user_fragments)
  31         self.room_left = queue.Queue()
  32
  33
  34 class SignalingClient:
  35     def __init__(self, events: Events, uri):
  36         self.events = events
  37         self.uri = uri
  38         self.ssl_context = ssl.SSLContext()
  39         self.ssl_context.check_hostname = False
  40         self.ssl_context.verify_mode = ssl.CERT_NONE
  41         self.session_id = None
  42
  43     async def receive(self, uri):
  44         async for msg in self.websocket:
  45             msg_json = json.loads(msg)
  46             msg_type = msg_json['Type']
  47             msg_value = msg_json['Value']
  48             assert self.session_id is None or self.session_id == msg_json['SessionID']
  49             if msg_type == 'newSession':
  50                 self.session_id = msg_json['SessionID']
  51                 log.info(f"New session {self.session_id}")
  52             elif msg_type == 'gotOffer':
  53                 value_json = json.loads(msg_value)
  54                 sdp = value_json['sdp']
  55                 log.info(f'Got SDP offer')
  56                 log.debug(f'SDP offer:\n{sdp}')
  57                 self.events.sdp_offer.put_nowait(sdp)
  58             elif msg_type == 'addCallerIceCandidate':
  59                 value_json = json.loads(msg_value)
  60                 log.info(f'Got ICE candidate')
  61                 log.debug(f'ICE candidate: {value_json}')
  62                 self.events.received_ice_candidates.put_nowait(value_json)
  63             elif msg_type == 'roomNotFound':
  64                 log.error(f'The room was not found: {uri}')
  65                 return
  66             elif msg_type == 'roomClosed':
  67                 log.info(f'Oh noes, the room went away (session {self.session_id})!')
  68                 self.events.room_left.put_nowait(True)
  69                 return
  70             else:
  71                 log.error(f'Unknown message type {msg_type}')
  72
  73     async def send(self):
  74         sdp_mids = None
  75         user_fragments = None
  76         while True:
  77             if self.events.sdp_answer.qsize() > 0:
  78                 sdp_answer = self.events.sdp_answer.get_nowait()
  79                 sdp_answer_msg = json.dumps({
  80                     'SessionID': self.session_id,
  81                     'Type': "gotAnswer",
  82                     'Value': json.dumps({
  83                         'type': 'answer',
  84                         'sdp': sdp_answer
  85                     })
  86                 })
  87                 await self.websocket.send(sdp_answer_msg)
  88
  89             elif self.events.sdp_info.qsize() > 0:
  90                 sdp_mids, user_fragments = self.events.sdp_info.get_nowait()
  91
  92             elif self.events.generated_ice_candidates.qsize() > 0 \
  93                     and sdp_mids is not None and user_fragments is not None:
  94                 mlineindex, candidate = self.events.generated_ice_candidates.get_nowait()
  95                 sdp_mid = sdp_mids[mlineindex]
  96                 user_fragment = user_fragments[mlineindex]
  97                 icemsg_value = json.dumps({
  98                     "candidate": candidate,
  99                     "sdpMid": sdp_mid,
 100                     "sdpMLineIndex": mlineindex,
 101                     "usernameFragment": user_fragment,
 102                 })
 103                 icemsg = json.dumps({
 104                     'SessionID': self.session_id,
 105                     'Type': 'addCalleeIceCandidate',
 106                     'Value': icemsg_value,
 107                 })
 108                 log.info(f'Send ICE candidate')
 109                 log.debug(f'ICE candidate: {icemsg_value}')
 110                 await self.websocket.send(icemsg)
 111
 112             else:
 113                 await asyncio.sleep(0.2)
 114
 115     async def run(self):
 116         self.session_id = None
 117         async with websockets.connect(self.uri, ssl=self.ssl_context, close_timeout=0.5) as self.websocket:
 118             receive_task = asyncio.Task(self.receive(self.uri))
 119             send_task = asyncio.Task(self.send())
 120             done, pending = await asyncio.wait([receive_task, send_task], return_when=asyncio.FIRST_COMPLETED)
 121             for task in pending:
 122                 task.cancel()
 123
 124
 125 class WebRTCClient:
 126     def __init__(self, events: Events, rtmp_uri: str):
 127         self.events = events
 128         self.rtmp_uri = rtmp_uri
 129         self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
 130         self.pipe = Gst.Pipeline.new("pipeline")
 131         Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
 132         self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed)
 133         self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate)
 134         self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added)
 135
 136     def on_negotiation_needed(self, element):
 137         log.info('on_negotiation_needed')
 138
 139     def on_ice_candidate(self, element, mlineindex, candidate):
 140         log.info('on_ice_candidate')
 141         self.events.generated_ice_candidates.put_nowait((mlineindex, candidate))
 142
 143     def webrtcbin_pad_added(self, element, pad):
 144         log.info('webrtcbin_pad_added')
 145         if pad.direction != Gst.PadDirection.SRC:
 146             return
 147         demux = Gst.ElementFactory.make('rtpptdemux')
 148         assert demux
 149         self.pipe.add(demux)
 150         demux.connect('pad-added', self.demux_pad_added)
 151         demux.sync_state_with_parent()
 152         ok = self.webrtcbin.link(demux)
 153         assert ok
 154
 155     def demux_pad_added(self, element, pad):
 156         log.info('demux_pad_added')
 157         if pad.direction != Gst.PadDirection.SRC:
 158             return
 159         depay = Gst.ElementFactory.make('rtpvp8depay')
 160         assert depay
 161         self.pipe.add(depay)
 162         ok = element.link(depay)
 163         assert ok
 164
 165         q = Gst.ElementFactory.make('queue')
 166         sink = Gst.ElementFactory.make('rtspclientsink')
 167         sink.set_property('location', self.rtmp_uri)
 168         sink.set_property('debug', True)
 169         assert q and sink
 170
 171         self.pipe.add(q)
 172         self.pipe.add(sink)
 173         ok = depay.link(q)
 174         assert ok
 175         ok = q.link(sink)
 176         assert ok
 177
 178         self.pipe.sync_children_states()
 179         self.pipe.set_state(Gst.State.PLAYING)
 180         self.pipe.sync_children_states()
 181
 182     def set_remote_desciption_done(self, gst_promise):
 183         gst_promise = Gst.Promise.new_with_change_func(self.create_answer_done)
 184         self.webrtcbin.emit('create-answer', None, gst_promise)
 185
 186     def create_answer_done(self, gst_promise):
 187         reply = gst_promise.get_reply()
 188         answer = reply.get_value('answer')
 189         sdp_message = answer.sdp
 190         mids = [sdp_message.get_media(i).get_attribute_val('mid')
 191                 for i in range(sdp_message.medias_len())]
 192         user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag')
 193                           for i in range(sdp_message.medias_len())]
 194         self.events.sdp_info.put_nowait((mids, user_fragments))
 195         sdp_answer = sdp_message.as_text()
 196         log.info(f'Send SDP answer')
 197         log.debug(f'SDP answer:\n{sdp_answer}')
 198         self.events.sdp_answer.put_nowait(sdp_answer)
 199         gst_promise = Gst.Promise.new_with_change_func(self.set_local_description_done)
 200         self.webrtcbin.emit('set-local-description', answer, gst_promise)
 201
 202     def set_local_description_done(self, gst_promise):
 203         gst_promise.get_reply()
 204
 205     async def run(self):
 206         bus = Gst.Pipeline.get_bus(self.pipe)
 207         self.pipe.set_state(Gst.State.PLAYING)
 208         try:
 209             while True:
 210                 if bus.have_pending():
 211                     msg = bus.pop()
 212                     if msg.type == Gst.MessageType.ERROR:
 213                         log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
 214                         return
 215                     elif msg.type == Gst.MessageType.EOS:  # end of stream
 216                         log.info(f'Gstreamer message bus reports end of stream')
 217                         return
 218                 elif self.events.sdp_offer.qsize() > 0:
 219                     sdp_offer = self.events.sdp_offer.get_nowait()
 220                     res, sm = GstSdp.SDPMessage.new()
 221                     assert res == GstSdp.SDPResult.OK
 222                     GstSdp.sdp_message_parse_buffer(bytes(sdp_offer.encode()), sm)
 223                     # the three lines above can also be done this way in new versions of GStreamer:
 224                     # sm = GstSdp.SDPMessage.new_from_text(sdp_offer)
 225                     rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
 226                     gst_promise = Gst.Promise.new_with_change_func(self.set_remote_desciption_done)
 227                     self.webrtcbin.emit('set-remote-description', rd, gst_promise)
 228
 229                 elif self.events.received_ice_candidates.qsize() > 0:
 230                     ic = self.events.received_ice_candidates.get_nowait()
 231                     if ic['candidate'] != '':
 232                         self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
 233
 234                 elif self.events.room_left.qsize() > 0:
 235                     self.events.room_left.get_nowait()
 236                     return
 237
 238                 else:
 239                     await asyncio.sleep(0.1)
 240         finally:
 241             self.pipe.set_state(Gst.State.NULL)
 242
 243
 244 async def run_repeated(task):
 245     while True:
 246         await task()
 247         await asyncio.sleep(0.1)
 248
 249
 250 async def run(laplace_uri: str, rtmp_uri: str):
 251     try:
 252         events = Events()
 253         webrtc = WebRTCClient(events, rtmp_uri)
 254         signaling = SignalingClient(events, laplace_uri)
 255
 256         webrtc_task = asyncio.Task(webrtc.run())
 257         signaling_task = asyncio.Task(signaling.run())
 258
 259         done, pending = await asyncio.wait([webrtc_task, signaling_task], return_when=asyncio.FIRST_COMPLETED)
 260
 261         for task in done:
 262             task.result()
 263         for task in pending:
 264             task.cancel()
 265     except OSError as e:
 266         print(e)
 267
 268
 269 def main():
 270     logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s')
 271     default_source = 'wss://localhost:1234/ws_connect?id=cug'
 272     default_dest = 'rtsp://localhost:8554/cug'
 273     parser = argparse.ArgumentParser()
 274     parser.add_argument('-s', '--source', default=default_source,
 275                         help=f'Laplace signalling websocket URI, default: {default_source}')
 276     parser.add_argument('-d', '--destination', default=default_dest,
 277                         help=f'RTMP server URI, default: {default_dest}')
 278     args = parser.parse_args()
 279
 280     Gst.init(None)
 281     asyncio.run(run(args.source, args.destination))
 282
 283
 284 if __name__ == '__main__':
 285     main()