Still doesn't work.... :-(

[toast/stream2beamer.git] / lagarde.py

diff --git a/lagarde.py b/lagarde.py
index bc7a99b4c2c469db4348b74b023e514858a06bd5..fd8285273617185cc57b457e53ae1e704270f2b7 100755 (executable)
--- a/lagarde.py
+++ b/lagarde.py
@@ -1,18 +1,14 @@
 #!/usr/bin/python3
 #!/usr/bin/python3

-

 import argparse
 import asyncio
 import argparse
 import asyncio

-import datetime

 import json
 import logging
 import json
 import logging

-import pathlib

 import ssl
 import ssl

-import sys
+import queue

 from typing import Optional, List
 
 from typing import Optional, List
 

-import websockets
-

 import gi
 import gi

+import websockets

 
 gi.require_version('Gst', '1.0')
 from gi.repository import Gst
 
 gi.require_version('Gst', '1.0')
 from gi.repository import Gst


@@ -23,27 +19,148 @@ from gi.repository import GstWebRTC
 gi.require_version('GstSdp', '1.0')
 from gi.repository import GstSdp
 
 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__)
 
 
 log = logging.getLogger(__name__)
 
 

-class Lagarde:
+class RtspServer:
+    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 Events:

     def __init__(self):
     def __init__(self):

-        self.sdp_offer: Optional[str] = None
-        self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
+        self.sdp_offer = queue.Queue()
+        self.sdp_answer = queue.Queue()
+        self.generated_ice_candidates = queue.Queue()
+        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
         self.session_id = None

-        self.received_ice_candidates = []
-        self.generated_ice_candidates = []
-        self.user_fragments: Optional[List] = None
-        self.mids: Optional[List] = None
-        self.pipe = None
-        self.webrtcbin = 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):
+        self.events = events
+        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')
 
     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.append((mlineindex, candidate))
+        self.events.generated_ice_candidates.put_nowait((mlineindex, candidate))

 
     def webrtcbin_pad_added(self, element, pad):
         log.info('webrtcbin_pad_added')
 
     def webrtcbin_pad_added(self, element, pad):
         log.info('webrtcbin_pad_added')


@@ -60,161 +177,187 @@ class Lagarde:
         if not pad.has_current_caps():
             log.info(pad, 'has no caps, ignoring')
             return
         if not pad.has_current_caps():
             log.info(pad, 'has no caps, ignoring')
             return

-

         caps = pad.get_current_caps()
         caps = pad.get_current_caps()

-        assert (len(caps))
-        s = caps[0]
-        name = s.get_name()
-        if name.startswith('video'):
-            q = Gst.ElementFactory.make('queue')
-            conv = Gst.ElementFactory.make('videoconvert')
-            sink = Gst.ElementFactory.make('autovideosink')
-            self.pipe.add(q, conv, sink)
-            self.pipe.sync_children_states()
-            pad.link(q.get_static_pad('sink'))
-            q.link(conv)
-            conv.link(sink)
-        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, conv, resample, sink)
-            self.pipe.sync_children_states()
-            pad.link(q.get_static_pad('sink'))
-            q.link(conv)
-            conv.link(resample)
-            resample.link(sink)
-
-    async def listen_to_gstreamer_bus(self):
-        Gst.init(None)
-        self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
-        self.pipe = Gst.Pipeline.new("pipeline")
-        Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
+        padsize = caps.get_size()
+
+        log.info(f'>>>> {padsize} {caps}')
+
+        for i in range(padsize):
+            s = caps.get_structure(i)  # Gst.Structure
+            name = s.get_name()
+            log.info(f'###### {name}')
+            if name.startswith('video'):
+                q = Gst.ElementFactory.make('queue')
+                conv = Gst.ElementFactory.make('videoconvert')
+                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'))
+                flmux = Gst.ElementFactory.make('flvmux')
+                sink = Gst.ElementFactory.make('rtmpsink')
+                sink.set_property('location', 'rtmp://192.168.1.46:1935/gregoa')
+                # sink.set_property('location', 'rtmp://bla:1936/gregoa')
+                print(sink.props.location, dir(sink.props))
+                assert q and conv and enc and capsfilter and flmux 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(sink)
+                self.pipe.sync_children_states()
+
+                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 = element.link(q)
+                # assert ok
+
+                ok = q.link(conv)
+                assert ok
+                ok = conv.link(enc)
+                assert ok
+                ok = enc.link(capsfilter)
+                assert ok
+                ok = capsfilter.link(flmux)
+                assert ok
+                ok = flmux.link(sink)
+                assert ok
+                self.pipe.set_state(Gst.State.PLAYING)
+                #print(dir(Gst.DebugGraphDetails))
+                #Gst.debug_bin_to_dot_data(element, Gst.DebugGraphDetails.ALL)
+
+            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(resample)
+                resample.link(sink)
+
+    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')
+        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())]
+        self.events.sdp_info.put_nowait((mids, user_fragments))
+        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)
+
+    def set_local_description_done(self, gst_promise):
+        gst_promise.get_reply()
+
+    async def run(self):

         bus = Gst.Pipeline.get_bus(self.pipe)
         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:
                 if bus.have_pending():
         self.pipe.set_state(Gst.State.PLAYING)
         try:
             while True:
                 if bus.have_pending():

