matthew/enhanced-discord.py
1
0
Fork 0
Code Issues 4 Pull Requests 21 Projects Releases Wiki Activity

Fix code style issues with Black

Browse Source
This commit is contained in:
Lint Action
2021-09-05 21:34:20 +00:00
parent a23dae8604
commit 7513c2138f
108 changed files with 5369 additions and 4858 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
discord/opus.py
View File

@ -38,9 +38,10 @@ import sys
from .errors import DiscordException, InvalidArgument
if TYPE_CHECKING:
T = TypeVar('T')
BAND_CTL = Literal['narrow', 'medium', 'wide', 'superwide', 'full']
SIGNAL_CTL = Literal['auto', 'voice', 'music']
T = TypeVar("T")
BAND_CTL = Literal["narrow", "medium", "wide", "superwide", "full"]
SIGNAL_CTL = Literal["auto", "voice", "music"]
class BandCtl(TypedDict):
narrow: int
@ -49,81 +50,89 @@ class BandCtl(TypedDict):
superwide: int
full: int
class SignalCtl(TypedDict):
auto: int
voice: int
music: int
__all__ = (
'Encoder',
'OpusError',
'OpusNotLoaded',
"Encoder",
"OpusError",
"OpusNotLoaded",
)
_log = logging.getLogger(__name__)
c_int_ptr = ctypes.POINTER(ctypes.c_int)
c_int_ptr = ctypes.POINTER(ctypes.c_int)
c_int16_ptr = ctypes.POINTER(ctypes.c_int16)
c_float_ptr = ctypes.POINTER(ctypes.c_float)
_lib = None
class EncoderStruct(ctypes.Structure):
pass
class DecoderStruct(ctypes.Structure):
pass
EncoderStructPtr = ctypes.POINTER(EncoderStruct)
DecoderStructPtr = ctypes.POINTER(DecoderStruct)
## Some constants from opus_defines.h
# Error codes
OK = 0
OK = 0
BAD_ARG = -1
# Encoder CTLs
APPLICATION_AUDIO = 2049
APPLICATION_VOIP = 2048
APPLICATION_AUDIO = 2049
APPLICATION_VOIP = 2048
APPLICATION_LOWDELAY = 2051
CTL_SET_BITRATE = 4002
CTL_SET_BANDWIDTH = 4008
CTL_SET_FEC = 4012
CTL_SET_PLP = 4014
CTL_SET_SIGNAL = 4024
CTL_SET_BITRATE = 4002
CTL_SET_BANDWIDTH = 4008
CTL_SET_FEC = 4012
CTL_SET_PLP = 4014
CTL_SET_SIGNAL = 4024
# Decoder CTLs
CTL_SET_GAIN = 4034
CTL_SET_GAIN = 4034
CTL_LAST_PACKET_DURATION = 4039
band_ctl: BandCtl = {
'narrow': 1101,
'medium': 1102,
'wide': 1103,
'superwide': 1104,
'full': 1105,
"narrow": 1101,
"medium": 1102,
"wide": 1103,
"superwide": 1104,
"full": 1105,
}
signal_ctl: SignalCtl = {
'auto': -1000,
'voice': 3001,
'music': 3002,
"auto": -1000,
"voice": 3001,
"music": 3002,
}
def _err_lt(result: int, func: Callable, args: List) -> int:
if result < OK:
_log.info('error has happened in %s', func.__name__)
_log.info("error has happened in %s", func.__name__)
raise OpusError(result)
return result
def _err_ne(result: T, func: Callable, args: List) -> T:
ret = args[-1]._obj
if ret.value != OK:
_log.info('error has happened in %s', func.__name__)
_log.info("error has happened in %s", func.__name__)
raise OpusError(ret.value)
return result
# A list of exported functions.
# The first argument is obviously the name.
# The second one are the types of arguments it takes.
@ -131,54 +140,51 @@ def _err_ne(result: T, func: Callable, args: List) -> T:
# The fourth is the error handler.
exported_functions: List[Tuple[Any, ...]] = [
# Generic
('opus_get_version_string',
None, ctypes.c_char_p, None),
('opus_strerror',
[ctypes.c_int], ctypes.c_char_p, None),
("opus_get_version_string", None, ctypes.c_char_p, None),
("opus_strerror", [ctypes.c_int], ctypes.c_char_p, None),
# Encoder functions
('opus_encoder_get_size',
[ctypes.c_int], ctypes.c_int, None),
('opus_encoder_create',
[ctypes.c_int, ctypes.c_int, ctypes.c_int, c_int_ptr], EncoderStructPtr, _err_ne),
('opus_encode',
[EncoderStructPtr, c_int16_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int32, _err_lt),
('opus_encode_float',
[EncoderStructPtr, c_float_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int32, _err_lt),
('opus_encoder_ctl',
None, ctypes.c_int32, _err_lt),
('opus_encoder_destroy',
[EncoderStructPtr], None, None),
("opus_encoder_get_size", [ctypes.c_int], ctypes.c_int, None),
("opus_encoder_create", [ctypes.c_int, ctypes.c_int, ctypes.c_int, c_int_ptr], EncoderStructPtr, _err_ne),
(
"opus_encode",
[EncoderStructPtr, c_int16_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32],
ctypes.c_int32,
_err_lt,
),
(
"opus_encode_float",
[EncoderStructPtr, c_float_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32],
ctypes.c_int32,
_err_lt,
),
("opus_encoder_ctl", None, ctypes.c_int32, _err_lt),
("opus_encoder_destroy", [EncoderStructPtr], None, None),
# Decoder functions
('opus_decoder_get_size',
[ctypes.c_int], ctypes.c_int, None),
('opus_decoder_create',
[ctypes.c_int, ctypes.c_int, c_int_ptr], DecoderStructPtr, _err_ne),
('opus_decode',
("opus_decoder_get_size", [ctypes.c_int], ctypes.c_int, None),
("opus_decoder_create", [ctypes.c_int, ctypes.c_int, c_int_ptr], DecoderStructPtr, _err_ne),
(
"opus_decode",
[DecoderStructPtr, ctypes.c_char_p, ctypes.c_int32, c_int16_ptr, ctypes.c_int, ctypes.c_int],
ctypes.c_int, _err_lt),
('opus_decode_float',
ctypes.c_int,
_err_lt,
),
(
"opus_decode_float",
[DecoderStructPtr, ctypes.c_char_p, ctypes.c_int32, c_float_ptr, ctypes.c_int, ctypes.c_int],
ctypes.c_int, _err_lt),
('opus_decoder_ctl',
None, ctypes.c_int32, _err_lt),
('opus_decoder_destroy',
[DecoderStructPtr], None, None),
('opus_decoder_get_nb_samples',
[DecoderStructPtr, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int, _err_lt),
ctypes.c_int,
_err_lt,
),
("opus_decoder_ctl", None, ctypes.c_int32, _err_lt),
("opus_decoder_destroy", [DecoderStructPtr], None, None),
("opus_decoder_get_nb_samples", [DecoderStructPtr, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int, _err_lt),
# Packet functions
('opus_packet_get_bandwidth',
[ctypes.c_char_p], ctypes.c_int, _err_lt),
('opus_packet_get_nb_channels',
[ctypes.c_char_p], ctypes.c_int, _err_lt),
('opus_packet_get_nb_frames',
[ctypes.c_char_p, ctypes.c_int], ctypes.c_int, _err_lt),
('opus_packet_get_samples_per_frame',
[ctypes.c_char_p, ctypes.c_int], ctypes.c_int, _err_lt),
("opus_packet_get_bandwidth", [ctypes.c_char_p], ctypes.c_int, _err_lt),
("opus_packet_get_nb_channels", [ctypes.c_char_p], ctypes.c_int, _err_lt),
("opus_packet_get_nb_frames", [ctypes.c_char_p, ctypes.c_int], ctypes.c_int, _err_lt),
("opus_packet_get_samples_per_frame", [ctypes.c_char_p, ctypes.c_int], ctypes.c_int, _err_lt),
]
def libopus_loader(name: str) -> Any:
# create the library...
lib = ctypes.cdll.LoadLibrary(name)
@ -203,22 +209,24 @@ def libopus_loader(name: str) -> Any:
return lib
def _load_default() -> bool:
global _lib
try:
if sys.platform == 'win32':
if sys.platform == "win32":
_basedir = os.path.dirname(os.path.abspath(__file__))
_bitness = struct.calcsize('P') * 8
_target = 'x64' if _bitness > 32 else 'x86'
_filename = os.path.join(_basedir, 'bin', f'libopus-0.{_target}.dll')
_bitness = struct.calcsize("P") * 8
_target = "x64" if _bitness > 32 else "x86"
_filename = os.path.join(_basedir, "bin", f"libopus-0.{_target}.dll")
_lib = libopus_loader(_filename)
else:
_lib = libopus_loader(ctypes.util.find_library('opus'))
_lib = libopus_loader(ctypes.util.find_library("opus"))
except Exception:
_lib = None
return _lib is not None
def load_opus(name: str) -> None:
"""Loads the libopus shared library for use with voice.
@ -257,6 +265,7 @@ def load_opus(name: str) -> None:
global _lib
_lib = libopus_loader(name)
def is_loaded() -> bool:
"""Function to check if opus lib is successfully loaded either
via the :func:`ctypes.util.find_library` call of :func:`load_opus`.
@ -271,6 +280,7 @@ def is_loaded() -> bool:
global _lib
return _lib is not None
class OpusError(DiscordException):
"""An exception that is thrown for libopus related errors.
@ -282,19 +292,22 @@ class OpusError(DiscordException):
def __init__(self, code: int):
self.code: int = code
msg = _lib.opus_strerror(self.code).decode('utf-8')
msg = _lib.opus_strerror(self.code).decode("utf-8")
_log.info('"%s" has happened', msg)
super().__init__(msg)
class OpusNotLoaded(DiscordException):
"""An exception that is thrown for when libopus is not loaded."""
pass
class _OpusStruct:
SAMPLING_RATE = 48000
CHANNELS = 2
FRAME_LENGTH = 20 # in milliseconds
SAMPLE_SIZE = struct.calcsize('h') * CHANNELS
SAMPLE_SIZE = struct.calcsize("h") * CHANNELS
SAMPLES_PER_FRAME = int(SAMPLING_RATE / 1000 * FRAME_LENGTH)
FRAME_SIZE = SAMPLES_PER_FRAME * SAMPLE_SIZE
@ -304,7 +317,8 @@ class _OpusStruct:
if not is_loaded() and not _load_default():
raise OpusNotLoaded()
return _lib.opus_get_version_string().decode('utf-8')
return _lib.opus_get_version_string().decode("utf-8")
class Encoder(_OpusStruct):
def __init__(self, application: int = APPLICATION_AUDIO):
@ -315,14 +329,14 @@ class Encoder(_OpusStruct):
self.set_bitrate(128)
self.set_fec(True)
self.set_expected_packet_loss_percent(0.15)
self.set_bandwidth('full')
self.set_signal_type('auto')
self.set_bandwidth("full")
self.set_signal_type("auto")
def __del__(self) -> None:
if hasattr(self, '_state'):
if hasattr(self, "_state"):
_lib.opus_encoder_destroy(self._state)
# This is a destructor, so it's okay to assign None
self._state = None # type: ignore
self._state = None # type: ignore
def _create_state(self) -> EncoderStruct:
ret = ctypes.c_int()
@ -352,18 +366,19 @@ class Encoder(_OpusStruct):
_lib.opus_encoder_ctl(self._state, CTL_SET_FEC, 1 if enabled else 0)
def set_expected_packet_loss_percent(self, percentage: float) -> None:
_lib.opus_encoder_ctl(self._state, CTL_SET_PLP, min(100, max(0, int(percentage * 100)))) # type: ignore
_lib.opus_encoder_ctl(self._state, CTL_SET_PLP, min(100, max(0, int(percentage * 100)))) # type: ignore
def encode(self, pcm: bytes, frame_size: int) -> bytes:
max_data_bytes = len(pcm)
# bytes can be used to reference pointer
pcm_ptr = ctypes.cast(pcm, c_int16_ptr) # type: ignore
pcm_ptr = ctypes.cast(pcm, c_int16_ptr) # type: ignore
data = (ctypes.c_char * max_data_bytes)()
ret = _lib.opus_encode(self._state, pcm_ptr, frame_size, data, max_data_bytes)
# array can be initialized with bytes but mypy doesn't know
return array.array('b', data[:ret]).tobytes() # type: ignore
return array.array("b", data[:ret]).tobytes() # type: ignore
class Decoder(_OpusStruct):
def __init__(self):
@ -372,10 +387,10 @@ class Decoder(_OpusStruct):
self._state: DecoderStruct = self._create_state()
def __del__(self) -> None:
if hasattr(self, '_state'):
if hasattr(self, "_state"):
_lib.opus_decoder_destroy(self._state)
# This is a destructor, so it's okay to assign None
self._state = None # type: ignore
self._state = None # type: ignore
def _create_state(self) -> DecoderStruct:
ret = ctypes.c_int()
@ -411,12 +426,12 @@ class Decoder(_OpusStruct):
def set_gain(self, dB: float) -> int:
"""Sets the decoder gain in dB, from -128 to 128."""
dB_Q8 = max(-32768, min(32767, round(dB * 256))) # dB * 2^n where n is 8 (Q8)
dB_Q8 = max(-32768, min(32767, round(dB * 256))) # dB * 2^n where n is 8 (Q8)
return self._set_gain(dB_Q8)
def set_volume(self, mult: float) -> int:
"""Sets the output volume as a float percent, i.e. 0.5 for 50%, 1.75 for 175%, etc."""
return self.set_gain(20 * math.log10(mult)) # amplitude ratio
return self.set_gain(20 * math.log10(mult)) # amplitude ratio
def _get_last_packet_duration(self) -> int:
"""Gets the duration (in samples) of the last packet successfully decoded or concealed."""
@ -428,7 +443,7 @@ class Decoder(_OpusStruct):
@overload
def decode(self, data: bytes, *, fec: bool) -> bytes:
...
@overload
def decode(self, data: Literal[None], *, fec: Literal[False]) -> bytes:
...
@ -451,4 +466,4 @@ class Decoder(_OpusStruct):
ret = _lib.opus_decode(self._state, data, len(data) if data else 0, pcm_ptr, frame_size, fec)
return array.array('h', pcm[:ret * channel_count]).tobytes()
return array.array("h", pcm[: ret * channel_count]).tobytes()