Module pytgcalls.mtproto

class MTProtoBridgeBase(ABC):
    def __init__(self, client):
        self.client = client
        '''Any MTProto client. Pyrogram/Telethon and so on'''

        self.group_call_participants_update_callback = None
        '''Native handler of wrapped group call participants update'''
        self.group_call_update_callback = None
        '''Native handler of wrapped group call update'''

        self.full_chat = None
        '''Full chat information'''
        self.chat_peer = None
        '''Chat peer where bot is now'''

        self.group_call = None
        '''Instance of MTProto's group call'''

        self.my_ssrc = None
        '''Client SSRC (Synchronization Source)'''
        self.my_peer = None
        '''Client user peer'''

        self.join_as = None
        '''How to present yourself in participants list'''

    def reset(self):
        self.group_call = self.full_chat = self.chat_peer = self.join_as = self.my_peer = self.my_ssrc = None

    def register_group_call_native_callback(
        self, group_call_participants_update_callback: Callable, group_call_update_callback: Callable
    ):
        self.group_call_participants_update_callback = group_call_participants_update_callback
        self.group_call_update_callback = group_call_update_callback

    def re_register_update_handlers(self):
        """Delete and add pytgcalls handler in MTProto client."""
        self.unregister_update_handlers()
        self.register_update_handlers()

    async def check_group_call(self) -> bool:
        """Check if client is in a voice chat.

        Returns:
            `bool`: Is in voice chat by opinion of Telegram server.

        raise wrapped BadRequest if you got [400 GROUPCALL_JOIN_MISSING] response!
        """

        raise NotImplementedError

    async def leave_current_group_call(self):
        """
        call phone.LeaveGroupCall and handle returned updates
        """
        raise NotImplementedError

    async def edit_group_call_member(self, peer, volume: int = None, muted=False):
        """
        call phone.EditGroupCallParticipant
        """
        raise NotImplementedError

    async def get_and_set_self_peer(self):
        """
        resolve self peer and set to obj field
        """
        raise NotImplementedError

    async def get_and_set_group_call(self, group):
        """
        there is group arg can be peer, int, string with username and so on
        need to support all of them

        in this method be set chat_peer, full_chat and group_call class fields

        i think group_call dont need to wrap cuz it will used in phone.JoinGroupCall
        """
        raise NotImplementedError

    def unregister_update_handlers(self):
        """
        delete all registered handlers from MTProto client
        """
        raise NotImplementedError

    def register_update_handlers(self):
        """
        register handlers
        """
        raise NotImplementedError

    async def resolve_and_set_join_as(self, join_as):
        """
        join_as arg can be str on peer. if it str we need to resolve peer
        save join_as to class field
        """
        raise NotImplementedError

    async def send_speaking_group_call_action(self):
        """
        call messages.SetTyping with SpeakingInGroupCallAction by chat_peer
        """
        raise NotImplementedError

    async def join_group_call(self, invite_hash: str, params: dict, muted: bool, pre_update_processing: Callable):
        """
        call phone.JoinGroupCall with group_call, join_as, invite hash, muted and params

        handle updates from response!

        reraise wrapped GroupcallSsrcDuplicateMuch!
        """
        raise NotImplementedError

    def get_event_loop(self) -> AbstractEventLoop:
        """
        return MTProto client loop
        """
        raise NotImplementedError

    def set_my_ssrc(self, ssrc):
        self.my_ssrc = ssrc

Ancestors

• abc.ABC

Subclasses

• PyrogramBridge
• TelethonBridge

Instance variables

var chat_peer

Chat peer where bot is now

var client

Any MTProto client. Pyrogram/Telethon and so on

var full_chat

Full chat information

var group_call

Instance of MTProto's group call

var group_call_participants_update_callback

Native handler of wrapped group call participants update

var group_call_update_callback

Native handler of wrapped group call update

var join_as

How to present yourself in participants list

var my_peer

Client user peer

var my_ssrc

Client SSRC (Synchronization Source)

Methods

async def check_group_call(self) ‑> bool

Check if client is in a voice chat.

Returns

bool: Is in voice chat by opinion of Telegram server. raise wrapped BadRequest if you got [400 GROUPCALL_JOIN_MISSING] response!

Expand source code

