Module pytgcalls.implementation

class GroupCall(ABC, GroupCallDispatcherMixin, GroupCallNative):
    SEND_ACTION_UPDATE_EACH = 0.45
    '''How often to send speaking action to chat'''

    __ASYNCIO_TIMEOUT = 10

    def __init__(
        self,
        mtproto_bridge,
        enable_logs_to_console: bool,
        path_to_log_file: str,
        outgoing_audio_bitrate_kbit: int,
    ):
        GroupCallNative.__init__(
            self,
            self.__emit_join_payload_callback,
            self.__network_state_updated_callback,
            enable_logs_to_console,
            path_to_log_file,
            outgoing_audio_bitrate_kbit,
        )
        GroupCallDispatcherMixin.__init__(self, GroupCallAction)

        self.mtproto = mtproto_bridge
        self.mtproto.register_group_call_native_callback(
            self._group_call_participants_update_callback, self._group_call_update_callback
        )

        self.invite_hash = None
        '''Hash from invite link to join as speaker'''

        self.enable_action = True
        '''Is enable sending of speaking action'''

        self.is_connected = False
        '''Is connected to voice chat via tgcalls'''

        self.__is_stop_requested = False
        self.__emit_join_payload_event = None

        self.__is_muted = True

    async def _group_call_participants_update_callback(self, update: UpdateGroupCallParticipantsWrapper):
        logger.debug('Group call participants update...')
        logger.debug(update)

        self.trigger_handlers(GroupCallAction.PARTICIPANT_LIST_UPDATED, self, update.participants)

        for participant in update.participants:
            ssrc = uint_ssrc(participant.source)

            # maybe (if needed) set unmute status on server side after allowing to speak by admin
            # also mb there is need a some delay after getting update cuz server sometimes cant handle editing properly
            if participant.is_self and participant.can_self_unmute:
                if not self.__is_muted:
                    await self.edit_group_call(muted=False)

            if participant.peer == self.mtproto.join_as and ssrc != self.mtproto.my_ssrc:
                logger.debug(f'Not equal ssrc. Expected: {ssrc}. Actual: {self.mtproto.my_ssrc}.')
                await self.reconnect()

    async def _group_call_update_callback(self, update: UpdateGroupCallWrapper):
        logger.debug('Group call update...')
        logger.debug(update)

        if isinstance(update.call, GroupCallDiscardedWrapper):
            logger.debug('Group call discarded.')
            await self.stop()
        elif update.call.params:
            self.__set_join_response_payload(update.call.params.data)

    def __set_join_response_payload(self, payload):
        logger.debug('Set join response payload...')

        if self.__is_stop_requested:
            logger.debug('Set payload rejected by a stop request.')
            return

        self._set_connection_mode(tgcalls.GroupConnectionMode.GroupConnectionModeRtc)
        self._set_join_response_payload(payload)
        logger.debug('Join response payload was set.')

    def __emit_join_payload_callback(self, payload):
        logger.debug('Emit join payload callback...')

        if self.__is_stop_requested:
            logger.debug('Join group call rejected by a stop request.')
            return

        if self.mtproto.group_call is None:
            logger.debug('Group Call is None.')
            return

        async def _():
            try:

                def pre_update_processing():
                    logger.debug(f'Set my ssrc to {payload.audioSsrc}.')
                    self.mtproto.set_my_ssrc(payload.audioSsrc)

                await self.mtproto.join_group_call(
                    self.invite_hash, payload.json, muted=True, pre_update_processing=pre_update_processing
                )

                if self.__emit_join_payload_event:
                    self.__emit_join_payload_event.set()

                logger.debug(
                    f'Successfully connected to VC with '
                    f'ssrc={self.mtproto.my_ssrc} '
                    f'as {type(self.mtproto.join_as).__name__}.'
                )
            except GroupcallSsrcDuplicateMuch:
                logger.debug('Duplicate SSRC.')
                await self.reconnect()

        asyncio.ensure_future(_(), loop=self.mtproto.get_event_loop())

    def __network_state_updated_callback(self, state: bool):
        logger.debug('Network state updated...')

        if self.is_connected == state:
            logger.debug('Network state is same. Do nothing.')
            return

        self.is_connected = state
        if self.is_connected:
            asyncio.ensure_future(self.set_is_mute(False), loop=self.mtproto.get_event_loop())
            if self.enable_action:
                self.__start_status_worker()

        self.trigger_handlers(GroupCallAction.NETWORK_STATUS_CHANGED, self, state)

        logger.debug(f'New network state is {self.is_connected}.')

    def __start_status_worker(self):
        async def worker():
            logger.debug('Start status (call action) worker...')
            while self.is_connected:
                await self.mtproto.send_speaking_group_call_action()
                await asyncio.sleep(self.SEND_ACTION_UPDATE_EACH)

        asyncio.ensure_future(worker(), loop=self.mtproto.get_event_loop())

    async def start(self, group, join_as=None, invite_hash: Optional[str] = None, enable_action=True):
        """Start voice chat (join and play/record from initial values).

        Note:
            Disconnect from current voice chat and connect to the new one.
            Multiple instances of `GroupCall` must be created for multiple voice chats at the same time.
            Join as by default is personal account.

        Args:
            group (`InputPeerChannel` | `InputPeerChat` | `str` | `int`): Chat ID in any form.
            join_as (`InputPeer` | `str` | `int`, optional): How to present yourself in participants list.
            invite_hash (`str`, optional): Hash from speaker invite link.
            enable_action (`bool`, optional): Is enables sending of speaking action.
        """
        self.__is_stop_requested = False
        self.enable_action = enable_action

        group_call = await self.mtproto.get_and_set_group_call(group)
        if group_call is None:
            raise GroupCallNotFoundError('Chat without a voice chat')

        # mb move in other place. save plain join_as arg and use it in JoinGroupCall
        # but for now it works  as optimization of requests
        # we resolve join_as only when try to connect
        # it doesnt call resolve on reconnect
        await self.mtproto.resolve_and_set_join_as(join_as)

        self.invite_hash = invite_hash

        self.mtproto.re_register_update_handlers()

        # when trying to connect to another chat or with another join_as
        if self.is_group_call_native_created():
            await self.reconnect()
        # the first start
        else:
            self._setup_and_start_group_call()

    async def stop(self):
        """Properly stop tgcalls, remove MTProto handler, leave from server side."""
        if not self.is_group_call_native_created():
            logger.debug('Group call is not started, so there\'s nothing to stop.')
            return

        self.__is_stop_requested = True
        logger.debug('Stop requested.')

        self.mtproto.unregister_update_handlers()
        # to bypass recreating of outgoing audio channel
        self._set_is_mute(True)
        self._set_connection_mode(tgcalls.GroupConnectionMode.GroupConnectionModeNone)

        on_disconnect_event = asyncio.Event()

        async def post_disconnect():
            await self.leave_current_group_call()
            self.mtproto.reset()
            self.__is_stop_requested = False

        async def on_disconnect(obj, is_connected):
            if is_connected:
                return

            obj._stop_audio_device_module()

            # need for normal waiting of stopping audio devices
            # destroying of webrtc client during .stop not needed yet
            # because we a working in the same native instance
            # and can reuse tis client for another connections.
            # In any case now its possible to reset group call ptr
            # self.__native_instance.stopGroupCall()

            await post_disconnect()

            obj.remove_handler(on_disconnect, GroupCallAction.NETWORK_STATUS_CHANGED)
            on_disconnect_event.set()

        if self.is_connected:
            self.add_handler(on_disconnect, GroupCallAction.NETWORK_STATUS_CHANGED)
            await asyncio.wait_for(on_disconnect_event.wait(), timeout=self.__ASYNCIO_TIMEOUT)
        else:
            await post_disconnect()

        logger.debug('GroupCall stopped properly.')

    async def reconnect(self):
        """Connect to voice chat using the same native instance."""
        logger.debug('Reconnecting...')
        if not self.mtproto.group_call:
            raise NotConnectedError("You don't connected to voice chat.")

        self._set_connection_mode(tgcalls.GroupConnectionMode.GroupConnectionModeNone)
        self._emit_join_payload(self.__emit_join_payload_callback)

        # during the .stop we stop audio device module. Need to restart
        self.restart_recording()
        self.restart_playout()

        # cuz native instance doesnt block python
        self.__emit_join_payload_event = asyncio.Event()
        await asyncio.wait_for(self.__emit_join_payload_event.wait(), timeout=self.__ASYNCIO_TIMEOUT)

    async def leave_current_group_call(self):
        """Leave group call from server side (MTProto part)."""
        logger.debug('Try to leave the current group call...')
        try:
            await self.mtproto.leave_current_group_call()
        except Exception as e:
            logger.warning("Couldn't leave the group call. But no worries, you'll get removed from it in seconds.")
            logger.debug(e)
        else:
            logger.debug('Completely left the current group call.')

    async def edit_group_call(self, volume: int = None, muted=False):
        """Edit own settings of group call.

        Note:
            There is bug where you can try to pass `volume=100`.

        Args:
            volume (`int`): Volume.
            muted (`bool`): Is muted.
        """

        await self.edit_group_call_member(self.mtproto.join_as, volume, muted)

    async def edit_group_call_member(self, peer, volume: int = None, muted=False):
        """Edit setting of user in voice chat (required voice chat management permission).

        Note:
            There is bug where you can try to pass `volume=100`.

        Args:
            peer (`InputPeer`): Participant of voice chat.
            volume (`int`): Volume.
            muted (`bool`): Is muted.
        """

        volume = max(1, volume * 100) if volume is not None else None
        await self.mtproto.edit_group_call_member(peer, volume, muted)

    async def set_is_mute(self, is_muted: bool):
        """Set is mute.

        Args:
            is_muted (`bool`): Is muted.
        """

        self.__is_muted = is_muted
        self._set_is_mute(is_muted)

        logger.debug(f'Set is muted on server side. New value: {is_muted}.')
        await self.edit_group_call(muted=is_muted)

    async def set_my_volume(self, volume):
        """Set volume for current client.

        Note:
            Volume value only can be in 1-200 range. There is auto normalization.

        Args:
            volume (`int` | `str` | `float`): Volume.
        """
        # Required "Manage Voice Chats" admin permission

        volume = max(1, min(int(volume), 200))
        logger.debug(f'Set volume to: {volume}.')

        await self.edit_group_call(volume)
        self._set_volume(uint_ssrc(self.mtproto.my_ssrc), volume / 100)

    # shortcuts for easy access in callbacks of events

    @property
    def client(self):
        return self.mtproto.client

    @property
    def full_chat(self):
        return self.mtproto.full_chat

    @property
    def chat_peer(self):
        return self.mtproto.chat_peer

    @property
    def group_call(self):
        return self.mtproto.group_call

    @property
    def my_ssrc(self):
        return self.mtproto.my_ssrc

    @property
    def my_peer(self):
        return self.mtproto.my_peer

    @property
    def join_as(self):
        return self.mtproto.join_as

