Rename laplace_client_2.py to lagarde.py
authorPhilipp Spitzer <philipp@spitzer.priv.at>
Wed, 8 Jul 2020 21:41:18 +0000 (23:41 +0200)
committerPhilipp Spitzer <philipp@spitzer.priv.at>
Wed, 8 Jul 2020 21:41:18 +0000 (23:41 +0200)
lagarde.py [new file with mode: 0755]
laplace_client_2.py [deleted file]

diff --git a/lagarde.py b/lagarde.py
new file mode 100755 (executable)
index 0000000..a658a4e
--- /dev/null
@@ -0,0 +1,222 @@
+#!/usr/bin/python3
+
+import argparse
+import asyncio
+import datetime
+import json
+import logging
+import pathlib
+import ssl
+import sys
+from typing import Optional, List
+
+import websockets
+
+import gi
+
+gi.require_version('Gst', '1.0')
+from gi.repository import Gst
+
+gi.require_version('GstWebRTC', '1.0')
+from gi.repository import GstWebRTC
+
+gi.require_version('GstSdp', '1.0')
+from gi.repository import GstSdp
+
+log = logging.getLogger(__name__)
+
+
+class Lagarde:
+    def __init__(self):
+        self.sdp_offer: Optional[str] = None
+        self.websocket: Optional[websockets.client.WebSocketClientProtocol] = 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
+
+    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))
+
+    def webrtcbin_pad_added(self, element, pad):
+        log.info('webrtcbin_pad_added')
+        if pad.direction != Gst.PadDirection.SRC:
+            return
+        decodebin = Gst.ElementFactory.make('decodebin')
+        decodebin.connect('pad-added', self.decodebin_pad_added)
+        self.pipe.add(decodebin)
+        decodebin.sync_state_with_parent()
+        self.webrtcbin.link(decodebin)
+
+    def decodebin_pad_added(self, element, pad):
+        log.info('decodebin_pad_added')
+        if not pad.has_current_caps():
+            log.info(pad, 'has no caps, ignoring')
+            return
+
+        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)
+        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():
+                    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:
+                    res, sm = GstSdp.SDPMessage.new()
+                    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()
+                    self.webrtcbin.emit('set-remote-description', rd, gst_promise)
+                    gst_promise.wait()
+                    print(gst_promise.get_reply())
+                    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)
+                    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)
+
+                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],
+            return_when=asyncio.FIRST_COMPLETED)
+        for d in done:
+            d.result()
+        for p in pending:
+            p.cancel()
+
+
+def main():
+    logging.basicConfig(level=logging.INFO, 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()
+    lagarde = Lagarde()
+    asyncio.run(lagarde.run(args.uri), debug=True)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/laplace_client_2.py b/laplace_client_2.py
deleted file mode 100755 (executable)
index a658a4e..0000000
+++ /dev/null
@@ -1,222 +0,0 @@
-#!/usr/bin/python3
-
-import argparse
-import asyncio
-import datetime
-import json
-import logging
-import pathlib
-import ssl
-import sys
-from typing import Optional, List
-
-import websockets
-
-import gi
-
-gi.require_version('Gst', '1.0')
-from gi.repository import Gst
-
-gi.require_version('GstWebRTC', '1.0')
-from gi.repository import GstWebRTC
-
-gi.require_version('GstSdp', '1.0')
-from gi.repository import GstSdp
-
-log = logging.getLogger(__name__)
-
-
-class Lagarde:
-    def __init__(self):
-        self.sdp_offer: Optional[str] = None
-        self.websocket: Optional[websockets.client.WebSocketClientProtocol] = 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
-
-    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))
-
-    def webrtcbin_pad_added(self, element, pad):
-        log.info('webrtcbin_pad_added')
-        if pad.direction != Gst.PadDirection.SRC:
-            return
-        decodebin = Gst.ElementFactory.make('decodebin')
-        decodebin.connect('pad-added', self.decodebin_pad_added)
-        self.pipe.add(decodebin)
-        decodebin.sync_state_with_parent()
-        self.webrtcbin.link(decodebin)
-
-    def decodebin_pad_added(self, element, pad):
-        log.info('decodebin_pad_added')
-        if not pad.has_current_caps():
-            log.info(pad, 'has no caps, ignoring')
-            return
-
-        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)
-        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():
-                    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:
-                    res, sm = GstSdp.SDPMessage.new()
-                    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()
-                    self.webrtcbin.emit('set-remote-description', rd, gst_promise)
-                    gst_promise.wait()
-                    print(gst_promise.get_reply())
-                    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)
-                    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)
-
-                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],
-            return_when=asyncio.FIRST_COMPLETED)
-        for d in done:
-            d.result()
-        for p in pending:
-            p.cancel()
-
-
-def main():
-    logging.basicConfig(level=logging.INFO, 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()
-    lagarde = Lagarde()
-    asyncio.run(lagarde.run(args.uri), debug=True)
-
-
-if __name__ == '__main__':
-    main()