"""
The MIT License (MIT)
Copyright (c) 2015-2021 Rapptz
Copyright (c) 2021-present Pycord Development
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.
Some documentation to refer to:
- Our main web socket (mWS) sends opcode 4 with a guild ID and channel ID.
- The mWS receives VOICE_STATE_UPDATE and VOICE_SERVER_UPDATE.
- We pull the session_id from VOICE_STATE_UPDATE.
- We pull the token, endpoint and server_id from VOICE_SERVER_UPDATE.
- Then we initiate the voice web socket (vWS) pointing to the endpoint.
- We send opcode 0 with the user_id, server_id, session_id and token using the vWS.
- The vWS sends back opcode 2 with an ssrc, port, modes(array) and heartbeat_interval.
- We send a UDP discovery packet to endpoint:port and receive our IP and our port in LE.
- Then we send our IP and port via vWS with opcode 1.
- When that's all done, we receive opcode 4 from the vWS.
- Finally we can transmit data to endpoint:port.
"""
from __future__ import annotations
import asyncio
import logging
import select
import socket
import struct
import threading
import time
from typing import TYPE_CHECKING, Any, Callable, Literal, overload
from . import opus, utils
from .backoff import ExponentialBackoff
from .errors import ClientException, ConnectionClosed
from .gateway import *
from .player import AudioPlayer, AudioSource
from .sinks import RawData, RecordingException, Sink
from .utils import MISSING
if TYPE_CHECKING:
from . import abc
from .client import Client
from .guild import Guild
from .opus import Encoder
from .state import ConnectionState
from .types.voice import GuildVoiceState as GuildVoiceStatePayload
from .types.voice import SupportedModes
from .types.voice import VoiceServerUpdate as VoiceServerUpdatePayload
from .user import ClientUser
has_nacl: bool
try:
import nacl.secret # type: ignore
has_nacl = True
except ImportError:
has_nacl = False
__all__ = (
"VoiceProtocol",
"VoiceClient",
)
_log = logging.getLogger(__name__)
[docs]class VoiceProtocol:
"""A class that represents the Discord voice protocol.
This is an abstract class. The library provides a concrete implementation
under :class:`VoiceClient`.
This class allows you to implement a protocol to allow for an external
method of sending voice, such as Lavalink_ or a native library implementation.
These classes are passed to :meth:`abc.Connectable.connect <VoiceChannel.connect>`.
.. _Lavalink: https://github.com/freyacodes/Lavalink
Parameters
----------
client: :class:`Client`
The client (or its subclasses) that started the connection request.
channel: :class:`abc.Connectable`
The voice channel that is being connected to.
"""
def __init__(self, client: Client, channel: abc.Connectable) -> None:
self.client: Client = client
self.channel: abc.Connectable = channel
[docs] async def on_voice_state_update(self, data: GuildVoiceStatePayload) -> None:
"""|coro|
An abstract method that is called when the client's voice state
has changed. This corresponds to ``VOICE_STATE_UPDATE``.
Parameters
----------
data: :class:`dict`
The raw `voice state payload`__.
.. _voice_state_update_payload: https://discord.com/developers/docs/resources/voice#voice-state-object
__ voice_state_update_payload_
"""
raise NotImplementedError
[docs] async def on_voice_server_update(self, data: VoiceServerUpdatePayload) -> None:
"""|coro|
An abstract method that is called when initially connecting to voice.
This corresponds to ``VOICE_SERVER_UPDATE``.
Parameters
----------
data: :class:`dict`
The raw `voice server update payload`__.
.. _voice_server_update_payload: https://discord.com/developers/docs/topics/gateway#voice-server-update-voice-server-update-event-fields
__ voice_server_update_payload_
"""
raise NotImplementedError
[docs] async def connect(self, *, timeout: float, reconnect: bool) -> None:
"""|coro|
An abstract method called when the client initiates the connection request.
When a connection is requested initially, the library calls the constructor
under ``__init__`` and then calls :meth:`connect`. If :meth:`connect` fails at
some point then :meth:`disconnect` is called.
Within this method, to start the voice connection flow it is recommended to
use :meth:`Guild.change_voice_state` to start the flow. After which,
:meth:`on_voice_server_update` and :meth:`on_voice_state_update` will be called.
The order that these two are called is unspecified.
Parameters
----------
timeout: :class:`float`
The timeout for the connection.
reconnect: :class:`bool`
Whether reconnection is expected.
"""
raise NotImplementedError
[docs] async def disconnect(self, *, force: bool) -> None:
"""|coro|
An abstract method called when the client terminates the connection.
See :meth:`cleanup`.
Parameters
----------
force: :class:`bool`
Whether the disconnection was forced.
"""
raise NotImplementedError
[docs] def cleanup(self) -> None:
"""This method *must* be called to ensure proper clean-up during a disconnect.
It is advisable to call this from within :meth:`disconnect` when you are
completely done with the voice protocol instance.
This method removes it from the internal state cache that keeps track of
currently alive voice clients. Failure to clean-up will cause subsequent
connections to report that it's still connected.
"""
key_id, _ = self.channel._get_voice_client_key()
self.client._connection._remove_voice_client(key_id)
[docs]class VoiceClient(VoiceProtocol):
"""Represents a Discord voice connection.
You do not create these, you typically get them from
e.g. :meth:`VoiceChannel.connect`.
Attributes
----------
session_id: :class:`str`
The voice connection session ID.
token: :class:`str`
The voice connection token.
endpoint: :class:`str`
The endpoint we are connecting to.
channel: :class:`abc.Connectable`
The voice channel connected to.
loop: :class:`asyncio.AbstractEventLoop`
The event loop that the voice client is running on.
Warning
-------
In order to use PCM based AudioSources, you must have the opus library
installed on your system and loaded through :func:`opus.load_opus`.
Otherwise, your AudioSources must be opus encoded (e.g. using :class:`FFmpegOpusAudio`)
or the library will not be able to transmit audio.
"""
endpoint_ip: str
voice_port: int
secret_key: list[int]
ssrc: int
def __init__(self, client: Client, channel: abc.Connectable):
if not has_nacl:
raise RuntimeError("PyNaCl library needed in order to use voice")
super().__init__(client, channel)
state = client._connection
self.token: str = MISSING
self.socket = MISSING
self.loop: asyncio.AbstractEventLoop = state.loop
self._state: ConnectionState = state
# this will be used in the AudioPlayer thread
self._connected: threading.Event = threading.Event()
self._handshaking: bool = False
self._potentially_reconnecting: bool = False
self._voice_state_complete: asyncio.Event = asyncio.Event()
self._voice_server_complete: asyncio.Event = asyncio.Event()
self.mode: str = MISSING
self._connections: int = 0
self.sequence: int = 0
self.timestamp: int = 0
self.timeout: float = 0
self._runner: asyncio.Task = MISSING
self._player: AudioPlayer | None = None
self.encoder: Encoder = MISSING
self.decoder = None
self._lite_nonce: int = 0
self.ws: DiscordVoiceWebSocket = MISSING
self.paused = False
self.recording = False
self.user_timestamps = {}
self.sink = None
self.starting_time = None
self.stopping_time = None
warn_nacl = not has_nacl
supported_modes: tuple[SupportedModes, ...] = (
"xsalsa20_poly1305_lite",
"xsalsa20_poly1305_suffix",
"xsalsa20_poly1305",
)
@property
def guild(self) -> Guild | None:
"""The guild we're connected to, if applicable."""
return getattr(self.channel, "guild", None)
@property
def user(self) -> ClientUser:
"""The user connected to voice (i.e. ourselves)."""
return self._state.user
def checked_add(self, attr, value, limit):
val = getattr(self, attr)
if val + value > limit:
setattr(self, attr, 0)
else:
setattr(self, attr, val + value)
# connection related
async def on_voice_state_update(self, data: GuildVoiceStatePayload) -> None:
self.session_id = data["session_id"]
channel_id = data["channel_id"]
if not self._handshaking or self._potentially_reconnecting:
# If we're done handshaking then we just need to update ourselves
# If we're potentially reconnecting due to a 4014, then we need to differentiate
# a channel move and an actual force disconnect
if channel_id is None:
# We're being disconnected so cleanup
await self.disconnect()
else:
guild = self.guild
self.channel = channel_id and guild and guild.get_channel(int(channel_id)) # type: ignore
else:
self._voice_state_complete.set()
async def on_voice_server_update(self, data: VoiceServerUpdatePayload) -> None:
if self._voice_server_complete.is_set():
_log.info("Ignoring extraneous voice server update.")
return
self.token = data.get("token")
self.server_id = int(data["guild_id"])
endpoint = data.get("endpoint")
if endpoint is None or self.token is None:
_log.warning(
"Awaiting endpoint... This requires waiting. "
"If timeout occurred considering raising the timeout and reconnecting."
)
return
self.endpoint, _, _ = endpoint.rpartition(":")
if self.endpoint.startswith("wss://"):
# Just in case, strip it off since we're going to add it later
self.endpoint = self.endpoint[6:]
# This gets set later
self.endpoint_ip = MISSING
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.socket.setblocking(False)
if not self._handshaking:
# If we're not handshaking then we need to terminate our previous connection in the websocket
await self.ws.close(4000)
return
self._voice_server_complete.set()
async def voice_connect(self) -> None:
await self.channel.guild.change_voice_state(channel=self.channel)
async def voice_disconnect(self) -> None:
_log.info(
"The voice handshake is being terminated for Channel ID %s (Guild ID %s)",
self.channel.id,
self.guild.id,
)
await self.channel.guild.change_voice_state(channel=None)
def prepare_handshake(self) -> None:
self._voice_state_complete.clear()
self._voice_server_complete.clear()
self._handshaking = True
_log.info(
"Starting voice handshake... (connection attempt %d)", self._connections + 1
)
self._connections += 1
def finish_handshake(self) -> None:
_log.info("Voice handshake complete. Endpoint found %s", self.endpoint)
self._handshaking = False
self._voice_server_complete.clear()
self._voice_state_complete.clear()
async def connect_websocket(self) -> DiscordVoiceWebSocket:
ws = await DiscordVoiceWebSocket.from_client(self)
self._connected.clear()
while ws.secret_key is None:
await ws.poll_event()
self._connected.set()
return ws
async def connect(self, *, reconnect: bool, timeout: float) -> None:
_log.info("Connecting to voice...")
self.timeout = timeout
for i in range(5):
self.prepare_handshake()
# This has to be created before we start the flow.
futures = [
self._voice_state_complete.wait(),
self._voice_server_complete.wait(),
]
# Start the connection flow
await self.voice_connect()
try:
await utils.sane_wait_for(futures, timeout=timeout)
except asyncio.TimeoutError:
await self.disconnect(force=True)
raise
self.finish_handshake()
try:
self.ws = await self.connect_websocket()
break
except (ConnectionClosed, asyncio.TimeoutError):
if reconnect:
_log.exception("Failed to connect to voice... Retrying...")
await asyncio.sleep(1 + i * 2.0)
await self.voice_disconnect()
continue
else:
raise
if self._runner is MISSING:
self._runner = self.loop.create_task(self.poll_voice_ws(reconnect))
async def potential_reconnect(self) -> bool:
# Attempt to stop the player thread from playing early
self._connected.clear()
self.prepare_handshake()
self._potentially_reconnecting = True
try:
# We only care about VOICE_SERVER_UPDATE since VOICE_STATE_UPDATE can come before we get disconnected
await asyncio.wait_for(
self._voice_server_complete.wait(), timeout=self.timeout
)
except asyncio.TimeoutError:
self._potentially_reconnecting = False
await self.disconnect(force=True)
return False
self.finish_handshake()
self._potentially_reconnecting = False
try:
self.ws = await self.connect_websocket()
except (ConnectionClosed, asyncio.TimeoutError):
return False
else:
return True
@property
def latency(self) -> float:
"""Latency between a HEARTBEAT and a HEARTBEAT_ACK in seconds.
This could be referred to as the Discord Voice WebSocket latency and is
an analogue of user's voice latencies as seen in the Discord client.
.. versionadded:: 1.4
"""
ws = self.ws
return float("inf") if not ws else ws.latency
@property
def average_latency(self) -> float:
"""Average of most recent 20 HEARTBEAT latencies in seconds.
.. versionadded:: 1.4
"""
ws = self.ws
return float("inf") if not ws else ws.average_latency
async def poll_voice_ws(self, reconnect: bool) -> None:
backoff = ExponentialBackoff()
while True:
try:
await self.ws.poll_event()
except (ConnectionClosed, asyncio.TimeoutError) as exc:
if isinstance(exc, ConnectionClosed):
# The following close codes are undocumented, so I will document them here.
# 1000 - normal closure (obviously)
# 4014 - voice channel has been deleted.
# 4015 - voice server has crashed
if exc.code in (1000, 4015):
_log.info(
"Disconnecting from voice normally, close code %d.",
exc.code,
)
await self.disconnect()
break
if exc.code == 4014:
_log.info(
"Disconnected from voice by force... potentially"
" reconnecting."
)
successful = await self.potential_reconnect()
if successful:
continue
_log.info(
"Reconnect was unsuccessful, disconnecting from voice"
" normally..."
)
await self.disconnect()
break
if not reconnect:
await self.disconnect()
raise
retry = backoff.delay()
_log.exception(
"Disconnected from voice... Reconnecting in %.2fs.", retry
)
self._connected.clear()
await asyncio.sleep(retry)
await self.voice_disconnect()
try:
await self.connect(reconnect=True, timeout=self.timeout)
except asyncio.TimeoutError:
# at this point we've retried 5 times... let's continue the loop.
_log.warning("Could not connect to voice... Retrying...")
continue
[docs] async def disconnect(self, *, force: bool = False) -> None:
"""|coro|
Disconnects this voice client from voice.
"""
if not force and not self.is_connected():
return
self.stop()
self._connected.clear()
try:
if self.ws:
await self.ws.close()
await self.voice_disconnect()
finally:
self.cleanup()
if self.socket:
self.socket.close()
[docs] async def move_to(self, channel: abc.Snowflake) -> None:
"""|coro|
Moves you to a different voice channel.
Parameters
----------
channel: :class:`abc.Snowflake`
The channel to move to. Must be a voice channel.
"""
await self.channel.guild.change_voice_state(channel=channel)
[docs] def is_connected(self) -> bool:
"""Indicates if the voice client is connected to voice."""
return self._connected.is_set()
# audio related
def _get_voice_packet(self, data):
header = bytearray(12)
# Formulate rtp header
header[0] = 0x80
header[1] = 0x78
struct.pack_into(">H", header, 2, self.sequence)
struct.pack_into(">I", header, 4, self.timestamp)
struct.pack_into(">I", header, 8, self.ssrc)
encrypt_packet = getattr(self, f"_encrypt_{self.mode}")
return encrypt_packet(header, data)
def _encrypt_xsalsa20_poly1305(self, header: bytes, data) -> bytes:
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce = bytearray(24)
nonce[:12] = header
return header + box.encrypt(bytes(data), bytes(nonce)).ciphertext
def _encrypt_xsalsa20_poly1305_suffix(self, header: bytes, data) -> bytes:
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce = nacl.utils.random(nacl.secret.SecretBox.NONCE_SIZE)
return header + box.encrypt(bytes(data), nonce).ciphertext + nonce
def _encrypt_xsalsa20_poly1305_lite(self, header: bytes, data) -> bytes:
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce = bytearray(24)
nonce[:4] = struct.pack(">I", self._lite_nonce)
self.checked_add("_lite_nonce", 1, 4294967295)
return header + box.encrypt(bytes(data), bytes(nonce)).ciphertext + nonce[:4]
def _decrypt_xsalsa20_poly1305(self, header, data):
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce = bytearray(24)
nonce[:12] = header
return self.strip_header_ext(box.decrypt(bytes(data), bytes(nonce)))
def _decrypt_xsalsa20_poly1305_suffix(self, header, data):
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce_size = nacl.secret.SecretBox.NONCE_SIZE
nonce = data[-nonce_size:]
return self.strip_header_ext(box.decrypt(bytes(data[:-nonce_size]), nonce))
def _decrypt_xsalsa20_poly1305_lite(self, header, data):
box = nacl.secret.SecretBox(bytes(self.secret_key))
nonce = bytearray(24)
nonce[:4] = data[-4:]
data = data[:-4]
return self.strip_header_ext(box.decrypt(bytes(data), bytes(nonce)))
@staticmethod
def strip_header_ext(data):
if data[0] == 0xBE and data[1] == 0xDE and len(data) > 4:
_, length = struct.unpack_from(">HH", data)
offset = 4 + length * 4
data = data[offset:]
return data
def get_ssrc(self, user_id):
return {info["user_id"]: ssrc for ssrc, info in self.ws.ssrc_map.items()}[
user_id
]
@overload
def play(
self,
source: AudioSource,
*,
after: Callable[[Exception | None], Any] | None = None,
wait_finish: Literal[False] = False,
) -> None: ...
@overload
def play(
self,
source: AudioSource,
*,
after: Callable[[Exception | None], Any] | None = None,
wait_finish: Literal[True],
) -> asyncio.Future: ...
[docs] def play(
self,
source: AudioSource,
*,
after: Callable[[Exception | None], Any] | None = None,
wait_finish: bool = False,
) -> None | asyncio.Future:
"""Plays an :class:`AudioSource`.
The finalizer, ``after`` is called after the source has been exhausted
or an error occurred.
If an error happens while the audio player is running, the exception is
caught and the audio player is then stopped. If no after callback is
passed, any caught exception will be displayed as if it were raised.
Parameters
----------
source: :class:`AudioSource`
The audio source we're reading from.
after: Callable[[Optional[:class:`Exception`]], Any]
The finalizer that is called after the stream is exhausted.
This function must have a single parameter, ``error``, that
denotes an optional exception that was raised during playing.
wait_finish: bool
If True, an awaitable will be returned, which can be used to wait for
audio to stop playing. This awaitable will return an exception if raised,
or None when no exception is raised.
If False, None is returned and the function does not block.
.. versionadded:: v2.5
Raises
------
ClientException
Already playing audio or not connected.
TypeError
Source is not a :class:`AudioSource` or after is not a callable.
OpusNotLoaded
Source is not opus encoded and opus is not loaded.
"""
if not self.is_connected():
raise ClientException("Not connected to voice.")
if self.is_playing():
raise ClientException("Already playing audio.")
if not isinstance(source, AudioSource):
raise TypeError(
f"source must be an AudioSource not {source.__class__.__name__}"
)
if not self.encoder and not source.is_opus():
self.encoder = opus.Encoder()
future = None
if wait_finish:
future = asyncio.Future()
after_callback = after
def _after(exc: Exception | None):
if callable(after_callback):
after_callback(exc)
future.set_result(exc)
after = _after
self._player = AudioPlayer(source, self, after=after)
self._player.start()
return future
[docs] def unpack_audio(self, data):
"""Takes an audio packet received from Discord and decodes it into pcm audio data.
If there are no users talking in the channel, `None` will be returned.
You must be connected to receive audio.
.. versionadded:: 2.0
Parameters
----------
data: :class:`bytes`
Bytes received by Discord via the UDP connection used for sending and receiving voice data.
"""
if 200 <= data[1] <= 204:
# RTCP received.
# RTCP provides information about the connection
# as opposed to actual audio data, so it's not
# important at the moment.
return
if self.paused:
return
data = RawData(data, self)
if data.decrypted_data == b"\xf8\xff\xfe": # Frame of silence
return
self.decoder.decode(data)
[docs] def start_recording(self, sink, callback, *args, sync_start: bool = False):
"""The bot will begin recording audio from the current voice channel it is in.
This function uses a thread so the current code line will not be stopped.
Must be in a voice channel to use.
Must not be already recording.
.. versionadded:: 2.0
Parameters
----------
sink: :class:`.Sink`
A Sink which will "store" all the audio data.
callback: :ref:`coroutine <coroutine>`
A function which is called after the bot has stopped recording.
*args:
Args which will be passed to the callback function.
sync_start: :class:`bool`
If True, the recordings of subsequent users will start with silence.
This is useful for recording audio just as it was heard.
Raises
------
RecordingException
Not connected to a voice channel.
RecordingException
Already recording.
RecordingException
Must provide a Sink object.
"""
if not self.is_connected():
raise RecordingException("Not connected to voice channel.")
if self.recording:
raise RecordingException("Already recording.")
if not isinstance(sink, Sink):
raise RecordingException("Must provide a Sink object.")
self.empty_socket()
self.decoder = opus.DecodeManager(self)
self.decoder.start()
self.recording = True
self.sync_start = sync_start
self.sink = sink
sink.init(self)
t = threading.Thread(
target=self.recv_audio,
args=(
sink,
callback,
*args,
),
)
t.start()
[docs] def stop_recording(self):
"""Stops the recording.
Must be already recording.
.. versionadded:: 2.0
Raises
------
RecordingException
Not currently recording.
"""
if not self.recording:
raise RecordingException("Not currently recording audio.")
self.decoder.stop()
self.recording = False
self.paused = False
[docs] def toggle_pause(self):
"""Pauses or unpauses the recording.
Must be already recording.
.. versionadded:: 2.0
Raises
------
RecordingException
Not currently recording.
"""
if not self.recording:
raise RecordingException("Not currently recording audio.")
self.paused = not self.paused
def empty_socket(self):
while True:
ready, _, _ = select.select([self.socket], [], [], 0.0)
if not ready:
break
for s in ready:
s.recv(4096)
def recv_audio(self, sink, callback, *args):
# Gets data from _recv_audio and sorts
# it by user, handles pcm files and
# silence that should be added.
self.user_timestamps: dict[int, tuple[int, float]] = {}
self.starting_time = time.perf_counter()
self.first_packet_timestamp: float
while self.recording:
ready, _, err = select.select([self.socket], [], [self.socket], 0.01)
if not ready:
if err:
print(f"Socket error: {err}")
continue
try:
data = self.socket.recv(4096)
except OSError:
self.stop_recording()
continue
self.unpack_audio(data)
self.stopping_time = time.perf_counter()
self.sink.cleanup()
callback = asyncio.run_coroutine_threadsafe(callback(sink, *args), self.loop)
result = callback.result()
if result is not None:
print(result)
def recv_decoded_audio(self, data: RawData):
# Add silence when they were not being recorded.
if data.ssrc not in self.user_timestamps: # First packet from user
if (
not self.user_timestamps or not self.sync_start
): # First packet from anyone
self.first_packet_timestamp = data.receive_time
silence = 0
else: # Previously received a packet from someone else
silence = (
(data.receive_time - self.first_packet_timestamp) * 48000
) - 960
else: # Previously received a packet from user
dRT = (
data.receive_time - self.user_timestamps[data.ssrc][1]
) * 48000 # delta receive time
dT = data.timestamp - self.user_timestamps[data.ssrc][0] # delta timestamp
diff = abs(100 - dT * 100 / dRT)
if (
diff > 60 and dT != 960
): # If the difference in change is more than 60% threshold
silence = dRT - 960
else:
silence = dT - 960
self.user_timestamps.update({data.ssrc: (data.timestamp, data.receive_time)})
data.decoded_data = (
struct.pack("<h", 0) * max(0, int(silence)) * opus._OpusStruct.CHANNELS
+ data.decoded_data
)
while data.ssrc not in self.ws.ssrc_map:
time.sleep(0.05)
self.sink.write(data.decoded_data, self.ws.ssrc_map[data.ssrc]["user_id"])
[docs] def is_playing(self) -> bool:
"""Indicates if we're currently playing audio."""
return self._player is not None and self._player.is_playing()
[docs] def is_paused(self) -> bool:
"""Indicates if we're playing audio, but if we're paused."""
return self._player is not None and self._player.is_paused()
[docs] def stop(self) -> None:
"""Stops playing audio."""
if self._player:
self._player.stop()
self._player = None
[docs] def pause(self) -> None:
"""Pauses the audio playing."""
if self._player:
self._player.pause()
[docs] def resume(self) -> None:
"""Resumes the audio playing."""
if self._player:
self._player.resume()
@property
def source(self) -> AudioSource | None:
"""The audio source being played, if playing.
This property can also be used to change the audio source currently being played.
"""
return self._player.source if self._player else None
@source.setter
def source(self, value: AudioSource) -> None:
if not isinstance(value, AudioSource):
raise TypeError(f"expected AudioSource not {value.__class__.__name__}.")
if self._player is None:
raise ValueError("Not playing anything.")
self._player._set_source(value)
[docs] def send_audio_packet(self, data: bytes, *, encode: bool = True) -> None:
"""Sends an audio packet composed of the data.
You must be connected to play audio.
Parameters
----------
data: :class:`bytes`
The :term:`py:bytes-like object` denoting PCM or Opus voice data.
encode: :class:`bool`
Indicates if ``data`` should be encoded into Opus.
Raises
------
ClientException
You are not connected.
opus.OpusError
Encoding the data failed.
"""
self.checked_add("sequence", 1, 65535)
if encode:
if not self.encoder:
self.encoder = opus.Encoder()
encoded_data = self.encoder.encode(data, self.encoder.SAMPLES_PER_FRAME)
else:
encoded_data = data
packet = self._get_voice_packet(encoded_data)
try:
self.socket.sendto(packet, (self.endpoint_ip, self.voice_port))
except BlockingIOError:
_log.warning(
"A packet has been dropped (seq: %s, timestamp: %s)",
self.sequence,
self.timestamp,
)
self.checked_add("timestamp", opus.Encoder.SAMPLES_PER_FRAME, 4294967295)