class GroupCallAction:
    NETWORK_STATUS_CHANGED = Action()
    '''When a status of network will be changed.'''
    PARTICIPANT_LIST_UPDATED = Action()
    '''When a list of participant will be updated.'''

class GroupCallDevice(GroupCall):
    def __init__(
        self,
        mtproto_bridge,
        audio_input_device: Optional[str] = None,
        audio_output_device: Optional[str] = None,
        enable_logs_to_console=False,
        path_to_log_file=None,
        outgoing_audio_bitrate_kbit=128,
    ):
        super().__init__(mtproto_bridge, enable_logs_to_console, path_to_log_file, outgoing_audio_bitrate_kbit)

        self.__is_playout_paused = False
        self.__is_recording_paused = False

        self.__audio_input_device = audio_input_device or ''
        self.__audio_output_device = audio_output_device or ''

    def _setup_and_start_group_call(self):
        self._start_native_group_call(self.__audio_input_device, self.__audio_output_device)

    @property
    def audio_input_device(self):
        """Get audio input device name or GUID

        Note:
            To get system recording device list you can use `get_recording_devices()` method.
        """

        return self.__audio_input_device

    @audio_input_device.setter
    def audio_input_device(self, name=None):
        self.set_audio_input_device(name)

    @property
    def audio_output_device(self):
        """Get audio output device name or GUID

        Note:
            To get system playout device list you can use `get_playout_devices()` method.
        """

        return self.__audio_output_device

    @audio_output_device.setter
    def audio_output_device(self, name=None):
        self.set_audio_output_device(name)

