# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*- import sys import threading import traceback import logging from qtpy import QtCore import av.calls from messenger.messages import * from ui import av_widgets import common.event as event import utils.ui as util_ui from toxygen_wrapper.tests import support_testing as ts global LOG LOG = logging.getLogger('app.'+__name__) class CallsManager: def __init__(self, toxav, settings, main_screen, contacts_manager, app=None): self._callav = av.calls.AV(toxav, settings) # object with data about calls self._call = self._callav self._call_widgets = {} # dict of incoming call widgets self._incoming_calls = set() self._settings = settings self._main_screen = main_screen self._contacts_manager = contacts_manager self._call_started_event = event.Event() # friend_number, audio, video, is_outgoing self._call_finished_event = event.Event() # friend_number, is_declined self._app = app def set_toxav(self, toxav) -> None: self._callav.set_toxav(toxav) # Events def get_call_started_event(self): return self._call_started_event call_started_event = property(get_call_started_event) def get_call_finished_event(self): return self._call_finished_event call_finished_event = property(get_call_finished_event) # AV support def call_click(self, audio=True, video=False) -> None: """User clicked audio button in main window""" num = self._contacts_manager.get_active_number() if not self._contacts_manager.is_active_a_friend(): return if num not in self._callav and self._contacts_manager.is_active_online(): # start call if not self._settings['audio']['enabled']: return self._callav(num, audio, video) self._main_screen.active_call() self._call_started_event(num, audio, video, True) elif num in self._callav: # finish or cancel call if you call with active friend self.stop_call(num, False) def incoming_call(self, audio, video, friend_number) -> None: """ Incoming call from friend. """ LOG.debug(f"CM incoming_call {friend_number}") # if not self._settings['audio']['enabled']: return friend = self._contacts_manager.get_friend_by_number(friend_number) self._call_started_event(friend_number, audio, video, False) self._incoming_calls.add(friend_number) if friend_number == self._contacts_manager.get_active_number(): self._main_screen.incoming_call() else: friend.actions = True text = util_ui.tr("Incoming video call") if video else util_ui.tr("Incoming audio call") self._call_widgets[friend_number] = self._get_incoming_call_widget(friend_number, text, friend.name) self._call_widgets[friend_number].set_pixmap(friend.get_pixmap()) self._call_widgets[friend_number].show() def accept_call(self, friend_number, audio, video) -> None: """ Accept incoming call with audio or video Called from a thread """ LOG.debug(f"CM accept_call from friend_number={friend_number} {audio} {video}") sys.stdout.flush() try: self._main_screen.active_call() # failsafe added somewhere this was being left up self.close_call(friend_number) QtCore.QCoreApplication.processEvents() self._callav.call_accept_call(friend_number, audio, video) LOG.debug(f"accept_call _call.accept_call CALLED f={friend_number}") except Exception as e: # LOG.error(f"accept_call _call.accept_call ERROR for {friend_number} {e}") LOG.debug(traceback.print_exc()) self._main_screen.call_finished() if hasattr(self._main_screen, '_settings') and \ 'audio' in self._main_screen._settings and \ 'input' in self._main_screen._settings['audio']: iInput = self._settings['audio']['input'] iOutput = self._settings['audio']['output'] iVideo = self._settings['video']['device'] LOG.debug(f"iInput={iInput} iOutput={iOutput} iVideo={iVideo}") elif hasattr(self._main_screen, '_settings') and \ hasattr(self._main_screen._settings, 'audio') and \ 'input' not in self._main_screen._settings['audio']: LOG.warn(f"'audio' not in {self._main_screen._settings}") elif hasattr(self._main_screen, '_settings') and \ hasattr(self._main_screen._settings, 'audio') and \ 'input' not in self._main_screen._settings['audio']: LOG.warn(f"'audio' not in {self._main_screen._settings}") else: LOG.warn(f"_settings not in self._main_screen") util_ui.message_box(str(e), util_ui.tr('ERROR Accepting call from {friend_number}')) finally: # does not terminate call - just the av_widget LOG.debug(f"CM.accept_call close av_widget") self.close_call(friend_number) LOG.debug(f" closed self._call_widgets[{friend_number}]") def close_call(self, friend_number:int) -> None: # refactored out from above because the accept window not getting # taken down in some accept audio calls LOG.debug(f"close_call {friend_number}") try: if friend_number in self._call_widgets: self._call_widgets[friend_number].close() del self._call_widgets[friend_number] if friend_number in self._incoming_calls: self._incoming_calls.remove(friend_number) except Exception as e: # RuntimeError: wrapped C/C++ object of type IncomingCallWidget has been deleted LOG.warn(f" closed self._call_widgets[{friend_number}] {e}") # invoke_in_main_thread(QtCore.QCoreApplication.processEvents) QtCore.QCoreApplication.processEvents() def stop_call(self, friend_number, by_friend) -> None: """ Stop call with friend """ LOG.debug(f"CM.stop_call friend={friend_number}") if friend_number in self._incoming_calls: self._incoming_calls.remove(friend_number) is_declined = True else: is_declined = False if friend_number in self._call_widgets: LOG.debug(f"CM.stop_call _call_widgets close") self.close_call(friend_number) LOG.debug(f"CM.stop_call _main_screen.call_finished") self._main_screen.call_finished() self._callav.finish_call(friend_number, by_friend) # finish or decline call is_video = self._callav.is_video_call(friend_number) if is_video: def destroy_window(): #??? FixMe with ts.ignoreStdout(): import cv2 cv2.destroyWindow(str(friend_number)) LOG.debug(f"CM.stop_call destroy_window") threading.Timer(2.0, destroy_window).start() LOG.debug(f"CM.stop_call _call_finished_event") self._call_finished_event(friend_number, is_declined) def friend_exit(self, friend_number:int) -> None: if friend_number in self._callav: self._callav.finish_call(friend_number, True) # Private methods def _get_incoming_call_widget(self, friend_number, text, friend_name): return av_widgets.IncomingCallWidget(self._settings, self, friend_number, text, friend_name)