X-Git-Url: https://git.toastfreeware.priv.at/toast/stream2beamer.git/blobdiff_plain/7a5738f7c5cf9ff96c5ac58839223fcc738482be..e97741140341d1c98f6cb350d819a988c8b73bbd:/lagarde.py diff --git a/lagarde.py b/lagarde.py index b829595..f31bc2d 100755 --- a/lagarde.py +++ b/lagarde.py @@ -1,12 +1,11 @@ #!/usr/bin/python3 - import argparse import asyncio import json import logging import ssl import queue -from typing import Optional, List +from typing import List import gi import websockets @@ -20,49 +19,127 @@ from gi.repository import GstWebRTC gi.require_version('GstSdp', '1.0') from gi.repository import GstSdp -gi.require_version('GstRtspServer', '1.0') -from gi.repository import Gst, GstRtspServer, GObject, GLib - log = logging.getLogger(__name__) -class GstreamerRtspServer(): - def __init__(self): - server = GstRtspServer.RTSPServer() - server.set_address("::") - server.set_service('8554') # port as string - factory = GstRtspServer.RTSPMediaFactory() - # factory.set_launch("intervideosrc ! decodebin ! theoraenc ! queue ! rtptheorapay name=pay0") - # factory.set_launch("intervideosrc ! decodebin ! videoconvert ! video/x-raw,format=I420 ! theoraenc ! queue ! rtptheorapay name=pay0") - # factory.set_launch("intervideosrc ! decodebin ! videorate ! videoconvert ! video/x-raw,format=I420,framerate=10/1 ! theoraenc ! queue ! rtptheorapay name=pay0") - # factory.set_launch("intervideosrc ! decodebin ! videoconvert ! video/x-raw,format=I420 ! x264enc ! queue ! rtph264pay pt=96 name=pay0") - factory.set_launch("intervideosrc ! decodebin ! x264enc ! queue ! rtph264pay pt=96 name=pay0") - # factory.set_launch("intervideosrc ! decodebin ! videorate ! videoconvert ! video/x-raw,framerate=10/1 ! x264enc ! queue ! rtph264pay pt=96 name=pay0") - factory.set_shared(True) - mountPoints = server.get_mount_points() - mountPoints.add_factory("/cug", factory) - server.attach() - self.server = server - - -class Lagarde: +class Events: def __init__(self): - self.sdp_offer: Optional[str] = None - self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None - self.session_id = None - self.received_ice_candidates = queue.Queue() + self.sdp_offer = queue.Queue() + self.sdp_answer = queue.Queue() self.generated_ice_candidates = queue.Queue() - self.user_fragments: Optional[List] = None - self.mids: Optional[List] = None - self.pipe = None - self.webrtcbin = None + self.received_ice_candidates = queue.Queue() + self.sdp_info = queue.Queue() # (sdp_mids, user_fragments) + self.room_left = queue.Queue() + + +class SignalingClient: + def __init__(self, events: Events, uri): + self.events = events + self.uri = uri + self.ssl_context = ssl.SSLContext() + self.ssl_context.check_hostname = False + self.ssl_context.verify_mode = ssl.CERT_NONE + self.session_id = None + + async def receive(self, uri): + async for msg in self.websocket: + msg_json = json.loads(msg) + msg_type = msg_json['Type'] + msg_value = msg_json['Value'] + assert self.session_id is None or self.session_id == msg_json['SessionID'] + if msg_type == 'newSession': + self.session_id = msg_json['SessionID'] + log.info(f"New session {self.session_id}") + elif msg_type == 'gotOffer': + value_json = json.loads(msg_value) + sdp = value_json['sdp'] + log.info(f'Got SDP offer') + log.debug(f'SDP offer:\n{sdp}') + self.events.sdp_offer.put_nowait(sdp) + elif msg_type == 'addCallerIceCandidate': + value_json = json.loads(msg_value) + log.info(f'Got ICE candidate') + log.debug(f'ICE candidate: {value_json}') + self.events.received_ice_candidates.put_nowait(value_json) + elif msg_type == 'roomNotFound': + log.error(f'The room was not found: {uri}') + return + elif msg_type == 'roomClosed': + log.info(f'Oh noes, the room went away (session {self.session_id})!') + self.events.room_left.put_nowait(True) + return + else: + log.error(f'Unknown message type {msg_type}') + + async def send(self): + sdp_mids = None + user_fragments = None + while True: + if self.events.sdp_answer.qsize() > 0: + sdp_answer = self.events.sdp_answer.get_nowait() + sdp_answer_msg = json.dumps({ + 'SessionID': self.session_id, + 'Type': "gotAnswer", + 'Value': json.dumps({ + 'type': 'answer', + 'sdp': sdp_answer + }) + }) + await self.websocket.send(sdp_answer_msg) + + elif self.events.sdp_info.qsize() > 0: + sdp_mids, user_fragments = self.events.sdp_info.get_nowait() + + elif self.events.generated_ice_candidates.qsize() > 0 \ + and sdp_mids is not None and user_fragments is not None: + mlineindex, candidate = self.events.generated_ice_candidates.get_nowait() + sdp_mid = sdp_mids[mlineindex] + user_fragment = user_fragments[mlineindex] + icemsg_value = json.dumps({ + "candidate": candidate, + "sdpMid": sdp_mid, + "sdpMLineIndex": mlineindex, + "usernameFragment": user_fragment, + }) + icemsg = json.dumps({ + 'SessionID': self.session_id, + 'Type': 'addCalleeIceCandidate', + 'Value': icemsg_value, + }) + log.info(f'Send ICE candidate') + log.debug(f'ICE candidate: {icemsg_value}') + await self.websocket.send(icemsg) + + else: + await asyncio.sleep(0.2) + + async def run(self): + self.session_id = None + async with websockets.connect(self.uri, ssl=self.ssl_context, close_timeout=0.5) as self.websocket: + receive_task = asyncio.Task(self.receive(self.uri)) + send_task = asyncio.Task(self.send()) + done, pending = await asyncio.wait([receive_task, send_task], return_when=asyncio.FIRST_COMPLETED) + for task in pending: + task.cancel() + + +class WebRTCClient: + def __init__(self, events: Events, rtmp_uri: str): + self.events = events + self.rtmp_uri = rtmp_uri + self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace') + self.pipe = Gst.Pipeline.new("pipeline") + Gst.Bin.do_add_element(self.pipe, self.webrtcbin) + 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) 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.put_nowait((mlineindex, candidate)) + self.events.generated_ice_candidates.put_nowait((mlineindex, candidate)) def webrtcbin_pad_added(self, element, pad): log.info('webrtcbin_pad_added') @@ -79,33 +156,97 @@ class Lagarde: if not pad.has_current_caps(): log.info(pad, 'has no caps, ignoring') return - caps = pad.get_current_caps() padsize = caps.get_size() + for i in range(padsize): 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('intervideosink') + enc = Gst.ElementFactory.make('x264enc') + enc.set_property('bitrate', 1000) + enc.set_property('tune', 'zerolatency') + capsfilter = Gst.ElementFactory.make('capsfilter') + capsfilter.set_properties(Gst.Caps.from_string('video/x-h264,stream-format=(string)avc')) + flvmux = Gst.ElementFactory.make('flvmux') + flvmux.set_property('streamable', True) + sink = Gst.ElementFactory.make('rtmpsink') + sink.set_property('location', self.rtmp_uri) + assert q and conv and enc and capsfilter and flvmux and sink + + self.pipe.add(q) + self.pipe.add(conv) + self.pipe.add(enc) + self.pipe.add(capsfilter) + self.pipe.add(flvmux) + self.pipe.add(sink) + + q_pad_sink = q.get_static_pad('sink') + assert q_pad_sink + pad_link_return = pad.link(q_pad_sink) + assert pad_link_return == Gst.PadLinkReturn.OK + + ok = q.link(conv) + assert ok + ok = conv.link(enc) + assert ok + ok = enc.link(capsfilter) + assert ok + ok = capsfilter.link(flvmux) + assert ok + ok = flvmux.link(sink) + assert ok + self.pipe.set_state(Gst.State.PLAYING) + self.pipe.sync_children_states() + + 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(sink) - self.pipe.set_state(Gst.State.PLAYING) + conv.link(resample) + resample.link(sink) - async def listen_to_gstreamer_bus(self): - self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace') - self.pipe = Gst.Pipeline.new("pipeline") - Gst.Bin.do_add_element(self.pipe, self.webrtcbin) + def set_remote_desciption_done(self, gst_promise): + gst_promise = Gst.Promise.new_with_change_func(self.create_answer_done) + self.webrtcbin.emit('create-answer', None, gst_promise) + + def create_answer_done(self, gst_promise): + reply = gst_promise.get_reply() + answer = reply.get_value('answer') + gst_promise = Gst.Promise.new_with_change_func(self.set_local_description_done) + self.webrtcbin.emit('set-local-description', answer, gst_promise) + + sdp_message = answer.sdp + mids = [sdp_message.get_media(i).get_attribute_val('mid') + for i in range(sdp_message.medias_len())] + user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag') + for i in range(sdp_message.medias_len())] + sdp_answer = sdp_message.as_text() + self.mids_uf = mids, user_fragments + self.answer = sdp_answer + + def set_local_description_done(self, gst_promise): + gst_promise.get_reply() + + sdp_answer = self.answer + log.info(f'Send SDP answer') + log.debug(f'SDP answer:\n{sdp_answer}') + self.events.sdp_answer.put_nowait(sdp_answer) + mids, user_fragments = self.mids_uf + self.events.sdp_info.put_nowait((mids, user_fragments)) + + async def run(self): 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: @@ -117,151 +258,85 @@ class Lagarde: elif msg.type == Gst.MessageType.EOS: # end of stream log.info(f'Gstreamer message bus reports end of stream') return - elif self.sdp_offer is not None: - res, sm = GstSdp.SDPMessage.new() + elif self.events.sdp_offer.qsize() > 0: + sdp_offer = self.events.sdp_offer.get_nowait() + res, sm = GstSdp.SDPMessage.new_from_text(sdp_offer) 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() + gst_promise = Gst.Promise.new_with_change_func(self.set_remote_desciption_done) self.webrtcbin.emit('set-remote-description', rd, gst_promise) - await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait) - self.sdp_offer = None - - gst_promise = Gst.Promise.new() - self.webrtcbin.emit('create-answer', None, gst_promise) - result = await asyncio.get_event_loop().run_in_executor(None, 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(f'Send SDP answer') - log.debug(f'SDP answer:\n{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) - await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait) - gst_promise.get_reply() - await self.websocket.send(sdp_answer_msg) - - elif self.received_ice_candidates.qsize() > 0: - ic = self.received_ice_candidates.get_nowait() + + elif self.events.received_ice_candidates.qsize() > 0: + ic = self.events.received_ice_candidates.get_nowait() if ic['candidate'] != '': self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate']) - elif self.generated_ice_candidates.qsize() > 0: - mlineindex, candidate = self.generated_ice_candidates.get_nowait() - icemsg_value = json.dumps({ - "candidate": candidate, - "sdpMid": self.mids[mlineindex], - "sdpMLineIndex": mlineindex, - "usernameFragment": self.user_fragments[mlineindex], - }) - icemsg = json.dumps({ - 'SessionID': self.session_id, - 'Type': 'addCalleeIceCandidate', - 'Value': icemsg_value, - }) - log.info(f'Send ICE candidate') - log.debug(f'ICE candidate: {icemsg_value}') - await self.websocket.send(icemsg) + elif self.events.room_left.qsize() > 0: + self.events.room_left.get_nowait() + return else: await asyncio.sleep(0.1) finally: self.pipe.set_state(Gst.State.NULL) - async def talk_to_websocket(self, uri, ssl_context): - async with websockets.connect(uri, ssl=ssl_context, close_timeout=0.5) as self.websocket: - async for msg in self.websocket: - msg_json = json.loads(msg) - msg_type = msg_json['Type'] - msg_value = msg_json['Value'] - assert self.session_id is None or self.session_id == msg_json['SessionID'] - if msg_type == 'newSession': - self.session_id = msg_json['SessionID'] - log.info(f"New session {self.session_id}") - elif msg_type == 'gotOffer': - value_json = json.loads(msg_value) - sdp = value_json['sdp'] - log.info(f'Got SDP offer') - log.debug(f'SDP offer:\n{sdp}') - self.sdp_offer = sdp - elif msg_type == 'addCallerIceCandidate': - value_json = json.loads(msg_value) - log.info(f'Got ICE candidate') - log.debug(f'ICE candidate: {value_json}') - self.received_ice_candidates.put_nowait(value_json) - elif msg_type == 'roomNotFound': - log.error(f'The room was not found: {uri}') - return - elif msg_type == 'roomClosed': - log.info(f'Oh noes, the room went away (session {self.session_id})!') - self.session_id = None - return - else: - log.error(f'Unknown message type {msg_type}') - async def talk_to_signaling_server(self, uri): - ssl_context = ssl.SSLContext() - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - while True: - await self.talk_to_websocket(uri, ssl_context) - await asyncio.sleep(0.1) +async def run_room(laplace_uri: str, rtmp_uri: str): + try: + events = Events() + webrtc = WebRTCClient(events, rtmp_uri) + signaling = SignalingClient(events, laplace_uri) - async def run(self, uri): - try: - talk_to_signaling_server_task = asyncio.Task(self.talk_to_signaling_server(uri)) - listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus()) - main_loop = asyncio.Task(gstreamer_main_loop()) - done, pending = await asyncio.wait( - [talk_to_signaling_server_task, listen_to_gstreamer_bus_task, main_loop], - return_when=asyncio.FIRST_COMPLETED) - for d in done: - d.result() - for p in pending: - p.cancel() - except OSError as e: - print(e) - - -async def gstreamer_main_loop(): - """Does the equivalent of the following lines in an async friendly way: - loop = GLib.MainLoop() - loop.run() - """ - gst_loop = GLib.MainLoop() - context = gst_loop.get_context() + webrtc_task = asyncio.Task(webrtc.run()) + signaling_task = asyncio.Task(signaling.run()) + + done, pending = await asyncio.wait([webrtc_task, signaling_task], return_when=asyncio.FIRST_COMPLETED) + + for task in done: + task.result() + for task in pending: + task.cancel() + except OSError as e: + print(e) + + +async def run_room_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float): while True: - events_dispatched = context.iteration(False) - await asyncio.sleep(0. if events_dispatched else 0.01) + await run_room(laplace_uri, rtmp_uri) + await asyncio.sleep(sleep_time) + + +async def run_rooms(laplace_base_uri: str, rtmp_base_uri: str, rooms: List[str], retry: bool): + tasks = [] + for room in rooms: + laplace_uri = laplace_base_uri + room # TODO: encode + rtmp_uri = rtmp_base_uri + room # TODO: encode + if retry: + tasks.append(run_room_repeated(laplace_uri, rtmp_uri, 2.)) + else: + tasks.append(run_room(laplace_uri, rtmp_uri)) + await asyncio.gather(*tasks) def main(): - logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s') + logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s') + default_source = 'wss://localhost:1234/ws_connect?id=' + default_dest = 'rtmp://localhost:1935/' + default_room = 'cug' parser = argparse.ArgumentParser() - parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug', - help='Signalling server URI') + parser.add_argument('-s', '--source', default=default_source, + help=f'Laplace signalling websocket base URI, default: {default_source}') + parser.add_argument('-d', '--destination', default=default_dest, + help=f'RTMP server base URI, default: {default_dest}') + parser.add_argument('-r', '--retry', action='store_true', help=f'Retry forever if room not found or closed') + parser.add_argument('room', nargs='*', help=f'Room names to be used, "{default_room}" if omitted') args = parser.parse_args() Gst.init(None) - rtsp = GstreamerRtspServer() - lagarde = Lagarde() - asyncio.run(lagarde.run(args.uri), debug=True) + rooms = args.room + if len(rooms) == 0: + rooms = [default_room] + asyncio.run(run_rooms(args.source, args.destination, rooms, args.retry)) if __name__ == '__main__':