Active voice state

attrs class ActiveVoiceState (VoiceState)

Attr attributes:

Name	Type	Description
ws	Optional[naff.api.voice.voice_gateway.VoiceGateway]	The websocket for this voice state
player	Optional[naff.api.voice.player.Player]	The playback task that broadcasts audio data to discord

Source code in naff/models/naff/active_voice_state.py

@define()
class ActiveVoiceState(VoiceState):
    ws: Optional[VoiceGateway] = field(default=None)
    """The websocket for this voice state"""
    player: Optional[Player] = field(default=None)
    """The playback task that broadcasts audio data to discord"""
    _volume: float = field(default=0.5)

    # standard voice states expect this data, this voice state lacks it initially; so we make them optional
    user_id: "Snowflake_Type" = field(default=MISSING, converter=optional(to_snowflake))
    _guild_id: Optional["Snowflake_Type"] = field(default=None, converter=optional(to_snowflake))
    _member_id: Optional["Snowflake_Type"] = field(default=None, converter=optional(to_snowflake))

    def __attrs_post_init__(self) -> None:
        # jank line to handle the two inherently incompatible data structures
        self._member_id = self.user_id = self._client.user.id

    def __del__(self) -> None:
        if self.connected:
            self.ws.close()
        if self.player:
            self.player.stop()

    def __repr__(self) -> str:
        return f"<ActiveVoiceState: channel={self.channel} guild={self.guild} volume={self.volume} playing={self.playing} audio={self.current_audio}>"

    @property
    def current_audio(self) -> Optional["BaseAudio"]:
        """The current audio being played"""
        if self.player:
            return self.player.current_audio

    @property
    def volume(self) -> float:
        """Get the volume of the player"""
        return self._volume

    @volume.setter
    def volume(self, value) -> None:
        """Set the volume of the player"""
        if value < 0.0:
            raise ValueError("Volume may not be negative.")
        self._volume = value
        if self.player and hasattr(self.player.current_audio, "volume"):
            self.player.current_audio.volume = value

    @property
    def paused(self) -> bool:
        """Is the player currently paused"""
        if self.player:
            return self.player.paused
        return False

    @property
    def playing(self) -> bool:
        """Are we currently playing something?"""
        # noinspection PyProtectedMember
        if not self.player or not self.current_audio or self.player.stopped or not self.player._resume.is_set():
            # if any of the above are truthy, we aren't playing
            return False
        return True

    @property
    def stopped(self) -> bool:
        """Is the player stopped?"""
        if self.player:
            return self.player.stopped
        return True

    @property
    def connected(self) -> bool:
        """Is this voice state currently connected?"""
        # noinspection PyProtectedMember
        if self.ws is None:
            return False
        return self.ws._closed.is_set()

    @property
    def gateway(self) -> "GatewayClient":
        return self._client.get_guild_websocket(self._guild_id)

    async def wait_for_stopped(self) -> None:
        """Wait for the player to stop playing."""
        if self.player:
            # noinspection PyProtectedMember
            await self.player._stopped.wait()

    async def _ws_connect(self) -> None:
        """Runs the voice gateway connection"""
        async with self.ws:
            try:
                await self.ws.run()
            finally:
                if self.playing:
                    await self.stop()

    async def ws_connect(self) -> None:
        """Connect to the voice gateway for this voice state"""
        self.ws = VoiceGateway(self._client._connection_state, self._voice_state.data, self._voice_server.data)

        asyncio.create_task(self._ws_connect())
        await self.ws.wait_until_ready()

    def _guild_predicate(self, event) -> bool:
        return int(event.data["guild_id"]) == self._guild_id

    async def connect(self, timeout: int = 5) -> None:
        """
        Establish the voice connection.

        Args:
            timeout: How long to wait for state and server information from discord

        Raises:
            VoiceAlreadyConnected: if the voice state is already connected to the voice channel
            VoiceConnectionTimeout: if the voice state fails to connect

        """
        if self.connected:
            raise VoiceAlreadyConnected
        await self.gateway.voice_state_update(self._guild_id, self._channel_id, self.self_mute, self.self_deaf)

        logger.debug("Waiting for voice connection data...")

        try:
            self._voice_state, self._voice_server = await asyncio.gather(
                self._client.wait_for("raw_voice_state_update", self._guild_predicate, timeout=timeout),
                self._client.wait_for("raw_voice_server_update", self._guild_predicate, timeout=timeout),
            )
        except asyncio.TimeoutError:
            raise VoiceConnectionTimeout from None

        logger.debug("Attempting to initialise voice gateway...")
        await self.ws_connect()

    async def disconnect(self) -> None:
        """Disconnect from the voice channel."""
        await self.gateway.voice_state_update(self._guild_id, None)

    async def move(self, channel: "Snowflake_Type", timeout: int = 5) -> None:
        """
        Move to another voice channel.

        Args:
            channel: The channel to move to
            timeout: How long to wait for state and server information from discord

        """
        target_channel = to_snowflake(channel)
        if target_channel != self._channel_id:
            already_paused = self.paused
            if self.player:
                self.player.pause()

            self._channel_id = target_channel
            await self.gateway.voice_state_update(self._guild_id, self._channel_id, self.self_mute, self.self_deaf)

            logger.debug("Waiting for voice connection data...")
            try:
                await self._client.wait_for("raw_voice_state_update", self._guild_predicate, timeout=timeout)
            except asyncio.TimeoutError:
                await self._close_connection()
                raise VoiceConnectionTimeout from None

            if self.player and not already_paused:
                self.player.resume()

    async def stop(self) -> None:
        """Stop playback."""
        self.player.stop()
        await self.player._stopped.wait()

    def pause(self) -> None:
        """Pause playback"""
        self.player.pause()

    def resume(self) -> None:
        """Resume playback."""
        self.player.resume()

    async def play(self, audio: "BaseAudio") -> None:
        """
        Start playing an audio object.

        Waits for the player to stop before returning.

        Args:
            audio: The audio object to play
        """
        if self.player:
            await self.stop()

        with Player(audio, self, asyncio.get_running_loop()) as self.player:
            self.player.play()
            await self.wait_for_stopped()

    def play_no_wait(self, audio: "BaseAudio") -> None:
        """
        Start playing an audio object, but don't wait for playback to finish.

        Args:
            audio: The audio object to play
        """
        asyncio.create_task(self.play(audio))

    async def _voice_server_update(self, data) -> None:
        """
        An internal receiver for voice server events.

        Args:
            data: voice server data
        """
        self.ws.set_new_voice_server(data)

    async def _voice_state_update(
        self, before: Optional[VoiceState], after: Optional[VoiceState], data: Optional[VoiceStateData]
    ) -> None:
        """
        An internal receiver for voice server state events.

        Args:
            before: The previous voice state
            after: The current voice state
            data: Raw data from gateway
        """
        if after is None:
            # bot disconnected
            logger.info(f"Disconnecting from voice channel {self._channel_id}")
            await self._close_connection()
            self._client.cache.delete_bot_voice_state(self._guild_id)
            return

        self.update_from_dict(data)

    async def _close_connection(self) -> None:
        """Close the voice connection."""
        if self.playing:
            await self.stop()
        if self.connected:
            self.ws.close()

