<s> """<STR_LIT>""" <EOL> import ctypes <EOL> import ctypes . util <EOL> import array <EOL> from . errors import DiscordException <EOL> import logging <EOL> log = logging . getLogger ( __name__ ) <EOL> c_int_ptr = ctypes . POINTER ( ctypes . c_int ) <EOL> c_int16_ptr = ctypes . POINTER ( ctypes . c_int16 ) <EOL> c_float_ptr = ctypes . POINTER ( ctypes . c_float ) <EOL> class EncoderStruct ( ctypes . Structure ) : <EOL> pass <EOL> EncoderStructPtr = ctypes . POINTER ( EncoderStruct ) <EOL> exported_functions = [ <EOL> ( '<STR_LIT>' , [ ctypes . c_int ] , ctypes . c_char_p ) , <EOL> ( '<STR_LIT>' , [ ctypes . c_int ] , ctypes . c_int ) , <EOL> ( '<STR_LIT>' , [ ctypes . c_int , ctypes . c_int , ctypes . c_int , c_int_ptr ] , EncoderStructPtr ) , <EOL> ( '<STR_LIT>' , [ EncoderStructPtr , c_int16_ptr , ctypes . c_int , ctypes . c_char_p , ctypes . c_int32 ] , ctypes . c_int32 ) , <EOL> ( '<STR_LIT>' , None , ctypes . c_int32 ) , <EOL> ( '<STR_LIT>' , [ EncoderStructPtr ] , None ) <EOL> ] <EOL> def libopus_loader ( name ) : <EOL> lib = ctypes . cdll . LoadLibrary ( name ) <EOL> for item in exported_functions : <EOL> try : <EOL> func = getattr ( lib , item [ <NUM_LIT:0> ] ) <EOL> except Exception as e : <EOL> raise e <EOL> try : <EOL> if item [ <NUM_LIT:1> ] : <EOL> func . argtypes = item [ <NUM_LIT:1> ] <EOL> func . restype = item [ <NUM_LIT:2> ] <EOL> except KeyError : <EOL> pass <EOL> return lib <EOL> try : <EOL> _lib = libopus_loader ( ctypes . util . find_library ( '<STR_LIT>' ) ) <EOL> except : <EOL> _lib = None <EOL> def load_opus ( name ) : <EOL> """<STR_LIT>""" <EOL> global _lib <EOL> _lib = libopus_loader ( name ) <EOL> def is_loaded ( ) : <EOL> """<STR_LIT>""" <EOL> global _lib <EOL> return _lib is not None <EOL> class OpusError ( DiscordException ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , code ) : <EOL> self . code = code <EOL> msg = _lib . opus_strerror ( self . code ) . decode ( '<STR_LIT:utf-8>' ) <EOL> log . info ( '<STR_LIT>' . format ( msg ) ) <EOL> super ( ) . __init__ ( msg ) <EOL> class OpusNotLoaded ( DiscordException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> OK = <NUM_LIT:0> <EOL> APPLICATION_AUDIO = <NUM_LIT> <EOL> APPLICATION_VOIP = <NUM_LIT> <EOL> APPLICATION_LOWDELAY = <NUM_LIT> <EOL> CTL_SET_BITRATE = <NUM_LIT> <EOL> CTL_SET_BANDWIDTH = <NUM_LIT> <EOL> band_ctl = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> class Encoder : <EOL> def __init__ ( self , sampling , channels , application = APPLICATION_AUDIO ) : <EOL> self . sampling_rate = sampling <EOL> self . channels = channels <EOL> self . application = application <EOL> self . frame_length = <NUM_LIT:20> <EOL> self . sample_size = <NUM_LIT:2> * self . channels <EOL> self . samples_per_frame = int ( self . sampling_rate / <NUM_LIT:1000> * self . frame_length ) <EOL> self . frame_size = self . samples_per_frame * self . sample_size <EOL> if not is_loaded ( ) : <EOL> raise OpusNotLoaded ( ) <EOL> self . _state = self . _create_state ( ) <EOL> self . set_bitrate ( <NUM_LIT> ) <EOL> self . set_bandwidth ( '<STR_LIT>' ) <EOL> def __del__ ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> _lib . opus_encoder_destroy ( self . _state ) <EOL> self . _state = None <EOL> def _create_state ( self ) : <EOL> ret = ctypes . c_int ( ) <EOL> result = _lib . opus_encoder_create ( self . sampling_rate , self . channels , self . application , ctypes . byref ( ret ) ) <EOL> if ret . value != <NUM_LIT:0> : <EOL> log . info ( '<STR_LIT>' ) <EOL> raise OpusError ( ret . value ) <EOL> return result <EOL> def set_bitrate ( self , kbps ) : <EOL> kbps = min ( <NUM_LIT> , max ( <NUM_LIT:16> , int ( kbps ) ) ) <EOL> ret = _lib . opus_encoder_ctl ( self . _state , CTL_SET_BITRATE , kbps * <NUM_LIT> ) <EOL> if ret < <NUM_LIT:0> : <EOL> log . info ( '<STR_LIT>' ) <EOL> raise OpusError ( ret ) <EOL> return kbps <EOL> def set_bandwidth ( self , req ) : <EOL> if req not in band_ctl : <EOL> raise KeyError ( '<STR_LIT>' % ( req , '<STR_LIT:U+002C>' . join ( band_ctl ) ) ) <EOL> k = band_ctl [ req ] <EOL> ret = _lib . opus_encoder_ctl ( self . _state , CTL_SET_BANDWIDTH , k ) <EOL> if ret < <NUM_LIT:0> : <EOL> log . info ( '<STR_LIT>' ) <EOL> raise OpusError ( ret ) <EOL> def encode ( self , pcm , frame_size ) : <EOL> max_data_bytes = len ( pcm ) <EOL> pcm = ctypes . cast ( pcm , c_int16_ptr ) <EOL> data = ( ctypes . c_char * max_data_bytes ) ( ) <EOL> ret = _lib . opus_encode ( self . _state , pcm , frame_size , data , max_data_bytes ) <EOL> if ret < <NUM_LIT:0> : <EOL> log . info ( '<STR_LIT>' ) <EOL> raise OpusError ( ret ) <EOL> return array . array ( '<STR_LIT:b>' , data [ : ret ] ) . tobytes ( ) </s>
