recorder.py (39985B)
1 # SPDX-FileCopyrightText: 2023-present MTRNord <support@nordgedanken.dev> 2 # 3 # SPDX-License-Identifier: AGPL-3.0-or-later 4 5 import asyncio 6 import os 7 import random 8 import string 9 import sys 10 import time 11 from dataclasses import dataclass 12 from typing import Dict, List, Optional, Tuple, Union 13 14 # This one is actually used. Sadly we cant tell py that 15 import av # type: ignore 16 import logbook # type: ignore 17 from aioice.candidate import Candidate 18 from aiortc import ( 19 MediaStreamTrack, 20 RTCConfiguration, 21 RTCIceCandidate, 22 RTCIceGatherer, 23 RTCIceServer, 24 RTCPeerConnection, 25 RTCSessionDescription, 26 ) 27 from aiortc.contrib.media import Frame, MediaPlayer, MediaRecorder, MediaStreamError 28 from aiortc.rtcicetransport import candidate_from_aioice, candidate_to_aioice 29 30 # This one is actually used. Sadly we cant tell py that 31 from av.filter import Filter, Graph # type: ignore 32 from logbook import Logger, StreamHandler 33 from nio import ( # ToDeviceCallNegotiateEvent,; CallNegotiateEvent, 34 AsyncClient, 35 CallCandidatesEvent, 36 CallHangupEvent, 37 CallInviteEvent, 38 MatrixRoom, 39 RoomGetStateError, 40 ToDeviceCallAnswerEvent, 41 ToDeviceCallCandidatesEvent, 42 ToDeviceCallHangupEvent, 43 ToDeviceCallInviteEvent, 44 ToDeviceMessage, 45 ) 46 47 from .utils.future_map import FutureMap 48 from .utils.misc_types import InputTracks 49 50 # import logging 51 52 # logging.basicConfig(level=logging.INFO) 53 # logging.getLogger("libav").setLevel(logging.DEBUG) 54 55 56 StreamHandler(sys.stdout).push_application() 57 logger = Logger(__name__) 58 logger.level = logbook.INFO 59 60 # directory to store recordings 61 RECORDING_PATH = "./recordings/" # local directory 62 63 UserID = str 64 ConfID = str 65 RoomID = str 66 UniqueCallID = Tuple[UserID, ConfID] 67 68 # TODO: Make via config 69 #STUN = RTCConfiguration(iceServers=[RTCIceServer(urls="stun:turn.matrix.org")]) 70 71 72 @dataclass 73 class Others: 74 """ 75 Data we keep about other users in a call. 76 Mostly needed for to_device messaging when setting up a call late. 77 """ 78 79 user_id: str 80 device_id: str 81 session_id: str 82 83 84 class ProxyTrack(MediaStreamTrack): 85 """ 86 Wrapper for the MediaPlayer to be able to show the duration 87 """ 88 89 __source: MediaStreamTrack 90 __graph: Optional[Graph] 91 92 def __init__( 93 self, 94 source: MediaPlayer, 95 ) -> None: 96 super().__init__() 97 self.kind = source.video.kind 98 self.__source = source.video 99 self.__graph = None 100 101 async def start(self, frame: Frame) -> None: 102 """ 103 This function is used to setup the filter graph that displays the 104 duration of the recording. 105 106 Note that type errors are expected since we interact with ffmpeg here. 107 So graph is actually a C pointer and not a python object. 108 """ 109 110 self.__graph = Graph() 111 graph_source = self.__graph.add_buffer(template=frame) 112 113 graph_filter = self.__graph.add( 114 "drawtext", 115 r"text='Recording Duration: %{pts:gmtime:0:%H\:%M\:%S}':x=(w-text_w)/2:y=(h-text_h)/2:fontcolor=white:fontsize=100", 116 ) 117 graph_sink = self.__graph.add("buffersink") 118 119 graph_source.link_to(graph_filter, 0, 0) 120 graph_filter.link_to(graph_sink, 0, 0) 121 self.__graph.configure() 122 123 async def recv(self) -> Frame: 124 """ 125 We handle the next Frame here. First we setup the graph and then use it. 126 we fall back to sending the source image. (Might fail due to ffmpeg. 127 It then results in a grey image.) 128 """ 129 130 frame = await self.__source.recv() 131 try: 132 if not self.__graph: 133 await self.start(frame) 134 if self.__graph: 135 self.__graph.push(frame) 136 filtered_frame = self.__graph.pull() 137 return filtered_frame 138 else: 139 logger.error("Video Init failed!") 140 return frame 141 except MediaStreamError as error: 142 logger.warning(f"MediaStreamError in recv: {error}") 143 return frame 144 except Exception as error: 145 logger.warning(f"Exception in recv: {error}") 146 return frame 147 148 def stop(self) -> None: 149 self.__source.stop() 150 super().stop() 151 152 153 @dataclass 154 class WrappedConn: 155 """ 156 The state data of a WebRTC Connection we have running 157 """ 158 159 pc: RTCPeerConnection 160 prepare_waiter: Optional[asyncio.Future] 161 candidate_waiter: Optional[asyncio.Future] 162 room_id: Optional[str] 163 input_tracks: InputTracks 164 165 166 class Recorder: 167 """ 168 Core handling of the bot. 169 This does both the webrtc handshake as well as the recording handling currently. 170 """ 171 172 __conns: FutureMap[UniqueCallID, WrappedConn] 173 __outputs: dict[RoomID, ProxyTrack] 174 party_id: str 175 client: AsyncClient 176 loop: asyncio.AbstractEventLoop 177 conf_room: Dict[ConfID, MatrixRoom] 178 recording_rooms: List[ConfID] 179 room_conf: Dict[RoomID, ConfID] 180 others: Dict[ConfID, List[Others]] 181 session_id: str 182 183 def __init__(self, client) -> None: 184 self.client = client 185 self.loop = asyncio.get_event_loop() 186 self.__conns = FutureMap() 187 self.conf_room = {} 188 self.room_conf = {} 189 self.others = {} 190 self.recording_rooms = [] 191 self.__outputs = {} 192 self.party_id = "".join( 193 random.choices(string.ascii_letters + string.digits, k=8) 194 ) 195 self.session_id = "".join( 196 random.choices(string.ascii_letters + string.digits, k=8) 197 ) 198 199 if not os.path.exists(RECORDING_PATH): 200 os.makedirs(RECORDING_PATH) 201 202 logger.info("Starting recording handler") 203 204 async def stop(self) -> None: 205 """ 206 Stops and cleans up the call. 207 This mainly means sending out all hangup events. 208 """ 209 210 logger.info("Stopping recording handler") 211 for (_, conf_or_call_id), conn in await self.__conns.items(): 212 await self.hangup(conf_or_call_id, conn) 213 214 def add_call(self, conf_id: ConfID, room: MatrixRoom) -> None: 215 """ 216 Adds the Call room to the internal state. 217 """ 218 219 logger.info(f"Adding conf {conf_id} to room {room.room_id}") 220 self.conf_room[conf_id] = room 221 self.room_conf[room.room_id] = conf_id 222 223 async def hangup(self, conf_or_call_id: ConfID, conn: WrappedConn) -> None: 224 """ 225 This handles the hangup negotiation of a call. 226 It also makes sure to close the WebRTC connection 227 """ 228 229 hangup = { 230 "call_id": conf_or_call_id, 231 "version": "1", 232 "party_id": self.party_id, 233 "conf_id": conf_or_call_id, 234 } 235 if conn.room_id and conf_or_call_id not in self.others: 236 # We are lazy and send it to the room and as to_device message 237 await self.client.room_send( 238 conn.room_id, 239 "m.call.hangup", 240 hangup, 241 ignore_unverified_devices=True, 242 ) 243 244 already_sent_to = [] 245 if conf_or_call_id in self.others: 246 # Send it as to_device message 247 others = self.others[conf_or_call_id] 248 249 for data in others: 250 if data.user_id == self.client.user_id: 251 continue 252 if data in already_sent_to: 253 continue 254 255 hangup["device_id"] = str(self.client.device_id) 256 hangup["sender_session_id"] = self.session_id 257 hangup["dest_session_id"] = data.session_id 258 259 message = ToDeviceMessage( 260 "m.call.hangup", 261 data.user_id, 262 data.device_id, 263 hangup, 264 ) 265 logger.info("Sending hangup via to_device") 266 await self.client.to_device(message) 267 already_sent_to.append(data) 268 269 # If as this might be not the case due to races 270 if conf_or_call_id in self.conf_room: 271 await self.client.room_put_state( 272 self.conf_room[conf_or_call_id].room_id, 273 "org.matrix.msc3401.call.member", 274 {"m.calls": []}, 275 state_key=self.client.user_id, 276 ) 277 278 await conn.pc.close() 279 280 async def leave_call(self, room: MatrixRoom) -> None: 281 """ 282 Handler for the stop command. 283 This sends the hangup events, cleans up the state we have, and resets the media. 284 """ 285 286 if room.room_id in self.room_conf: 287 conf_id = self.room_conf[room.room_id] 288 289 for other in self.others[conf_id]: 290 # Send Hangups 291 unique_id = (other.user_id, conf_id) 292 try: 293 conn = await asyncio.wait_for(self.__conns[unique_id], timeout=3) 294 await self.hangup(conf_id, conn) 295 except asyncio.TimeoutError: 296 logger.warning("Gave up waiting for call on leave, task canceled") 297 298 # End the connection 299 await self.remove_connection(room) 300 301 if room.room_id in self.__outputs: 302 del self.__outputs[room.room_id] 303 304 # Notify user 305 await self.client.room_send( 306 room.room_id, 307 "m.room.message", 308 { 309 "msgtype": "m.notice", 310 "body": "Recording stopped", 311 }, 312 ignore_unverified_devices=True, 313 ) 314 else: 315 # Notify user that we didnt have a running recording 316 await self.client.room_send( 317 room.room_id, 318 "m.room.message", 319 { 320 "msgtype": "m.notice", 321 "body": "No running recording was found.", 322 }, 323 ignore_unverified_devices=True, 324 ) 325 326 async def remove_connection(self, room: MatrixRoom) -> Optional[ConfID]: 327 """ 328 This resets the connection for a call as well as removing the state. 329 """ 330 if room.room_id in self.room_conf: 331 call_id = self.room_conf[room.room_id] 332 unique_id = (self.client.user_id, call_id) 333 334 try: 335 conn = await asyncio.wait_for(self.__conns[unique_id], timeout=3) 336 await conn.pc.close() 337 del self.__conns[unique_id] 338 except asyncio.TimeoutError: 339 logger.warning( 340 "Gave up waiting for call on remove connection, task canceled" 341 ) 342 343 # del self.room_conf[room.room_id] 344 # del self.conf_room[call_id] 345 346 return call_id 347 return None 348 349 def track_others( 350 self, conf_id: ConfID, device_id: str, user_id: UserID, session_id: str 351 ) -> None: 352 """ 353 Adds members of a call to the internal state. 354 """ 355 356 if conf_id not in self.others: 357 self.others[conf_id] = [] 358 self.others[conf_id].append(Others(user_id, device_id, session_id)) 359 360 def remove_other(self, conf_id: ConfID, user_id: UserID) -> None: 361 """ 362 Removes people that are tracked. 363 364 This should be called if a member leaves. 365 """ 366 self.others[conf_id] = [o for o in self.others[conf_id] if o.user_id != user_id] 367 368 async def join_call(self, room: MatrixRoom) -> None: 369 """ 370 Joins a Call. 371 372 This usually only is used when !start is pressed in a element-call room. 373 374 It sets up a new invite and candidates. 375 """ 376 377 if room.room_id not in self.room_conf: 378 logger.warning( 379 f""" 380 Room {room.room_id} is not a call room or we forgot. 381 Trying to get call id from state. 382 """ 383 ) 384 room_state = await self.client.room_get_state(room.room_id) 385 if isinstance(room_state, RoomGetStateError): 386 logger.warning(f"Could not get state for {room.room_id}: {room_state}") 387 return 388 389 # Find "org.matrix.msc3401.call" state event 390 call_state = next( 391 ( 392 event 393 for event in room_state.events 394 if event["type"] == "org.matrix.msc3401.call" 395 ), 396 None, 397 ) 398 if call_state is None: 399 logger.warning(f"No call state event in {room.room_id}") 400 return 401 402 member_states = [ 403 event 404 for event in room_state.events 405 if event["type"] == "org.matrix.msc3401.call.member" 406 ] 407 408 call_id = call_state["state_key"] 409 logger.info(f"Found call id {call_id} for {room.room_id}") 410 self.add_call(call_id, room) 411 for member in member_states: 412 for call in member["content"]["m.calls"]: 413 for device in call["m.devices"]: 414 self.track_others( 415 call["m.call_id"], 416 device["device_id"], 417 member["sender"], 418 device["session_id"], 419 ) 420 421 logger.info(f"Joining call in {room.room_id}") 422 await self.client.room_put_state( 423 room.room_id, 424 "org.matrix.msc3401.call.member", 425 {"m.calls": []}, 426 state_key=self.client.user_id, 427 ) 428 429 # Borked without GLARE 430 # await self.send_offer(room) 431 432 conf_id = self.room_conf[room.room_id] 433 await self.client.room_put_state( 434 room.room_id, 435 "org.matrix.msc3401.call.member", 436 { 437 "m.calls": [ 438 { 439 "m.call_id": conf_id, 440 "m.devices": [ 441 { 442 "device_id": self.client.device_id, 443 "expires_ts": int(time.time() * 1000) 444 + (1000 * 60 * 60), 445 "session_id": self.session_id, 446 "feeds": [ 447 { 448 "purpose": "m.usermedia", 449 } 450 ], 451 } 452 ], 453 } 454 ] 455 }, 456 state_key=self.client.user_id, 457 ) 458 459 async def send_offer(self, room: MatrixRoom) -> None: 460 # Send offer to others 461 conf_id = self.room_conf[room.room_id] 462 logger.info(f"Sending offer to others: {conf_id}") 463 if self.room_conf[room.room_id] in self.others: 464 # Send it as to_device message 465 others = self.others[conf_id] 466 logger.info(f"Sending offers to {others}") 467 468 call_id = "".join(random.choices(string.ascii_letters + string.digits, k=8)) 469 for data in others: 470 if data.user_id == self.client.user_id: 471 continue 472 473 logger.info(f"Making offer for {data.user_id}") 474 475 # Create offer 476 #pc = RTCPeerConnection(STUN) 477 pc = RTCPeerConnection() 478 logger.info(f"Started ice for {call_id}") 479 unique_id = (data.user_id, self.room_conf[room.room_id]) 480 conn = WrappedConn( 481 pc=pc, 482 candidate_waiter=None, 483 prepare_waiter=None, 484 room_id=room.room_id if room else None, 485 input_tracks={}, 486 ) 487 488 if room.room_id not in self.__outputs: 489 self.__outputs[room.room_id] = ProxyTrack( 490 MediaPlayer( 491 "./black.png", 492 options={"loop": "1", "framerate": "1", "hwaccel": "auto", "c:v": "h264", "preset:v": "ultrafast"}, 493 ) 494 ) 495 496 logger.info(f"Created connection {unique_id}") 497 498 conn.pc.addTrack(self.__outputs[room.room_id]) 499 conn.pc.addTransceiver("audio", "sendrecv") 500 501 logger.info("Adding tracks") 502 503 base_path = os.path.join(RECORDING_PATH, f"{conf_id}") 504 if not os.path.exists(base_path): 505 os.mkdir(base_path) 506 base_name_audio = f"{data.user_id}_{call_id}" 507 base_name_video = f"{data.user_id}_{call_id}" 508 509 (wav_file, mp4_file) = self.get_filenames( 510 base_path, base_name_audio, base_name_video 511 ) 512 513 logger.info("Setting up callbacks") 514 515 pc.on( 516 "connectionstatechange", 517 lambda conn=conn, unique_id=unique_id, wav_file=wav_file, mp4_file=mp4_file, conf_id=conf_id: self.on_connectionstatechange( 518 conn, unique_id, wav_file, mp4_file, conf_id 519 ), 520 ) 521 522 pc.on( 523 "track", 524 lambda track, conn=conn, user_id=data.user_id: self.on_track( 525 track, conn, user_id 526 ), 527 ) 528 529 offer = await conn.pc.createOffer() 530 await conn.pc.setLocalDescription(offer) 531 532 logger.info(f"Got local candidates for {call_id}") 533 candidates: List[Tuple[RTCIceCandidate, str]] = [] 534 for transceiver in conn.pc.getTransceivers(): 535 gatherer: RTCIceGatherer = ( 536 transceiver.sender.transport.transport.iceGatherer 537 ) 538 for candidate in gatherer.getLocalCandidates(): 539 candidate.sdpMid = transceiver.mid 540 candidates.append( 541 ( 542 candidate, 543 str(gatherer.getLocalParameters().usernameFragment), 544 ) 545 ) 546 547 # We set this late to make sure the connection is actually set up before we handle the answer. 548 # Order of operation is weird otherwise. 549 self.__conns[unique_id] = conn 550 551 offer_message = ToDeviceMessage( 552 "m.call.invite", 553 recipient=data.user_id, 554 recipient_device=data.device_id, 555 content={ 556 "lifetime": 60000, 557 "invitee": data.user_id, 558 "offer": { 559 "sdp": conn.pc.localDescription.sdp, 560 "type": conn.pc.localDescription.type, 561 }, 562 "version": "1", 563 "conf_id": conf_id, 564 "call_id": call_id, 565 "party_id": self.party_id, 566 "seq": 0, 567 "device_id": self.client.device_id, 568 "sender_session_id": self.session_id, 569 "dest_session_id": data.session_id, 570 "capabilities": { 571 "m.call.transferee": False, 572 "m.call.dtmf": False, 573 }, 574 "org.matrix.msc3077.sdp_stream_metadata": { 575 pc._RTCPeerConnection__stream_id: { # type: ignore 576 "purpose": "m.usermedia", 577 "audio_muted": True, 578 "video_muted": False, 579 } 580 }, 581 }, 582 ) 583 await self.client.to_device(offer_message) 584 585 logger.info(f"Sending candidates to {data.user_id} {data.device_id}") 586 candidates_message = ToDeviceMessage( 587 "m.call.candidates", 588 recipient=data.user_id, 589 recipient_device=data.device_id, 590 content={ 591 "candidates": [ 592 { 593 "candidate": f"candidate:{candidate_to_aioice(c).to_sdp()}", 594 "sdpMid": c.sdpMid, 595 # "sdpMLineIndex": c.sdpMLineIndex, 596 "usernameFragment": usernameFragment, 597 } 598 for (c, usernameFragment) in candidates 599 ], 600 "call_id": call_id, 601 "party_id": self.party_id, 602 "version": "1", 603 "seq": 1, 604 "conf_id": conf_id, 605 "device_id": self.client.device_id, 606 "sender_session_id": self.session_id, 607 "dest_session_id": data.session_id, 608 }, 609 ) 610 await self.client.to_device(candidates_message) 611 612 # async def handle_negotiation( 613 # self, 614 # room: Optional[MatrixRoom], 615 # event: [ToDeviceCallNegotiateEvent, CallNegotiateEvent], 616 # ): 617 # pass 618 619 def on_track( 620 self, 621 track: MediaStreamTrack, 622 conn: WrappedConn, 623 user_id: str, 624 output_track: Optional[ProxyTrack] = None, 625 ) -> None: 626 if track.kind == "audio": 627 logger.info( 628 f"Adding audio track to recording for {user_id} when state {conn.pc.connectionState}" 629 ) 630 conn.input_tracks["audio"] = track 631 elif track.kind == "video": 632 logger.info( 633 f"Adding video track to recording for {user_id} when state {conn.pc.connectionState}" 634 ) 635 conn.input_tracks["video"] = track 636 if conn.pc and output_track: 637 conn.pc.addTrack(output_track) 638 639 async def on_connectionstatechange( 640 self, 641 conn: WrappedConn, 642 unique_id: UniqueCallID, 643 wav_file: str, 644 mp4_file: str, 645 conf_id: ConfID, 646 ) -> None: 647 if conn.pc.connectionState == "failed": 648 logger.warn(f'State changed to "failed" for {unique_id}') 649 await conn.pc.close() 650 del self.__conns[unique_id] 651 if conn.pc.connectionState == "connected": 652 logger.info(f'State changed to "connected" for {unique_id}') 653 asyncio.create_task( 654 self.start_recording(wav_file, mp4_file, conn.input_tracks, conf_id) 655 ) 656 657 async def handle_call_answer(self, event: ToDeviceCallAnswerEvent) -> None: 658 """ 659 Handles the ToDevice call answer event. 660 661 We only handle the to_device variant since we only initiate the call within 1:1 rooms. 662 """ 663 logger.info(f"Received call answer from {event.sender}") 664 665 unique_id: UniqueCallID = (event.sender, event.conf_id) 666 logger.info(f"Received call answer for {unique_id}") 667 668 if event.conf_id not in self.conf_room: 669 logger.warning("Got invalid to-device call answer") 670 return 671 672 try: 673 conn = await asyncio.wait_for(self.__conns[unique_id], timeout=3) 674 except asyncio.TimeoutError: 675 logger.warning("Gave up waiting for call on call answer, task canceled") 676 return 677 except KeyError: 678 logger.warning("Received answer for unknown call") 679 return 680 681 logger.info(f"Setting remote description for {unique_id}") 682 await conn.pc.setRemoteDescription( 683 RTCSessionDescription( 684 sdp=str(event.answer.get("sdp")), type=str(event.answer.get("type")) 685 ) 686 ) 687 688 stats = await conn.pc.getStats() 689 print(f"Stats for {unique_id}: {stats}") 690 receivers = conn.pc.getTransceivers() 691 for receiver in receivers: 692 print( 693 f"Transceiver kind for {unique_id}: {receiver.kind} - {receiver.currentDirection}" 694 ) 695 696 logger.info("Adding tracks") 697 698 if isinstance(event, CallInviteEvent): 699 base_path = RECORDING_PATH 700 else: 701 base_path = os.path.join(RECORDING_PATH, f"{event.conf_id}") 702 if not os.path.exists(base_path): 703 os.mkdir(base_path) 704 base_name_audio = f"{event.sender}_{event.call_id}" 705 base_name_video = f"{event.sender}_{event.call_id}" 706 707 (wav_file, mp4_file) = self.get_filenames( 708 base_path, base_name_audio, base_name_video 709 ) 710 711 logger.info("Setting up callbacks") 712 713 conn.pc.on( 714 "track", 715 lambda track, conn=conn, user_id=event.sender: self.on_track( 716 track, conn, user_id 717 ), 718 ) 719 720 conn.pc.on( 721 "connectionstatechange", 722 lambda conn=conn, unique_id=unique_id, wav_file=wav_file, mp4_file=mp4_file, conf_id=event.call_id if isinstance( 723 event, CallInviteEvent 724 ) else event.conf_id: self.on_connectionstatechange( 725 conn, unique_id, wav_file, mp4_file, conf_id 726 ), 727 ) 728 729 others = self.others[event.conf_id] 730 data = next((x for x in others if x.user_id == event.sender), None) 731 732 if not data: 733 logger.warning("Received answer for unknown call") 734 return 735 736 logger.info(f"Sending select_answer for {unique_id}") 737 message = ToDeviceMessage( 738 "m.call.select_answer", 739 recipient=event.sender, 740 recipient_device=event.source["content"]["device_id"], 741 content={ 742 "selected_party_id": event.party_id, 743 "call_id": event.call_id, 744 "party_id": self.party_id, 745 "version": "1", 746 "seq": 2, 747 "conf_id": event.conf_id, 748 "device_id": self.client.device_id, 749 "sender_session_id": self.session_id, 750 "dest_session_id": data.session_id, 751 }, 752 ) 753 await self.client.to_device(message) 754 755 def get_filenames( 756 self, base_path: str, base_name_audio: str, base_name_video: str 757 ) -> Tuple[str, str]: 758 if os.path.exists(os.path.join(base_path, f"{base_name_audio}.wav")): 759 i = 1 760 while os.path.exists(os.path.join(base_path, f"{base_name_audio}_{i}.wav")): 761 i += 1 762 base_name_audio = f"{base_name_audio}_{i}" 763 764 if os.path.exists(os.path.join(base_path, f"{base_name_video}.mp4")): 765 i = 1 766 while os.path.exists(os.path.join(base_path, f"{base_name_video}_{i}.mp4")): 767 i += 1 768 base_name_video = f"{base_name_video}_{i}" 769 wav_file = os.path.join(base_path, f"{base_name_audio}") 770 mp4_file = os.path.join(base_path, f"{base_name_video}") 771 return (wav_file, mp4_file) 772 773 async def start_recording( 774 self, wav_file: str, mp4_file: str, input_tracks: InputTracks, conf_id: ConfID 775 ) -> None: 776 if "audio" in input_tracks: 777 track_id = input_tracks["audio"].id 778 wav_file_track_id = f"{wav_file}_{track_id}.wav" 779 logger.info(f"Starting audio recorder for {wav_file_track_id}") 780 audio_recorder = MediaRecorder(wav_file_track_id, format="wav") 781 audio_recorder.addTrack(input_tracks["audio"]) 782 await audio_recorder.start() 783 logger.info(f"Started audio recorder for {wav_file_track_id}") 784 if "video" in input_tracks: 785 track_id = input_tracks["video"].id 786 mp4_file_track_id = f"{mp4_file}_{track_id}.mp4" 787 logger.info(f"Starting video recorder for {mp4_file_track_id}") 788 video_recorder = MediaRecorder(mp4_file_track_id, format="mp4") 789 video_recorder.addTrack(input_tracks["video"]) 790 await video_recorder.start() 791 logger.info(f"Started video recorder for {mp4_file_track_id}") 792 else: 793 video_recorder = None 794 795 if "audio" in input_tracks: 796 audio_track = input_tracks["audio"] 797 798 @audio_track.on("ended") 799 async def on_ended_audio(): 800 if audio_recorder: 801 logger.info(f"Audio ended for {wav_file}") 802 await audio_recorder.stop() 803 804 if "video" in input_tracks: 805 video_track = input_tracks["video"] 806 807 @video_track.on("ended") 808 async def on_ended_video(): 809 if video_recorder: 810 logger.info(f"Video ended for {mp4_file}") 811 await video_recorder.stop() 812 813 if conf_id not in self.recording_rooms: 814 self.recording_rooms.append(conf_id) 815 await self.client.room_send( 816 self.conf_room[conf_id].room_id, 817 "m.room.message", 818 { 819 "msgtype": "m.notice", 820 "body": "Successfully started recording", 821 }, 822 ignore_unverified_devices=True, 823 ) 824 825 async def handle_call_invite( 826 self, 827 event: Union[CallInviteEvent, ToDeviceCallInviteEvent], 828 room: Optional[MatrixRoom], 829 ) -> None: 830 """ 831 Handle any invite we get. The main caller is 1:1 calls. 832 """ 833 if room: 834 logger.info(f"Received call invite from {event.sender} in {room.room_id}") 835 else: 836 logger.info(f"Received call invite from {event.sender}") 837 if event.expired: 838 logger.warning("Call invite expired") 839 return 840 if event.version != "1": 841 logger.warning("Call invite version not supported") 842 return 843 844 if isinstance(event, ToDeviceCallInviteEvent): 845 if event.conf_id not in self.conf_room: 846 logger.warning("Got invalid to-device call invite") 847 return 848 849 logger.info("Preparing call") 850 offer = RTCSessionDescription( 851 sdp=str(event.offer.get("sdp")), type=str(event.offer.get("type")) 852 ) 853 854 #pc = RTCPeerConnection(STUN) 855 pc = RTCPeerConnection() 856 room_id = "" 857 if room: 858 room_id = room.room_id 859 else: 860 if isinstance(event, ToDeviceCallInviteEvent): 861 room_id = self.conf_room[event.conf_id].room_id 862 863 if room_id not in self.__outputs: 864 self.__outputs[room_id] = ProxyTrack( 865 MediaPlayer( 866 "./black.png", 867 options={"loop": "1", "framerate": "1", "hwaccel": "auto"}, 868 ) 869 ) 870 pc.addTrack(self.__outputs[room_id]) 871 if isinstance(event, CallInviteEvent): 872 unique_id: UniqueCallID = (event.sender, event.call_id) 873 else: 874 unique_id = (event.sender, event.conf_id) 875 conn = self.__conns[unique_id] = WrappedConn( 876 pc=pc, 877 candidate_waiter=self.loop.create_future(), 878 prepare_waiter=self.loop.create_future(), 879 input_tracks={}, 880 room_id=room.room_id if room else None, 881 ) 882 883 logger.info("Adding tracks") 884 885 if isinstance(event, CallInviteEvent): 886 base_path = RECORDING_PATH 887 else: 888 base_path = os.path.join(RECORDING_PATH, f"{event.conf_id}") 889 if not os.path.exists(base_path): 890 os.mkdir(base_path) 891 base_name_audio = f"{event.sender}_{event.call_id}" 892 base_name_video = f"{event.sender}_{event.call_id}" 893 894 (wav_file, mp4_file) = self.get_filenames( 895 base_path, base_name_audio, base_name_video 896 ) 897 898 logger.info("Setting up callbacks") 899 900 conn.pc.on( 901 "track", 902 lambda track, conn=conn, user_id=event.sender: self.on_track( 903 track, conn, user_id 904 ), 905 ) 906 907 conn.pc.on( 908 "connectionstatechange", 909 lambda conn=conn, unique_id=unique_id, wav_file=wav_file, mp4_file=mp4_file, conf_id=event.call_id if isinstance( 910 event, CallInviteEvent 911 ) else event.conf_id: self.on_connectionstatechange( 912 conn, unique_id, wav_file, mp4_file, conf_id 913 ), 914 ) 915 916 logger.info("Waiting for prepare") 917 await pc.setRemoteDescription(offer) 918 919 logger.info("Ready to receive candidates") 920 if conn.prepare_waiter: 921 conn.prepare_waiter.set_result(None) 922 923 if room and isinstance(event, CallInviteEvent): 924 logger.info("Sending receipt") 925 await self.client.update_receipt_marker(room.room_id, event.event_id) 926 927 if conn.candidate_waiter: 928 await conn.candidate_waiter 929 logger.info("Got candidates") 930 931 logger.info("Creating answer") 932 answer = await pc.createAnswer() 933 if not answer: 934 logger.warning("Failed to create answer") 935 await pc.close() 936 del self.__conns[unique_id] 937 if room: 938 await self.hangup(event.call_id, conn) 939 elif isinstance(event, ToDeviceCallInviteEvent): 940 await self.hangup(event.conf_id, conn) 941 return 942 943 await pc.setLocalDescription(answer) 944 945 logger.info("Sending answer") 946 if room: 947 answer = { 948 "call_id": event.call_id, 949 "version": "1", 950 "party_id": self.party_id, 951 "answer": { 952 "type": pc.localDescription.type, 953 "sdp": pc.localDescription.sdp, 954 }, 955 } 956 await self.client.room_send( 957 room.room_id, "m.call.answer", answer, ignore_unverified_devices=True 958 ) 959 else: 960 answer = { 961 "call_id": event.call_id, 962 "version": "1", 963 "party_id": self.party_id, 964 "conf_id": event.source["content"]["conf_id"], 965 "capabilities": { 966 "m.call.transferee": False, 967 "m.call.dtmf": False, 968 }, 969 "org.matrix.msc3077.sdp_stream_metadata": { 970 pc._RTCPeerConnection__stream_id: { # type: ignore 971 "purpose": "m.usermedia", 972 "audio_muted": True, 973 "video_muted": False, 974 } 975 }, 976 "answer": { 977 "type": pc.localDescription.type, 978 "sdp": pc.localDescription.sdp, 979 }, 980 "device_id": self.client.device_id, 981 "dest_session_id": event.source["content"]["sender_session_id"], 982 "sender_session_id": self.session_id, 983 "seq": event.source["content"]["seq"], 984 } 985 to_device_message = ToDeviceMessage( 986 "m.call.answer", 987 event.source["sender"], 988 event.source["sender_device"], 989 answer, 990 ) 991 await self.client.to_device(to_device_message) 992 if room: 993 logger.info(f"Sent answer to {event.sender} in {room.room_id}") 994 elif isinstance(event, ToDeviceCallInviteEvent): 995 logger.info( 996 f"Sent answer to {event.sender} with device {event.source['sender_device']}" 997 ) 998 999 async def handle_call_candidates( 1000 self, 1001 room: Optional[MatrixRoom], 1002 event: Union[CallCandidatesEvent, ToDeviceCallCandidatesEvent], 1003 ) -> None: 1004 """ 1005 Handle call candidates we get and add them to the connection. 1006 """ 1007 1008 if room: 1009 logger.info( 1010 f"Received call candidates from {event.sender} in {room.room_id}" 1011 ) 1012 else: 1013 logger.info(f"Received call candidates from {event.sender}") 1014 1015 if isinstance(event, CallCandidatesEvent): 1016 unique_id: UniqueCallID = (event.sender, event.call_id) 1017 else: 1018 unique_id = (event.sender, event.conf_id) 1019 1020 if event.conf_id not in self.conf_room: 1021 logger.warning("Got invalid to-device call candidates") 1022 return 1023 1024 try: 1025 conn = await asyncio.wait_for(self.__conns[unique_id], timeout=10) 1026 except asyncio.TimeoutError: 1027 logger.warning("Gave up waiting for call candidates, task canceled") 1028 return 1029 except KeyError: 1030 logger.warning("Received candidates for unknown call") 1031 return 1032 1033 logger.info("Waiting for prepare") 1034 if conn.prepare_waiter: 1035 await conn.prepare_waiter 1036 logger.info("Adding candidates") 1037 for raw_candidate in event.candidates: 1038 if not raw_candidate.get("candidate"): 1039 # End of candidates 1040 try: 1041 if conn.candidate_waiter: 1042 conn.candidate_waiter.set_result(None) 1043 except asyncio.InvalidStateError: 1044 pass 1045 break 1046 try: 1047 candidate = candidate_from_aioice( 1048 Candidate.from_sdp(raw_candidate.get("candidate")) 1049 ) 1050 except ValueError as error: 1051 logger.warning(f"Received invalid candidate: {error}") 1052 continue 1053 candidate.sdpMid = raw_candidate.get("sdpMid") 1054 candidate.sdpMLineIndex = raw_candidate.get("sdpMLineIndex") 1055 1056 logger.info(f"Adding candidate {candidate} for {event.call_id}") 1057 await conn.pc.addIceCandidate(candidate) 1058 logger.info("Done adding candidates") 1059 try: 1060 if conn.candidate_waiter: 1061 conn.candidate_waiter.set_result(None) 1062 except asyncio.InvalidStateError: 1063 pass 1064 if room and isinstance(event, CallCandidatesEvent): 1065 await self.client.update_receipt_marker(room.room_id, event.event_id) 1066 1067 async def handle_call_hangup( 1068 self, 1069 room: Optional[MatrixRoom], 1070 event: Union[CallHangupEvent, ToDeviceCallHangupEvent], 1071 ) -> None: 1072 """ 1073 Handles the hangup. 1074 1075 Currently this only closes the connection. 1076 """ 1077 1078 if event.sender == self.client.user_id: 1079 return 1080 reason = None 1081 if "reason" in event.source["content"]: 1082 reason = event.source["content"]["reason"] 1083 if room: 1084 logger.info( 1085 f"Received call hangup from {event.sender} in {room.room_id} with reason {reason}" 1086 ) 1087 else: 1088 logger.info( 1089 f"Received call hangup from {event.sender} with reason {reason}" 1090 ) 1091 1092 # TODO: This is incorrect: 1093 # The session ele-web sends on new_session is dead 1094 if reason == "replaced" or reason == "new_session": 1095 logger.warning("Call was replaced but we ignore that for now") 1096 return 1097 1098 try: 1099 if isinstance(event, CallHangupEvent): 1100 unique_id: UniqueCallID = (event.sender, event.call_id) 1101 else: 1102 unique_id = (event.sender, event.conf_id) 1103 try: 1104 conn = await asyncio.wait_for(self.__conns[unique_id], timeout=3) 1105 await conn.pc.close() 1106 except asyncio.TimeoutError: 1107 logger.warning("Gave up waiting for call on hangup, task canceled") 1108 if room and isinstance(event, CallHangupEvent): 1109 await self.client.update_receipt_marker(room.room_id, event.event_id) 1110 1111 except KeyError: 1112 logger.warning("Received hangup for unknown call") 1113 return