<s> """<STR_LIT>""" <EOL> from lantz import Feat , Action <EOL> from lantz . errors import InstrumentError <EOL> from lantz . foreign import LibraryDriver , RetValue , RetStr <EOL> from . constants import Constants , Types <EOL> default_buf_size = <NUM_LIT> <EOL> _SAMPLE_MODES = { '<STR_LIT>' : Constants . Val_FiniteSamps , <EOL> '<STR_LIT>' : Constants . Val_ContSamps , <EOL> '<STR_LIT>' : Constants . Val_HWTimedSinglePoint } <EOL> _BUS_TYPES = { '<STR_LIT>' : Constants . Val_PCI , '<STR_LIT>' : Constants . Val_PCIe , <EOL> '<STR_LIT>' : Constants . Val_PXI , '<STR_LIT>' : Constants . Val_SCXI , <EOL> '<STR_LIT>' : Constants . Val_PCCard , '<STR_LIT>' : Constants . Val_USB , <EOL> '<STR_LIT>' : Constants . Val_Unknown } <EOL> _SIGNAL_TYPES = { '<STR_LIT>' : Constants . Val_SampleClock , <EOL> '<STR_LIT>' : Constants . Val_SampleCompleteEvent , <EOL> '<STR_LIT>' : Constants . Val_ChangeDetectionEvent , <EOL> '<STR_LIT>' : Constants . Val_CounterOutputEvent } <EOL> _EDGE_TYPES = { '<STR_LIT>' : Constants . Val_Rising , '<STR_LIT>' : Constants . Val_Falling } <EOL> _SLOPE_TYPES = { '<STR_LIT>' : Constants . Val_RisingSlope , '<STR_LIT>' : Constants . Val_FallingSlope } <EOL> _WHEN_WINDOW = { '<STR_LIT>' : Constants . Val_EnteringWin , '<STR_LIT>' : Constants . Val_LeavingWin } <EOL> _WHEN_TRIGGER_DIG = { '<STR_LIT>' : Constants . Val_High , '<STR_LIT>' : Constants . Val_Low } <EOL> _WHEN_TRIGGER_ALVL = { '<STR_LIT>' : Constants . Val_AboveLvl , '<STR_LIT>' : Constants . Val_BelowLvl } <EOL> _WHEN_TRIGGER_AWIN = { '<STR_LIT>' : Constants . Val_InsideWin , '<STR_LIT>' : Constants . Val_OutsideWin } <EOL> _WHEN_MATCH = { True : Constants . Val_PatternMatches , False : Constants . Val_PatternDoesNotMatch } <EOL> _TRIGGER_TYPES = { '<STR_LIT>' : Constants . Val_DigLvl , <EOL> '<STR_LIT>' : Constants . Val_AnlgLvl , <EOL> '<STR_LIT>' : Constants . Val_AnlgWin } <EOL> _CHANNEL_TYPES = { '<STR_LIT>' : Constants . Val_AI , '<STR_LIT>' : Constants . Val_AO , <EOL> '<STR_LIT>' : Constants . Val_DI , '<STR_LIT>' : Constants . Val_DO , <EOL> '<STR_LIT>' : Constants . Val_CI , '<STR_LIT>' : Constants . Val_CO } <EOL> class _Base ( LibraryDriver ) : <EOL> """<STR_LIT>""" <EOL> LIBRARY_NAME = '<STR_LIT>' , '<STR_LIT>' <EOL> LIBRARY_PREFIX = '<STR_LIT>' <EOL> _DEVICES = { } <EOL> _TASKS = { } <EOL> _CHANNELS = { } <EOL> def _get_error_string ( self , error_code ) : <EOL> size = self . lib . GetErrorString ( error_code , None , <NUM_LIT:0> ) <EOL> if size <= <NUM_LIT:0> : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> err , msg = self . lib . GetErrorString ( error_code , * RetStr ( size ) ) <EOL> if err < <NUM_LIT:0> : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> return msg <EOL> def _get_error_extended_error_info ( self ) : <EOL> size = self . lib . GetExtendedErrorInfo ( None , <NUM_LIT:0> ) <EOL> if size <= <NUM_LIT:0> : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> err , msg = self . lib . GetExtendedErrorInfo ( * RetStr ( size ) ) <EOL> if err < <NUM_LIT:0> : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> return msg <EOL> def _return_handler ( self , func_name , ret_value ) : <EOL> if ret_value < <NUM_LIT:0> and func_name not in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> msg = self . _get_error_string ( ret_value ) <EOL> raise InstrumentError ( msg ) <EOL> return ret_value <EOL> def __get_fun ( self , name ) : <EOL> return getattr ( self . lib , name . format ( self . operation_direction . title ( ) ) ) <EOL> def _add_types ( self ) : <EOL> super ( ) . _add_types ( ) <EOL> T = Types <EOL> self . lib . CreateAIVoltageChan . argtypes = [ T . TaskHandle , T . string , T . string , T . int32 , T . float64 , T . float64 , T . int32 , T . string ] <EOL> self . lib . ReadAnalogScalarF64 . argtypes = [ T . TaskHandle , T . float64 , T . _ , T . _ ] <EOL> class _ObjectDict ( object ) : <EOL> def __init__ ( self , key_fun , obj_creator , dictionary = None ) : <EOL> self . key_fun = key_fun <EOL> self . obj_creator = obj_creator <EOL> self . _internal = { } if dictionary is None else dictionary <EOL> def __getitem__ ( self , item ) : <EOL> if item in self . _internal : <EOL> return self . _internal [ item ] <EOL> if item not in self : <EOL> raise KeyError ( '<STR_LIT>' . format ( item ) ) <EOL> value = self . obj_creator ( item ) <EOL> self . _internal [ item ] = value <EOL> return value <EOL> def __len__ ( self ) : <EOL> return sum ( ( <NUM_LIT:1> for key in self . keys ( ) ) ) <EOL> def __contains__ ( self , item ) : <EOL> return item in self . key_fun ( ) <EOL> def keys ( self ) : <EOL> for key in self . key_fun ( ) : <EOL> yield key <EOL> def values ( self ) : <EOL> for key in self . keys ( ) : <EOL> yield self [ key ] <EOL> def items ( self ) : <EOL> for key in self . keys ( ) : <EOL> yield key , self [ key ] <EOL> class System ( _Base ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( ) . __init__ ( * args , ** kwargs ) <EOL> self . devices = _ObjectDict ( self . _device_names , Device , self . _DEVICES ) <EOL> self . tasks = _ObjectDict ( self . _task_names , Task , self . _TASKS ) <EOL> self . channels = _ObjectDict ( self . _channel_names , Channel , self . _CHANNELS ) <EOL> @ Feat ( read_once = True ) <EOL> def version ( self ) : <EOL> """<STR_LIT>""" <EOL> err , major = self . lib . GetSysNIDAQMajorVersion ( RetValue ( '<STR_LIT>' ) ) <EOL> err , minor = self . lib . GetSysNIDAQMinorVersion ( RetValue ( '<STR_LIT>' ) ) <EOL> return major , minor <EOL> def _device_names ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetSysDevNames ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> def _task_names ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetSysTasks ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> def _channel_names ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetSysGlobalChans ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> class Device ( _Base ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , device_name , * args , ** kwargs ) : <EOL> super ( ) . __init__ ( * args , ** kwargs ) <EOL> self . device_name = device_name <EOL> def _preprocess_args ( self , name , * args ) : <EOL> """<STR_LIT>""" <EOL> if name in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> return super ( ) . _preprocess_args ( name , * args ) <EOL> else : <EOL> return super ( ) . _preprocess_args ( name , * ( ( self . device_name , ) + args ) ) <EOL> @ Feat ( read_once = True ) <EOL> def is_simulated ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevIsSimulated ( RetValue ( '<STR_LIT>' ) ) <EOL> return value != <NUM_LIT:0> <EOL> @ Feat ( read_once = True ) <EOL> def product_type ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevProductType ( * RetStr ( default_buf_size ) ) <EOL> return buf <EOL> @ Feat ( read_once = True ) <EOL> def product_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevProductNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def serial_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevSerialNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def analog_input_channels ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevAIPhysicalChans ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def analog_output_channels ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevAOPhysicalChans ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def digital_input_lines ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevDILines ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def digital_output_lines ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevDOLines ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def digital_input_ports ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevDIPorts ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def digital_output_ports ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevDOPorts ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def counter_input_channels ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevCIPhysicalChans ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True ) <EOL> def counter_output_channels ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetDevCOPhysicalChans ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( read_once = True , values = _BUS_TYPES ) <EOL> def bus_type ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevBusType ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def pci_bus_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevPCIBusNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def pci_device_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevPCIDevNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def pxi_slot_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevPXISlotNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def pxi_chassis_number ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDevPXIChassisNum ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def bus_info ( self ) : <EOL> t = self . bus_type <EOL> if t in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> return '<STR_LIT>' % ( t , self . pci_bus_number , self . pci_device_number ) <EOL> if t == '<STR_LIT>' : <EOL> return '<STR_LIT>' % ( t , self . pxi_chassis_number , self . pxi_slot_number ) <EOL> return t <EOL> @ Action ( ) <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . lib . ResetDevice ( ) <EOL> class Task ( _Base ) : <EOL> """<STR_LIT>""" <EOL> _REGISTRY = { } <EOL> @ classmethod <EOL> def register_class ( cls , klass ) : <EOL> cls . _REGISTRY [ klass . IO_TYPE ] = klass <EOL> @ classmethod <EOL> def typed_task ( cls , io_type ) : <EOL> return cls . _REGISTRY [ io_type ] <EOL> def _load_task ( self , name ) : <EOL> err , self . __task_handle = self . lib . LoadTask ( name , RetValue ( '<STR_LIT>' ) ) <EOL> self . name = name <EOL> self . log_debug ( '<STR_LIT>' . format ( self . name , self . __task_handle ) ) <EOL> def _create_task ( self , name ) : <EOL> err , self . __task_handle = self . lib . CreateTask ( name , RetValue ( '<STR_LIT>' ) ) <EOL> err , self . name = self . lib . GetTaskName ( * RetStr ( default_buf_size ) ) <EOL> self . log_debug ( '<STR_LIT>' . format ( self . name , self . __task_handle ) ) <EOL> def __init__ ( self , name = '<STR_LIT>' , * args , ** kwargs ) : <EOL> super ( ) . __init__ ( name , * args , ** kwargs ) <EOL> if not name : <EOL> self . _create_task ( name ) <EOL> else : <EOL> try : <EOL> self . _load_task ( name ) <EOL> except Exception as e : <EOL> self . _create_task ( name ) <EOL> self . sample_mode = None <EOL> self . channels = _ObjectDict ( self . _channel_names , self . _create_channel_from_name , self . _CHANNELS ) <EOL> self . devices = _ObjectDict ( self . _device_names , Device , self . _DEVICES ) <EOL> @ property <EOL> def task_handle ( self ) : <EOL> return self . __task_handle <EOL> def _preprocess_args ( self , name , * args ) : <EOL> """<STR_LIT>""" <EOL> if name in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> return super ( ) . _preprocess_args ( name , * args ) <EOL> else : <EOL> return super ( ) . _preprocess_args ( name , * ( ( self . task_handle , ) + args ) ) <EOL> def _create_channel_from_name ( self , name ) : <EOL> return Channel ( self , name = name ) <EOL> def _channel_names ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetTaskChannels ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> def _device_names ( self ) : <EOL> """<STR_LIT>""" <EOL> err , buf = self . lib . GetTaskDevices ( * RetStr ( default_buf_size ) ) <EOL> names = tuple ( n . strip ( ) for n in buf . split ( '<STR_LIT:U+002C>' ) if n . strip ( ) ) <EOL> return names <EOL> @ Feat ( ) <EOL> def io_type ( self ) : <EOL> for name , channel in self . channels . items ( ) : <EOL> return channel . io_type <EOL> else : <EOL> return None <EOL> def add_channel ( self , channel ) : <EOL> if not isinstance ( channel , Channel ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if channel . task is self . channels : <EOL> return <EOL> elif channel . task is None : <EOL> channel . task = self <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def execute_fun ( self , func_name , * args ) : <EOL> return getattr ( self . lib , func_name ) ( * args ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . task_handle : <EOL> self . lib . ClearTask ( ) <EOL> self . __task_handle = None <EOL> __del__ = clear <EOL> @ Feat ( ) <EOL> def is_done ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . IsTaskDone ( RetValue ( '<STR_LIT>' ) ) <EOL> return value != <NUM_LIT:0> <EOL> @ Action ( ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . lib . StartTask ( ) <EOL> @ Action ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . lib . StopTask ( ) <EOL> @ Action ( ) <EOL> def verify ( self ) : <EOL> """<STR_LIT>""" <EOL> self . alter_state ( '<STR_LIT>' ) <EOL> @ Action ( ) <EOL> def commit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . alter_state ( '<STR_LIT>' ) <EOL> @ Action ( ) <EOL> def reserve ( self ) : <EOL> """<STR_LIT>""" <EOL> self . alter_state ( '<STR_LIT>' ) <EOL> @ Action ( ) <EOL> def unreserve ( self ) : <EOL> """<STR_LIT>""" <EOL> self . alter_state ( '<STR_LIT>' ) <EOL> @ Action ( ) <EOL> def abort ( self ) : <EOL> """<STR_LIT>""" <EOL> self . alter_state ( '<STR_LIT>' ) <EOL> @ Action ( values = { '<STR_LIT:start>' : Constants . Val_Task_Start , '<STR_LIT>' : Constants . Val_Task_Stop , <EOL> '<STR_LIT>' : Constants . Val_Task_Verify , '<STR_LIT>' : Constants . Val_Task_Commit , <EOL> '<STR_LIT>' : Constants . Val_Task_Reserve , '<STR_LIT>' : Constants . Val_Task_Unreserve , <EOL> '<STR_LIT>' : Constants . Val_Task_Abort } ) <EOL> def alter_state ( self , new_state ) : <EOL> """<STR_LIT>""" <EOL> self . lib . TaskControl ( new_state ) <EOL> _register_every_n_samples_event_cache = None <EOL> def register_every_n_samples_event ( self , func , samples = <NUM_LIT:1> , options = <NUM_LIT:0> , cb_data = None ) : <EOL> """<STR_LIT>""" <EOL> if self . operation_direction == '<STR_LIT:input>' : <EOL> event_type = Constants . Val_Acquired_Into_Buffer <EOL> else : <EOL> event_type = Constants . Val_Transferred_From_Buffer <EOL> if options == '<STR_LIT>' : <EOL> options = Constants . Val_SynchronousEventCallbacks <EOL> if func is None : <EOL> c_func = None <EOL> else : <EOL> if self . _register_every_n_samples_event_cache is not None : <EOL> self . register_every_n_samples_event ( None , samples = samples , options = options , cb_data = cb_data ) <EOL> c_func = EveryNSamplesEventCallback_map [ self . CHANNEL_TYPE ] ( func ) <EOL> self . _register_every_n_samples_event_cache = c_func <EOL> self . lib . RegisterEveryNSamplesEvent ( event_type , uInt32 ( samples ) , uInt32 ( options ) , c_func , cb_data ) <EOL> _register_done_event_cache = None <EOL> def register_done_event ( self , func , options = <NUM_LIT:0> , cb_data = None ) : <EOL> """<STR_LIT>""" <EOL> if options == '<STR_LIT>' : <EOL> options = Constants . Val_SynchronousEventCallbacks <EOL> if func is None : <EOL> c_func = None <EOL> else : <EOL> if self . _register_done_event_cache is not None : <EOL> self . register_done_event ( None , options = options , cb_data = cb_data ) <EOL> c_func = DoneEventCallback_map [ self . CHANNEL_TYPE ] ( func ) <EOL> self . _register_done_event_cache = c_func <EOL> self . lib . RegisterDoneEvent ( uInt32 ( options ) , c_func , cb_data ) <EOL> def operation_direction ( self ) : <EOL> return '<STR_LIT:input>' if self . CHANNEL_TYPE else '<STR_LIT>' <EOL> _register_signal_event_cache = None <EOL> @ Action ( values = ( None , _SIGNAL_TYPES , None , None ) ) <EOL> def register_signal_event ( self , func , signal , options = <NUM_LIT:0> , cb_data = None ) : <EOL> """<STR_LIT>""" <EOL> if options == '<STR_LIT>' : <EOL> options = Constants . Val_SynchronousEventCallbacks <EOL> if func is None : <EOL> c_func = None <EOL> else : <EOL> if self . _register_signal_event_cache is not None : <EOL> self . _register_signal_event ( None , signal = signal , options = options , cb_data = cb_data ) <EOL> c_func = SignalEventCallback_map [ self . CHANNEL_TYPE ] ( func ) <EOL> self . _register_signal_event_cache = c_func <EOL> self . lib . RegisterSignalEvent ( signal , uInt32 ( options ) , c_func , cb_data ) <EOL> @ Action ( values = ( str , str , _SAMPLE_MODES , None ) ) <EOL> def configure_timing_change_detection ( self , rising_edge_channel = '<STR_LIT>' , falling_edge_channel = '<STR_LIT>' , <EOL> sample_mode = '<STR_LIT>' , samples_per_channel = <NUM_LIT:1000> ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgChangeDetectionTiming ( rising_edge_channel , falling_edge_channel , <EOL> sample_mode , samples_per_channel ) <EOL> @ Action ( values = ( _SAMPLE_MODES , None ) ) <EOL> def configure_timing_handshaking ( self , sample_mode = '<STR_LIT>' , samples_per_channel = <NUM_LIT:1000> ) : <EOL> """<STR_LIT>""" <EOL> self . samples_per_channel = samples_per_channel <EOL> self . sample_mode = sample_mode <EOL> self . lib . CfgHandshakingTiming ( sample_mode , samples_per_channel ) <EOL> @ Action ( values = ( _SAMPLE_MODES , None ) ) <EOL> def configure_timing_implicit ( self , sample_mode = '<STR_LIT>' , samples_per_channel = <NUM_LIT:1000> ) : <EOL> """<STR_LIT>""" <EOL> self . samples_per_channel = samples_per_channel <EOL> self . sample_mode = sample_mode <EOL> self . lib . CfgImplicitTiming ( self , sample_mode , samples_per_channel ) <EOL> @ Action ( values = ( str , None , _EDGE_TYPES , _SAMPLE_MODES , None ) ) <EOL> def configure_timing_sample_clock ( self , source = '<STR_LIT>' , rate = <NUM_LIT:1> , active_edge = '<STR_LIT>' , <EOL> sample_mode = '<STR_LIT>' , samples_per_channel = <NUM_LIT:1000> ) : <EOL> """<STR_LIT>""" <EOL> if source == '<STR_LIT>' : <EOL> source = None <EOL> self . samples_per_channel = samples_per_channel <EOL> self . sample_mode = sample_mode <EOL> self . lib . CfgSampClkTiming ( source , float64 ( rate ) , active_edge , sample_mode , samples_per_channel ) <EOL> def configure_timing_burst_handshaking_export_clock ( self , * args , ** kws ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def configure_timing_burst_handshaking_import_clock ( self , * args , ** kws ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ Action ( values = ( str , _SLOPE_TYPES , None ) ) <EOL> def configure_trigger_analog_edge_start ( self , source , slope = '<STR_LIT>' , level = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgAnlgEdgeStartTrig ( source , slope , level ) <EOL> @ Action ( values = ( str , _WHEN_WINDOW , None , None ) ) <EOL> def configure_trigger_analog_window_start ( self , source , when = '<STR_LIT>' , top = <NUM_LIT:1.0> , bottom = - <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgAnlgWindowStartTrig ( self , source , when , top , bottom ) <EOL> @ Action ( values = ( str , _EDGE_TYPES ) ) <EOL> def configure_trigger_digital_edge_start ( self , source , edge = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgDigEdgeStartTrig ( self , source , edge ) <EOL> @ Action ( values = ( str , str , _WHEN_MATCH ) ) <EOL> def configure_trigger_digital_pattern_start ( self , source , pattern , match = True ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgDigPatternStartTrig ( self , source , pattern , match ) <EOL> def configure_trigger_disable_start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . lib . DisableStartTrig ( ) <EOL> @ Action ( values = ( str , _SLOPE_TYPES , None , None ) ) <EOL> def configure_analog_edge_reference_trigger ( self , source , slope = '<STR_LIT>' , level = <NUM_LIT:1.0> , pre_trigger_samps = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgAnlgEdgeRefTrig ( source , slope , level , pre_trigger_samps ) <EOL> @ Feat ( values = ( str , _WHEN_WINDOW , None , None , None ) ) <EOL> def configure_analog_window_reference_trigger ( self , source , when = '<STR_LIT>' , top = <NUM_LIT:1.0> , bottom = <NUM_LIT:1.0> , pre_trigger_samps = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . lib . CfgAnlgWindowRefTrig ( source , when , top , bottom , pre_trigger_samps ) <EOL> @ Action ( values = ( str , _SLOPE_TYPES , None ) ) <EOL> def configure_digital_edge_reference_trigger ( self , source , slope = '<STR_LIT>' , pre_trigger_samps = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if not source . startswith ( '<STR_LIT:/>' ) : <EOL> source = '<STR_LIT:/>' + source <EOL> self . lib . CfgDigEdgeRefTrig ( source , slope , pre_trigger_samps ) <EOL> @ Action ( values = ( str , str , _WHEN_MATCH , None ) ) <EOL> def configure_digital_pattern_reference_trigger ( self , source , pattern , match = True , pre_trigger_samps = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if not source . startswith ( '<STR_LIT:/>' ) : <EOL> source = '<STR_LIT:/>' + source <EOL> self . lib . CfgDigPatternRefTrig ( self , source , pattern , match , pre_trigger_samps ) <EOL> @ Action ( ) <EOL> def disable_reference_trigger ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . lib . DisableRefTrig ( self ) == <NUM_LIT:0> <EOL> def set_buffer ( self , samples_per_channel ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ Feat ( units = '<STR_LIT>' ) <EOL> def sample_clock_rate ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetSampClkRate ( self , RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ sample_clock_rate . setter <EOL> def sample_clock_rate ( self , value ) : <EOL> if value is None : <EOL> self . lib . ResetSampClkRate ( ) <EOL> else : <EOL> self . lib . SetSampClkRate ( value ) <EOL> @ Feat ( ) <EOL> def convert_clock_rate ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetAIConvRate ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ convert_clock_rate . setter <EOL> def convert_clock_rate ( self , value ) : <EOL> if value is None : <EOL> self . lib . ResetAIConvRate ( ) <EOL> else : <EOL> self . lib . SetAIConvRate ( value ) <EOL> def sample_clock_max_rate ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetSampClkMaxRate ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( values = { True : Constants . Val_AllowRegen , False : Constants . Val_DoNotAllowRegen } ) <EOL> def regeneration_enabled ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetWriteRegenMode ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ regeneration_enabled . setter <EOL> def regeneration_enabled ( self , value ) : <EOL> if value is None : <EOL> self . lib . ResetWriteRegenMode ( ) <EOL> else : <EOL> self . lib . SetWriteRegenMode ( value ) <EOL> @ Feat ( values = { '<STR_LIT>' : Constants . Val_DigEdge , '<STR_LIT>' : Constants . Val_None , None : None } ) <EOL> def arm_start_trigger_type ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetArmStartTrigType ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ arm_start_trigger_type . setter <EOL> def arm_start_trigger_type ( self , trigger_type ) : <EOL> if trigger_type is None : <EOL> self . lib . ResetArmStartTrigType ( ) <EOL> else : <EOL> self . lib . SetArmStartTrigType ( trigger_type ) <EOL> @ Feat ( ) <EOL> def arm_start_trigger_source ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDigEdgeArmStartTrigSrc ( RetStr ( default_buf_size ) ) <EOL> return value <EOL> @ arm_start_trigger_source . setter <EOL> @ arm_start_trigger_source . setter <EOL> def arm_start_trigger_source ( self , source ) : <EOL> source = str ( source ) <EOL> if source is None : <EOL> self . lib . ResetDigEdgeArmStartTrigSrc ( ) <EOL> else : <EOL> self . lib . SetDigEdgeArmStartTrigSrc ( source ) <EOL> @ Feat ( values = { None : None } . update ( _EDGE_TYPES ) ) <EOL> def arm_start_trigger_edge ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetDigEdgeArmStartTrigEdge ( RetValue ( '<STR_LIT>' ) ) <EOL> @ arm_start_trigger_edge . setter <EOL> def arm_start_trigger_edge ( self , edge ) : <EOL> if edge is None : <EOL> self . lib . ResetDigEdgeArmStartTrigEdge ( ) <EOL> else : <EOL> self . lib . SetDigEdgeArmStartTrigEdge ( edge ) <EOL> @ Feat ( values = { None : None } . update ( _TRIGGER_TYPES ) ) <EOL> def pause_trigger_type ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetPauseTrigType ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ pause_trigger_type . setter <EOL> def pause_trigger_type ( self , trigger_type ) : <EOL> if trigger_type is None : <EOL> self . lib . ResetPauseTrigType ( ) <EOL> else : <EOL> self . lib . SetPauseTrigType ( trigger_type ) <EOL> @ Feat ( ) <EOL> def pause_trigger_source ( self ) : <EOL> """<STR_LIT>""" <EOL> type = self . pause_trigger_type <EOL> if type == '<STR_LIT>' : <EOL> fun = self . lib . GetDigLvlPauseTrigSrc <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . GetAnlgLvlPauseTrigSrc <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . GetAnlgWinPauseTrigSrc <EOL> else : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> err , value = fun ( * RetStr ( default_buf_size ) ) <EOL> return value <EOL> @ pause_trigger_source . setter <EOL> def pause_trigger_source ( self , source ) : <EOL> type = self . pause_trigger_type <EOL> if type == '<STR_LIT>' : <EOL> fun = self . lib . SetDigLvlPauseTrigSrc <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgLvlPauseTrigSrc <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgWinPauseTrigSrc <EOL> else : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> fun ( source ) <EOL> @ Feat ( ) <EOL> def pause_trigger_when ( self ) : <EOL> """<STR_LIT>""" <EOL> type = self . pause_trigger_type <EOL> if type == '<STR_LIT>' : <EOL> fun = self . lib . SetDigLvlPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_DIG <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgLvlPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_ALVL <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgWinPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_AWIN <EOL> else : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> err , val = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> for key , value in convert . items ( ) : <EOL> if key == val : <EOL> return val <EOL> else : <EOL> raise ValueError ( val ) <EOL> @ pause_trigger_when . setter <EOL> def pause_trigger_when ( self , when = None ) : <EOL> if when is None : <EOL> self . lib . ResetDigLvlPauseTrigWhen ( ) <EOL> return <EOL> type = self . pause_trigger_type <EOL> if type == '<STR_LIT>' : <EOL> fun = self . lib . SetDigLvlPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_DIG <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgLvlPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_ALVL <EOL> elif type == '<STR_LIT>' : <EOL> fun = self . lib . SetAnlgWinPauseTrigWhen <EOL> convert = _WHEN_TRIGGER_AWIN <EOL> else : <EOL> raise InstrumentError ( '<STR_LIT>' ) <EOL> fun ( convert [ when ] ) <EOL> def read_current_position ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetReadCurrReadPos ( * RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> def samples_per_channel_available ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetReadAvailSampPerChan ( * RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> def samples_per_channel_acquired ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetReadTotalSampPerChanAcquired ( * RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Action ( units = '<STR_LIT>' ) <EOL> def wait_until_done ( self , timeout = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if timeout < <NUM_LIT:0> : <EOL> timeout = Constants . Val_WaitInfinitely <EOL> return self . lib . WaitUntilTaskDone ( timeout ) <EOL> class Channel ( _Base ) : <EOL> """<STR_LIT>""" <EOL> IO_TYPE = None <EOL> def __init__ ( self , task = None , * args , ** kwargs ) : <EOL> super ( ) . __init__ ( * args , ** kwargs ) <EOL> self . _task = None <EOL> if task == '<STR_LIT>' : <EOL> task = Task . typed_task ( self . IO_TYPE ) ( ) <EOL> print ( task ) <EOL> self . task = task <EOL> @ property <EOL> def task ( self ) : <EOL> return self . _task <EOL> @ task . setter <EOL> def task ( self , value ) : <EOL> if not self . _task is None : <EOL> raise Exception <EOL> self . _task = value <EOL> if value is not None : <EOL> self . log_debug ( '<STR_LIT>' . format ( self . CREATE_FUN , self . _create_args ) ) <EOL> value . execute_fun ( self . CREATE_FUN , * self . _create_args ) <EOL> def _preprocess_args ( self , name , * args ) : <EOL> """<STR_LIT>""" <EOL> return super ( ) . _preprocess_args ( name , * ( ( self . task . task_handle , self . name ) + args ) ) <EOL> @ Feat ( read_once = True , values = _CHANNEL_TYPES ) <EOL> def io_type ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetChanType ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( read_once = True ) <EOL> def physical_channel_name ( self , ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetPhysicalChanName ( * RetStr ( default_buf_size ) ) <EOL> return value <EOL> def operation_direction ( self ) : <EOL> return '<STR_LIT:input>' if self . CHANNEL_TYPE else '<STR_LIT>' <EOL> @ Feat ( read_once = True ) <EOL> def is_global ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetChanIsGlobal ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( ) <EOL> def buffer_size_on_board ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ Feat ( ) <EOL> def buffer_size ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ buffer_size . setter <EOL> def buffer_size ( self , size ) : <EOL> if size is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( size ) <EOL> @ Feat ( ) <EOL> def max ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ max . setter <EOL> def max ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( ) <EOL> def min ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ min . setter <EOL> def min ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( ) <EOL> def range_high ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ range_high . setter <EOL> def range_high ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( ) <EOL> def range_low ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ range_low . setter <EOL> def range_low ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( ) <EOL> def gain ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ gain . setter <EOL> def gain ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( ) <EOL> def number_of_lines ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . __get_fun ( '<STR_LIT>' ) ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> def get_units ( self , channel_name ) : <EOL> """<STR_LIT>""" <EOL> channel_name = str ( channel_name ) <EOL> mt = self . get_measurment_type ( channel_name ) <EOL> channel_type = self . channel_type <EOL> if mt == '<STR_LIT>' : <EOL> d = int32 ( <NUM_LIT:0> ) <EOL> CALL ( '<STR_LIT>' % ( channel_type ) , self , channel_name , ctypes . byref ( d ) ) <EOL> units_map = { Contants . Val_Volts : '<STR_LIT>' , <EOL> } <EOL> return units_map [ d . value ] <EOL> raise NotImplementedError ( '<STR_LIT>' . format ( channel_name , mt ) ) <EOL> @ Feat ( values = { '<STR_LIT:none>' : Constants . Val_None , '<STR_LIT>' : Constants . Val_Once , <EOL> '<STR_LIT>' : Constants . Val_EverySample } ) <EOL> def auto_zero_mode ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ auto_zero_mode . setter <EOL> def auto_zero_mode ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( values = { '<STR_LIT>' : Constants . Val_DMA , '<STR_LIT>' : Constants . Val_Interrupts , <EOL> '<STR_LIT>' : Constants . Val_ProgrammedIO , <EOL> '<STR_LIT>' : Constants . Val_USBbulk } ) <EOL> def data_transfer_mode ( self ) : <EOL> """<STR_LIT>""" <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err , value = fun ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ data_transfer_mode . setter <EOL> def data_transfer_mode ( self , value ) : <EOL> if value is None : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( ) <EOL> else : <EOL> fun = self . __get_fun ( '<STR_LIT>' ) <EOL> err = fun ( value ) <EOL> @ Feat ( values = { True : <NUM_LIT:1> , False : <NUM_LIT:0> , None : None } ) <EOL> def duplicate_count_prevention_enabled ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetCIDupCountPrevent ( RetValue ( '<STR_LIT>' ) ) <EOL> return value != <NUM_LIT:0> <EOL> @ duplicate_count_prevention_enabled . setter <EOL> def duplicate_count_prevention_enabled ( self , value ) : <EOL> if value is None : <EOL> err = self . lib . ResetCIDupCountPrevent ( ) <EOL> else : <EOL> err = self . lib . SetCIDupCountPrevent ( value ) <EOL> @ Feat ( ) <EOL> def timebase_rate ( self ) : <EOL> """<STR_LIT>""" <EOL> err , value = self . lib . GetCICtrTimebaseRate ( RetValue ( '<STR_LIT>' ) ) <EOL> return value <EOL> @ timebase_rate . setter <EOL> def timebase_rate ( self , value ) : <EOL> if value is None : <EOL> err = self . lib . ResetCICtrTimebaseRate ( ) <EOL> else : <EOL> err = self . lib . SetCICtrTimebaseRate ( value ) <EOL> @ Feat ( None ) <EOL> def terminal_pulse ( self , terminal ) : <EOL> """<STR_LIT>""" <EOL> self . lib . SetCOPulseTerm ( terminal ) <EOL> @ Feat ( None ) <EOL> def terminal_count_edges ( self , channel , terminal ) : <EOL> """<STR_LIT>""" <EOL> self . lib . SetCICountEdgesTerm ( terminal ) <EOL> """<STR_LIT>""" <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import lantz . log <EOL> lantz . log . log_to_screen ( lantz . log . DEBUG ) <EOL> inst = System ( ) <EOL> print ( inst . version ) <EOL> print ( inst . device_names ) </s>