-                    msg = bus.pop()  # Gst.Message, has to be unref'ed.
-                    if msg.type != Gst.MessageType.STATE_CHANGED:
-                        # log.info(f'Receive Gst.Message: {msg.type}, {msg.seqnum}, {msg.get_structure()}')
-                        # log.info(f'{webrtcbin.props.signaling_state} {webrtcbin.props.ice_gathering_state} {webrtcbin.props.ice_connection_state}')
-                        # Gst.Message.unref(msg)
-                        pass
-                elif self.sdp_offer is not None:
+                    msg = bus.pop()
+                    if msg.type == Gst.MessageType.ERROR:
+                        log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
+                        return
+                    elif msg.type == Gst.MessageType.EOS:  # end of stream
+                        log.info(f'Gstreamer message bus reports end of stream')
+                        return
+                elif self.events.sdp_offer.qsize() > 0:
+                    sdp_offer = self.events.sdp_offer.get_nowait()

                     res, sm = GstSdp.SDPMessage.new()
                     assert res == GstSdp.SDPResult.OK
                     res, sm = GstSdp.SDPMessage.new()
                     assert res == GstSdp.SDPResult.OK

-                    GstSdp.sdp_message_parse_buffer(bytes(self.sdp_offer.encode()), sm)
+                    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)
                     # 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)
                     self.webrtcbin.emit('set-remote-description', rd, gst_promise)

-                    gst_promise.wait()
-                    self.sdp_offer = None
-
-                    log.info('create-answer')
-                    gst_promise = Gst.Promise.new()
-                    self.webrtcbin.emit('create-answer', None, gst_promise)
-                    result = 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(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)
-                    gst_promise.wait()
-                    gst_promise.get_reply()
-                    await self.websocket.send(sdp_answer_msg)
-
-                elif len(self.received_ice_candidates) > 0:
-                    ic = self.received_ice_candidates.pop(0)
+
+                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'])
 
                     if ic['candidate'] != '':
                         self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
 

-                elif len(self.generated_ice_candidates) > 0:
-                    mlineindex, candidate = self.generated_ice_candidates.pop(0)
-                    icemsg = json.dumps({
-                        'SessionID': self.session_id,
-                        'Type': 'addCalleeIceCandidate',
-                        'Value': json.dumps({
-                            "candidate": candidate,
-                            "sdpMid": self.mids[mlineindex],
-                            "sdpMLineIndex": mlineindex,
-                            "usernameFragment": self.user_fragments[mlineindex],
-                        })
-                    })
-                    log.info(f'send_ice_candidate_message with {icemsg}')
-                    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)
 
 
                 else:
                     await asyncio.sleep(0.1)
         finally:
             self.pipe.set_state(Gst.State.NULL)
 

-    async def talk_to_websocket(self, uri):
-        ssl_context = ssl.SSLContext()
-        ssl_context.check_hostname = False
-        ssl_context.verify_mode = ssl.CERT_NONE
-        async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
-            async for msg in self.websocket:
-                msg_json = json.loads(msg)
-                msg_type = msg_json['Type']
-                msg_value = msg_json['Value']
-                self.session_id = msg_json['SessionID']
-                log.info(f"receive for session {self.session_id} type {msg_type}")
-                if msg_type == 'newSession':
-                    pass
-                elif msg_type == 'gotOffer':
-                    value_json = json.loads(msg_value)
-                    sdp = value_json['sdp']
-                    log.info(f'SDP: {sdp}')
-                    self.sdp_offer = sdp
-                elif msg_type == 'addCallerIceCandidate':
-                    value_json = json.loads(msg_value)
-                    log.info(f'ICE: {value_json}')
-                    self.received_ice_candidates.append(value_json)
-                else:
-                    log.error(f'Unknown message type {msg_type}')

 
 

-    async def run(self, uri):
-        talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
-        listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
-        done, pending = await asyncio.wait(
-            [talk_to_websocket_task, listen_to_gstreamer_bus_task],
+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()
+    while True:
+        events_dispatched = context.iteration(False)
+        await asyncio.sleep(0. if events_dispatched else 0.01)
+
+
+async def run_repeated(task):
+    while True:
+        await task()
+        await asyncio.sleep(0.1)
+
+
+async def run(uri):
+    try:
+        events = Events()
+
+        # rtsp = RtspServer()
+        webrtc = WebRTCClient(events)
+        signaling = SignalingClient(events, uri)
+
+        main_loop_task = asyncio.Task(gstreamer_main_loop())
+        webrtc_task = asyncio.Task(run_repeated(webrtc.run))
+        signaling_task = asyncio.Task(run_repeated(signaling.run))
+
+        done, pending = await asyncio.wait([main_loop_task, webrtc_task, signaling_task],

             return_when=asyncio.FIRST_COMPLETED)
             return_when=asyncio.FIRST_COMPLETED)

-        for d in done:
-            d.result()
-        for p in pending:
-            p.cancel()
+
+        for task in done:
+            task.result()
+        for task in pending:
+            task.cancel()
+    except OSError as e:
+        print(e)

 
 
 def main():
 
 
 def main():

-    logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s')
+    logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s')

     parser = argparse.ArgumentParser()
     parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
                         help='Signalling server URI')
     args = parser.parse_args()
     parser = argparse.ArgumentParser()
     parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
                         help='Signalling server URI')
     args = parser.parse_args()

-    lagarde = Lagarde()
-    asyncio.run(lagarde.run(args.uri), debug=True)
+
+    Gst.init(None)
+    asyncio.run(run(args.uri), debug=True)

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':