class GroupCallDispatcherMixin(DispatcherMixin):
    def on_network_status_changed(self, func: Callable) -> Callable:
        """When a status of network will be changed.

        Args:
            func (`Callable`): A functions that accept group_call and is_connected args.

        Returns:
            `Callable`: passed to args callback function.
        """

        return self.add_handler(func, GroupCallAction.NETWORK_STATUS_CHANGED)

    def on_participant_list_updated(self, func: Callable) -> Callable:
        """When a list of participant will be updated.

        Args:
            func (`Callable`): A functions that accept group_call and participants args.

        Note:
            The `participants` arg is a `list` of `GroupCallParticipantWrapper`.
            It contains only updated participants! It's not a list of all participants!

        Returns:
            `Callable`: passed to args callback function.
        """

        return self.add_handler(func, GroupCallAction.PARTICIPANT_LIST_UPDATED)

class GroupCallFile(GroupCall, GroupCallFileDispatcherMixin):
    def __init__(
        self,
        mtproto_bridge,
        input_filename: str = None,
        output_filename: str = None,
        play_on_repeat=True,
        enable_logs_to_console=False,
        path_to_log_file=None,
        outgoing_audio_bitrate_kbit=128,
    ):
        super().__init__(mtproto_bridge, enable_logs_to_console, path_to_log_file, outgoing_audio_bitrate_kbit)
        super(GroupCallFileDispatcherMixin, self).__init__(GroupCallFileAction)

        self.play_on_repeat = play_on_repeat
        '''When the file ends, play it again'''
        self.__is_playout_paused = False
        self.__is_recording_paused = False

        self.__input_filename = input_filename or ''
        self.__output_filename = output_filename or ''

        self.__file_audio_device_descriptor = None

    def __create_and_return_file_audio_device_descriptor(self):
        self.__file_audio_device_descriptor = tgcalls.FileAudioDeviceDescriptor()
        self.__file_audio_device_descriptor.getInputFilename = self.__get_input_filename_callback
        self.__file_audio_device_descriptor.getOutputFilename = self.__get_output_filename_callback
        self.__file_audio_device_descriptor.isEndlessPlayout = self.__is_endless_playout_callback
        self.__file_audio_device_descriptor.isPlayoutPaused = self.__is_playout_paused_callback
        self.__file_audio_device_descriptor.isRecordingPaused = self.__is_recording_paused_callback
        self.__file_audio_device_descriptor.playoutEndedCallback = self.__playout_ended_callback

        return self.__file_audio_device_descriptor

    def _setup_and_start_group_call(self):
        self._start_native_group_call(self.__create_and_return_file_audio_device_descriptor())

    def stop_playout(self):
        """Stop playing of file."""

        self.input_filename = ''

    def stop_output(self):
        """Stop recording to file."""

        self.output_filename = ''

    @property
    def input_filename(self):
        """Input filename (or path) to play."""

        return self.__input_filename

    @input_filename.setter
    def input_filename(self, filename):
        self.__input_filename = filename or ''
        if self.is_connected:
            self.restart_playout()

    @property
    def output_filename(self):
        """Output filename (or path) to record."""

        return self.__output_filename

    @output_filename.setter
    def output_filename(self, filename):
        self.__output_filename = filename or ''
        if self.is_connected:
            self.restart_recording()

    def pause_playout(self):
        """Pause playout (playing from file)."""
        self.__is_playout_paused = True

    def resume_playout(self):
        """Resume playout (playing from file)."""
        self.__is_playout_paused = False

    def pause_recording(self):
        """Pause recording (output to file)."""
        self.__is_recording_paused = True

    def resume_recording(self):
        """Resume recording (output to file)."""
        self.__is_recording_paused = False

    def __get_input_filename_callback(self):
        return self.__input_filename

    def __get_output_filename_callback(self):
        return self.__output_filename

    def __is_endless_playout_callback(self):
        return self.play_on_repeat

    def __is_playout_paused_callback(self):
        return self.__is_playout_paused

    def __is_recording_paused_callback(self):
        return self.__is_recording_paused

    def __playout_ended_callback(self, input_filename: str):
        self.trigger_handlers(GroupCallFileAction.PLAYOUT_ENDED, self, input_filename)

