#!/usr/bin/python3
-
import argparse
import asyncio
import json
log = logging.getLogger(__name__)
-class GstreamerRtspServer():
+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 ! 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)
self.server = server
-class Lagarde:
+class Events:
def __init__(self):
- self.sdp_offer: Optional[str] = None
- self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
- self.session_id = None
- self.received_ice_candidates = queue.Queue()
+ self.sdp_offer = queue.Queue()
+ self.sdp_answer = queue.Queue()
self.generated_ice_candidates = queue.Queue()
- self.user_fragments: Optional[List] = None
- self.mids: Optional[List] = None
- self.pipe = None
- self.webrtcbin = None
+ 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
+ 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.debug('on_negotiation_needed')
+ log.info('on_negotiation_needed')
def on_ice_candidate(self, element, mlineindex, candidate):
- log.debug('on_ice_candidate')
- self.generated_ice_candidates.put_nowait((mlineindex, candidate))
+ log.info('on_ice_candidate')
+ self.events.generated_ice_candidates.put_nowait((mlineindex, candidate))
def webrtcbin_pad_added(self, element, pad):
- log.debug('webrtcbin_pad_added')
+ log.info('webrtcbin_pad_added')
if pad.direction != Gst.PadDirection.SRC:
return
decodebin = Gst.ElementFactory.make('decodebin')
self.webrtcbin.link(decodebin)
def decodebin_pad_added(self, element, pad):
- log.debug('decodebin_pad_added')
+ log.info('decodebin_pad_added')
if not pad.has_current_caps():
- log.debug(pad, 'has no caps, ignoring')
+ log.info(pad, 'has no caps, ignoring')
return
-
caps = pad.get_current_caps()
padsize = caps.get_size()
+
+ log.info(f'>>>> {padsize} {caps}')
+
for i in range(padsize):
- s = caps.get_structure(i) # Gst.Structure
+ 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')
- sink = Gst.ElementFactory.make('intervideosink')
+ 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)
+
+ 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)
self.pipe.sync_children_states()
- pad.link(q.get_static_pad('sink'))
- q.link(conv)
- conv.link(sink)
+ #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')
conv.link(resample)
resample.link(sink)
- async def listen_to_gstreamer_bus(self):
- self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
- self.pipe = Gst.Pipeline.new("pipeline")
- Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
+ 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)
- 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:
elif msg.type == Gst.MessageType.EOS: # end of stream
log.info(f'Gstreamer message bus reports end of stream')
return
- elif self.sdp_offer is not None:
+ 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
- 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)
- 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)
- await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
- self.sdp_offer = None
-
- gst_promise = Gst.Promise.new()
- self.webrtcbin.emit('create-answer', None, gst_promise)
- result = await asyncio.get_event_loop().run_in_executor(None, 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(f'Send SDP answer:\n{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)
- await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
- gst_promise.get_reply()
- await self.websocket.send(sdp_answer_msg)
-
- elif self.received_ice_candidates.qsize() > 0:
- ic = self.received_ice_candidates.get_nowait()
+
+ 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'])
- elif self.generated_ice_candidates.qsize() > 0:
- mlineindex, candidate = self.generated_ice_candidates.get_nowait()
- icemsg_value = json.dumps({
- "candidate": candidate,
- "sdpMid": self.mids[mlineindex],
- "sdpMLineIndex": mlineindex,
- "usernameFragment": self.user_fragments[mlineindex],
- })
- icemsg = json.dumps({
- 'SessionID': self.session_id,
- 'Type': 'addCalleeIceCandidate',
- 'Value': icemsg_value,
- })
- log.info(f'Send ICE candidate: {icemsg_value}')
- 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)
- async def talk_to_websocket(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:\n{sdp}')
- self.sdp_offer = sdp
- elif msg_type == 'addCallerIceCandidate':
- value_json = json.loads(msg_value)
- log.info(f'Got ICE candidate: {value_json}')
- self.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.session_id = None
- return
- else:
- log.error(f'Unknown message type {msg_type}')
- async def run(self, uri):
- ssl_context = ssl.SSLContext()
- ssl_context.check_hostname = False
- ssl_context.verify_mode = ssl.CERT_NONE
- try:
- async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
- talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
- listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
- main_loop = asyncio.Task(gstreamer_main_loop())
- done, pending = await asyncio.wait(
- [talk_to_websocket_task, listen_to_gstreamer_bus_task, main_loop],
- return_when=asyncio.FIRST_COMPLETED)
- for d in done:
- d.result()
- for p in pending:
- p.cancel()
- except OSError as e:
- print(e)
-
-
-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()
+async def run_repeated(task):
while True:
- events_dispatched = context.iteration(False)
- await asyncio.sleep(0. if events_dispatched else 0.01)
+ await task()
+ await asyncio.sleep(0.1)
+
+
+async def run(uri):
+ try:
+ events = Events()
+ # rtsp = RtspServer()
+ webrtc = WebRTCClient(events)
+ signaling = SignalingClient(events, uri)
+
+ webrtc_task = asyncio.Task(webrtc.run())
+ signaling_task = asyncio.Task(signaling.run())
+
+ done, pending = await asyncio.wait([webrtc_task, signaling_task],
+ return_when=asyncio.FIRST_COMPLETED)
+
+ for task in done:
+ task.result()
+ for task in pending:
+ task.cancel()
+ except OSError as e:
+ print(e)
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()
Gst.init(None)
- rtsp = GstreamerRtspServer()
- lagarde = Lagarde()
- asyncio.run(lagarde.run(args.uri), debug=True)
+ asyncio.run(run(args.uri), debug=True)
if __name__ == '__main__':
ToastFreeware / toast / stream2beamer.git / blobdiff