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 from typing import 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 Events:
  26     def __init__(self):
  27         self.sdp_offer = queue.Queue()
  28         self.sdp_answer = queue.Queue()
  29         self.generated_ice_candidates = queue.Queue()
  30         self.received_ice_candidates = queue.Queue()
  31         self.sdp_info = queue.Queue()  # (sdp_mids, user_fragments)
  32         self.room_left = queue.Queue()
  33
  34
  35 class SignalingClient:
  36     def __init__(self, events: Events, uri):
  37         self.events = events
  38         self.uri = uri
  39         self.ssl_context = ssl.SSLContext()
  40         self.ssl_context.check_hostname = False
  41         self.ssl_context.verify_mode = ssl.CERT_NONE
  42         self.session_id = None
  43
  44     async def receive(self, uri):
  45         async for msg in self.websocket:
  46             msg_json = json.loads(msg)
  47             msg_type = msg_json['Type']
  48             msg_value = msg_json['Value']
  49             assert self.session_id is None or self.session_id == msg_json['SessionID']
  50             if msg_type == 'newSession':
  51                 self.session_id = msg_json['SessionID']
  52                 log.info(f"New session {self.session_id}")
  53             elif msg_type == 'gotOffer':
  54                 value_json = json.loads(msg_value)
  55                 sdp = value_json['sdp']
  56                 log.info(f'Got SDP offer')
  57                 log.debug(f'SDP offer:\n{sdp}')
  58                 self.events.sdp_offer.put_nowait(sdp)
  59             elif msg_type == 'addCallerIceCandidate':
  60                 value_json = json.loads(msg_value)
  61                 log.info(f'Got ICE candidate')
  62                 log.debug(f'ICE candidate: {value_json}')
  63                 self.events.received_ice_candidates.put_nowait(value_json)
  64             elif msg_type == 'roomNotFound':
  65                 log.error(f'The room was not found: {uri}')
  66                 return
  67             elif msg_type == 'roomClosed':
  68                 log.info(f'Oh noes, the room went away (session {self.session_id})!')
  69                 self.events.room_left.put_nowait(True)
  70                 return
  71             else:
  72                 log.error(f'Unknown message type {msg_type}')
  73
  74     async def send(self):
  75         sdp_mids = None
  76         user_fragments = None
  77         while True:
  78             if self.events.sdp_answer.qsize() > 0:
  79                 sdp_answer = self.events.sdp_answer.get_nowait()
  80                 sdp_answer_msg = json.dumps({
  81                     'SessionID': self.session_id,
  82                     'Type': "gotAnswer",
  83                     'Value': json.dumps({
  84                         'type': 'answer',
  85                         'sdp': sdp_answer
  86                     })
  87                 })
  88                 await self.websocket.send(sdp_answer_msg)
  89
  90             elif self.events.sdp_info.qsize() > 0:
  91                 sdp_mids, user_fragments = self.events.sdp_info.get_nowait()
  92
  93             elif self.events.generated_ice_candidates.qsize() > 0 \
  94                     and sdp_mids is not None and user_fragments is not None:
  95                 mlineindex, candidate = self.events.generated_ice_candidates.get_nowait()
  96                 sdp_mid = sdp_mids[mlineindex]
  97                 user_fragment = user_fragments[mlineindex]
  98                 icemsg_value = json.dumps({
  99                     "candidate": candidate,
 100                     "sdpMid": sdp_mid,
 101                     "sdpMLineIndex": mlineindex,
 102                     "usernameFragment": user_fragment,
 103                 })
 104                 icemsg = json.dumps({
 105                     'SessionID': self.session_id,
 106                     'Type': 'addCalleeIceCandidate',
 107                     'Value': icemsg_value,
 108                 })
 109                 log.info(f'Send ICE candidate')
 110                 log.debug(f'ICE candidate: {icemsg_value}')
 111                 await self.websocket.send(icemsg)
 112
 113             else:
 114                 await asyncio.sleep(0.2)
 115
 116     async def run(self):
 117         self.session_id = None
 118         async with websockets.connect(self.uri, ssl=self.ssl_context, close_timeout=0.5) as self.websocket:
 119             receive_task = asyncio.Task(self.receive(self.uri))
 120             send_task = asyncio.Task(self.send())
 121             done, pending = await asyncio.wait([receive_task, send_task], return_when=asyncio.FIRST_COMPLETED)
 122             for task in pending:
 123                 task.cancel()
 124
 125
 126 class WebRTCClient:
 127     def __init__(self, events: Events, rtmp_uri: str):
 128         self.events = events
 129         self.rtmp_uri = rtmp_uri
 130         self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
 131         self.pipe = Gst.Pipeline.new("pipeline")
 132         Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
 133         self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed)
 134         self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate)
 135         self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added)
 136
 137     def on_negotiation_needed(self, element):
 138         log.info('on_negotiation_needed')
 139
 140     def on_ice_candidate(self, element, mlineindex, candidate):
 141         log.info('on_ice_candidate')
 142         self.events.generated_ice_candidates.put_nowait((mlineindex, candidate))
 143
 144     def webrtcbin_pad_added(self, element, pad):
 145         log.info('webrtcbin_pad_added')
 146         if pad.direction != Gst.PadDirection.SRC:
 147             return
 148         decodebin = Gst.ElementFactory.make('decodebin')
 149         decodebin.connect('pad-added', self.decodebin_pad_added)
 150         self.pipe.add(decodebin)
 151         decodebin.sync_state_with_parent()
 152         self.webrtcbin.link(decodebin)
 153
 154     def decodebin_pad_added(self, element, pad):
 155         log.info('decodebin_pad_added')
 156         if not pad.has_current_caps():
 157             log.info(pad, 'has no caps, ignoring')
 158             return
 159         caps = pad.get_current_caps()
 160         padsize = caps.get_size()
 161
 162         for i in range(padsize):
 163             s = caps.get_structure(i)  # Gst.Structure
 164             name = s.get_name()
 165             if name.startswith('video'):
 166                 q = Gst.ElementFactory.make('queue')
 167                 conv = Gst.ElementFactory.make('videoconvert')
 168                 enc = Gst.ElementFactory.make('x264enc')
 169                 enc.set_property('bitrate', 1000)
 170                 enc.set_property('tune', 'zerolatency')
 171                 capsfilter = Gst.ElementFactory.make('capsfilter')
 172                 capsfilter.set_properties(Gst.Caps.from_string('video/x-h264,stream-format=(string)avc'))
 173                 flvmux = Gst.ElementFactory.make('flvmux')
 174                 flvmux.set_property('streamable', True)
 175                 sink = Gst.ElementFactory.make('rtmpsink')
 176                 sink.set_property('location', self.rtmp_uri)
 177                 assert q and conv and enc and capsfilter and flvmux and sink
 178
 179                 self.pipe.add(q)
 180                 self.pipe.add(conv)
 181                 self.pipe.add(enc)
 182                 self.pipe.add(capsfilter)
 183                 self.pipe.add(flvmux)
 184                 self.pipe.add(sink)
 185
 186                 q_pad_sink = q.get_static_pad('sink')
 187                 assert q_pad_sink
 188                 pad_link_return = pad.link(q_pad_sink)
 189                 assert pad_link_return == Gst.PadLinkReturn.OK
 190
 191                 ok = q.link(conv)
 192                 assert ok
 193                 ok = conv.link(enc)
 194                 assert ok
 195                 ok = enc.link(capsfilter)
 196                 assert ok
 197                 ok = capsfilter.link(flvmux)
 198                 assert ok
 199                 ok = flvmux.link(sink)
 200                 assert ok
 201                 self.pipe.set_state(Gst.State.PLAYING)
 202                 self.pipe.sync_children_states()
 203
 204             elif name.startswith('audio'):
 205                 q = Gst.ElementFactory.make('queue')
 206                 conv = Gst.ElementFactory.make('audioconvert')
 207                 resample = Gst.ElementFactory.make('audioresample')
 208                 sink = Gst.ElementFactory.make('autoaudiosink')
 209                 self.pipe.add(q)
 210                 self.pipe.add(conv)
 211                 self.pipe.add(resample)
 212                 self.pipe.add(sink)
 213                 self.pipe.sync_children_states()
 214                 pad.link(q.get_static_pad('sink'))
 215                 q.link(conv)
 216                 conv.link(resample)
 217                 resample.link(sink)
 218
 219     def set_remote_desciption_done(self, gst_promise):
 220         gst_promise = Gst.Promise.new_with_change_func(self.create_answer_done)
 221         self.webrtcbin.emit('create-answer', None, gst_promise)
 222
 223     def create_answer_done(self, gst_promise):
 224         reply = gst_promise.get_reply()
 225         answer = reply.get_value('answer')
 226         sdp_message = answer.sdp
 227         mids = [sdp_message.get_media(i).get_attribute_val('mid')
 228                 for i in range(sdp_message.medias_len())]
 229         user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag')
 230                           for i in range(sdp_message.medias_len())]
 231         self.events.sdp_info.put_nowait((mids, user_fragments))
 232         sdp_answer = sdp_message.as_text()
 233         log.info(f'Send SDP answer')
 234         log.debug(f'SDP answer:\n{sdp_answer}')
 235         self.events.sdp_answer.put_nowait(sdp_answer)
 236         gst_promise = Gst.Promise.new_with_change_func(self.set_local_description_done)
 237         self.webrtcbin.emit('set-local-description', answer, gst_promise)
 238
 239     def set_local_description_done(self, gst_promise):
 240         gst_promise.get_reply()
 241
 242     async def run(self):
 243         bus = Gst.Pipeline.get_bus(self.pipe)
 244         self.pipe.set_state(Gst.State.PLAYING)
 245         try:
 246             while True:
 247                 if bus.have_pending():
 248                     msg = bus.pop()
 249                     if msg.type == Gst.MessageType.ERROR:
 250                         log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
 251                         return
 252                     elif msg.type == Gst.MessageType.EOS:  # end of stream
 253                         log.info(f'Gstreamer message bus reports end of stream')
 254                         return
 255                 elif self.events.sdp_offer.qsize() > 0:
 256                     sdp_offer = self.events.sdp_offer.get_nowait()
 257                     res, sm = GstSdp.SDPMessage.new()
 258                     assert res == GstSdp.SDPResult.OK
 259                     GstSdp.sdp_message_parse_buffer(bytes(sdp_offer.encode()), sm)
 260                     # the three lines above can also be done this way in new versions of GStreamer:
 261                     # sm = GstSdp.SDPMessage.new_from_text(sdp_offer)
 262                     rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
 263                     gst_promise = Gst.Promise.new_with_change_func(self.set_remote_desciption_done)
 264                     self.webrtcbin.emit('set-remote-description', rd, gst_promise)
 265
 266                 elif self.events.received_ice_candidates.qsize() > 0:
 267                     ic = self.events.received_ice_candidates.get_nowait()
 268                     if ic['candidate'] != '':
 269                         self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
 270
 271                 elif self.events.room_left.qsize() > 0:
 272                     self.events.room_left.get_nowait()
 273                     return
 274
 275                 else:
 276                     await asyncio.sleep(0.1)
 277         finally:
 278             self.pipe.set_state(Gst.State.NULL)
 279
 280
 281 async def run_room(laplace_uri: str, rtmp_uri: str):
 282     try:
 283         events = Events()
 284         webrtc = WebRTCClient(events, rtmp_uri)
 285         signaling = SignalingClient(events, laplace_uri)
 286
 287         webrtc_task = asyncio.Task(webrtc.run())
 288         signaling_task = asyncio.Task(signaling.run())
 289
 290         done, pending = await asyncio.wait([webrtc_task, signaling_task], return_when=asyncio.FIRST_COMPLETED)
 291
 292         for task in done:
 293             task.result()
 294         for task in pending:
 295             task.cancel()
 296     except OSError as e:
 297         print(e)
 298
 299
 300 async def run_room_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float):
 301     while True:
 302         await run_room(laplace_uri, rtmp_uri)
 303         await asyncio.sleep(sleep_time)
 304
 305
 306 async def run_rooms(laplace_base_uri: str, rtmp_base_uri: str, rooms: List[str], retry: bool):
 307     tasks = []
 308     for room in rooms:
 309         laplace_uri = laplace_base_uri + room  # TODO: encode
 310         rtmp_uri = rtmp_base_uri + room  # TODO: encode
 311         if retry:
 312             tasks.append(run_room_repeated(laplace_uri, rtmp_uri, 2.))
 313         else:
 314             tasks.append(run_room(laplace_uri, rtmp_uri))
 315     await asyncio.gather(*tasks)
 316
 317
 318 def main():
 319     logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s')
 320     default_source = 'wss://localhost:1234/ws_connect?id='
 321     default_dest = 'rtmp://localhost:1935/'
 322     default_room = 'cug'
 323     parser = argparse.ArgumentParser()
 324     parser.add_argument('-s', '--source', default=default_source,
 325                         help=f'Laplace signalling websocket base URI, default: {default_source}')
 326     parser.add_argument('-d', '--destination', default=default_dest,
 327                         help=f'RTMP server base URI, default: {default_dest}')
 328     parser.add_argument('-r', '--retry', action='store_true', help=f'Retry forever if room not found or closed')
 329     parser.add_argument('room', nargs='*', help=f'Room names to be used, "{default_room}" if omitted')
 330     args = parser.parse_args()
 331
 332     Gst.init(None)
 333     rooms = args.room
 334     if len(rooms) == 0:
 335         rooms = [default_room]
 336     asyncio.run(run_rooms(args.source, args.destination, rooms, args.retry))
 337
 338
 339 if __name__ == '__main__':
 340     main()