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