Package `pytgcalls`

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/>.

import logging
import warnings

from pytgcalls.exceptions import PytgcallsError
from pytgcalls.group_call_factory import GroupCallFactory
from pytgcalls.implementation.group_call_file import GroupCallFileAction
from pytgcalls.implementation.group_call import GroupCallAction

# backward compatibility below. Dont use it in new projects

logger = logging.getLogger(__name__)


_deprecation_warning_text = (
    'You use deprecated import from backward compatibility suite. '
    'Please update you code. Backward compatibility will be deleted at any time! '
    'For more info visit https://github.com/MarshalX/tgcalls/discussions/101'
)


def backward_compatibility_helper(group_call_type, client, *args, **kwargs):
    clazz = GroupCallFactory.GROUP_CALL_CLASS_TO_TYPE.get(group_call_type)

    try:
        import pyrogram
    except ImportError:
        raise PytgcallsError(
            'To use this backward compatibility you need to install Pyrogram. '
            'Run this command: pip3 install -U pytgcalls[pyrogram]'
        )

    from pytgcalls.mtproto.pyrogram_bridge import PyrogramBridge

    wrapped_client = PyrogramBridge(client)

    return clazz(wrapped_client, *args, **kwargs)


def GroupCall(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.FILE, client, *args, **kwargs)


def GroupCallDevice(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.DEVICE, client, *args, **kwargs)


def GroupCallRaw(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.RAW, client, *args, **kwargs)


__all__ = [
    'GroupCallFactory',
    'GroupCallFileAction',
    'GroupCallAction',
    # below backward compatibility
    'GroupCall',
    'GroupCallDevice',
    'GroupCallRaw',
]
__version__ = '2.1.0'
__pdoc__ = {
    # files
    'utils': False,
    'dispatcher': False,
}

Sub-modules

pytgcalls.exceptions
pytgcalls.group_call_factory
pytgcalls.group_call_type
pytgcalls.implementation
pytgcalls.mtproto
pytgcalls.mtproto_client_type

Functions

def GroupCall(client, *args, **kwargs)

Deprecated method

Expand source code

def GroupCall(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.FILE, client, *args, **kwargs)

def GroupCallDevice(client, *args, **kwargs)

Deprecated method

Expand source code

def GroupCallDevice(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.DEVICE, client, *args, **kwargs)

def GroupCallRaw(client, *args, **kwargs)

Deprecated method

Expand source code

def GroupCallRaw(client, *args, **kwargs):
    """Deprecated method"""
    warnings.warn(_deprecation_warning_text, DeprecationWarning, 2)
    return backward_compatibility_helper(GroupCallFactory.GROUP_CALL_TYPE.RAW, client, *args, **kwargs)

Classes

class GroupCallAction