inherited method update_from_dict(self, data)

Updates object attribute(s) with new json data received from discord api.

Parameters:

Name	Type	Description	Default
data		The json data received from discord api.	required

Returns:

Type	Description
~T	The updated object class instance.

Source code in naff/models/naff/active_voice_state.py

def update_from_dict(self, data) -> T:
    data = self._process_dict(data, self._client)
    for key, value in self._filter_kwargs(data, self._get_keys()).items():
        # todo improve
        setattr(self, key, value)

    return self

property readonly current_audio: Optional[BaseAudio]

The current audio being played

inherited property readonly guild: Guild

The guild this voice state is for.

property writable volume: float

Get the volume of the player

inherited property readonly channel: TYPE_VOICE_CHANNEL

The channel the user is connected to.

property readonly paused: bool

Is the player currently paused

property readonly playing: bool

Are we currently playing something?

inherited property readonly member: Member

The member this voice state is for.

property readonly stopped: bool

Is the player stopped?

property readonly connected: bool

Is this voice state currently connected?

inherited method to_dict(self)

Exports object into dictionary representation, ready to be sent to discord api.

Returns:

Type	Description
Dict[str, Any]	The exported dictionary.

Source code in naff/models/naff/active_voice_state.py

def to_dict(self) -> Dict[str, Any]:
    """
    Exports object into dictionary representation, ready to be sent to discord api.

    Returns:
        The exported dictionary.

    """
    self._check_object()
    return serializer.to_dict(self)

async method wait_for_stopped(self)

Wait for the player to stop playing.

Source code in naff/models/naff/active_voice_state.py

async def wait_for_stopped(self) -> None:
    """Wait for the player to stop playing."""
    if self.player:
        # noinspection PyProtectedMember
        await self.player._stopped.wait()

async method ws_connect(self)

Connect to the voice gateway for this voice state