async def check_group_call(self) -> bool:
    """Check if client is in a voice chat.

    Returns:
        `bool`: Is in voice chat by opinion of Telegram server.

    raise wrapped BadRequest if you got [400 GROUPCALL_JOIN_MISSING] response!
    """

    raise NotImplementedError

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

call phone.EditGroupCallParticipant

Expand source code

async def edit_group_call_member(self, peer, volume: int = None, muted=False):
    """
    call phone.EditGroupCallParticipant
    """
    raise NotImplementedError

async def get_and_set_group_call(self, group)

there is group arg can be peer, int, string with username and so on need to support all of them

in this method be set chat_peer, full_chat and group_call class fields

i think group_call dont need to wrap cuz it will used in phone.JoinGroupCall

Expand source code

async def get_and_set_group_call(self, group):
    """
    there is group arg can be peer, int, string with username and so on
    need to support all of them

    in this method be set chat_peer, full_chat and group_call class fields

    i think group_call dont need to wrap cuz it will used in phone.JoinGroupCall
    """
    raise NotImplementedError

async def get_and_set_self_peer(self)

resolve self peer and set to obj field

Expand source code

async def get_and_set_self_peer(self):
    """
    resolve self peer and set to obj field
    """
    raise NotImplementedError

def get_event_loop(self) ‑> asyncio.events.AbstractEventLoop

return MTProto client loop

Expand source code

def get_event_loop(self) -> AbstractEventLoop:
    """
    return MTProto client loop
    """
    raise NotImplementedError

async def join_group_call(self, invite_hash: str, params: dict, muted: bool, pre_update_processing: Callable)

call phone.JoinGroupCall with group_call, join_as, invite hash, muted and params

handle updates from response!

reraise wrapped GroupcallSsrcDuplicateMuch!

Expand source code

async def join_group_call(self, invite_hash: str, params: dict, muted: bool, pre_update_processing: Callable):
    """
    call phone.JoinGroupCall with group_call, join_as, invite hash, muted and params

    handle updates from response!

    reraise wrapped GroupcallSsrcDuplicateMuch!
    """
    raise NotImplementedError

async def leave_current_group_call(self)

call phone.LeaveGroupCall and handle returned updates

Expand source code

async def leave_current_group_call(self):
    """
    call phone.LeaveGroupCall and handle returned updates
    """
    raise NotImplementedError

def re_register_update_handlers(self)

Delete and add pytgcalls handler in MTProto client.

Expand source code

def re_register_update_handlers(self):
    """Delete and add pytgcalls handler in MTProto client."""
    self.unregister_update_handlers()
    self.register_update_handlers()

def register_group_call_native_callback(self, group_call_participants_update_callback: Callable, group_call_update_callback: Callable)

Expand source code

def register_group_call_native_callback(
    self, group_call_participants_update_callback: Callable, group_call_update_callback: Callable
):
    self.group_call_participants_update_callback = group_call_participants_update_callback
    self.group_call_update_callback = group_call_update_callback

def register_update_handlers(self)

register handlers

Expand source code

def register_update_handlers(self):
    """
    register handlers
    """
    raise NotImplementedError

def reset(self)

Expand source code

def reset(self):
    self.group_call = self.full_chat = self.chat_peer = self.join_as = self.my_peer = self.my_ssrc = None

async def resolve_and_set_join_as(self, join_as)

join_as arg can be str on peer. if it str we need to resolve peer save join_as to class field

Expand source code

async def resolve_and_set_join_as(self, join_as):
    """
    join_as arg can be str on peer. if it str we need to resolve peer
    save join_as to class field
    """
    raise NotImplementedError

async def send_speaking_group_call_action(self)

call messages.SetTyping with SpeakingInGroupCallAction by chat_peer

Expand source code

async def send_speaking_group_call_action(self):
    """
    call messages.SetTyping with SpeakingInGroupCallAction by chat_peer
    """
    raise NotImplementedError

def set_my_ssrc(self, ssrc)

Expand source code

def set_my_ssrc(self, ssrc):
    self.my_ssrc = ssrc

def unregister_update_handlers(self)

delete all registered handlers from MTProto client

Expand source code

def unregister_update_handlers(self):
    """
    delete all registered handlers from MTProto client
    """
    raise NotImplementedError