<s> import ctypes <EOL> import logging <EOL> import time <EOL> log = logging . getLogger ( __name__ ) <EOL> class Interface ( object ) : <EOL> def __init__ ( self , config , debug = False ) : <EOL> self . debug = bool ( debug ) <EOL> self . config = config <EOL> self . streams = [ ] <EOL> self . lib = None <EOL> def load ( self , name ) : <EOL> self . lib = ctypes . CDLL ( name ) <EOL> assert self . _error_string ( <NUM_LIT:0> ) == b'<STR_LIT>' <EOL> version = self . call ( '<STR_LIT>' , restype = ctypes . c_char_p ) <EOL> log . info ( '<STR_LIT>' , version ) <EOL> return self <EOL> def _error_string ( self , code ) : <EOL> return self . call ( '<STR_LIT>' , code , restype = ctypes . c_char_p ) <EOL> def call ( self , name , * args , ** kwargs ) : <EOL> assert self . lib is not None <EOL> func_name = '<STR_LIT>' . format ( name ) <EOL> if self . debug : <EOL> log . debug ( '<STR_LIT>' , name , args ) <EOL> func = getattr ( self . lib , func_name ) <EOL> func . restype = kwargs . get ( '<STR_LIT>' , self . _error_check ) <EOL> return func ( * args ) <EOL> def _error_check ( self , res ) : <EOL> if res != <NUM_LIT:0> : <EOL> raise Exception ( res , self . _error_string ( res ) ) <EOL> def __enter__ ( self ) : <EOL> self . call ( '<STR_LIT>' ) <EOL> return self <EOL> def __exit__ ( self , * args ) : <EOL> for s in self . streams : <EOL> s . close ( ) <EOL> self . call ( '<STR_LIT>' ) <EOL> def recorder ( self ) : <EOL> return Stream ( self , config = self . config , read = True ) <EOL> def player ( self ) : <EOL> return Stream ( self , config = self . config , write = True ) <EOL> class Stream ( object ) : <EOL> timer = time . time <EOL> class Parameters ( ctypes . Structure ) : <EOL> _fields_ = [ <EOL> ( '<STR_LIT>' , ctypes . c_int ) , <EOL> ( '<STR_LIT>' , ctypes . c_int ) , <EOL> ( '<STR_LIT>' , ctypes . c_ulong ) , <EOL> ( '<STR_LIT>' , ctypes . c_double ) , <EOL> ( '<STR_LIT>' , ctypes . POINTER ( None ) ) <EOL> ] <EOL> def __init__ ( self , interface , config , read = False , write = False ) : <EOL> self . interface = interface <EOL> self . stream = ctypes . POINTER ( ctypes . c_void_p ) ( ) <EOL> self . user_data = ctypes . c_void_p ( None ) <EOL> self . stream_callback = ctypes . c_void_p ( None ) <EOL> self . bytes_per_sample = config . sample_size <EOL> self . latency = float ( config . latency ) <EOL> self . bufsize = int ( self . latency * config . Fs * self . bytes_per_sample ) <EOL> assert config . bits_per_sample == <NUM_LIT:16> <EOL> read = bool ( read ) <EOL> write = bool ( write ) <EOL> assert read != write <EOL> direction = '<STR_LIT>' if read else '<STR_LIT>' <EOL> api_name = '<STR_LIT>' . format ( direction ) <EOL> index = interface . call ( api_name , restype = ctypes . c_int ) <EOL> self . params = Stream . Parameters ( <EOL> device = index , <EOL> channelCount = <NUM_LIT:1> , <EOL> sampleFormat = <NUM_LIT> , <EOL> suggestedLatency = self . latency , <EOL> hostApiSpecificStreamInfo = None ) <EOL> self . interface . call ( <EOL> '<STR_LIT>' , <EOL> ctypes . byref ( self . stream ) , <EOL> ctypes . byref ( self . params ) if read else None , <EOL> ctypes . byref ( self . params ) if write else None , <EOL> ctypes . c_double ( config . Fs ) , <EOL> ctypes . c_ulong ( <NUM_LIT:0> ) , <EOL> ctypes . c_ulong ( <NUM_LIT:0> ) , <EOL> self . stream_callback , <EOL> self . user_data ) <EOL> self . interface . streams . append ( self ) <EOL> self . interface . call ( '<STR_LIT>' , self . stream ) <EOL> self . start_time = self . timer ( ) <EOL> self . io_time = <NUM_LIT:0> <EOL> def close ( self ) : <EOL> if self . stream : <EOL> self . interface . call ( '<STR_LIT>' , self . stream ) <EOL> self . interface . call ( '<STR_LIT>' , self . stream ) <EOL> self . stream = None <EOL> def read ( self , size ) : <EOL> assert size % self . bytes_per_sample == <NUM_LIT:0> <EOL> buf = ctypes . create_string_buffer ( size ) <EOL> frames = ctypes . c_ulong ( size // self . bytes_per_sample ) <EOL> t0 = self . timer ( ) <EOL> self . interface . call ( '<STR_LIT>' , self . stream , buf , frames ) <EOL> t1 = self . timer ( ) <EOL> self . io_time += ( t1 - t0 ) <EOL> if self . interface . debug : <EOL> io_wait = self . io_time / ( t1 - self . start_time ) <EOL> log . debug ( '<STR_LIT>' , io_wait * <NUM_LIT:100> ) <EOL> return buf . raw <EOL> def write ( self , data ) : <EOL> data = bytes ( data ) <EOL> assert len ( data ) % self . bytes_per_sample == <NUM_LIT:0> <EOL> buf = ctypes . c_char_p ( data ) <EOL> frames = ctypes . c_ulong ( len ( data ) // self . bytes_per_sample ) <EOL> self . interface . call ( '<STR_LIT>' , self . stream , buf , frames ) </s>
