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