class GroupCallNative:
    def __init__(
        self,
        emit_join_payload_callback,
        network_state_updated_callback,
        enable_logs_to_console: bool,
        path_to_log_file: str,
        outgoing_audio_bitrate_kbit: int,
    ):
        """Create NativeInstance of tgcalls C++ part.

        Args:
            enable_logs_to_console (`bool`): Is enable logs to stderr from tgcalls.
            path_to_log_file (`str`, optional): Path to log file for logs of tgcalls.
        """

        # bypass None value
        if not path_to_log_file:
            path_to_log_file = ''

        logger.debug('Create a new native instance...')
        self.__native_instance = tgcalls.NativeInstance(enable_logs_to_console, path_to_log_file)

        self.__native_instance.setupGroupCall(
            emit_join_payload_callback,
            network_state_updated_callback,
            outgoing_audio_bitrate_kbit,
        )

        logger.debug('Native instance created.')

    def is_group_call_native_created(self):
        return self.__native_instance.isGroupCallNativeCreated()

    def _setup_and_start_group_call(self):
        raise NotImplementedError()

    @if_native_instance_created
    def _set_connection_mode(self, mode: tgcalls.GroupConnectionMode, keep_broadcast_if_was_enabled=False):
        logger.debug(f'Set native connection mode {mode}.')
        self.__native_instance.setConnectionMode(mode, keep_broadcast_if_was_enabled)

    @if_native_instance_created
    def _emit_join_payload(self, callback):
        logger.debug(f'Trigger native emit join payload.')
        self.__native_instance.emitJoinPayload(callback)

    def _start_native_group_call(self, *args):
        logger.debug('Start native group call...')
        self.__native_instance.startGroupCall(*args)

    @if_native_instance_created
    def _emit_join_payload(self, callback):
        logger.debug('Emit native join payload.')
        self.__native_instance.emitJoinPayload(callback)

    @if_native_instance_created
    def _set_join_response_payload(self, payload):
        logger.debug('Set native join response payload...')
        self.__native_instance.setJoinResponsePayload(payload)

    @if_native_instance_created
    def _set_is_mute(self, is_muted: bool):
        logger.debug(f'Set is muted on native instance side. New value: {is_muted}.')
        self.__native_instance.setIsMuted(is_muted)

    @if_native_instance_created
    def _set_volume(self, ssrc, volume):
        logger.debug(f'Set native volume for {ssrc} to {volume}.')
        self.__native_instance.setVolume(ssrc, volume)

    @if_native_instance_created
    def _stop_audio_device_module(self):
        logger.debug(f'Stop audio device module.')
        self.__native_instance.stopAudioDeviceModule()

    @if_native_instance_created
    def _start_audio_device_module(self):
        logger.debug(f'Start audio device module.')
        self.__native_instance.startAudioDeviceModule()

    @if_native_instance_created
    def get_playout_devices(self) -> List['tgcalls.AudioDevice']:
        """Get available playout audio devices in the system.

        Note:
            `tgcalls.AudioDevice` have 2 attributes: name, guid.
        """

        logger.debug('Get native playout devices.')
        return self.__native_instance.getPlayoutDevices()

    @if_native_instance_created
    def get_recording_devices(self) -> List['tgcalls.AudioDevice']:
        """Get available recording audio devices in the system.

        Note:
            `tgcalls.AudioDevice` have 2 attributes: name, guid.
        """

        logger.debug('Get native recording devices.')
        return self.__native_instance.getRecordingDevices()

    @if_native_instance_created
    def set_audio_input_device(self, name: Optional[str] = None):
        """Set audio input device.

        Note:
            If `name` is `None`, will use default system device.
            And this is works only at first device initialization time!

        Args:
            name (`str`): Name or GUID of device.
        """

        logger.debug(f'Set native audio input device to {name}.')
        self.__native_instance.setAudioInputDevice(name or '')

    @if_native_instance_created
    def set_audio_output_device(self, name: Optional[str] = None):
        """Set audio output device.

        Note:
            If `name` is `None`, will use default system device.
            And this is works only at first device initialization time!

        Args:
            name (`str`): Name or GUID of device.
        """

        logger.debug(f'Set native audio output device to {name}.')
        self.__native_instance.setAudioOutputDevice(name or '')

    @if_native_instance_created
    def restart_playout(self):
        """Start play current input file from start or just reload file audio device.

        Note:
            Device restart needed to apply new filename in tgcalls.
        """

        logger.debug(f'Restart native audio input device.')
        self.__native_instance.restartAudioInputDevice()

    @if_native_instance_created
    def restart_recording(self):
        """Start recording to output file from begin or just restart recording device.

        Note:
            Device restart needed to apply new filename in tgcalls.
        """

        logger.debug(f'Restart native audio output device.')
        self.__native_instance.restartAudioOutputDevice()

    # legacy below

    def print_available_playout_devices(self):
        """Print name and guid of available playout audio devices in system. Just helper method

        Note:
            You should use this method after calling .start()!
        """

        warnings.warn("It's a deprecated method. Use .get_recording_devices() instead", DeprecationWarning, 2)

        for device in self.get_playout_devices():
            print(f'Playout device \n name: {device.name} \n guid: {device.guid}')

    def print_available_recording_devices(self):
        """Print name and guid of available recording audio devices in system. Just helper method

        Note:
            You should use this method after calling .start()!
        """

        warnings.warn("It's a deprecated method. Use .get_playout_devices() instead", DeprecationWarning, 2)

        for device in self.get_recording_devices():
            print(f'Recording device \n name: {device.name} \n guid: {device.guid}')

