From fe0087291391be2dc8815f09ff22388d18993794 Mon Sep 17 00:00:00 2001 From: dolfies Date: Tue, 9 Nov 2021 19:16:08 -0500 Subject: [PATCH] Migrate calls.py --- discord/calls.py | 315 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 315 insertions(+) create mode 100644 discord/calls.py diff --git a/discord/calls.py b/discord/calls.py new file mode 100644 index 000000000..4ff671601 --- /dev/null +++ b/discord/calls.py @@ -0,0 +1,315 @@ +""" +The MIT License (MIT) + +Copyright (c) 2015-present Rapptz + +Permission is hereby granted, free of charge, to any person obtaining a +copy of this software and associated documentation files (the "Software"), +to deal in the Software without restriction, including without limitation +the rights to use, copy, modify, merge, publish, distribute, sublicense, +and/or sell copies of the Software, and to permit persons to whom the +Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. +""" + +import datetime +from typing import List, Optional, TYPE_CHECKING + +from . import utils +from .enums import VoiceRegion, try_enum +from .errors import ClientException +from .utils import MISSING + +if TYPE_CHECKING: + from .abc import PrivateChannel + from .channel import DMChannel, GroupChannel + from .member import VoiceState + from .message import Message + from .state import ConnectionState + from .types.snowflake import Snowflake, SnowflakeList + from .types.voice import GuildVoiceState + from .user import User + from .voice_client import VoiceProtocol + + +def _running_only(func): + def decorator(self, *args, **kwargs): + if self._ended: + raise ClientException('Call is over') + else: + return func(self, *args, **kwargs) + return decorator + + +class CallMessage: + """Represents a group call message from Discord. + + This is only received in cases where the message type is equivalent to + :attr:`MessageType.call`. + + Attributes + ----------- + ended_timestamp: Optional[:class:`datetime.datetime`] + A naive UTC datetime object that represents the time that the call has ended. + participants: List[:class:`User`] + A list of users that participated in the call. + message: :class:`Message` + The message associated with this call message. + """ + + def __init__( + self, message: Message, *, participants: List[User], ended_timestamp: str + ) -> None: + self.message = message + self.ended_timestamp = utils.parse_time(ended_timestamp) + self.participants = participants + + @property + def call_ended(self) -> bool: + """:class:`bool`: Indicates if the call has ended.""" + return self.ended_timestamp is not None + + @property + def initiator(self) -> User: + """:class:`User`: Returns the user that started the call.""" + return self.message.author + + @property + def channel(self) -> PrivateChannel: + r""":class:`PrivateChannel`\: The private channel associated with this message.""" + return self.message.channel + + @property + def duration(self) -> datetime.timedelta: + """Queries the duration of the call. + + If the call has not ended then the current duration will + be returned. + + Returns + --------- + :class:`datetime.timedelta` + The timedelta object representing the duration. + """ + if self.ended_timestamp is None: + return datetime.datetime.utcnow() - self.message.created_at + else: + return self.ended_timestamp - self.message.created_at + + +class PrivateCall: + """Represents the actual group call from Discord. + + This is accompanied with a :class:`CallMessage` denoting the information. + + Attributes + ----------- + channel: :class:`DMChannel` + The channel the call is in. + message: Optional[:class:`Message`] + The message associated with this call (if available). + unavailable: :class:`bool` + Denotes if this call is unavailable. + ringing: List[:class:`User`] + A list of users that are currently being rung to join the call. + region: :class:`VoiceRegion` + The region the call is being hosted at. + """ + + if TYPE_CHECKING: + channel: DMChannel + ringing: List[User] + region: VoiceRegion + + def __init__( + self, + state: ConnectionState, + *, + message_id: Snowflake, + channel_id: Snowflake, + message: Message = None, + channel: PrivateChannel, + unavailable: bool, + voice_states: List[GuildVoiceState] = [], + **kwargs, + ) -> None: + self._state = state + self._message_id: int = int(message_id) + self._channel_id: int = int(channel_id) + self.message: Optional[Message] = message + self.channel = channel # type: ignore + self.unavailable: bool = unavailable + self._ended: bool = False + + for vs in voice_states: + state._update_voice_state(vs) + + self._update(**kwargs) + + def _deleteup(self) -> None: + self.ringing = [] + self._ended = True + + def _update( + self, *, ringing: SnowflakeList = {}, region: VoiceRegion = MISSING + ) -> None: + if region is not MISSING: + self.region = try_enum(VoiceRegion, region) + channel = self.channel + recipients = {channel.me, channel.recipient} + lookup = {u.id: u for u in recipients} + self.ringing = list(filter(None, map(lookup.get, ringing))) + + @property + def initiator(self) -> Optional[User]: + """Optional[:class:`User`]: Returns the user that started the call. The call message must be available to obtain this information.""" + if self.message: + return self.message.author + + @property + def connected(self) -> bool: + """:class:`bool`: Returns whether you're in the call (this does not mean you're in the call through the lib).""" + return self.voice_state_for(self.channel.me).channel.id == self._channel_id + + @property + def members(self) -> List[User]: + """List[:class:`User`]: Returns all users that are currently in this call.""" + channel = self.channel + recipients = {channel.me, channel.recipient} + ret = [u for u in recipients if self.voice_state_for(u).channel.id == self._channel_id] + + return ret + + @property + def voice_states(self) -> List[VoiceState]: + """Mapping[:class:`int`, :class:`VoiceState`]: Returns a mapping of user IDs who have voice states in this call.""" + return set(self._voice_states) + + async def fetch_message(self) -> Optional[Message]: + message = await self.channel.fetch_message(self._message_id) + if message is not None and self.message is None: + self.message = message + return message + + @_running_only + async def change_region(self, region) -> None: + """|coro| + + Changes the channel's voice region. + + Parameters + ----------- + region: :class:`VoiceRegion` + A :class:`VoiceRegion` to change the voice region to. + + Raises + ------- + HTTPException + Failed to change the channel's voice region. + """ + await self._state.http.change_call_voice_region(self.channel.id, str(region)) + + @_running_only + async def ring(self) -> None: + channel = self.channel + await self._state.http.ring(channel.id, channel.recipient.id) + + @_running_only + async def stop_ringing(self) -> None: + channel = self.channel + await self._state.http.stop_ringing(channel.id, channel.recipient.id) + + @_running_only + async def join(self, **kwargs) -> VoiceProtocol: + return await self.channel._connect(**kwargs) + + connect = join + + @_running_only + async def leave(self, **kwargs) -> None: + state = self._state + if not (client := state._get_voice_client(self.channel.me.id)): + return + + return await client.disconnect(**kwargs) + + disconnect = leave + + def voice_state_for(self, user) -> Optional[VoiceState]: + """Retrieves the :class:`VoiceState` for a specified :class:`User`. + + If the :class:`User` has no voice state then this function returns + ``None``. + + Parameters + ------------ + user: :class:`User` + The user to retrieve the voice state for. + + Returns + -------- + Optional[:class:`VoiceState`] + The voice state associated with this user. + """ + return self._state._voice_state_for(user.id) + + +class GroupCall(PrivateCall): + """Represents a Discord group call. + + This is accompanied with a :class:`CallMessage` denoting the information. + + Attributes + ----------- + channel: :class:`GroupChannel` + The channel the group call is in. + message: Optional[:class:`Message`] + The message associated with this group call (if available). + unavailable: :class:`bool` + Denotes if this group call is unavailable. + ringing: List[:class:`User`] + A list of users that are currently being rung to join the call. + region: :class:`VoiceRegion` + The region the group call is being hosted in. + """ + + if TYPE_CHECKING: + channel: GroupChannel + + def _update( + self, *, ringing: SnowflakeList = [], region: VoiceRegion = MISSING + ) -> None: + if region is not MISSING: + self.region = try_enum(VoiceRegion, region) + lookup = {u.id: u for u in self.channel.recipients} + me = self.channel.me + lookup[me.id] = me + self.ringing = list(filter(None, map(lookup.get, ringing))) + + @property + def members(self) -> List[User]: + """List[:class:`User`]: Returns all users that are currently in this call.""" + ret = [u for u in self.channel.recipients if self.voice_state_for(u).channel.id == self._channel_id] + me = self.channel.me + if self.voice_state_for(me).channel.id == self._channel_id: + ret.append(me) + + return ret + + @_running_only + async def ring(self, *recipients) -> None: + await self._state.http.ring(self._channel_id, *{r.id for r in recipients}) + + @_running_only + async def stop_ringing(self, *recipients) -> None: + await self._state.http.stop_ringing(self._channel_id, *{r.id for r in recipients})