The two pipelines now successfully communicate via intervideosink/src.
[toast/stream2beamer.git] / lagarde.py
1 #!/usr/bin/python3
2
3 import argparse
4 import asyncio
5 import json
6 import logging
7 import ssl
8 import queue
9 from typing import Optional, List
10
11 import gi
12 import websockets
13
14 gi.require_version('Gst', '1.0')
15 from gi.repository import Gst
16
17 gi.require_version('GstWebRTC', '1.0')
18 from gi.repository import GstWebRTC
19
20 gi.require_version('GstSdp', '1.0')
21 from gi.repository import GstSdp
22
23 gi.require_version('GstRtspServer', '1.0')
24 from gi.repository import Gst, GstRtspServer, GObject, GLib
25
26 log = logging.getLogger(__name__)
27
28
29 class GstreamerRtspServer():
30     def __init__(self):
31         server = GstRtspServer.RTSPServer()
32         server.set_address("::")
33         server.set_service('8554')  # port as string
34         factory = GstRtspServer.RTSPMediaFactory()
35         factory.set_launch("intervideosrc ! decodebin ! theoraenc ! queue ! rtptheorapay name=pay0")
36         factory.set_shared(True)
37         mountPoints = server.get_mount_points()
38         mountPoints.add_factory("/cug", factory)
39         server.attach()
40         self.server = server
41
42
43 class Lagarde:
44     def __init__(self):
45         self.sdp_offer: Optional[str] = None
46         self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
47         self.session_id = None
48         self.received_ice_candidates = queue.Queue()
49         self.generated_ice_candidates = queue.Queue()
50         self.user_fragments: Optional[List] = None
51         self.mids: Optional[List] = None
52         self.pipe = None
53         self.webrtcbin = None
54
55     def on_negotiation_needed(self, element):
56         log.debug('on_negotiation_needed')
57
58     def on_ice_candidate(self, element, mlineindex, candidate):
59         log.debug('on_ice_candidate')
60         self.generated_ice_candidates.put_nowait((mlineindex, candidate))
61
62     def webrtcbin_pad_added(self, element, pad):
63         log.debug('webrtcbin_pad_added')
64         if pad.direction != Gst.PadDirection.SRC:
65             return
66         decodebin = Gst.ElementFactory.make('decodebin')
67         decodebin.connect('pad-added', self.decodebin_pad_added)
68         self.pipe.add(decodebin)
69         decodebin.sync_state_with_parent()
70         self.webrtcbin.link(decodebin)
71
72     def decodebin_pad_added(self, element, pad):
73         log.debug('decodebin_pad_added')
74         if not pad.has_current_caps():
75             log.debug(pad, 'has no caps, ignoring')
76             return
77
78         caps = pad.get_current_caps()
79         padsize = caps.get_size()
80         for i in range(padsize):
81             s = caps.get_structure(i) # Gst.Structure
82             name = s.get_name()
83             if name.startswith('video'):
84                 q = Gst.ElementFactory.make('queue')
85                 conv = Gst.ElementFactory.make('videoconvert')
86                 sink = Gst.ElementFactory.make('intervideosink')
87                 self.pipe.add(q)
88                 self.pipe.add(conv)
89                 self.pipe.add(sink)
90                 self.pipe.sync_children_states()
91                 pad.link(q.get_static_pad('sink'))
92                 q.link(conv)
93                 conv.link(sink)
94             elif name.startswith('audio'):
95                 q = Gst.ElementFactory.make('queue')
96                 conv = Gst.ElementFactory.make('audioconvert')
97                 resample = Gst.ElementFactory.make('audioresample')
98                 sink = Gst.ElementFactory.make('autoaudiosink')
99                 self.pipe.add(q)
100                 self.pipe.add(conv)
101                 self.pipe.add(resample)
102                 self.pipe.add(sink)
103                 self.pipe.sync_children_states()
104                 pad.link(q.get_static_pad('sink'))
105                 q.link(conv)
106                 conv.link(resample)
107                 resample.link(sink)
108
109     async def listen_to_gstreamer_bus(self):
110         self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
111         self.pipe = Gst.Pipeline.new("pipeline")
112         Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
113         bus = Gst.Pipeline.get_bus(self.pipe)
114         self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed)
115         self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate)
116         self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added)
117         self.pipe.set_state(Gst.State.PLAYING)
118         try:
119             while True:
120                 if bus.have_pending():
121                     msg = bus.pop()
122                     if msg.type == Gst.MessageType.ERROR:
123                         log.error(f'Error from gstreamer message bus: {msg.get_structure()}')
124                         return
125                     elif msg.type == Gst.MessageType.EOS:  # end of stream
126                         log.info(f'Gstreamer message bus reports end of stream')
127                         return
128                 elif self.sdp_offer is not None:
129                     res, sm = GstSdp.SDPMessage.new()
130                     assert res == GstSdp.SDPResult.OK
131                     GstSdp.sdp_message_parse_buffer(bytes(self.sdp_offer.encode()), sm)
132                     # the three lines above can also be done this way in new versions of GStreamer:
133                     # sm = GstSdp.SDPMessage.new_from_text(sdp_offer)
134                     rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
135                     gst_promise = Gst.Promise.new()
136                     self.webrtcbin.emit('set-remote-description', rd, gst_promise)
137                     await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
138                     self.sdp_offer = None
139
140                     gst_promise = Gst.Promise.new()
141                     self.webrtcbin.emit('create-answer', None, gst_promise)
142                     result = await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
143                     assert result == Gst.PromiseResult.REPLIED
144                     reply = gst_promise.get_reply()
145                     answer = reply.get_value('answer')
146                     sdp_message = answer.sdp
147                     self.user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag')
148                                           for i in range(sdp_message.medias_len())]
149                     self.mids = [sdp_message.get_media(i).get_attribute_val('mid')
150                                           for i in range(sdp_message.medias_len())]
151                     sdp_answer = sdp_message.as_text()
152                     log.info(f'Send SDP answer:\n{sdp_answer}')
153                     sdp_answer_msg = json.dumps({
154                         'SessionID': self.session_id,
155                         'Type': "gotAnswer",
156                         'Value': json.dumps({
157                             'type': 'answer',
158                             'sdp': sdp_answer
159                         })
160                     })
161                     gst_promise = Gst.Promise.new()
162                     self.webrtcbin.emit('set-local-description', answer, gst_promise)
163                     await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
164                     gst_promise.get_reply()
165                     await self.websocket.send(sdp_answer_msg)
166
167                 elif self.received_ice_candidates.qsize() > 0:
168                     ic = self.received_ice_candidates.get_nowait()
169                     if ic['candidate'] != '':
170                         self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
171
172                 elif self.generated_ice_candidates.qsize() > 0:
173                     mlineindex, candidate = self.generated_ice_candidates.get_nowait()
174                     icemsg_value = json.dumps({
175                         "candidate": candidate,
176                         "sdpMid": self.mids[mlineindex],
177                         "sdpMLineIndex": mlineindex,
178                         "usernameFragment": self.user_fragments[mlineindex],
179                     })
180                     icemsg = json.dumps({
181                         'SessionID': self.session_id,
182                         'Type': 'addCalleeIceCandidate',
183                         'Value': icemsg_value,
184                     })
185                     log.info(f'Send ICE candidate: {icemsg_value}')
186                     await self.websocket.send(icemsg)
187
188                 else:
189                     await asyncio.sleep(0.1)
190         finally:
191             self.pipe.set_state(Gst.State.NULL)
192
193     async def talk_to_websocket(self, uri):
194         async for msg in self.websocket:
195             msg_json = json.loads(msg)
196             msg_type = msg_json['Type']
197             msg_value = msg_json['Value']
198             assert self.session_id is None or self.session_id == msg_json['SessionID']
199             if msg_type == 'newSession':
200                 self.session_id = msg_json['SessionID']
201                 log.info(f"New session {self.session_id}")
202             elif msg_type == 'gotOffer':
203                 value_json = json.loads(msg_value)
204                 sdp = value_json['sdp']
205                 log.info(f'Got SDP offer:\n{sdp}')
206                 self.sdp_offer = sdp
207             elif msg_type == 'addCallerIceCandidate':
208                 value_json = json.loads(msg_value)
209                 log.info(f'Got ICE candidate: {value_json}')
210                 self.received_ice_candidates.put_nowait(value_json)
211             elif msg_type == 'roomNotFound':
212                 log.error(f'The room was not found: {uri}')
213                 return
214             elif msg_type == 'roomClosed':
215                 log.info(f'Oh noes, the room went away (session {self.session_id})!')
216                 self.session_id = None
217                 return
218             else:
219                 log.error(f'Unknown message type {msg_type}')
220
221     async def run(self, uri):
222         ssl_context = ssl.SSLContext()
223         ssl_context.check_hostname = False
224         ssl_context.verify_mode = ssl.CERT_NONE
225         try:
226             async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
227                 talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
228                 listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
229                 main_loop = asyncio.Task(gstreamer_main_loop())
230                 done, pending = await asyncio.wait(
231                     [talk_to_websocket_task, listen_to_gstreamer_bus_task, main_loop],
232                     return_when=asyncio.FIRST_COMPLETED)
233                 for d in done:
234                     d.result()
235                 for p in pending:
236                     p.cancel()
237         except OSError as e:
238             print(e)
239
240
241 async def gstreamer_main_loop():
242     """Does the equivalent of the following lines in an async friendly way:
243         loop = GLib.MainLoop()
244         loop.run()
245     """
246     gst_loop = GLib.MainLoop()
247     context = gst_loop.get_context()
248     while True:
249         events_dispatched = context.iteration(False)
250         await asyncio.sleep(0. if events_dispatched else 0.01)
251
252
253 def main():
254     logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(message)s')
255     parser = argparse.ArgumentParser()
256     parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
257                         help='Signalling server URI')
258     args = parser.parse_args()
259
260     Gst.init(None)
261     rtsp = GstreamerRtspServer()
262     lagarde = Lagarde()
263     asyncio.run(lagarde.run(args.uri), debug=True)
264
265
266 if __name__ == '__main__':
267     main()