import argparse
import asyncio
+import datetime
import json
import logging
import pathlib
import ssl
import sys
-from typing import Optional
+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__)
-sdp_offer: Optional[str] = None
-ice_candidates = []
-
-
-async def listen_to_gstreamer_bus():
- global sdp_offer, ice_candidates
- Gst.init(None)
- webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
- pipe = Gst.Pipeline.new("pipeline")
- Gst.Bin.do_add_element(pipe, webrtcbin)
- bus = Gst.Pipeline.get_bus(pipe)
- pipe.set_state(Gst.State.PLAYING)
- try:
- while True:
- if bus.have_pending():
- msg = bus.pop() # Gst.Message, has to be unref'ed.
- log.info(f'Receive Gst.Message: {msg.type}')
- # Gst.Message.unref(msg)
- elif sdp_offer is not None:
- res, sm = GstSdp.SDPMessage.new()
- 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)
- webrtcbin.emit('set-remote-description', rd, None)
- sdp_offer = None
- elif len(ice_candidates) > 0:
- ic = ice_candidates.pop(0)
- webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
- else:
- await asyncio.sleep(0.1)
- finally:
- pipe.set_state(Gst.State.NULL)
-
-
-async def talk_to_websocket(uri):
- global sdp_offer, ice_candidates
- 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 websocket:
- async for msg in websocket:
- msg_json = json.loads(msg)
- msg_type = msg_json['Type']
- msg_value = msg_json['Value']
- session_id = msg_json['SessionID']
- log.info(f"receive for session {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}')
- sdp_offer = sdp
- elif msg_type == 'addCallerIceCandidate':
- value_json = json.loads(msg_value)
- log.info(f'ICE: {value_json}')
- ice_candidates.append(value_json)
- else:
- log.error(f'Unknown message type {msg_type}')
-
-
-async def run(uri):
- talk_to_websocket_task = asyncio.Task(talk_to_websocket(uri))
- listen_to_gstreamer_bus_task = asyncio.Task(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()
+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')
+ help='Signalling server URI')
args = parser.parse_args()
- asyncio.run(run(args.uri), debug=True)
+ lagarde = Lagarde()
+ asyncio.run(lagarde.run(args.uri), debug=True)
if __name__ == '__main__':
ToastFreeware / toast / stream2beamer.git / commitdiff