Module pytgcalls.implementation.group_call_file

Expand source code
#  tgcalls - a Python binding for C++ library by Telegram
#  pytgcalls - a library connecting the Python binding with MTProto
#  Copyright (C) 2020-2021 Il`ya (Marshal) <https://github.com/MarshalX>
#
#  This file is part of tgcalls and pytgcalls.
#
#  tgcalls and pytgcalls is free software: you can redistribute it and/or modify
#  it under the terms of the GNU Lesser General Public License as published
#  by the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  tgcalls and pytgcalls is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU Lesser General Public License for more details.
#
#  You should have received a copy of the GNU Lesser General Public License v3
#  along with tgcalls. If not, see <http://www.gnu.org/licenses/>.

from typing import Callable

import tgcalls
from pytgcalls.implementation import GroupCall, GroupCallAction, GroupCallDispatcherMixin
from pytgcalls.dispatcher import Action


class GroupCallFileAction(GroupCallAction):
    PLAYOUT_ENDED = Action()
    '''When a input file is ended.'''


class GroupCallFileDispatcherMixin(GroupCallDispatcherMixin):
    def on_playout_ended(self, func: Callable) -> Callable:
        """When a input file is ended.

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

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

        return self.add_handler(func, GroupCallFileAction.PLAYOUT_ENDED)


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)

Classes

class GroupCallFile (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)

Helper class that provides a standard way to create an ABC using inheritance.

Expand source code
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)

Ancestors

Class variables

var SEND_ACTION_UPDATE_EACH

Inherited from: GroupCall.SEND_ACTION_UPDATE_EACH

How often to send speaking action to chat

Instance variables

var enable_action

Inherited from: GroupCall.enable_action

Is enable sending of speaking action

var input_filename

Input filename (or path) to play.

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

    return self.__input_filename
var invite_hash

Inherited from: GroupCall.invite_hash

Hash from invite link to join as speaker

var is_connected

Inherited from: GroupCall.is_connected

Is connected to voice chat via tgcalls

var output_filename

Output filename (or path) to record.

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

    return self.__output_filename
var play_on_repeat

When the file ends, play it again

Methods

async def edit_group_call(self, volume: int = None, muted=False)

Inherited from: GroupCall.edit_group_call

Edit own settings of group call …

async def edit_group_call_member(self, peer, volume: int = None, muted=False)

Inherited from: GroupCall.edit_group_call_member

Edit setting of user in voice chat (required voice chat management permission) …

async def leave_current_group_call(self)

Inherited from: GroupCall.leave_current_group_call

Leave group call from server side (MTProto part).

def on_network_status_changed(self, func: Callable) ‑> Callable

Inherited from: GroupCall.on_network_status_changed

When a status of network will be changed …

def on_participant_list_updated(self, func: Callable) ‑> Callable

Inherited from: GroupCall.on_participant_list_updated

When a list of participant will be updated …

def on_playout_ended(self, func: Callable) ‑> Callable

Inherited from: GroupCallFileDispatcherMixin.on_playout_ended

When a input file is ended …

def pause_playout(self)

Pause playout (playing from file).

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

Pause recording (output to file).

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

Inherited from: GroupCall.print_available_playout_devices

Print name and guid of available playout audio devices in system. Just helper method …

def print_available_recording_devices(self)

Inherited from: GroupCall.print_available_recording_devices

Print name and guid of available recording audio devices in system. Just helper method …

async def reconnect(self)

Inherited from: GroupCall.reconnect

Connect to voice chat using the same native instance.

def resume_playout(self)

Resume playout (playing from file).

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

Resume recording (output to file).

Expand source code
def resume_recording(self):
    """Resume recording (output to file)."""
    self.__is_recording_paused = False
async def set_is_mute(self, is_muted: bool)

Inherited from: GroupCall.set_is_mute

Set is mute …

async def set_my_volume(self, volume)

Inherited from: GroupCall.set_my_volume

Set volume for current client …

async def start(self, group, join_as=None, invite_hash: Optional[str] = None, enable_action=True)

Inherited from: GroupCall.start

Start voice chat (join and play/record from initial values) …

async def stop(self)

Inherited from: GroupCall.stop

Properly stop tgcalls, remove MTProto handler, leave from server side.

def stop_output(self)

Stop recording to file.

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

    self.output_filename = ''
def stop_playout(self)

Stop playing of file.

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

    self.input_filename = ''
class GroupCallFileAction
Expand source code
class GroupCallFileAction(GroupCallAction):
    PLAYOUT_ENDED = Action()
    '''When a input file is ended.'''

Ancestors

Class variables

var NETWORK_STATUS_CHANGED

Inherited from: GroupCallAction.NETWORK_STATUS_CHANGED

When a status of network will be changed.

var PARTICIPANT_LIST_UPDATED

Inherited from: GroupCallAction.PARTICIPANT_LIST_UPDATED

When a list of participant will be updated.

var PLAYOUT_ENDED

When a input file is ended.

class GroupCallFileDispatcherMixin (actions)
Expand source code
class GroupCallFileDispatcherMixin(GroupCallDispatcherMixin):
    def on_playout_ended(self, func: Callable) -> Callable:
        """When a input file is ended.

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

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

        return self.add_handler(func, GroupCallFileAction.PLAYOUT_ENDED)

Ancestors

Subclasses

Methods

def on_network_status_changed(self, func: Callable) ‑> Callable

Inherited from: GroupCallDispatcherMixin.on_network_status_changed

When a status of network will be changed …

def on_participant_list_updated(self, func: Callable) ‑> Callable

Inherited from: GroupCallDispatcherMixin.on_participant_list_updated

When a list of participant will be updated …

def on_playout_ended(self, func: Callable) ‑> Callable

When a input file is ended.

Args

func (Callable): A functions that accept group_call and filename args.

Returns

Callable: passed to args callback function.

Expand source code
def on_playout_ended(self, func: Callable) -> Callable:
    """When a input file is ended.

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

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

    return self.add_handler(func, GroupCallFileAction.PLAYOUT_ENDED)