From: Philipp Spitzer Date: Wed, 8 Jul 2020 21:41:18 +0000 (+0200) Subject: Rename laplace_client_2.py to lagarde.py X-Git-Url: https://git.toastfreeware.priv.at/toast/stream2beamer.git/commitdiff_plain/ff2de464ec867b72e4d3d3abd0a84dc6ff54afad Rename laplace_client_2.py to lagarde.py --- diff --git a/lagarde.py b/lagarde.py new file mode 100755 index 0000000..a658a4e --- /dev/null +++ b/lagarde.py @@ -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 index a658a4e..0000000 --- a/laplace_client_2.py +++ /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()