diff --git a/README.rst b/README.rst
index a86176f..a77ef4c 100644
--- a/README.rst
+++ b/README.rst
@@ -71,7 +71,7 @@ do not share their instrument plugins (``NIDAQmx`` here, ``DAQmx_...`` there).
* - Python interface
- `nidaqmx `_, the package of National Instruments
(``NIDAQmx`` object in ``daqmxni.py``)
- - `PyDAQmx `_ (``DAQmx`` object in ``hardware/national_instruments/daqmx.py``)
+ - `PyDAQmx `_ (``DAQmx`` object in ``hardware/national_instruments/daqmx.py`` of that branch)
* - Instruments
- ``NIDAQmx`` actuator, 0D viewer and 1D viewer
- generic ``DAQmx`` actuator and viewers (0D, 1D), ``DAQmxAI``, ``DAQmxDualAI``, ``DAQmx_PLcounter``,
@@ -85,8 +85,10 @@ do not share their instrument plugins (``NIDAQmx`` here, ``DAQmx_...`` there).
``NIDAQmx`` actuator is not finished yet (its movements still raise ``NotImplementedError``)
- the ``PyDAQmx`` based plugins, kept as they are
-This branch still contains the ``PyDAQmx`` based ``DAQmx`` object and the user guides of the photon counter and the
-scanner control, but not the instrument plugins of the ``pydaqmx`` branch: use that branch if you need them.
+This branch does not contain the ``PyDAQmx`` based ``DAQmx`` object (``daqmx.py``, ``daqmx_objects.py``) nor the
+instrument plugins of the ``pydaqmx`` branch: use that branch (or the 1.1.x releases) if you need them. The
+user guides of the photon counter and the scanner control remain here as a reference, they describe those deprecated
+plugins.
Requirements
============
@@ -94,4 +96,4 @@ Requirements
* PyMoDAQ >= 5.3.7
* the NI-DAQmx driver of National Instruments, installed on the computer connected to the device (use NI MAX to test
your hardware). Where the driver cannot be found, the modules of this plugin are not usable and PyMoDAQ skips them.
-* the ``nidaqmx`` and ``pydaqmx`` Python packages, installed with the plugin.
+* the ``nidaqmx`` Python package, installed with the plugin.
diff --git a/pyproject.toml b/pyproject.toml
index d1e9f24..7cee548 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,7 +16,6 @@ dependencies = [
"pymodaq_utils>=5.3.1",
"pymodaq_gui>=5.3.0",
"pymodaq_data>=5.3.0",
- "pydaqmx",
"nidaqmx",
"aenum",
]
diff --git a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daq_NIDAQmx.py b/src/pymodaq_plugins_daqmx/hardware/national_instruments/daq_NIDAQmx.py
deleted file mode 100644
index 003479e..0000000
--- a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daq_NIDAQmx.py
+++ /dev/null
@@ -1,862 +0,0 @@
-from qtpy import QtWidgets, QtCore
-from qtpy.QtCore import Signal, QThread
-from pymodaq.utils.daq_utils import ThreadCommand, getLineInfo
-from pymodaq.utils.data import DataFromPlugins, Axis, DataActuator, DataToExport
-import numpy as np
-from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base
-from pymodaq.control_modules.move_utility_classes import DAQ_Move_base
-from easydict import EasyDict as edict
-
-from pymodaq.control_modules.viewer_utility_classes import comon_parameters as viewer_params
-from pymodaq.control_modules.move_utility_classes import comon_parameters_fun as actuator_params
-
-from pymodaq.utils.parameter import Parameter
-from pymodaq.utils.parameter.pymodaq_ptypes import registerParameterType, GroupParameter
-
-from .daqmx import DAQmx, DAQ_analog_types, DAQ_thermocouples, DAQ_termination, Edge, DAQ_NIDAQ_source, \
- ClockSettings, AIChannel, Counter, AIThermoChannel, AOChannel, TriggerSettings, DOChannel, DIChannel
-
-
-class ScalableGroupAI(GroupParameter):
- """
- |
-
- ================ =============
- **Attributes** **Type**
- *opts* dictionnary
- ================ =============
-
- See Also
- --------
- hardware.DAQ_Move_Stage_type
- """
-
- params = [{'title': 'AI type:', 'name': 'ai_type', 'type': 'list', 'limits': DAQ_analog_types.names()},
- {'title': 'Voltages:', 'name': 'voltage_settings', 'type': 'group', 'children': [
- {'title': 'Voltage Min:', 'name': 'volt_min', 'type': 'list', 'value': -10.},
- {'title': 'Voltage Max:', 'name': 'volt_max', 'type': 'list', 'value': 10.},
- ]},
- {'title': 'Current:', 'name': 'current_settings', 'type': 'group', 'visible': False, 'children': [
- {'title': 'Current Min:', 'name': 'curr_min', 'type': 'float', 'value': -1, 'suffix': 'A'},
- {'title': 'Current Max:', 'name': 'curr_max', 'type': 'float', 'value': 1, 'suffix': 'A'},
- ]},
- {'title': 'Thermocouple:', 'name': 'thermoc_settings', 'type': 'group', 'visible': False, 'children': [
- {'title': 'Thc. type:', 'name': 'thermoc_type', 'type': 'list', 'limits': DAQ_thermocouples.names(),
- 'value': 'K'},
- {'title': 'Temp. Min (°C):', 'name': 'T_min', 'type': 'float', 'value': 0, 'suffix': '°C'},
- {'title': 'Temp. Max (°C):', 'name': 'T_max', 'type': 'float', 'value': 50, 'suffix': '°C'},
- ]},
- {'title': 'Termination:', 'name': 'termination', 'type': 'list', 'limits': DAQ_termination.names()},
- ]
-
- def __init__(self, **opts):
- opts['type'] = 'groupai'
- opts['addText'] = "Add"
- opts['addList'] = opts['limits']
- GroupParameter.__init__(self, **opts)
-
- def addNew(self, typ):
- """
- Add a child.
-
- =============== ===========
- **Parameters** **Type**
- *typ* string
- =============== ===========
- """
- childnames = [par.name() for par in self.children()]
- if childnames == []:
- newindex = 0
- else:
- newindex = len(childnames)
-
- child = {'title': typ, 'name': 'ai{:02.0f}'.format(newindex), 'type': 'group', 'children': self.params,
- 'removable': True, 'renamable': False}
-
- self.addChild(child)
-registerParameterType('groupai', ScalableGroupAI, override=True)
-
-
-class ScalableGroupAO(GroupParameter):
- """
- |
-
- ================ =============
- **Attributes** **Type**
- *opts* dictionnary
- ================ =============
-
- See Also
- --------
- hardware.DAQ_Move_Stage_type
- """
-
- params = [{'title': 'AO type:', 'name': 'ao_type', 'type': 'list', 'limits': DAQ_analog_types.names()[0:2]},
- {'title': 'Voltages:', 'name': 'voltage_settings', 'type': 'group', 'children': [
- {'title': 'Voltage Min:', 'name': 'volt_min', 'type': 'list', 'value': -10., },
- {'title': 'Voltage Max:', 'name': 'volt_max', 'type': 'list', 'value': 10., },
- ]},
- {'title': 'Current:', 'name': 'current_settings', 'type': 'group', 'visible': False, 'children': [
- {'title': 'Current Min:', 'name': 'curr_min', 'type': 'float', 'value': -1, 'suffix': 'A'},
- {'title': 'Current Max:', 'name': 'curr_max', 'type': 'float', 'value': 1, 'suffix': 'A'},
- ]},
- ]
-
- def __init__(self, **opts):
- opts['type'] = 'groupao'
- opts['addText'] = "Add"
- opts['addList'] = opts['limits']
- GroupParameter.__init__(self, **opts)
-
- def addNew(self, typ):
- """
- Add a child.
-
- =============== ===========
- **Parameters** **Type**
- *typ* string
- =============== ===========
- """
- childnames = [par.name() for par in self.children()]
- if childnames == []:
- newindex = 0
- else:
- newindex = len(childnames)
-
- child = {'title': typ, 'name': 'ao{:02.0f}'.format(newindex), 'type': 'group', 'children': self.params,
- 'removable': True, 'renamable': False}
-
- self.addChild(child)
-registerParameterType('groupao', ScalableGroupAO, override=True)
-
-
-class ScalableGroupCounter(GroupParameter):
- """
- |
-
- ================ =============
- **Attributes** **Type**
- *opts* dictionnary
- ================ =============
-
- See Also
- --------
- hardware.DAQ_Move_Stage_type
- """
-
- params = [{'title': 'Edge type:', 'name': 'edge', 'type': 'list', 'limits': Edge.names()}, ]
-
- def __init__(self, **opts):
- opts['type'] = 'groupcounter'
- opts['addText'] = "Add"
- opts['addList'] = opts['limits']
- GroupParameter.__init__(self, **opts)
-
- def addNew(self, typ):
- """
- Add a child.
-
- =============== ===========
- **Parameters** **Type**
- *typ* string
- =============== ===========
- """
- childnames = [par.name() for par in self.children()]
- if childnames == []:
- newindex = 0
- else:
- newindex = len(childnames)
-
- child = {'title': typ, 'name': 'counter{:02.0f}'.format(newindex), 'type': 'group', 'children': self.params,
- 'removable': True, 'renamable': False}
-
- self.addChild(child)
-registerParameterType('groupcounter', ScalableGroupCounter, override=True)
-
-
-class ScalableGroupDI(GroupParameter):
- """
- """
-
- params = []
-
- def __init__(self, **opts):
- opts['type'] = 'groupdi'
- opts['addText'] = "Add"
- opts['addList'] = opts['limits']
- GroupParameter.__init__(self, **opts)
-
- def addNew(self, typ):
- """
- Add a child.
-
- =============== ===========
- **Parameters** **Type**
- *typ* string
- =============== ===========
- """
- childnames = [par.name() for par in self.children()]
- if childnames == []:
- newindex = 0
- else:
- newindex = len(childnames)
-
- child = {'title': typ, 'name': 'di{:02.0f}'.format(newindex), 'type': 'group', 'children': self.params,
- 'removable': True, 'renamable': False}
- self.addChild(child)
-registerParameterType('groupdi', ScalableGroupDI, override=True)
-
-
-class ScalableGroupDO(GroupParameter):
- """
- """
-
- params = []
-
- def __init__(self, **opts):
- opts['type'] = 'groupdo'
- opts['addText'] = "Add"
- opts['addList'] = opts['limits']
- GroupParameter.__init__(self, **opts)
-
- def addNew(self, typ):
- """
- Add a child.
-
- =============== ===========
- **Parameters** **Type**
- *typ* string
- =============== ===========
- """
- childnames = [par.name() for par in self.children()]
- if childnames == []:
- newindex = 0
- else:
- newindex = len(childnames)
-
- child = {'title': typ, 'name': 'counter{:02.0f}'.format(newindex), 'type': 'group', 'children': self.params,
- 'removable': True, 'renamable': False}
- self.addChild(child)
-registerParameterType('groupdo', ScalableGroupDO, override=True)
-
-
-class DAQ_NIDAQmx_base(DAQmx):
- data_grabed_signal = Signal(list)
-
- params =[{'title': 'Refresh hardware:', 'name': 'refresh_hardware', 'type': 'bool', 'value': False},
- {'title': 'Signal type:', 'name': 'NIDAQ_type', 'type': 'list', 'limits': DAQ_NIDAQ_source.names()},
- {'title': 'AO Settings:', 'name': 'ao_settings', 'type': 'group', 'children': [
- {'title': 'Waveform:', 'name': 'waveform', 'type': 'list', 'value': 'DC', 'limits': ['DC', 'Sinus', 'Ramp']},
-
- {'title': 'Controlled param:', 'name': 'cont_param', 'type': 'list', 'value': 'offset',
- 'limits': ['offset', 'amplitude', 'frequency']},
- {'title': 'Waveform Settings:', 'name': 'waveform_settings', 'type': 'group', 'visible': False, 'children': [
- {'title': 'Offset:', 'name': 'offset', 'type': 'float', 'value': 0., },
- {'title': 'Amplitude:', 'name': 'amplitude', 'type': 'float', 'value': 1., },
- {'title': 'Frequency:', 'name': 'frequency', 'type': 'float', 'value': 10., },
- ]},
- ]},
- {'title': 'Clock Settings:', 'name': 'clock_settings', 'type': 'group', 'children': [
- {'title': 'Nsamples:', 'name': 'Nsamples', 'type': 'int', 'value': 1000, 'default': 1000, 'min': 1},
- {'title': 'Frequency:', 'name': 'frequency', 'type': 'float', 'value': 1000., 'default': 1000.,
- 'min': 0., 'suffix': 'Hz'},
- {'title': 'Repetition?:', 'name': 'repetition', 'type': 'bool', 'value': False, },
- ]
- },
- {'title': 'AI Channels:', 'name': 'ai_channels', 'type': 'groupai',
- 'limits': DAQmx.get_NIDAQ_channels(source_type='Analog_Input')},
- {'title': 'AO Channels:', 'name': 'ao_channels', 'type': 'groupao',
- 'limits': DAQmx.get_NIDAQ_channels(source_type='Analog_Output')},
- {'title': 'DO Channels:', 'name': 'do_channels', 'type': 'groupdo',
- 'limits': DAQmx.get_NIDAQ_channels(source_type='Digital_Output')},
- {'title': 'DI Channels:', 'name': 'di_channels', 'type': 'groupdi',
- 'limits': DAQmx.get_NIDAQ_channels(source_type='Digital_Input')},
- {'title': 'Counter Settings:', 'name': 'counter_settings', 'type': 'group', 'visible': True, 'children': [
- {'title': 'Counting time (ms):', 'name': 'counting_time', 'type': 'float', 'value': 100.,
- 'default': 100., 'min': 0.},
- {'title': 'Counting Channels:', 'name': 'counter_channels', 'type': 'groupcounter',
- 'limits': DAQmx.get_NIDAQ_channels(source_type='Counter')},
- ]},
- {'title': 'Trigger Settings:', 'name': 'trigger_settings', 'type': 'group', 'visible': True, 'children': [
- {'title': 'Enable?:', 'name': 'enable', 'type': 'bool', 'value': False, },
- {'title': 'Trigger Source:', 'name': 'trigger_channel', 'type': 'list',
- 'limits': DAQmx.getTriggeringSources()},
- {'title': 'Edge type:', 'name': 'edge', 'type': 'list', 'limits': Edge.names(), 'visible': False},
- {'title': 'Level:', 'name': 'level', 'type': 'float', 'value': 1., 'visible': False}
- ]}
- ]
-
- def __init__(self):
- super().__init__()
-
- self.timer = None
- self.channels = None
- self.clock_settings = None
- self.trigger_settings = None
- self.live = False
- self.refresh_hardware()
-
- def commit_settings(self, param: Parameter):
- """
- Activate the parameters changes in the hardware.
-
- =============== ================================ ===========================
- **Parameters** **Type** **Description**
- *param* instance of pyqtgraph.parameter the parameter to activate
- =============== ================================ ===========================
-
- See Also
- --------
- update_NIDAQ_channels, update_task, DAQ_NIDAQ_source, refresh_hardware
- """
- # if param.name() == 'NIDAQ_devices':
- # self.update_NIDAQ_channels()
- # self.update_task()
-
- if param.name() == 'NIDAQ_type':
- self.update_NIDAQ_channels(param.value())
- if param.value() == DAQ_NIDAQ_source(0).name: #analog input
- self.settings.child('clock_settings').show()
- self.settings.child('ai_channels').show()
- self.settings.child('ao_channels').hide()
- self.settings.child('ao_settings').hide()
- self.settings.child('counter_settings').hide()
- self.settings.child('do_channels').hide()
- self.settings.child('di_channels').hide()
-
- elif param.value() == DAQ_NIDAQ_source(1).name: #counter input
- self.settings.child('clock_settings').hide()
- self.settings.child('ai_channels').hide()
- self.settings.child('ao_channels').hide()
- self.settings.child('ao_settings').hide()
- self.settings.child('counter_settings').show()
- self.settings.child('do_channels').hide()
- self.settings.child('di_channels').hide()
-
- elif param.value() == DAQ_NIDAQ_source(2).name: #analog output
- self.settings.child('clock_settings').show()
- self.settings.child('ai_channels').hide()
- self.settings.child('ao_channels').show()
- self.settings.child('ao_settings').show()
- self.settings.child('counter_settings').hide()
- self.settings.child('do_channels').hide()
- self.settings.child('di_channels').hide()
-
- elif param.value() == DAQ_NIDAQ_source(3).name: # digital output
- self.settings.child('clock_settings').show()
- self.settings.child('ai_channels').hide()
- self.settings.child('ao_channels').hide()
- self.settings.child('ao_settings').hide()
- self.settings.child('counter_settings').hide()
- self.settings.child('do_channels').show()
- self.settings.child('di_channels').hide()
-
- elif param.value() == DAQ_NIDAQ_source(4).name: # Digital_Input
- self.settings.child('clock_settings').show()
- self.settings.child('ai_channels').hide()
- self.settings.child('ao_channels').show()
- self.settings.child('ao_settings').show()
- self.settings.child('counter_settings').hide()
- self.settings.child('do_channels').hide()
- self.settings.child('di_channels').show()
- self.update_task()
-
- elif param.name() == 'refresh_hardware':
- if param.value():
- self.refresh_hardware()
- QtWidgets.QApplication.processEvents()
- self.settings.child('refresh_hardware').setValue(False)
-
- elif param.name() == 'ai_type':
- param.parent().child('voltage_settings').show(param.value() == 'Voltage')
- param.parent().child('current_settings').show(param.value() == 'Current')
- param.parent().child('thermoc_settings').show(param.value() == 'Thermocouple')
- self.update_task()
-
- elif param.name() == 'ao_type':
- param.parent().child('voltage_settings').show(param.value() == 'Voltage')
- param.parent().child('current_settings').show(param.value() == 'Current')
- self.update_task()
-
- elif param.name() == 'trigger_channel':
- param.parent().child('level').show('PF' not in param.opts['title'])
-
- else:
- self.update_task()
-
- def update_task(self):
- self.channels = self.get_channels_from_settings()
- self.clock_settings = ClockSettings(frequency=self.settings['clock_settings', 'frequency'],
- Nsamples=self.settings['clock_settings', 'Nsamples'],
- repetition=self.live,)
- self.trigger_settings = \
- TriggerSettings(trig_source=self.settings['trigger_settings', 'trigger_channel'],
- enable=self.settings['trigger_settings', 'enable'],
- edge=self.settings['trigger_settings', 'edge'],
- level=self.settings['trigger_settings', 'level'],)
-
- if not not self.channels:
- super().update_task(self.channels, self.clock_settings, trigger_settings=self.trigger_settings)
-
-
- def get_channels_from_settings(self):
- channels = []
- if self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(0).name: # analog input
- for channel in self.settings.child('ai_channels').children():
- analog_type = channel['ai_type']
- if analog_type == 'Voltage':
- channels.append(AIChannel(name=channel.opts['title'],
- source='Analog_Input', analog_type=analog_type,
- value_min=channel['voltage_settings', 'volt_min'],
- value_max=channel['voltage_settings', 'volt_max'],
- termination=channel['termination'],))
- elif analog_type == 'Current':
- channels.append(AIChannel(name=channel.opts['title'],
- source='Analog_Input', analog_type=analog_type,
- value_min=channel['current_settings', 'curr_min'],
- value_max=channel['current_settings', 'curr_max'],
- termination=channel['termination'], ))
- elif analog_type == 'Thermocouple':
- channels.append(AIThermoChannel(name=channel.opts['title'],
- source='Analog_Input', analog_type=analog_type,
- value_min=channel['thermoc_settings', 'T_min'],
- value_max=channel['thermoc_settings', 'T_max'],
- termination=channel['termination'],
- thermo_type=channel['thermoc_settings', 'thermoc_type'],))
-
- elif self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(1).name: # counter input
- for channel in self.settings.child('counter_settings', 'counter_channels').children():
- channels.append(Counter(name=channel.opts['title'],
- source='Counter', edge=channel['edge']))
-
- elif self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(2).name: # analog output
- for channel in self.settings.child('ao_channels').children():
- analog_type = channel['ao_type']
- channels.append(AOChannel(name=channel.opts['title'],
- source='Analog_Output', analog_type=analog_type,
- value_min=channel['voltage_settings', 'volt_min'],
- value_max=channel['voltage_settings', 'volt_max'],
- ))
- elif self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(3).name: # Digital output
- for channel in self.settings.child('do_channels').children():
- channels.append(DOChannel(name=channel.opts['title'],
- source='Digital_Output'))
- elif self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(4).name: # digital input
- for channel in self.settings.child('di_channels').children():
- channels.append(DIChannel(name=channel.opts['title'],
- source='Digital_Input'))
- return channels
-
- def stop(self):
- """
- """
- if not not self.timer:
- self.timer.stop()
- QtWidgets.QApplication.processEvents()
- DAQmx.stop(self)
-
-
-class DAQ_NIDAQmx_Viewer(DAQ_Viewer_base, DAQ_NIDAQmx_base):
- """
- ==================== ========================
- **Attributes** **Type**
- *data_grabed_signal* instance of Signal
- *params* dictionnary list
- *task*
- ==================== ========================
-
- See Also
- --------
- refresh_hardware
- """
-
- live_mode_available = True
- params = viewer_params + DAQ_NIDAQmx_base.params
-
- def __init__(self, parent=None, params_state=None, control_type="0D"):
- DAQ_Viewer_base.__init__(self, parent, params_state) #defines settings attribute and various other methods
- DAQ_NIDAQmx_base.__init__(self)
-
- self.live = False
- self.control_type = control_type # could be "0D", "1D" or "Actuator"
- if self.control_type == "0D":
- self.settings.child('NIDAQ_type').setLimits(['Analog_Input', 'Counter', 'Digital_Input']) # analog input and counter
- elif self.control_type == "1D":
- self.settings.child('NIDAQ_type').setLimits(['Analog_Input'])
- elif self.control_type == "Actuator":
- self.settings.child('NIDAQ_type').setLimits(['Analog_Output'])
-
- self.settings.child('ao_channels').hide()
-
- #timer used for the counter
- self.timer = QtCore.QTimer()
- self.timer.setSingleShot(True)
- self.timer.timeout.connect(self.counter_done)
-
- def stop(self):
- try:
- self.controller['ai'].task.StopTask()
- except:
- pass
- ##############################
-
- return ''
-
- def commit_settings(self, param):
- """
- Activate the parameters changes in the hardware.
-
- =============== ================================ ===========================
- **Parameters** **Type** **Description**
- *param* instance of pyqtgraph.parameter the parameter to activate
- =============== ================================ ===========================
-
- See Also
- --------
- update_NIDAQ_channels, update_task, DAQ_NIDAQ_source, refresh_hardware
- """
-
- if param.parent() is not None:
- if param.parent().name() == 'ai_channels':
- device = param.opts['title'].split('/')[0]
- self.settings.child('clock_settings', 'frequency').setOpts(max=self.getAIMaxRate(device))
-
- ranges = self.getAIVoltageRange(device)
- param.child('voltage_settings', 'volt_min').setOpts(limits=[r[0] for r in ranges])
- param.child('voltage_settings', 'volt_max').setOpts(limits=[r[1] for r in ranges])
-
- DAQ_NIDAQmx_base.commit_settings(self, param)
-
- def ini_detector(self, controller=None):
- """
- Initialisation procedure of the detector.
-
- See Also
- --------
- daq_utils.ThreadCommand
- """
- self.status.update(edict(initialized=False, info="", x_axis=None, y_axis=None, controller=None))
- try:
- if self.settings['controller_status'] == "Slave":
- if controller is None:
- raise Exception('no controller has been defined externally while this detector is a slave one')
- else:
- self.controller = 'A Nidaqmx task'
- else:
- self.update_task()
-
- #actions to perform in order to set properly the settings tree options
- self.commit_settings(self.settings.child('NIDAQ_type'))
-
- self.status.info = "Plugin Initialized"
- self.status.initialized = True
- self.status.controller = controller
- return self.status
-
- except Exception as e:
- self.emit_status(ThreadCommand('Update_Status', [str(e), 'log']))
- self.status.info = str(e)
- self.status.initialized = False
- return self.status
-
- def grab_data(self, Naverage=1, **kwargs):
- """
- | grab the current values with NIDAQ profile procedure.
- |
- | Send the data_grabed_signal once done.
-
- =============== ======== ===============================================
- **Parameters** **Type** **Description**
- *Naverage* int Number of values to average
- =============== ======== ===============================================
-
- See Also
- --------
- DAQ_NIDAQ_source
- """
- update = False
- if 'live' in kwargs:
- if kwargs['live'] != self.live:
- update = True
- self.live = kwargs['live']
- if update:
- self.update_task()
-
- if self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(0).name: #analog input
- if self.c_callback is None:
- self.register_callback(self.emit_data)
- elif self.settings['NIDAQ_type'] == DAQ_NIDAQ_source(1).name: #counter input
- self.timer.start(self.settings['counter_settings', 'counting_time'])
- self.waitTaskDone()
- self.start()
-
- def emit_data(self, taskhandle, status, callbackdata):
- channels_name = [ch.name for ch in self.channels]
-
- data_tot = []
- data = self.readAnalog(len(self.channels), self.clock_settings)
- N = self.clock_settings.Nsamples
- if self.control_type == "0D":
- for ind in range(len(self.channels)):
- data_tot.append(np.array([np.mean(data[ind*N:(ind+1)*N])]))
- self.data_grabed_signal.emit([DataFromPlugins(name='NI AI', data=data_tot, dim='Data0D',
- labels=channels_name)])
- else:
- for ind in range(len(self.channels)):
- data_tot.append(data[ind*N:(ind+1)*N])
- self.data_grabed_signal.emit([
- DataFromPlugins(name='NI AI', data=data_tot, dim='Data1D',
- labels=channels_name,
- axes=[Axis(data=np.linspace(0, N / self.clock_settings.frequency, N),
- index=0)])])
- return 0 #mandatory for the PyDAQmx callback
-
- def counter_done(self):
- channels_name = [ch.name for ch in self.channels]
- data_counter = self.readCounter(len(self.channels),
- self.settings['counter_settings', 'counting_time'] * 1e-3)
- self.data_grabed_signal.emit([DataFromPlugins(name='NI Counter', data=[data_counter / 1e-3], dim='Data0D',
- labels=channels_name,)])
- #y_axis=Axis(label='Count Number', units='1/s'))])
- self.task.StopTask()
-
-
-class DAQ_NIDAQmx_Actuator(DAQ_Move_base, DAQ_NIDAQmx_base):
- """
- Wrapper object to access the Mock fonctionnalities, similar wrapper for all controllers.
-
- =============== ==============
- **Attributes** **Type**
- *params* dictionnary
- =============== ==============
- """
- _controller_units = 'Volts'
- is_multiaxes = False # set to True if this plugin is controlled for a multiaxis controller (with a unique communication link)
- stage_names = [] # "list of strings of the multiaxes
-
- params = DAQ_NIDAQmx_base.params +[
- # elements to be added here as dicts in order to control your custom stage
- ############
- {'title': 'MultiAxes:', 'name': 'multiaxes', 'type': 'group', 'visible': is_multiaxes,
- 'children': [
- {'title': 'is Multiaxes:', 'name': 'ismultiaxes', 'type': 'bool', 'value': is_multiaxes,
- 'default': False},
- {'title': 'Status:', 'name': 'multi_status', 'type': 'list', 'value': 'Master',
- 'limits': ['Master', 'Slave']},
- {'title': 'Axis:', 'name': 'axis', 'type': 'list', 'limits': stage_names},
-
- ]}] + actuator_params
-
- def __init__(self, parent=None, params_state=None, control_type="Actuator"):
- DAQ_Move_base.__init__(self, parent, params_state) # defines settings attribute and various other methods
- DAQ_NIDAQmx_base.__init__(self)
-
- self.control_type = "Actuator" # could be "0D", "1D" or "Actuator"
- self.settings.child('NIDAQ_type').setLimits(['Analog_Output', 'Digital_Output'])
-
- self.settings.child('clock_settings', 'Nsamples').setValue(1)
-
- def get_actuator_value(self) -> DataActuator:
- """Get the current position from the hardware with scaling conversion.
-
- Returns
- -------
- float: The position obtained after scaling conversion.
- """
-
- pos = self.target_position
- ##
-
- pos = self.get_position_with_scaling(pos)
- self.emit_status(ThreadCommand('check_position', [pos]))
- return pos
-
-
- def commit_settings(self, param):
- """
- | Activate any parameter changes on the PI_GCS2 hardware.
- |
- | Called after a param_tree_changed signal from DAQ_Move_main.
-
- """
-
- DAQ_NIDAQmx_base.commit_settings(self, param)
- if param.name() == 'waveform':
- if param.value() == 'DC':
- self.settings.child('ao_settings', 'cont_param').setValue('offset')
- self.settings.child('ao_settings', 'cont_param').show(not param.value() == 'DC')
- self.settings.child('ao_settings', 'waveform_settings').show(not param.value() == 'DC')
-
- if param.parent() is not None:
- if param.parent().name() == 'ao_channels':
- device = param.opts['title'].split('/')[0]
- self.settings.child('clock_settings', 'frequency').setOpts(max=self.getAOMaxRate(device))
-
- ranges = self.getAOVoltageRange(device)
- param.child('voltage_settings', 'volt_min').setOpts(limits=[r[0] for r in ranges])
- param.child('voltage_settings', 'volt_max').setOpts(limits=[r[1] for r in ranges])
-
- def ini_stage(self, controller=None):
- """Actuator communication initialization
-
- Parameters
- ----------
- controller: (object) custom object of a PyMoDAQ plugin (Slave case). None if only one actuator by controller (Master case)
-
- Returns
- -------
- self.status (edict): with initialization status: three fields:
- * info (str)
- * controller (object) initialized controller
- *initialized: (bool): False if initialization failed otherwise True
- """
-
- try:
- # initialize the stage and its controller status
- # controller is an object that may be passed to other instances of DAQ_Move_Mock in case
- # of one controller controlling multiactuators (or detector)
-
- self.status.update(edict(info="", controller=None, initialized=False))
-
- # check whether this stage is controlled by a multiaxe controller (to be defined for each plugin)
- # if multiaxes then init the controller here if Master state otherwise use external controller
- if self.settings['multiaxes', 'ismultiaxes'] and self.settings['multiaxes',
- 'multi_status'] == "Slave":
- if controller is None:
- raise Exception('no controller has been defined externally while this axe is a slave one')
- else:
- self.controller = controller
- else:
- self.controller = 'A Nidaqmx task'
- self.update_task()
-
- # actions to perform in order to set properly the settings tree options
- self.commit_settings(self.settings.child('NIDAQ_type'))
-
- self.status.info = "Plugin Initialized"
- self.status.controller = self.controller
- self.status.initialized = True
- return self.status
-
- except Exception as e:
- self.emit_status(ThreadCommand('Update_Status', [getLineInfo() + str(e), 'log']))
- self.status.info = getLineInfo() + str(e)
- self.status.initialized = False
- return self.status
-
- def calulate_waveform(self, value):
- waveform = self.settings['ao_settings', 'waveform']
- if waveform == 'DC':
- values = np.array([value])
- else:
- Nsamples = self.settings['clock_settings', 'Nsamples']
- freq = self.settings['clock_settings', 'frequency']
- time = np.linspace(0, Nsamples / freq, Nsamples, endpoint=False)
-
- freq0 = self.settings['ao_settings', 'waveform_settings', 'frequency']
- amp = self.settings['ao_settings', 'waveform_settings', 'amplitude']
- offset = self.settings['ao_settings', 'waveform_settings', 'offset']
- if waveform == 'Sinus':
- values = offset + amp * np.sin(2*np.pi*freq0*time)
- elif waveform == 'Ramp':
- values = offset + amp * np.linspace(0, 1, Nsamples)
-
- return values
-
-
- def move_Abs(self, position):
- """ Move the actuator to the absolute target defined by position
-
- Parameters
- ----------
- position: (flaot) value of the absolute target positioning
- """
-
- position = self.check_bound(position) # if user checked bounds, the defined bounds are applied here
- position = self.set_position_with_scaling(position) # apply scaling if the user specified one
- if self.settings['NIDAQ_type'] == 'Analog_Output':
- self.settings.child('ao_settings', 'waveform_settings',
- self.settings['ao_settings', 'cont_param']).setValue(position)
- values = self.calulate_waveform(position)
- self.target_position = position
-
- self.stop()
-
- if len(values) == 1:
- self.writeAnalog(len(values), 1, values, autostart=True)
- self.current_position = self.check_position()
- self.move_done()
- else:
- if self.c_callback is None:
- self.register_callback(self.move_done_callback)
- self.writeAnalog(len(values), 1, values, autostart=False)
- self.task.StartTask()
- elif self.settings['NIDAQ_type'] == 'Digital_Output':
- self.writeDigital(1, np.array([position], dtype=np.uint8), autostart=True)
-
- def move_done_callback(self, taskhandle, status, callbackdata):
- self.current_position = self.check_position()
- QtWidgets.QApplication.processEvents()
- self.move_done()
- self.task.StopTask()
- return 0
-
- def move_Rel(self, position):
- """ Move the actuator to the relative target actuator value defined by position
-
- Parameters
- ----------
- position: (flaot) value of the relative target positioning
- """
-
- position = self.check_bound(self.current_position + position) - self.current_position
- self.target_position = position + self.current_position
- if self.settings['NIDAQ_type'] == 'Analog_Output':
- self.settings.child('ao_settings', 'waveform_settings',
- self.settings['ao_settings', 'cont_param']).setValue(self.target_position)
-
- values = self.calulate_waveform(self.target_position)
-
- self.stop()
-
- if len(values) == 1:
- self.writeAnalog(len(values), 1, values, autostart=True)
- self.current_position = self.check_position()
- self.move_done()
- else:
- if self.c_callback is None:
- self.register_callback(self.move_done_callback)
- self.writeAnalog(len(values), 1, values, autostart=False)
- self.task.StartTask()
- elif self.settings['NIDAQ_type'] == 'Digital_Output':
- self.writeDigital(1, np.array([self.target_position], dtype=np.uint8), autostart=True)
-
- def move_Home(self):
- """
- Send the update status thread command.
- See Also
- --------
- daq_utils.ThreadCommand
- """
-
- self.stop()
-
- if self.c_callback is None:
- self.register_callback(self.move_done_callback)
- self.writeAnalog(1, 1, np.array([0.]))
- self.task.StartTask()
-
- def stop_motion(self):
- """
- Call the specific move_done function (depending on the hardware).
-
- See Also
- --------
- move_done
- """
-
- ## TODO for your custom plugin
- self.stop()
- self.emit_status(ThreadCommand('Update_Status', ['Some info you want to log']))
- self.move_done() # to let the interface know the actuator stopped
- ##############################
\ No newline at end of file
diff --git a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx.py b/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx.py
deleted file mode 100644
index 7b43c41..0000000
--- a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx.py
+++ /dev/null
@@ -1,782 +0,0 @@
-import PyDAQmx
-import ctypes
-from enum import IntEnum
-import numpy as np
-from pymodaq.utils.logger import set_logger, get_module_name
-
-logger = set_logger(get_module_name(__file__))
-
-
-class DAQ_NIDAQ_source(IntEnum):
- """
- Enum class of NIDAQ_source
-
- =============== ==========
- **Attributes** **Type**
- *Analog_Input* int
- *Counter* int
- =============== ==========
- """
- Analog_Input = 0
- Counter = 1
- Analog_Output = 2
- Digital_Output = 3
- Digital_Input = 4
- Terminals = 5
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
-
-class DAQ_analog_types(IntEnum):
- """
- Enum class of Ai types
-
- =============== ==========
- **Attributes** **Type**
- =============== ==========
- """
- Voltage = PyDAQmx.DAQmx_Val_Voltage
- Current = PyDAQmx.DAQmx_Val_Current
- Thermocouple = PyDAQmx.DAQmx_Val_Temp_TC
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
- @classmethod
- def values(cls):
- return [cls[name].value for name, member in cls.__members__.items()]
-
-
-class DAQ_thermocouples(IntEnum):
- """
- Enum class of thermocouples type
-
- =============== ==========
- **Attributes** **Type**
- =============== ==========
- """
- J = PyDAQmx.DAQmx_Val_J_Type_TC
- K = PyDAQmx.DAQmx_Val_K_Type_TC
- N = PyDAQmx.DAQmx_Val_N_Type_TC
- R = PyDAQmx.DAQmx_Val_R_Type_TC
- S = PyDAQmx.DAQmx_Val_S_Type_TC
- T = PyDAQmx.DAQmx_Val_T_Type_TC
- B = PyDAQmx.DAQmx_Val_B_Type_TC
- E = PyDAQmx.DAQmx_Val_E_Type_TC
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
-
-class DAQ_termination(IntEnum):
- """
- Enum class of thermocouples type
-
- =============== ==========
- **Attributes** **Type**
- =============== ==========
- """
- Auto = PyDAQmx.DAQmx_Val_Cfg_Default
- RSE = PyDAQmx.DAQmx_Val_RSE
- NRSE = PyDAQmx.DAQmx_Val_NRSE
- Diff = PyDAQmx.DAQmx_Val_Diff
- Pseudodiff = PyDAQmx.DAQmx_Val_PseudoDiff
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
-
-class Edge(IntEnum):
- """
- """
- Rising = PyDAQmx.DAQmx_Val_Rising
- Falling = PyDAQmx.DAQmx_Val_Falling
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
-
-class ClockMode(IntEnum):
- """
- """
- Finite = PyDAQmx.DAQmx_Val_FiniteSamps
- Continuous = PyDAQmx.DAQmx_Val_ContSamps
-
-
- @classmethod
- def names(cls):
- return [name for name, member in cls.__members__.items()]
-
-
-class ClockSettingsBase:
- def __init__(self, Nsamples=1000, repetition=False):
-
- self.Nsamples = Nsamples
- self.repetition = repetition
-
-
-class ClockSettings(ClockSettingsBase):
- def __init__(self, source=None, frequency=1000, Nsamples=1000, edge=Edge.names()[0], repetition=False):
- super().__init__(Nsamples, repetition)
- self.source = source
- assert edge in Edge.names()
- self.frequency = frequency
- self.edge = edge
-
-
-class ChangeDetectionSettings(ClockSettingsBase):
- def __init__(self, Nsamples=1000, rising_channel='', falling_channel='',
- repetition=False):
- super().__init__(Nsamples, repetition)
- self.rising_channel = rising_channel
- self.falling_channel = falling_channel
-
-
-class TriggerSettings:
- def __init__(self, trig_source='', enable=False, edge=Edge.names()[0], level=0.1):
- assert edge in Edge.names()
- self.trig_source = trig_source
- self.enable = enable
- self.edge = edge
- self.level = level
-
-
-class Channel:
- def __init__(self, name='', source=DAQ_NIDAQ_source.names()[0]):
- """
- Parameters
- ----------
-
- """
- self.name = name
- assert source in DAQ_NIDAQ_source.names()
- self.source = source
-
-
-class AChannel(Channel):
- def __init__(self, analog_type=DAQ_analog_types.names()[0], value_min=-10., value_max=+10., **kwargs):
- """
- Parameters
- ----------
- min: (float) minimum value for the configured input channel (could be voltage, amps, temperature...)
- max: (float) maximum value for the configured input channel
- """
- super().__init__(**kwargs)
- self.value_min = value_min
- self.value_max = value_max
- self.analog_type = analog_type
-
-
-class AIChannel(AChannel):
- def __init__(self, termination=DAQ_termination.names()[0], **kwargs):
- super().__init__(**kwargs)
- assert termination in DAQ_termination.names()
- self.termination = termination
-
-
-class AIThermoChannel(AIChannel):
- def __init__(self, thermo_type=DAQ_thermocouples.names()[0], **kwargs):
- super().__init__(**kwargs)
- assert thermo_type in DAQ_thermocouples.names()
- self.thermo_type = thermo_type
-
-
-class AOChannel(AChannel):
- def __init__(self, **kwargs):
- super().__init__(**kwargs)
-
-
-class Counter(Channel):
- def __init__(self, edge=Edge.names()[0], **kwargs):
- assert edge in Edge.names()
- super().__init__(**kwargs)
- self.edge = edge
- self.counter_type = "Edge Counter"
-
-
-class ClockCounter(Counter):
- def __init__(self, clock_frequency, **kwargs):
- super().__init__(**kwargs)
- self.clock_frequency = clock_frequency
- self.counter_type = "Clock Output"
-
-
-class SemiPeriodCounter(Counter):
- def __init__(self, value_max, **kwargs):
- super().__init__(**kwargs)
- self.value_max = value_max
- self.counter_type = "SemiPeriod Input"
-
-
-class DigitalChannel(Channel):
- def __init__(self, **kwargs):
- super().__init__(**kwargs)
-
-
-class DOChannel(DigitalChannel):
- def __init__(self, **kwargs):
- super().__init__(**kwargs)
-
-
-class DIChannel(DigitalChannel):
- def __init__(self, **kwargs):
- super().__init__(**kwargs)
-
-
-def try_string_buffer(fun, *args):
- """
- generic function to read string from a PyDAQmx function making sure the chosen buffer is large enough
- Parameters
- ----------
- fun: (PyDAQmx function pointer) e.g. PyDAQmx.DAQmxGetSysDevNames
- kwargs
-
- Returns
- -------
-
- """
- buff_size = 1024
- while True:
- buff = PyDAQmx.create_string_buffer(buff_size)
- try:
- if not not len(args):
- fun(args[0], buff, buff_size)
- else:
- fun(buff, buff_size)
- break
-
- except Exception as e:
- if isinstance(e, PyDAQmx.DAQmxFunctions.DAQException):
- if e.error == -200228: # BufferTooSmallForStringError
- buff_size = 2 * buff_size
- else:
- raise e
- else:
- raise e
- return buff.value.decode()
-
-
-class DAQmx:
- """Wrapper around the PyDAQmx package giving an easy to use object to instantiate channels and tasks"""
- def __init__(self):
- self.devices = []
- self.channels = []
- self._device = None
- self._task = None
- self.update_NIDAQ_devices()
- self.update_NIDAQ_channels()
- self.c_callback = None
- self.callback_data = None
- self.is_scalar = True
- self.write_buffer = np.array([0.])
-
- @property
- def task(self):
- return self._task
-
- @property
- def device(self):
- return self._device
-
- @device.setter
- def device(self, device):
- if device not in self.devices:
- raise IOError(f'your device: {device} is not known or connected')
- self._device = device
-
- def update_NIDAQ_devices(self):
- self.devices = self.get_NIDAQ_devices()
-
- @classmethod
- def get_NIDAQ_devices(cls):
- """Get list of NI connected devices
-
- Returns
- -------
- list
- list of devices as strings to be used in subsequent commands
- """
- try:
- string = try_string_buffer(PyDAQmx.DAQmxGetSysDevNames)
- devices = string.split(', ')
- if devices == ['']:
- devices = []
- return devices
- except:
- return []
-
- def update_NIDAQ_channels(self, source_type=None):
- self.channels = self.get_NIDAQ_channels(self.devices, source_type=source_type)
-
- @classmethod
- def get_NIDAQ_channels(cls, devices=None, source_type=None):
- """Get the list of available channels for all NiDAq connected devices
-
- Parameters
- ----------
- devices: list
- list of strings, each one being a connected device
- source_type: str
- One of the entries of DAQ_NIDAQ_source enum
-
- Returns
- -------
- List of str containing device and channel names
-
- """
- if devices is None:
- devices = cls.get_NIDAQ_devices()
-
- if source_type is None:
- source_type = DAQ_NIDAQ_source.names()
- if not isinstance(source_type, list):
- source_type = [source_type]
- channels_tot = []
- if not not devices:
- for device in devices:
- for source in source_type:
- if source == DAQ_NIDAQ_source.Analog_Input.name: # analog input
- string = try_string_buffer(PyDAQmx.DAQmxGetDevAIPhysicalChans, device)
- elif source == DAQ_NIDAQ_source.Counter.name: # counter
- string = try_string_buffer(PyDAQmx.DAQmxGetDevCIPhysicalChans, device)
- elif source == DAQ_NIDAQ_source.Analog_Output.name: # analog output
- string = try_string_buffer(PyDAQmx.DAQmxGetDevAOPhysicalChans, device)
- elif source == DAQ_NIDAQ_source.Digital_Output.name: # digital output
- string = try_string_buffer(PyDAQmx.DAQmxGetDevDOLines, device)
- elif source == DAQ_NIDAQ_source.Digital_Input.name: # digital output
- string = try_string_buffer(PyDAQmx.DAQmxGetDevDILines, device)
- elif source == DAQ_NIDAQ_source.Terminals.name: # digital output
- string = try_string_buffer(PyDAQmx.DAQmxGetDevTerminals, device)
-
- channels = string.split(', ')
- if channels != ['']:
- channels_tot.extend(channels)
-
- return channels_tot
-
- @classmethod
- def getAOMaxRate(cls, device):
- data = PyDAQmx.c_double()
- PyDAQmx.DAQmxGetDevAOMaxRate(device, PyDAQmx.byref(data))
- return data.value
-
- @classmethod
- def getAIMaxRate(cls, device):
- data = PyDAQmx.c_double()
- PyDAQmx.DAQmxGetDevAIMaxSingleChanRate(device, PyDAQmx.byref(data))
- return data.value
-
- @classmethod
- def isAnalogTriggeringSupported(cls, device):
- data = PyDAQmx.c_uint32()
- PyDAQmx.DAQmxGetDevAnlgTrigSupported(device, PyDAQmx.byref(data))
- return bool(data.value)
-
- @classmethod
- def isDigitalTriggeringSupported(cls, device):
- data = PyDAQmx.c_uint32()
- PyDAQmx.DAQmxGetDevDigTrigSupported(device, PyDAQmx.byref(data))
- return bool(data.value)
-
- @classmethod
- def getTriggeringSources(cls, devices=None):
- sources = []
- if devices is None:
- devices = cls.get_NIDAQ_devices()
-
- for device in devices:
- if cls.isDigitalTriggeringSupported(device):
- string = try_string_buffer(PyDAQmx.DAQmxGetDevTerminals, device)
- channels = [chan for chan in string.split(', ') if 'PFI' in chan]
- if channels != ['']:
- sources.extend(channels)
- if cls.isAnalogTriggeringSupported(device):
- channels = cls.get_NIDAQ_channels(devices=[device], source_type='Analog_Input')
- if channels != ['']:
- sources.extend(channels)
- return sources
-
- def update_task(self, channels=[], clock_settings=ClockSettings(), trigger_settings=TriggerSettings()):
-
- try:
- if self._task is not None:
- if isinstance(self._task, PyDAQmx.Task):
- self._task.ClearTask()
-
- self._task = None
- self.c_callback = None
-
- self._task = PyDAQmx.Task()
-
- ## create all channels one task for one type of channels
- for channel in channels:
- if channel.source == 'Analog_Input': #analog input
- if channel.analog_type == "Voltage":
- err_code = self._task.CreateAIVoltageChan(channel.name, "",
- DAQ_termination[channel.termination].value,
- channel.value_min,
- channel.value_max,
- PyDAQmx.DAQmx_Val_Volts, None)
-
- elif channel.analog_type == "Current":
- err_code = self._task.CreateAICurrentChan(channel.name, "",
- DAQ_termination[channel.termination].value,
- channel.value_min,
- channel.value_max,
- PyDAQmx.DAQmx_Val_Amps, PyDAQmx.DAQmx_Val_Internal,
- 0., None)
-
- elif channel.analog_type == "Thermocouple":
- err_code = self._task.CreateAIThrmcplChan(channel.name, "",
- channel.value_min,
- channel.value_max,
- PyDAQmx.DAQmx_Val_DegC,
- DAQ_termination[channel.thermo_type].value,
- PyDAQmx.DAQmx_Val_BuiltIn, 0., "")
-
- elif channel.source == 'Counter': #counter
- if channel.counter_type == "Edge Counter":
- err_code = self._task.CreateCICountEdgesChan(channel.name, "",
- Edge[channel.edge].value, 0,
- PyDAQmx.DAQmx_Val_CountUp)
- elif channel.counter_type == "Clock Output":
- err_code = self._task.CreateCOPulseChanFreq(channel.name, "clock task",
- # units, Hertz in our case
- PyDAQmx.DAQmx_Val_Hz,
- # idle state
- PyDAQmx.DAQmx_Val_Low,
- # initial delay
- 0,
- # pulse frequency
- channel.clock_frequency,
- # duty cycle of pulses, 0.5 such that
- # high and low duration are both
- # equal to count_interval
- 0.5)
- elif channel.counter_type == "SemiPeriod Input":
- err_code = self._task.CreateCISemiPeriodChan(channel.name, "counter task",
- 0, # expected min
- channel.value_max, # expected max
- PyDAQmx.DAQmx_Val_Ticks, "")
-
-
- if not not err_code:
- status = self.DAQmxGetErrorString(err_code)
- raise IOError(status)
-
- elif channel.source == 'Analog_Output': # Analog_Output
- if channel.analog_type == "Voltage":
- err_code = self._task.CreateAOVoltageChan(channel.name, "",
- channel.value_min,
- channel.value_max,
- PyDAQmx.DAQmx_Val_Volts, None)
-
- if channel.analog_type == "Current":
- err_code = self._task.CreateAOCurrentChan(channel.name, "",
- channel.value_min,
- channel.value_max,
- PyDAQmx.DAQmx_Val_Amps, None)
-
- elif channel.source == 'Digital_Output': #Digital_Output
- err_code = self._task.CreateDOChan(channel.name, "", PyDAQmx.DAQmx_Val_ChanPerLine)
- if not not err_code:
- status = self.DAQmxGetErrorString(err_code)
- raise IOError(status)
-
- elif channel.source == 'Digital_Input': #Digital_Input
- err_code = self._task.CreateDIChan(channel.name, "", PyDAQmx.DAQmx_Val_ChanPerLine)
- if not not err_code:
- status = self.DAQmxGetErrorString(err_code)
- raise IOError(status)
-
- ## configure the timing
- if clock_settings.repetition:
- mode = PyDAQmx.DAQmx_Val_ContSamps
- else:
- mode = PyDAQmx.DAQmx_Val_FiniteSamps
- if clock_settings.Nsamples > 1 and err_code == 0:
- if isinstance(clock_settings, ClockSettings):
- err_code =\
- self._task.CfgSampClkTiming(clock_settings.source, clock_settings.frequency,
- Edge[clock_settings.edge].value,
- mode,
- clock_settings.Nsamples)
- elif isinstance(clock_settings, ChangeDetectionSettings):
- err_code =\
- self._task.CfgChangeDetectionTiming(clock_settings.rising_channel,
- clock_settings.falling_channel,
- mode,
- clock_settings.Nsamples)
-
- if not not err_code:
- status = self.DAQmxGetErrorString(err_code)
- raise IOError(status)
-
- # if channel.source == 'Analog_Input': # analog input
- # if isinstance(clock_settings, ClockSettings):
- # err_code =\
- # self._task.CfgSampClkTiming(clock_settings.source, clock_settings.frequency,
- # Edge[clock_settings.edge].value,
- # mode,
- # clock_settings.Nsamples)
- # elif isinstance(clock_settings, ChangeDetectionSettings):
- # err_code =\
- # self._task.CfgChangeDetectionTiming(clock_settings.source,
- # clock_settings.rising_channel,
- # clock_settings.falling_channel,
- # mode,
- # clock_settings.Nsamples)
- #
- # if not not err_code:
- # status = self.DAQmxGetErrorString(err_code)
- # raise IOError(status)
- #
- # elif channel.source == 'Counter': # counter
- # pass
- #
- # elif channel.source == 'Analog_Output': # Analog_Output
- # if clock_settings.Nsamples > 1 and err_code == 0:
- #
- # if isinstance(clock_settings, ClockSettings):
- # err_code = self._task.CfgSampClkTiming(clock_settings.source,
- # clock_settings.frequency,
- # Edge[clock_settings.edge].value,
- # mode,
- # clock_settings.Nsamples)
- # elif isinstance(clock_settings, ChangeDetectionSettings):
- # err_code = \
- # self._task.CfgChangeDetectionTiming(clock_settings.source,
- # clock_settings.rising_channel,
- # clock_settings.falling_channel,
- # mode,
- # clock_settings.Nsamples)
- #
- # if not not err_code:
- # status = self.DAQmxGetErrorString(err_code)
- # raise IOError(status)
- #
- # else:
- # pass
-
- ##configure the triggering, except for counters
- if not trigger_settings.enable:
- if channel.source == 'Counter':
- pass
- else:
- err = self._task.DisableStartTrig()
- if err != 0:
- raise IOError(self.DAQmxGetErrorString(err))
- else:
- if 'PF' in trigger_settings.trig_source:
- self._task.CfgDigEdgeStartTrig(trigger_settings.trig_source,
- Edge[trigger_settings.edge].value)
- elif 'ai' in trigger_settings.trig_source:
- self._task.CfgAnlgEdgeStartTrig(trigger_settings.trig_source,
- Edge[trigger_settings.edge].value,
- PyDAQmx.c_double(trigger_settings.level))
- else:
- raise IOError('Unsupported Trigger source')
-
- except Exception as e:
- print(e)
-
- def register_callback(self, callback, event='done', nsamples=1):
-
- if event == 'done':
- self.c_callback = PyDAQmx.DAQmxDoneEventCallbackPtr(callback)
- self._task.RegisterDoneEvent(0, self.c_callback, None)
- elif event == 'sample':
- self.c_callback = PyDAQmx.DAQmxSignalEventCallbackPtr(callback)
- self._task.RegisterSignalEvent(PyDAQmx.DAQmx_Val_SampleCompleteEvent, 0, self.c_callback, None)
- elif event == 'Nsamples':
- self.c_callback = PyDAQmx.DAQmxEveryNSamplesEventCallbackPtr(callback)
- self._task.RegisterEveryNSamplesEvent(PyDAQmx.DAQmx_Val_Acquired_Into_Buffer, nsamples,
- 0, self.c_callback, None)
-
- def get_last_write_index(self):
- if self.task is not None:
- index_buffer = PyDAQmx.c_uint64()
- ret = self._task.GetWriteCurrWritePos(PyDAQmx.byref(index_buffer))
- if not not ret:
- raise IOError(self.DAQmxGetErrorString(ret))
- else:
- return index_buffer.value
- else:
- return -1
-
- def get_last_write(self):
- if self.is_scalar:
- return self.write_buffer[-1]
- else:
- index = self.get_last_write_index()
- if index != -1:
- return self.write_buffer[index % len(self.write_buffer)]
-
- else:
- return 0.
-
- def writeAnalog(self, Nsamples, Nchannels, values, autostart=False):
- """
- Write Nsamples on N analog output channels
- Parameters
- ----------
- Nsamples: (int) numver of samples to write on each channel
- Nchannels: (int) number of AO channels defined in the task
- values: (ndarray) 2D array (or flattened array) of size Nsamples * Nchannels
-
- Returns
- -------
-
- """
- if np.prod(values.shape) != Nsamples * Nchannels:
- raise ValueError(f'The shape of analog outputs values is incorrect, should be {Nsamples} x {Nchannels}')
-
- if len(values.shape) != 1:
- values = values.reshape((Nchannels * Nsamples))
- self.write_buffer = values
-
- timeout = -1
- if values.size == 1:
- self._task.WriteAnalogScalarF64(autostart, timeout, values[0], None)
- self.is_scalar = True
-
- else:
- self.is_scalar = False
- read = PyDAQmx.int32()
- self._task.WriteAnalogF64(Nsamples, autostart, timeout, PyDAQmx.DAQmx_Val_GroupByChannel, values,
- PyDAQmx.byref(read), None)
- if read.value != Nsamples:
- raise IOError(f'Insufficient number of samples have been written:{read.value}/{Nsamples}')
-
- def readAnalog(self, Nchannels, clock_settings):
- read = PyDAQmx.int32()
- N = clock_settings.Nsamples
- data = np.zeros(N * Nchannels, dtype=np.float64)
- timeout = N * Nchannels * 1 / clock_settings.frequency * 2 # set to twice the time it should take to acquire the data
-
- self._task.ReadAnalogF64(N, timeout, PyDAQmx.DAQmx_Val_GroupByChannel, data, len(data),
- PyDAQmx.byref(read), None)
- if read.value == N:
- return data
- else:
- raise IOError(f'Insufficient number of samples have been read:{read.value}/{N}')
-
- def readCounter(self, Nchannels, counting_time=10., read_function="Ex"):
-
- data_counter = np.zeros(Nchannels, dtype='uint32')
- read = PyDAQmx.int32()
- if read_function == "Ex":
- self._task.ReadCounterU32Ex(PyDAQmx.DAQmx_Val_Auto, 2*counting_time,
- PyDAQmx.DAQmx_Val_GroupByChannel,
- data_counter,
- Nchannels, PyDAQmx.byref(read), None)
- else:
- self._task.ReadCounterU32(PyDAQmx.DAQmx_Val_Auto, 2*counting_time,
- data_counter, Nchannels, PyDAQmx.byref(read), None)
-
- self._task.StopTask()
-
- if read.value == Nchannels:
- return data_counter
- else:
- raise IOError(f'Insufficient number of samples have been read:{read}/{Nchannels}')
-
- def readDigital(self, Nchannels):
- read = PyDAQmx.int32()
- bytes_sample = PyDAQmx.int32()
- data = np.zeros(Nchannels, dtype='uint8')
- self._task.ReadDigitalLines(Nchannels, 0, PyDAQmx.DAQmx_Val_GroupByChannel,
- data, Nchannels, PyDAQmx.byref(read), PyDAQmx.byref(bytes_sample), None);
-
- if read.value == Nchannels:
- return data
- else:
- raise IOError(f'Insufficient number of samples have been read:{read}/{Nchannels}')
-
- def writeDigital(self, Nchannels, values, autostart=False):
- if np.prod(values.shape) != Nchannels:
- raise ValueError(f'The shape of digital outputs values is incorrect, should be {Nchannels}')
- values.astype(np.uint8)
- written = PyDAQmx.int32()
- self._task.WriteDigitalLines(Nchannels, autostart, 0, PyDAQmx.DAQmx_Val_GroupByChannel,
- values, PyDAQmx.byref(written), None)
- if written.value != Nchannels:
- raise IOError(f'Insufficient number of samples have been written:{written}/{Nchannels}')
-
- @classmethod
- def getAIVoltageRange(cls, device='Dev1'):
- buff_size = 100
- ranges = ctypes.pointer((buff_size*ctypes.c_double)())
- ret = PyDAQmx.DAQmxGetDevAIVoltageRngs(device, ranges[0], buff_size)
- if ret == 0:
- return [tuple(ranges.contents[2*ind:2*(ind+1)]) for ind in range(int(buff_size/2-2))
- if np.abs(ranges.contents[2*ind]) > 1e-12]
- return [(-10., 10.)]
-
- @classmethod
- def getAOVoltageRange(cls, device='Dev1'):
- buff_size = 100
- ranges = ctypes.pointer((buff_size*ctypes.c_double)())
- ret = PyDAQmx.DAQmxGetDevAOVoltageRngs(device, ranges[0], buff_size)
- if ret == 0:
- return [tuple(ranges.contents[2*ind:2*(ind+1)]) for ind in range(int(buff_size/2-2))
- if np.abs(ranges.contents[2*ind]) > 1e-12]
- return [(-10., 10.)]
-
- def stop(self):
- if self._task is not None:
- self._task.StopTask()
-
- def start(self):
- if self._task is not None:
- self._task.StartTask()
-
- def close(self):
- """
- close the current task.
- """
- if self._task is not None:
- self._task.StopTask()
- self._task.ClearTask()
- self._task = None
-
- @classmethod
- def DAQmxGetErrorString(cls, error_code):
- if error_code is None:
- return ''
- else:
- buffer = PyDAQmx.create_string_buffer(1024)
- PyDAQmx.DAQmxGetErrorString(error_code, buffer, len(buffer))
- return buffer.value.decode()
-
- def isTaskDone(self):
- done = PyDAQmx.bool32(False)
- self._task.GetTaskComplete(PyDAQmx.byref(done))
- return bool(done.value)
-
- def waitTaskDone(self, timeout=10.):
- ret = self._task.WaitUntilTaskDone(timeout)
- if ret != 0:
- logger.info(self.DAQmxGetErrorString(ret))
-
- def refresh_hardware(self):
- """
- Refresh the NIDAQ hardware from settings values.
-
- See Also
- --------
- update_NIDAQ_devices, update_NIDAQ_channels
- """
- devices = self.update_NIDAQ_devices()
- self.update_NIDAQ_channels(devices)
-
-
-if __name__ == '__main__':
- print(DAQmx.get_NIDAQ_channels())
- pass
diff --git a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx_objects.py b/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx_objects.py
deleted file mode 100644
index 439cacc..0000000
--- a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmx_objects.py
+++ /dev/null
@@ -1,113 +0,0 @@
-from collections import OrderedDict
-import numpy as np
-from qtpy.QtCore import QObject, Signal
-import pymodaq_plugins_daqmx.hardware.national_instruments.daqmx as dq
-from PyDAQmx import DAQmx_Val_FiniteSamps
-
-from pymodaq.utils.logger import set_logger, get_module_name
-
-logger = set_logger(get_module_name(__file__))
-
-
-class AO_with_clock_DAQmx(QObject):
- """Object used to coordinate the use of several DAQmx by several modules.
- Its main intended use is to control the movement of several scanners (AO chans) with
- the timing given by the same clock channel."""
- ni_card_ready_for_moving = Signal()
- def __init__(self):
- QObject.__init__(self)
-
- self.clock = dq.DAQmx()
- self.analog = dq.DAQmx()
- self.clock_channel_name = ''
- self.clock_channel = None
- self.clock_frequency = 100.0
- self.AO_channels = OrderedDict()
- self.num_ch = 0
- self.voltage_array = None
- self.max_ch_nb = 1
- self.applied_voltages = OrderedDict()
- self.locked = False # True when busy already moving something
-
- def set_up_clock(self, nb_steps):
- """Prepare the clock with the desired frequency
- nb_steps (int) specifies the number of steps there will be in the movement."""
- self.clock_channel = dq.ClockCounter(self.clock_frequency,
- name=self.clock_channel_name,
- source="Counter")
- if self.clock.task is not None:
- self.clock.task.WaitUntilTaskDone(-1) # in case another scanner is still moving
- self.clock.update_task(channels=[self.clock_channel])
- # we need to set the rate again, I do not understand why
- self.clock.task.SetSampClkRate(self.clock_frequency)
- self.clock.task.CfgImplicitTiming(DAQmx_Val_FiniteSamps, nb_steps + 1)
-
- clock_settings_ao = dq.ClockSettings(source="/" + self.clock_channel_name + "InternalOutput",
- frequency=self.clock_frequency,
- edge=dq.Edge.names()[0],
- Nsamples=nb_steps + 1,
- repetition=False)
- self.get_max_ch_nb()
- return clock_settings_ao
-
- def update_ao_channels(self, channel, axis, clock_settings_ao):
- """Update the dict containing the AO channels, and the associated task in the
- corresponding DAQmx object."""
- self.AO_channels[axis] = channel
- if axis not in self.applied_voltages.keys():
- self.applied_voltages[axis] = 0.0
- self.num_ch = len(self.AO_channels)
- self.get_max_ch_nb()
- if self.num_ch > self.max_ch_nb:
- logger.info("Too many AO channels!")
- return
- if self.clock.task is not None: # we wait for slow movements
- self.clock.task.WaitUntilTaskDone(-1) # in case another scanner is still moving
- self.analog.update_task(channels=[self.AO_channels[ax] for ax in self.AO_channels.keys()],
- clock_settings=clock_settings_ao)
-
- def set_up_voltage_array(self, voltage_list, axis):
- if self.num_ch == 1:
- self.voltage_array = voltage_list
- elif self.num_ch > 1:
- self.voltage_array = np.zeros((self.num_ch, len(voltage_list)))
- for i, ax in enumerate(self.AO_channels.keys()):
- if ax == axis:
- self.voltage_array[i, :] = voltage_list
- else:
- # we need to keep the other scanners were they are
- self.voltage_array[i, :] = self.applied_voltages[ax] * np.ones(len(voltage_list))
-
- def write_voltages(self):
- """Actually writes the voltages"""
- if len(np.shape(self.voltage_array)) > 1:
- nb_steps = np.size(self.voltage_array, axis=1)
- else:
- nb_steps = len(self.voltage_array)
-
- # store last values of the lists as applied voltage
- axes = [ax for ax in self.applied_voltages.keys()]
- if len(axes) == 1:
- self.applied_voltages[axes[0]] = self.voltage_array[-1]
- elif len(axes) > 1:
- for i in range(len(axes)):
- self.applied_voltages[axes[i]] = self.voltage_array[i, -1]
-
- self.analog.start()
- self.analog.writeAnalog(nb_steps, self.num_ch, self.voltage_array)
-
- def get_max_ch_nb(self):
- # get the max number of available AO channels of the device with the clock
- dev = self.clock_channel_name.split('/')[0]
- chans = self.clock.get_NIDAQ_channels(devices=[dev], source_type="Analog_Output")
- self.max_ch_nb = len(chans)
-
- def stop(self):
- self.locked = False
- self.clock.stop()
- self.analog.stop()
-
- def received_move_done(self):
- self.locked = False
- self.ni_card_ready_for_moving.emit()
-
diff --git a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmxni.py b/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmxni.py
index 7280499..685a264 100644
--- a/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmxni.py
+++ b/src/pymodaq_plugins_daqmx/hardware/national_instruments/daqmxni.py
@@ -156,10 +156,9 @@ def __init__(self, edge=Edge.RISING, counter_type=UsageTypeCO.PULSE_FREQUENCY, *
class ClockCounter(COChannel):
def __init__(self, clock_frequency=100, counter_type=UsageTypeCO.PULSE_FREQUENCY, **kwargs):
- super().counter_type = counter_type
- super().__init__(**kwargs)
+ kwargs.setdefault('source', ChannelType.COUNTER_OUTPUT)
+ super().__init__(counter_type=counter_type, **kwargs)
self.clock_frequency = clock_frequency
- self.counter_type = counter_type
class DigitalChannel(Channel):
diff --git a/tests/test_channels.py b/tests/test_channels.py
new file mode 100644
index 0000000..158b07f
--- /dev/null
+++ b/tests/test_channels.py
@@ -0,0 +1,27 @@
+"""Tests of the description objects of the nidaqmx based interface (no hardware nor NI-DAQmx driver needed)."""
+import pytest
+
+pytest.importorskip('nidaqmx')
+pytest.importorskip('pymodaq')
+
+from nidaqmx.constants import ChannelType, Edge, UsageTypeCO
+
+from pymodaq_plugins_daqmx.hardware.national_instruments.daqmxni import ClockCounter
+
+
+class TestClockCounter:
+ def test_can_be_instantiated(self):
+ clock = ClockCounter(clock_frequency=250., name='Dev1/ctr0')
+
+ assert clock.name == 'Dev1/ctr0'
+ assert clock.clock_frequency == 250.
+ assert clock.counter_type == UsageTypeCO.PULSE_FREQUENCY
+ assert clock.edge == Edge.RISING
+
+ def test_is_a_counter_output_channel_by_default(self):
+ assert ClockCounter(name='Dev1/ctr0').source == ChannelType.COUNTER_OUTPUT
+
+ def test_source_can_be_overridden(self):
+ clock = ClockCounter(name='Dev1/ctr0', source=ChannelType.COUNTER_OUTPUT, edge=Edge.FALLING)
+
+ assert clock.edge == Edge.FALLING