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
- GroupCall
- abc.ABC
- GroupCallFileDispatcherMixin
- GroupCallDispatcherMixin
- pytgcalls.dispatcher.dispatcher_mixin.DispatcherMixin
- GroupCallNative
Class variables
var SEND_ACTION_UPDATE_EACH-
Inherited from:
GroupCall.SEND_ACTION_UPDATE_EACHHow often to send speaking action to chat
Instance variables
var enable_action-
Inherited from:
GroupCall.enable_actionIs 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_hashHash from invite link to join as speaker
var is_connected-
Inherited from:
GroupCall.is_connectedIs 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_callEdit own settings of group call …
async def edit_group_call_member(self, peer, volume: int = None, muted=False)-
Inherited from:
GroupCall.edit_group_call_memberEdit setting of user in voice chat (required voice chat management permission) …
async def leave_current_group_call(self)-
Inherited from:
GroupCall.leave_current_group_callLeave group call from server side (MTProto part).
def on_network_status_changed(self, func: Callable) ‑> Callable-
Inherited from:
GroupCall.on_network_status_changedWhen a status of network will be changed …
def on_participant_list_updated(self, func: Callable) ‑> Callable-
Inherited from:
GroupCall.on_participant_list_updatedWhen a list of participant will be updated …
def on_playout_ended(self, func: Callable) ‑> Callable-
Inherited from:
GroupCallFileDispatcherMixin.on_playout_endedWhen 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_devicesPrint 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_devicesPrint name and guid of available recording audio devices in system. Just helper method …
async def reconnect(self)-
Inherited from:
GroupCall.reconnectConnect 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_muteSet is mute …
async def set_my_volume(self, volume)-
Inherited from:
GroupCall.set_my_volumeSet volume for current client …
async def start(self, group, join_as=None, invite_hash: Optional[str] = None, enable_action=True)-
Inherited from:
GroupCall.startStart voice chat (join and play/record from initial values) …
async def stop(self)-
Inherited from:
GroupCall.stopProperly 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_CHANGEDWhen a status of network will be changed.
var PARTICIPANT_LIST_UPDATED-
Inherited from:
GroupCallAction.PARTICIPANT_LIST_UPDATEDWhen 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
- GroupCallDispatcherMixin
- pytgcalls.dispatcher.dispatcher_mixin.DispatcherMixin
Subclasses
Methods
def on_network_status_changed(self, func: Callable) ‑> Callable-
Inherited from:
GroupCallDispatcherMixin.on_network_status_changedWhen a status of network will be changed …
def on_participant_list_updated(self, func: Callable) ‑> Callable-
Inherited from:
GroupCallDispatcherMixin.on_participant_list_updatedWhen 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)