X-Git-Url: https://git.toastfreeware.priv.at/toast/stream2beamer.git/blobdiff_plain/39fa9dc97b66ab016f7cb4a617c809e42ee6c8b5..e97741140341d1c98f6cb350d819a988c8b73bbd:/lagarde.py?ds=sidebyside diff --git a/lagarde.py b/lagarde.py index 7c6ce9a..f31bc2d 100755 --- a/lagarde.py +++ b/lagarde.py @@ -5,6 +5,7 @@ import json import logging import ssl import queue +from typing import List import gi import websockets @@ -169,16 +170,17 @@ class WebRTCClient: enc.set_property('tune', 'zerolatency') capsfilter = Gst.ElementFactory.make('capsfilter') capsfilter.set_properties(Gst.Caps.from_string('video/x-h264,stream-format=(string)avc')) - flmux = Gst.ElementFactory.make('flvmux') + 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 flmux and sink + 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(flmux) + self.pipe.add(flvmux) self.pipe.add(sink) q_pad_sink = q.get_static_pad('sink') @@ -192,9 +194,9 @@ class WebRTCClient: assert ok ok = enc.link(capsfilter) assert ok - ok = capsfilter.link(flmux) + ok = capsfilter.link(flvmux) assert ok - ok = flmux.link(sink) + ok = flvmux.link(sink) assert ok self.pipe.set_state(Gst.State.PLAYING) self.pipe.sync_children_states() @@ -221,22 +223,28 @@ class WebRTCClient: 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())] + 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())] - self.events.sdp_info.put_nowait((mids, user_fragments)) + 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}') - self.events.sdp_answer.put_nowait(sdp_answer) - gst_promise = Gst.Promise.new_with_change_func(self.set_local_description_done) - self.webrtcbin.emit('set-local-description', answer, gst_promise) + 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.pipe.set_state(Gst.State.PLAYING) @@ -252,11 +260,8 @@ class WebRTCClient: return elif self.events.sdp_offer.qsize() > 0: sdp_offer = self.events.sdp_offer.get_nowait() - res, sm = GstSdp.SDPMessage.new() + res, sm = GstSdp.SDPMessage.new_from_text(sdp_offer) assert res == GstSdp.SDPResult.OK - GstSdp.sdp_message_parse_buffer(bytes(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_with_change_func(self.set_remote_desciption_done) self.webrtcbin.emit('set-remote-description', rd, gst_promise) @@ -276,13 +281,7 @@ class WebRTCClient: self.pipe.set_state(Gst.State.NULL) -async def run_repeated(task): - while True: - await task() - await asyncio.sleep(0.1) - - -async def run(laplace_uri: str, rtmp_uri: str): +async def run_room(laplace_uri: str, rtmp_uri: str): try: events = Events() webrtc = WebRTCClient(events, rtmp_uri) @@ -301,19 +300,43 @@ async def run(laplace_uri: str, rtmp_uri: str): print(e) +async def run_room_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float): + while True: + 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.DEBUG, format='%(asctime)-15s %(message)s') - default_source = 'wss://localhost:1234/ws_connect?id=cug' - default_dest = 'rtmp://localhost:1935/cug' + default_source = 'wss://localhost:1234/ws_connect?id=' + default_dest = 'rtmp://localhost:1935/' + default_room = 'cug' parser = argparse.ArgumentParser() parser.add_argument('-s', '--source', default=default_source, - help=f'Laplace signalling websocket URI, 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 URI, 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) - asyncio.run(run(args.source, args.destination), 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__':