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