pattern.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 numpy.lib.recfunctions as lrf

from PyDAQmx import *



class Pattern(object):
	"""
	Pattern class.
	Setup a generic pattern with both analog and digital outputs on a National
	Instruments DAQmx device (both buffered).

	Digital and analog output are buffered and synchronized, the digital
	task sharing the clock of the	analog task, if any.

	It is based on the Python interface to the NIDAQmx drivers PyDAQmx.

	The data to be written to the board is read from a numpy structured array,
	where a field 't' is expected to contain the timing for the pattern in
	milliseconds.

	See:
	http://www.ni.com/dataacquisition/nidaqmx
	https://pypi.python.org/pypi/PyDAQmx
	http://docs.scipy.org/doc/numpy/user/basics.rec.html




	Attributes
	----------
	analog_output:	PyDAQmx Task
		task for the analog output
	analog_channels: list of strings
		names of the analog channels
	analog_lines: list of strings
		names of the physical channels where the analog output is


	digital_output:	PyDAQmx Task
		task for the digital output
	digital_channels: list of strings
		names of the digital channels
	digital_lines: list of strings
		names of the physical channels where the digital output is


	sequence:	numpy structured array
		specify the pattern sequence
		the column 't' is expected to be the timing in milliseconds
		other columns are expected to be named as the analog_channels and
		digital_channels and to contain the output to be written
		dt:	numpy array
		array of the sequence time bins in milliseconds
	rate:	float
		sample per second of the output pattern




	Methods
	-------
	start() : sart the DAQmx tasks
	stop() : stop the DAQmx tasks
	clear() : clear the DAQmx tasks
	is_running(): return if the patter is running

	write(...) : write a numpy structured array to the National Instrument
				board

	ms2n(x) : convert an array in milliseconds to number of samples
	analog_word() : compute the analog word
	digital_word() : compute the digital word




	Usage
	-----
	# this is the basic usage
	# for a simpler and more powerful approach see the Experiment class
	# instantiate the pattern
	spam = Pattern()

	# prepare a numpy structured array containing the output and the timing,
	# in this example from string	using genfromtxt
	seqstr = \"""
	#t	do0	do1	ao0
	10.	0	0	1.
	10.	1	0	2.
	10.	0	1	1.5
	\"""
	seq = numpy.genfromtxt(StringIO.StringIO(seqstr), dtype=None, names=True)

	#if needed, prepare a dictionary of delays for some channels
	# again the delay is intended in milliseconds
	delays = {'do0': 5.}

	# write the numpy structured array to the National Instrument board
	# specifiyng lines (= NI output port line),
	# channels (= names in the structured array),
	# rate and delays
	# note that lines and names are matched by order

	spam.write(seq, alines = ['Dev1/ao0'], anames=['ao0'],
	 dlines = ['Dev1/port0/line0','Dev1/port0/line1'], dnames=['do0','do1'],
	 rate = 1000., delays = delays)

	# start!
	spam.start()

	"""
	def __init__(self):
		"""Initialize an analog output and digital output task."""
		self.analog_output = Task()
		self.digital_output = Task()

		self.read = ctypes.c_int32()


	def start(self):
		"""Start the PyDAQmx tasks. """
		# digital output starts first, because it shares the analog output clock, ensuring syncronization
		if self.digital_lines:
			self.digital_output.StartTask()
		if self.analog_lines:
			self.analog_output.StartTask()





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



	def clear(self):
		"""Eexplicitly clear the PyDAQmx tasks.

		Clearing is necessary to re-create the channels.
		Note that the tasks are automatically cleared when garbage-collected
		(for example deleting the Pattern).

		Note that after this the values on the physical channels are NOT
		reset to zero.
		"""
		self.analog_output.ClearTask()
		self.digital_output.ClearTask()



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

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

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

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


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

		assert rate > 0.

		return numpy.rint(ms*rate/1000.).astype(numpy.int)






	def write(self, seq, alines = [], anames = None, dlines = [], dnames = None,
			rate = 10000., minV = -10., maxV = 10., delays = None, ext_trig = None, retriggerable=False, counter_channel = "/Dev1/ctr1", counter_source = "/Dev1/Ctr1InternalOutput"):
		"""
		Write the pattern to the National Instruments device using PyDAQmx.

		Parameters
		----------
		seq : numpy structured array
			the data to write to the buffered output.

		alines : list of strings
			lines of the National Instruments board used as analog output

		anames : list of strings
			list of field in seq corresponding to output for analog lines

		dlines : list of strings
			lines of the National Instruments board used as digital output

		dnames : list of strings
			list of field in seq corresponding to output for digital lines

		rate : float
			the rate of the National Instrument clock in hertz

		minV : float
			minimum voltage output for the analog task

		maxV : float
			maximum voltage output for the analog task

		delays : dict or callable
			optional dictionary of delays to apply to the output for some
			channels. It will be called by items in anames and dnames.
		ext_trig : None or str
			optional name of the channel for ext. trigger, e.g. "Dev0/PFI0"
			if not None, the pattern will wait for a trigger on this port


		Note
		----
		The structured array seq contains the data and timing to write.
		Each column is interpretated as the output to write to an analog or
		digital channel.
		Please ensure a column 't' specifying timing in milliseconds.
		See the class usage fro an example.



		"""

		if seq.dtype.names:
			self.sequence = seq
		else:
			raise Exception('Please use a structured array for sequence!')


		# check for the time column
		try:
			self.dt = self.sequence['t']
		except:
			raise Exception("Please ensure a field 't' in the sequence structured array!")

		self.analog_lines = alines
		self.digital_lines = dlines
		self.analog_channels =  (anames if anames else alines)
		self.digital_channels = (dnames if dnames else dlines)
		self.rate = rate
		self.delays = delays


		# string specifying the output lines
		alines = ",".join(self.analog_lines)
		dlines = ",".join(self.digital_lines)

		# set the buffer size
		self.buffer_size = sum(self.ms2n(self.dt))
	
		if retriggerable==False:
			counter_source=""

		# Set the analog output task
		# 1. create ao channel
		# 2. set clock timing
		# 3. write bufferable ouput
		#
		# CreateAOVoltageChan(
		#			range of physical channels,
		#			names of the channels (empty = use physical names)
		#			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
		#			);
		#
		# WriteAnalogF64(
		#			samps per channel to generate or buffer_size per channel,
		#			auostart?,
		#			timeout in seconds,
		#			data layout (DAQmx_Val_GroupByScanNumber = interleaved),
		#			word in float 64,
		#			output = samps per channel written (pyDAQmx),
		#			None (reserved)
		# );
		#

		if self.analog_lines:
			self.analog_output.CreateAOVoltageChan(alines, "", minV, maxV, DAQmx_Val_Volts, None)
			self.analog_output.CfgSampClkTiming(counter_source, self.rate, DAQmx_Val_Rising, DAQmx_Val_ContSamps, self.buffer_size)
			
			if ext_trig is not None and retriggerable==False:
				self.analog_output.CfgDigEdgeStartTrig(ext_trig, DAQmx_Val_Rising)
		
			
			self.analog_output.WriteAnalogF64(self.buffer_size, 0, 10., DAQmx_Val_GroupByScanNumber, self.analog_word(), ctypes.byref(self.read), None)

		# Set the digital output task
		# 1. create do channel
		# 2. set clock timing (the same of ao for syncronization)
		# 3. write bufferable ouput
		#
		# CreateDOchan(
		#			range of physical channels,
		#			names of the channels (empty = use physical names),
		#			lineGrouping (DAQmx_Val_ChanPerLine = 1 channel per line)
		#			);
		#
		# CfgSampClkTiming(
		#			clock terminal ("" = use onboard clock, "/Dev1/ao/SampleClock" = use the same of analog out),
		#			sampling rate per second (ignored if you are using the analog out clock),
		#			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
		#			);
		#
		# WriteDigitalLines(
		#			samps per channel to generate or buffer_size per channel,
		#			auostart?,
		#			timeout in seconds,
		#			data layout (DAQmx_Val_GroupByScanNumber = interleaved),
		#			word in unsigned int 8,
		#			output = samps per channel written (pyDAQmx),
		#			None (reserved)
		# );
		#
		if self.digital_lines:
			# digital lines share the analog lines clock
			# note that to use an arbitrary rate (that is, not the "100kHzTimebase")
			# the digital channels need to exploit an external clock or the
			# sample clock of analog input/output

			if self.analog_lines:
				doclock = "/Dev1/ao/SampleClock"
			else:
				if retriggerable == False:
					#No analog lines found. Digital lines clock switched to 100kHzTimebase.
					# Rate set to 100 kHz. User rate ignored.
					doclock = "100kHzTimebase"
					self.rate = 100000.
				else:
					doclock = counter_source
				self.buffer_size = sum(self.ms2n(self.dt))


			
			self.digital_output.CreateDOChan(dlines, "", DAQmx_Val_ChanPerLine)
			self.digital_output.CfgSampClkTiming(doclock, self.rate, DAQmx_Val_Rising, DAQmx_Val_ContSamps, self.buffer_size)
			
			if (ext_trig is not None) and not self.analog_lines and retriggerable==False:
				self.digital_output.CfgDigEdgeStartTrig(ext_trig, DAQmx_Val_Rising)
			
			self.digital_output.WriteDigitalLines(self.buffer_size, 0, 10., DAQmx_Val_GroupByScanNumber, self.digital_word(), ctypes.byref(self.read), None)
		
		# set up a retriggerable clock similar to acuisition
		if retriggerable:
			if ext_trig is None:
				raise Exception("Please specify a trigger port when asking retriggerable patterns!")

			
			self.counter = Task()
			
			self.counter.CreateCOPulseChanFreq(counter_channel, "", DAQmx_Val_Hz, DAQmx_Val_Low, 0, self.rate, 0.5);
			self.counter.CfgImplicitTiming(DAQmx_Val_FiniteSamps, self.buffer_size)
			self.counter.CfgDigEdgeStartTrig(ext_trig, DAQmx_Val_Rising)
			self.counter.SetStartTrigRetriggerable(True)
			self.counter.StartTask() # counter starts only once




	def analog_word(self):
		"""Compute the analog word from the sequence structured array.
		The word is produced for interleaved channels.

		"""
		# analog only structured array
		x = self.sequence[self.analog_channels]
		# repeat according to the pattern rate
		x = x.repeat(self.ms2n(self.dt))

		# apply delay
		for channel in self.analog_channels:
			delay = (self.delays.get(channel,0.) if self.delays else 0.)
			# add delay, see http://stackoverflow.com/questions/4998587/shear-a-numpy-array
			x[channel] = numpy.roll(x[channel], self.ms2n(delay))


		# convert from structured array to array in the format required by NI DAQmx
		# see numpy structure array docs
		y = lrf.structured_to_unstructured(x).astype(numpy.float64)

		# bird is the word
		# this produces an interleaved word, to be used with DAQmx_Val_GroupByScanNumber
		bird = y.reshape(-1)
		return bird

	def digital_word(self):
		"""Compute the digital word from the sequence structured array.
		The word is produced for interleaved channels.

		"""
		# digital only structured array
		x = self.sequence[self.digital_channels]
		# repeat according to the pattern rate
		x = x.repeat(self.ms2n(self.dt))

		# apply delay
		for channel in self.digital_channels:
			delay = (self.delays.get(channel,0.) if self.delays else 0.)

			# add delay, see http://stackoverflow.com/questions/4998587/shear-a-numpy-array
			x[channel] = numpy.roll(x[channel], self.ms2n(delay))




		# convert from structured array to array
		# convert in the format required by NI DAQmx
		# the number type is the correct one to be written by WriteDigitalLines
		y = lrf.structured_to_unstructured(x).astype(numpy.uint8)

		# bird is the word
		# this produces an interleaved word, to be used with DAQmx_Val_GroupByScanNumber
		bird = y.reshape(-1)
		return bird


	def is_running(self):
		"""Return true if the patter is running, false if the pattern is
		cleared or stopped.

		"""
		read = ctypes.c_ulong()

		self.digital_output.IsTaskDone(ctypes.byref(read))
		is_digital_done = bool(read)

		self.analog_output.IsTaskDone(ctypes.byref(read))
		is_analog_done = bool(read)

		if is_digital_done and is_analog_done:
			return False
		else:
			return True