acquisition.py

# -*- coding: utf-8 -*-

#    Pyblab
#    Copyright (C) 2021  Marco Pizzocaro <m.pizzocaro@inrim.it>
#
#    This file is part of Pyblab.
#
#    Pyblab is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    Pyblab is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with Pyblab.  If not, see <https://www.gnu.org/licenses/>.


import ctypes
import numpy
import threading
from PyDAQmx import *




class Acquisition(object):
	"""Acquisition class.
	Set up a retriggerable Analog Input Task using a Counter Task  as clock
	source.
	See http://digital.ni.com/public.nsf/allkb/82DAECADE90CC56F8625659200675B2A

	Note that the Analog Input Task on the National Instrument PCI 6229 card
	is not retriggerable as is. Clearing and recreating the Task is a
	workaround but it is slow (it takes about 30 ms).

	Moreover this task exploits the method callback from PyDAQmx
	https://pythonhosted.org/PyDAQmx/callback.html

	It assures synchronization using the threading module, as in
	https://github.com/clade/PyDAQmx/blob/master/PyDAQmx/example/callback_Task_synchronous.py

	A nice tutorial on threading:
	http://pymotw.com/2/threading/

	"""
	def __init__(self):
		"""Initialize an analog input and counter task."""
		self.analog_input = Task()
		self.counter = Task()

		# number of acquisition per cycle
		self.num = 0
		# number of channels
		self.num_channel = 0

		# acquisition counter
		self.n = 0
		# output variable for some DAQmx functions
		self.read = ctypes.c_int32()


		# see https://github.com/clade/PyDAQmx/blob/master/PyDAQmx/example/callback_Task_synchronous.py
		# data lock and new data available event
		self._data_lock = threading.Lock()
		self._newdata_event = threading.Event()

		# callbacks for everyNsamples and Done signals
		self.every_N_c = DAQmxEveryNSamplesEventCallbackPtr(self.every_N_callback)
		self.done_c = DAQmxDoneEventCallbackPtr(self.done_callback)




	def ms2n(self, ms, rate = None):
		"""Covert from milliseconds to number of samples according to the
		Acquisition rate.

		Parameters
		----------
		ms : array_like
			data in milliseconds

		rate : float, optional
				rate used for convertion
				if unspecified uses the acquisition rate

		Returns
		----------
		out : array_like
			data in number of samples


		"""
		if not rate:
			rate = self.rate

		assert rate > 0.

		#Converting to number of samples at rate self.rate
		return numpy.rint(ms*rate/1000.).astype(numpy.int)



	def rread(self, alines, anames, num, trigger, duration, terminal = DAQmx_Val_Cfg_Default, minV = -10., maxV  = 10.,
			counter_channel = "/Dev1/ctr0", counter_source = "/Dev1/Ctr0InternalOutput",
			rate = 10000., trigger_edge = DAQmx_Val_Rising):
		"""Set up the acquisition.

		Parameters
		----------
		alines : list of string
			lines of the NI board to be read as analog input

		anames : list of string
			names of the acuired channels


		trigger : string
			line of the NI board to be used as trigger input

		duration : float
			duration in milliseconds of the acuisition

		terminal : NI-DAQmx input terminal configuration
			deafault to DAQmx_Val_Cfg_Defaul

		minV : float
			minimum voltage expected to be read

		maxV : float
			maximum voltage expected to be read

		counter_channel : string
			line on the NI board used as counter
			default to "/Dev1/ctr0"

		counter_source : string
			line on the NI board used as counter source
			default to /Dev1/Ctr0InternalOutput"

		rate : float
			the rate of the acquisition in hertz

		trigger_edge : NI-DAQmx trigger type
			default to DAQmx_Val_Risin


		Note
		----

		The counter is set-up as the clock of the analog input task, to fake
		a retriggerable input. Default counter is ctr0. User can change that
		to ctr1 but that should not be needed.

		"""
		self.rate = rate
		self.duration = duration
		self.num = num
		self.num_channel = len(alines)




		self.single_acq_size = self.ms2n(self.duration) #was buffer_size
		self.acq_size = self.single_acq_size*self.num
		self.data_size = self.acq_size*self.num_channel
		self._data = numpy.zeros(self.data_size)


		self.channels =  (anames if anames else alines)
		self.analog_lines = alines
		alines = ",".join(alines)



		# set the analog input task
		# 1. create AI Voltage channel
		# 2. set up counter as clock
		# 3. register everyNsamples and Done callbacks
		#
		# CreateAIVoltageChan (
		#			range of physical channels,
		#			names of the channels (empty = use physical names)
		#			terminal config (e.g. DAQmx_Val_RSE or DAQmx_Val_Diff)
		#			min Val, max val, Units
		#			None (instead of custom scale)
		#
		# CfgSampClkTiming(
		#			clock terminal ("" = use onboard clock),
		#			sampling rate per second,
		#			active edge (DAQmx_Val_Rising/DAQmx_Val_Falling),
		#			sample mode (DAQmx_Val_ContSamps = Acquire or generate samples until you stop the task),
		#			samps per channel to generate or buffer_size per channel
		#			);

		self.analog_input.CreateAIVoltageChan(alines, "", terminal, minV, maxV, DAQmx_Val_Volts, None)
		self.analog_input.CfgSampClkTiming(counter_source, self.rate, DAQmx_Val_Rising, DAQmx_Val_ContSamps, self.acq_size)

		# make buffer size an integer mupltiple of the sample acquired per channel01
		real_buffer_size = self.data_size * 10
		self.analog_input.CfgInputBuffer(real_buffer_size)


		# DAQmxRegisterEveryNSamplesEvent(
		#					everyNsamplesEventType (DAQmx_Val_Acquired_Into_Buffer/DAQmx_Val_Transferred_From_Buffer )
		#					number of samples *per channel* to read
		#					options (0 = The callback function is called in a DAQmx thread),
		#					pointer to callback function,
		#					data to pass to function)
		#
		#
		self.analog_input.RegisterEveryNSamplesEvent(DAQmx_Val_Acquired_Into_Buffer, self.acq_size, 0, self.every_N_c, None)
		self.analog_input.RegisterDoneEvent(0, self.done_c, None)

		# set the counter task
		# 1. create counter channel
		# 2. set implicit timing
		# 3. set trigger edge
		# 4. set retriggerable
		#
		# CreateCOPulseChanFreq(
		#			range of physical counter channels		,
		#			names of the channels (empty = use physical names),
		#			Units (DAQmx_Val_Hz),
		#			idle state (DAQmx_Val_Low/DAQmx_Val_High),
		#			initial delay in seconds,
		#			rate,
		#			duty cycle)
		#
		self.counter.CreateCOPulseChanFreq(counter_channel, "", DAQmx_Val_Hz, DAQmx_Val_Low, 0, self.rate, 0.5);
		self.counter.CfgImplicitTiming(DAQmx_Val_FiniteSamps, self.single_acq_size)
		self.counter.CfgDigEdgeStartTrig(trigger, trigger_edge)
		self.counter.SetStartTrigRetriggerable(True)
		self.counter.StartTask() # counter starts only once




	def start(self):
		"""Start the PyDAQmx tasks. """
		self.analog_input.StartTask()
		#self.counter.StartTask()

	def stop(self):
		"""Stop the PyDAQmx tasks. """
		self.analog_input.StopTask()

		# self.counter.ClearTask()
		# stopping the counter task gives an error:
		# Finite acquisition or generation has been stopped before the requested number of samples were acquired or generated.
		# in function DAQmxStopTask
		#
		# To avoid the issue clearing the task instead is a nice workaround
		# even better I am doing nothing and the acquisition is stopped anyway


	def clear(self):
		"""Clear the PyDAQmx tasks. """
		self.analog_input.ClearTask()
		self.counter.ClearTask()


	def every_N_callback(self, taskHandle, everyNsamplesEventType, nSamples, data):
		"""Callback to the everyNsamples event.

		Lock the data and save it in self._data.
		Set the newdata_event.
		"""
		with self._data_lock:
			# note the DAQmx_Val_GroupByChannel = non-interleaved
			# instead of DAQmx_Val_GroupByScanNumber = interleaved
			self.analog_input.ReadAnalogF64(self.acq_size,10.0,DAQmx_Val_GroupByChannel,
							self._data,len(self._data),ctypes.byref(self.read),None)
			self._newdata_event.set()
			self.n += 1


		return 0 # The function should return an integer

	def done_callback(self, status, data):
		"""Callback to the EventDone event.

		Just return 0."""
		#print "Status", status.value
		return 0 # The function should return an integer


	def get_data(self, timeout=None):
		""" Get the data acquired.

		Parameters
		----------

		timeout : float or None
			timeout in seconds

		Returns
		-------
		data : numpy array
			data acquired, where every channel is on a different column

		n : int
			number of acquisitions so far

		Note
		----
		The function waits for the newdata_event or for
		timeout seconds. After timeout is elapsed raise an error.



		"""
		#if blocking:
		while not self._newdata_event.isSet():
			event_is_set = self._newdata_event.wait(timeout)
			if not event_is_set:
				raise ValueError("timeout waiting for data from device")
		with self._data_lock:
			self._newdata_event.clear()
			#return self._data.reshape((-1,self.num_channel*self.num)).copy(), self.n-1
			#return self._data, self.n-1

			# reshape in a column for each channel and acquisition
			# for example if channels A and B are acquired 2 times
			# the columns are A1 A2 B1 B2
			# note that adding extra channels does not mess up with the order
			# of the previous ones
			# this works beacuse DAQmx_Val_GroupByChannel = non-interleaved

			res = self._data.reshape((-1, self.num_channel*self.num), order='F').copy()

			# return the data in mV, scratch the first point of each measurement, usually buggy
			return res[1:]*1000., self.n-1