Expand source code

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.'''

Subclasses

GroupCallFileAction

Class variables

var NETWORK_STATUS_CHANGED: When a status of network will be changed.
var PARTICIPANT_LIST_UPDATED: When a list of participant will be updated.

class GroupCallFactory (client, mtproto_backend=MTProtoClientType.PYROGRAM, enable_logs_to_console=False, path_to_log_file=None, outgoing_audio_bitrate_kbit=128)

Expand source code

class GroupCallFactory:
    MTPROTO_CLIENT_TYPE = MTProtoClientType
    GROUP_CALL_TYPE = GroupCallType

    GROUP_CALL_CLASS_TO_TYPE = {
        GROUP_CALL_TYPE.FILE: GroupCallFile,
        GROUP_CALL_TYPE.DEVICE: GroupCallDevice,
        GROUP_CALL_TYPE.RAW: GroupCallRaw,
    }

    def __init__(
        self,
        client,
        mtproto_backend=MTProtoClientType.PYROGRAM,
        enable_logs_to_console=False,
        path_to_log_file=None,
        outgoing_audio_bitrate_kbit=128,
    ):
        self.client = client

        if mtproto_backend is MTProtoClientType.PYROGRAM:
            hot_load_mtproto_lib_or_exception(MTProtoClientType.PYROGRAM.value)
            from pytgcalls.mtproto.pyrogram_bridge import PyrogramBridge

            self.__mtproto_bride_class = PyrogramBridge
        elif mtproto_backend is MTProtoClientType.TELETHON:
            hot_load_mtproto_lib_or_exception(MTProtoClientType.TELETHON.value)
            from pytgcalls.mtproto.telethon_bridge import TelethonBridge

            self.__mtproto_bride_class = TelethonBridge
        else:
            raise PytgcallsError('Unknown MTProto client type')

        self.enable_logs_to_console = enable_logs_to_console
        self.path_to_log_file = path_to_log_file
        self.outgoing_audio_bitrate_kbit = outgoing_audio_bitrate_kbit

    def get_mtproto_bridge(self):
        return self.__mtproto_bride_class(self.client)

    def get(self, group_call_type: GroupCallType, **kwargs) -> Union[GroupCallFile, GroupCallDevice, GroupCallRaw]:
        return GroupCallFactory.GROUP_CALL_CLASS_TO_TYPE[group_call_type](
            mtproto_bridge=self.get_mtproto_bridge(),
            enable_logs_to_console=self.enable_logs_to_console,
            path_to_log_file=self.path_to_log_file,
            outgoing_audio_bitrate_kbit=self.outgoing_audio_bitrate_kbit,
            **kwargs,
        )

    def get_file_group_call(
        self, input_filename: Optional[str] = None, output_filename: Optional[str] = None, play_on_repeat=True
    ) -> GroupCallFile:
        return GroupCallFile(
            self.get_mtproto_bridge(),
            input_filename,
            output_filename,
            play_on_repeat,
            self.enable_logs_to_console,
            self.path_to_log_file,
            self.outgoing_audio_bitrate_kbit,
        )

    def get_device_group_call(
        self, audio_input_device: Optional[str] = None, audio_output_device: Optional[str] = None
    ) -> GroupCallDevice:
        return GroupCallDevice(
            self.get_mtproto_bridge(),
            audio_input_device,
            audio_output_device,
            self.enable_logs_to_console,
            self.path_to_log_file,
            self.outgoing_audio_bitrate_kbit,
        )

    def get_raw_group_call(
        self,
        on_played_data: Callable[['GroupCallRaw', int], bytes] = None,
        on_recorded_data: Callable[['GroupCallRaw', bytes, int], None] = None,
    ) -> GroupCallRaw:
        return GroupCallRaw(
            self.get_mtproto_bridge(),
            on_played_data,
            on_recorded_data,
            self.enable_logs_to_console,
            self.path_to_log_file,
            self.outgoing_audio_bitrate_kbit,
        )

Class variables

var GROUP_CALL_CLASS_TO_TYPE
var GROUP_CALL_TYPE: An enumeration.
var MTPROTO_CLIENT_TYPE: An enumeration.

Methods

def get(self, group_call_type: GroupCallType, **kwargs) ‑> Union[GroupCallFile, GroupCallDevice, GroupCallRaw]

Expand source code

def get(self, group_call_type: GroupCallType, **kwargs) -> Union[GroupCallFile, GroupCallDevice, GroupCallRaw]:
    return GroupCallFactory.GROUP_CALL_CLASS_TO_TYPE[group_call_type](
        mtproto_bridge=self.get_mtproto_bridge(),
        enable_logs_to_console=self.enable_logs_to_console,
        path_to_log_file=self.path_to_log_file,
        outgoing_audio_bitrate_kbit=self.outgoing_audio_bitrate_kbit,
        **kwargs,
    )

def get_device_group_call(self, audio_input_device: Optional[str] = None, audio_output_device: Optional[str] = None) ‑> GroupCallDevice

Expand source code

def get_device_group_call(
    self, audio_input_device: Optional[str] = None, audio_output_device: Optional[str] = None
) -> GroupCallDevice:
    return GroupCallDevice(
        self.get_mtproto_bridge(),
        audio_input_device,
        audio_output_device,
        self.enable_logs_to_console,
        self.path_to_log_file,
        self.outgoing_audio_bitrate_kbit,
    )

def get_file_group_call(self, input_filename: Optional[str] = None, output_filename: Optional[str] = None, play_on_repeat=True) ‑> GroupCallFile

Expand source code

def get_file_group_call(
    self, input_filename: Optional[str] = None, output_filename: Optional[str] = None, play_on_repeat=True
) -> GroupCallFile:
    return GroupCallFile(
        self.get_mtproto_bridge(),
        input_filename,
        output_filename,
        play_on_repeat,
        self.enable_logs_to_console,
        self.path_to_log_file,
        self.outgoing_audio_bitrate_kbit,
    )

def get_mtproto_bridge(self)

Expand source code

def get_mtproto_bridge(self):
    return self.__mtproto_bride_class(self.client)

def get_raw_group_call(self, on_played_data: Callable[[_ForwardRef('GroupCallRaw()'), int], bytes] = None, on_recorded_data: Callable[[_ForwardRef('GroupCallRaw()'), bytes, int], NoneType] = None) ‑> GroupCallRaw

Expand source code

def get_raw_group_call(
    self,
    on_played_data: Callable[['GroupCallRaw', int], bytes] = None,
    on_recorded_data: Callable[['GroupCallRaw', bytes, int], None] = None,
) -> GroupCallRaw:
    return GroupCallRaw(
        self.get_mtproto_bridge(),
        on_played_data,
        on_recorded_data,
        self.enable_logs_to_console,
        self.path_to_log_file,
        self.outgoing_audio_bitrate_kbit,
    )

class GroupCallFileAction

Expand source code

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

Ancestors

GroupCallAction

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.