class GroupCallRaw(GroupCall):
    def __init__(
        self,
        mtproto_bridge,
        on_played_data: Callable[['GroupCallRaw', int], bytes] = None,
        on_recorded_data: Callable[['GroupCallRaw', bytes, int], None] = None,
        enable_logs_to_console=False,
        path_to_log_file=None,
        outgoing_audio_bitrate_kbit=128,
    ):
        super().__init__(mtproto_bridge, enable_logs_to_console, path_to_log_file, outgoing_audio_bitrate_kbit)

        self.__is_playout_paused = False
        self.__is_recording_paused = False

        self.__raw_audio_device_descriptor = None

        self.on_played_data = on_played_data
        self.on_recorded_data = on_recorded_data

    def __create_and_return_raw_audio_device_descriptor(self):
        self.__raw_audio_device_descriptor = tgcalls.RawAudioDeviceDescriptor()
        self.__raw_audio_device_descriptor.getPlayedBufferCallback = self.__get_played_buffer_callback
        self.__raw_audio_device_descriptor.setRecordedBufferCallback = self.__set_recorded_buffer_callback
        self.__raw_audio_device_descriptor.isPlayoutPaused = self.__is_playout_paused_callback
        self.__raw_audio_device_descriptor.isRecordingPaused = self.__is_recording_paused_callback

        return self.__raw_audio_device_descriptor

    def _setup_and_start_group_call(self):
        self._start_native_group_call(self.__create_and_return_raw_audio_device_descriptor())

    def __get_played_buffer_callback(self, length: int):
        frame = b''
        if self.on_played_data:
            data = self.on_played_data(self, length)
            if data:
                frame = data

        return frame.ljust(length, b'\0')

    def __set_recorded_buffer_callback(self, frame: bytes, length: int):
        if self.on_recorded_data:
            self.on_recorded_data(self, frame, length)

    def pause_playout(self):
        """Pause playout (playing from callback)."""
        self.__is_playout_paused = True

    def resume_playout(self):
        """Resume playout (playing from callback)."""
        self.__is_playout_paused = False

    def pause_recording(self):
        """Pause recording (output to callback)."""
        self.__is_recording_paused = True

    def resume_recording(self):
        """Resume recording (output to callback)."""
        self.__is_recording_paused = False

    def __is_playout_paused_callback(self):
        return self.__is_playout_paused

    def __is_recording_paused_callback(self):
        return self.__is_recording_paused

    """Stop requesting new data to send."""
    stop_playout = pause_playout
    """Stop getting raw data."""
    stop_output = pause_recording