Source code in naff/models/naff/active_voice_state.py

async def ws_connect(self) -> None:
    """Connect to the voice gateway for this voice state"""
    self.ws = VoiceGateway(self._client._connection_state, self._voice_state.data, self._voice_server.data)

    asyncio.create_task(self._ws_connect())
    await self.ws.wait_until_ready()

async method connect(self, timeout)

Establish the voice connection.

Parameters:

Name	Type	Description	Default
timeout	int	How long to wait for state and server information from discord	5

Exceptions:

Type	Description
VoiceAlreadyConnected	if the voice state is already connected to the voice channel
VoiceConnectionTimeout	if the voice state fails to connect

Source code in naff/models/naff/active_voice_state.py

async def connect(self, timeout: int = 5) -> None:
    """
    Establish the voice connection.

    Args:
        timeout: How long to wait for state and server information from discord

    Raises:
        VoiceAlreadyConnected: if the voice state is already connected to the voice channel
        VoiceConnectionTimeout: if the voice state fails to connect

    """
    if self.connected:
        raise VoiceAlreadyConnected
    await self.gateway.voice_state_update(self._guild_id, self._channel_id, self.self_mute, self.self_deaf)

    logger.debug("Waiting for voice connection data...")

    try:
        self._voice_state, self._voice_server = await asyncio.gather(
            self._client.wait_for("raw_voice_state_update", self._guild_predicate, timeout=timeout),
            self._client.wait_for("raw_voice_server_update", self._guild_predicate, timeout=timeout),
        )
    except asyncio.TimeoutError:
        raise VoiceConnectionTimeout from None

    logger.debug("Attempting to initialise voice gateway...")
    await self.ws_connect()

async method disconnect(self)

Disconnect from the voice channel.

Source code in naff/models/naff/active_voice_state.py

async def disconnect(self) -> None:
    """Disconnect from the voice channel."""
    await self.gateway.voice_state_update(self._guild_id, None)

async method move(self, channel, timeout)

Move to another voice channel.

Parameters:

Name	Type	Description	Default
channel	Snowflake_Type	The channel to move to	required
timeout	int	How long to wait for state and server information from discord	5

Source code in naff/models/naff/active_voice_state.py

async def move(self, channel: "Snowflake_Type", timeout: int = 5) -> None:
    """
    Move to another voice channel.

    Args:
        channel: The channel to move to
        timeout: How long to wait for state and server information from discord

    """
    target_channel = to_snowflake(channel)
    if target_channel != self._channel_id:
        already_paused = self.paused
        if self.player:
            self.player.pause()

        self._channel_id = target_channel
        await self.gateway.voice_state_update(self._guild_id, self._channel_id, self.self_mute, self.self_deaf)

        logger.debug("Waiting for voice connection data...")
        try:
            await self._client.wait_for("raw_voice_state_update", self._guild_predicate, timeout=timeout)
        except asyncio.TimeoutError:
            await self._close_connection()
            raise VoiceConnectionTimeout from None

        if self.player and not already_paused:
            self.player.resume()

async method stop(self)

Stop playback.

Source code in naff/models/naff/active_voice_state.py

async def stop(self) -> None:
    """Stop playback."""
    self.player.stop()
    await self.player._stopped.wait()

method pause(self)

Pause playback

Source code in naff/models/naff/active_voice_state.py

def pause(self) -> None:
    """Pause playback"""
    self.player.pause()

method resume(self)

Resume playback.

Source code in naff/models/naff/active_voice_state.py

def resume(self) -> None:
    """Resume playback."""
    self.player.resume()

async method play(self, audio)

Start playing an audio object.

Waits for the player to stop before returning.

Parameters:

Name	Type	Description	Default
audio	BaseAudio	The audio object to play	required

Source code in naff/models/naff/active_voice_state.py

async def play(self, audio: "BaseAudio") -> None:
    """
    Start playing an audio object.

    Waits for the player to stop before returning.

    Args:
        audio: The audio object to play
    """
    if self.player:
        await self.stop()

    with Player(audio, self, asyncio.get_running_loop()) as self.player:
        self.player.play()
        await self.wait_for_stopped()

method play_no_wait(self, audio)

Start playing an audio object, but don't wait for playback to finish.

Parameters:

Name	Type	Description	Default
audio	BaseAudio	The audio object to play	required

Source code in naff/models/naff/active_voice_state.py

def play_no_wait(self, audio: "BaseAudio") -> None:
    """
    Start playing an audio object, but don't wait for playback to finish.

    Args:
        audio: The audio object to play
    """
    asyncio.create_task(self.play(audio))