+++ /dev/null
-# Simultaneous, finite, buffered analog inpout/output using comedi drivers
-
-import comedi
-from numpy import array, fromstring, uint16, float32, pi, sin
-import int16_rw
-
-# imports for testing
-from time import sleep
-from scipy.stats import linregress
-from os import system
-
-#VERBOSE = True
-VERBOSE = False
-AO_TRIGGERS_OFF_AI_START = True
-#AO_TRIGGERS_OFF_AI_START = False
-
-class simAioError (Exception) :
- "Simultaneous Analog IO error"
- pass
-
-_example_array = array([0], dtype=uint16) # for typing, since I don't know what type(array) should be
-
-_cmdtest_message = ["success",
- "invalid source",
- "source conflict",
- "invalid argument",
- "argument conflict",
- "invalid chanlist"]
-
-def _print_cmdsrc(source) :
- if source & comedi.TRIG_NONE : print "none|",
- if source & comedi.TRIG_NOW : print "now|",
- if source & comedi.TRIG_FOLLOW : print "follow|",
- if source & comedi.TRIG_TIME : print "time|",
- if source & comedi.TRIG_TIMER : print "timer|",
- if source & comedi.TRIG_COUNT : print "count|",
- if source & comedi.TRIG_EXT : print "ext|",
- if source & comedi.TRIG_INT : print "int|",
- if source & comedi.TRIG_OTHER : print "other|",
-
-def _print_command(cmd) :
- print "subdevice: \t%d" % cmd.subdev
- print "flags: \t0x%x" % cmd.flags
- print "start: \t",
- _print_cmdsrc(cmd.start_src)
- print "\t%d" % cmd.start_arg
- print "scan_begin:\t",
- _print_cmdsrc(cmd.scan_begin_src)
- print "\t%d" % cmd.scan_begin_arg
- print "convert: \t",
- _print_cmdsrc(cmd.convert_src)
- print "\t%d" % cmd.convert_arg
- print "scan_end: \t",
- _print_cmdsrc(cmd.scan_end_src)
- print "\t%d" % cmd.scan_end_arg
- print "stop: \t",
- _print_cmdsrc(cmd.stop_src)
- print "\t%d" % cmd.stop_arg
-
-def _expand_tuple(tup, length) :
- "Expand an iterable TUP to a tuple of LENGTH by repeating the last element"
- if len(tup) > length :
- raise simAioError, "Tuple too long."
- elif len(tup) < length :
- temp_tup = tup + tuple((tup[-1],)*(length-len(tup)))
- tup = temp_tup
- return tup
-
-class aio_obj :
- def __init__(self, filename="/dev/comedi0",
- in_subdevice=-1, in_chan=(0,), in_aref=(0,), in_range=(0,),
- out_subdevice=-1, out_chan=(0,), out_aref=(0,), out_range=(0,)) :
- self._comedi = comedi
- self._filename = filename
- self.state = "Closed"
- self.open()
-
- self._iaref = _expand_tuple(in_aref, len(in_chan))
- self._irange = _expand_tuple(in_range, len(in_chan))
- temp = self._check_options(in_subdevice, in_chan, self._iaref, self._irange, output=False)
- self._isubdev = temp["subdevice"]
- self._ichan_params = temp["chan_params"]
- self._ichan = in_chan
- self.i_nchan = len(self._ichan)
- self._ichanlist = self._comedi.chanlist(self.i_nchan)
- for i in range(self.i_nchan) :
- self._ichanlist[i] = self._comedi.cr_pack(self._ichan[i], self._irange[i], self._iaref[i])
-
- self._oaref = _expand_tuple(out_aref, len(in_chan))
- self._orange = _expand_tuple(out_range, len(in_chan))
- temp = self._check_options(out_subdevice, out_chan, self._oaref, self._orange, output=True)
- self._osubdev = temp["subdevice"]
- self._ochan_params = temp["chan_params"]
- self._ochan = out_chan
- self.o_nchan = len(self._ochan)
- self._ochanlist = self._comedi.chanlist(self.o_nchan)
- for i in range(self.o_nchan) :
- self._ochanlist[i] = self._comedi.cr_pack(self._ochan[i], self._orange[i], self._oaref[i])
-
- self._gen_rough_output_cmd()
- self._gen_rough_input_cmd()
- self.state = "Initialized"
- def __del__(self) :
- self.close()
- def close(self) :
- if self.state != "Closed" :
- self.reset(force=True)
- rc = self._comedi.comedi_close(self._dev)
- if rc < 0 :
- self._comedi.comedi_perror("comedi_close")
- raise simAioError, "Cannot close %s" % self._filename
- if VERBOSE :
- print "Closed %s on fd %d" % (self._filename, self._fd)
- self.state = "Closed"
- def open(self) :
- if self.state != "Closed" :
- raise simAioError, "Invalid state %s" % self.state
- self._dev = self._comedi.comedi_open(self._filename)
- self._fd = self._comedi.comedi_fileno(self._dev)
- if VERBOSE :
- print "Opened %s on fd %d" % (self._filename, self._fd)
- self.state = "Initialized"
- def _check_options(self, subdevice, chan, aref, rnge, output=True) :
- subdevice = self._check_subdevice(subdevice, output=output)
- chan_params = []
- for i in range(len(chan)) :
- chan_params.append(self._check_chan(subdevice, chan[i], aref[i], rnge[i]))
- if VERBOSE :
- if output :
- print "Output",
- else :
- print "Input",
- print " subdevice with channels %s is valid" % (str(chan))
- return {"subdevice":subdevice,
- "chan_params":chan_params}
- def _check_subdevice(self, subdevice, output=True) :
- if output == True :
- target_type = self._comedi.COMEDI_SUBD_AO
- else :
- target_type = self._comedi.COMEDI_SUBD_AI
- if (subdevice < 0) : # autodetect an input device
- subdevice = self._comedi.comedi_find_subdevice_by_type(self._dev, target_type, 0) # 0 is starting subdevice
- else :
- type = self._comedi.comedi_get_subdevice_type(self._dev, subdevice)
- if type != target_type :
- raise simAioError, "Comedi subdevice %d has wrong type %d" % (subdevice, type)
- return subdevice
- def _check_chan(self, subdevice, chan, aref, range) :
- subdev_n_chan = self._comedi.comedi_get_n_channels(self._dev, subdevice)
- if chan >= subdev_n_chan :
- raise simAioError, "Channel %d > subdevice %d's largest chan %d" % (chan, subdevice, subdev_n_chan-1)
- n_range = self._comedi.comedi_get_n_ranges(self._dev, subdevice, chan)
- if range >= n_range :
- raise simAioError, "Range %d > subdevice %d, chan %d's largest range %d" % (range, subdevice, chan, n_range-1)
- maxdata = self._comedi.comedi_get_maxdata(self._dev, subdevice, chan)
- comrange = self._comedi.comedi_get_range(self._dev, subdevice, chan, range)
- return {"maxdata":maxdata, "comrange": comrange}
- def _gen_rough_output_cmd(self) :
- if VERBOSE :
- print "generate rough output command"
- cmd = self._comedi.comedi_cmd_struct()
- cmd.subdev = self._osubdev
- cmd.flags = self._comedi.CMDF_WRITE
- if AO_TRIGGERS_OFF_AI_START :
- cmd.start_src = self._comedi.TRIG_EXT
- cmd.start_arg = 18 # AI_START1 internal AI start signal
- else :
- cmd.start_src = self._comedi.TRIG_INT
- cmd.start_arg = 0
- cmd.scan_begin_src = self._comedi.TRIG_TIMER
- cmd.scan_begin_arg = 1 # temporary value for now
- cmd.convert_src = self._comedi.TRIG_NOW
- cmd.convert_arg = 0
- cmd.scan_end_src = self._comedi.TRIG_COUNT
- cmd.scan_end_arg = self.o_nchan
- cmd.stop_src = self._comedi.TRIG_COUNT
- cmd.stop_arg = 1 # temporary value for now
- cmd.chanlist = self._ochanlist
- cmd.chanlist_len = self.o_nchan
- self._test_cmd(cmd, max_passes=3)
- self._ocmd = cmd
- def _gen_rough_input_cmd(self) :
- if VERBOSE :
- print "generate rough input command"
- cmd = self._comedi.comedi_cmd_struct()
- cmd.subdev = self._isubdev
- cmd.flags = 0
- cmd.start_src = self._comedi.TRIG_INT
- cmd.start_arg = 0
- cmd.scan_begin_src = self._comedi.TRIG_TIMER
- cmd.scan_begin_arg = 1 # temporary value for now
- cmd.convert_src = self._comedi.TRIG_TIMER
- cmd.convert_arg = 1
- cmd.scan_end_src = self._comedi.TRIG_COUNT
- cmd.scan_end_arg = self.i_nchan
- cmd.stop_src = self._comedi.TRIG_COUNT
- cmd.stop_arg = 1 # temporary value for now
- cmd.chanlist = self._ichanlist
- cmd.chanlist_len = self.i_nchan
- self._test_cmd(cmd, max_passes=3)
- self._icmd = cmd
- def _test_cmd(self, cmd, max_passes=1) :
- very_verbose = False
- i = 0
- rc = 0
- if very_verbose :
- print "Testing command:"
- _print_command(cmd)
- while i < max_passes :
- rc = self._comedi.comedi_command_test(self._dev, cmd)
- if (rc == 0) :
- break
- if VERBOSE or very_verbose :
- print "test pass %d, %s" % (i, _cmdtest_message[rc])
- i += 1
- if (VERBOSE or very_verbose) and i < max_passes :
- print "Passing command:"
- _print_command(cmd)
- if i >= max_passes :
- print "Failing command:"
- _print_command(cmd)
- raise simAioError, "Invalid command: %s" % _cmdtest_message[rc]
- def setup(self, nsamps, freq, out_buffer) :
- if self.state != "Initialized" :
- raise simAioError, "Invalid state %s" % self.state
- if type(out_buffer) != type(_example_array) :
- raise simAioError, "out_buffer must be a numpy array, not a %s" % str(type(out_buffer))
- self._ocmd.scan_begin_arg = int(1e9/freq)
- self._ocmd.stop_arg = nsamps
- if VERBOSE :
- print "Configure the board (%d ns per scan, %d samps)" % (self._ocmd.scan_begin_arg, self._ocmd.stop_arg)
- self._onremain = nsamps
- self._test_cmd(self._ocmd)
- rc = self._comedi.comedi_command(self._dev, self._ocmd)
- if rc < 0 :
- self._comedi.comedi_perror("comedi_command")
- raise simAioError, "Error executing output command %d" % rc
- self._icmd.scan_begin_arg = int(1e9/freq)
- self._icmd.stop_arg = nsamps
- self._test_cmd(self._icmd)
- self._inremain = nsamps
- rc = self._comedi.comedi_command(self._dev, self._icmd)
- if rc < 0 :
- self._comedi.comedi_perror("comedi_command")
- raise simAioError, "Error executing input command"
-
- if VERBOSE :
- print "Write %d output samples to the card" % (nsamps*self.o_nchan)
- rc = int16_rw.write_samples(self._fd, nsamps*self.o_nchan, out_buffer, 1)
- if rc != nsamps*self.o_nchan :
- raise simAioError, "Error %d writing output buffer\n" % rc
- rc = int16_rw.write_samples(self._fd, self.o_nchan, out_buffer[-self.o_nchan:], 1) # HACK, add an extra sample for each channel to the output buffer
- if rc != self.o_nchan :
- raise simAioError, "Error %d writing hack output buffer\n" % rc
- # maybe will avoid resetting...
- self._nsamps = nsamps
- self.state = "Setup"
- def arm(self) :
- if self.state != "Setup" :
- raise simAioError, "Invalid state %s" % self.state
- if VERBOSE :
- print "Arm the analog ouptut"
- self._comedi_internal_trigger(self._osubdev)
- self.state = "Armed"
- def start_read(self, in_buffer) :
- if self.state != "Armed" :
- raise simAioError, "Invalid state %s" % self.state
- if VERBOSE :
- print "Start the run"
- self._comedi_internal_trigger(self._isubdev)
- if VERBOSE :
- print "Read %d input samples from the card" % (self._nsamps*self.i_nchan)
- rc = int16_rw.read_samples(self._fd, self._nsamps*self.i_nchan, in_buffer, -1)
- if rc != self._nsamps*self.i_nchan :
- raise simAioError, "Error %d reading input buffer\n" % rc
- self.state = "Read"
- def _comedi_internal_trigger(self, subdevice) :
- data = self._comedi.chanlist(1) # by luck, data is an array of lsampl_t (unsigned ints), as is chanlist
- insn = self._comedi.comedi_insn_struct()
- insn.insn = self._comedi.INSN_INTTRIG
- insn.subdev = subdevice
- insn.data = data
- insn.n = 1
- data[0] = 0
- rc = self._comedi.comedi_do_insn(self._dev, insn)
- def reset(self, force=False) :
- if VERBOSE :
- print "Reset the analog subdevices"
- # clean up after the read
- rc = self._comedi.comedi_cancel(self._dev, self._osubdev)
- if rc < 0 :
- self._comedi.comedi_perror("comedi_cancel")
- raise simAioError, "Error cleaning up output command"
- rc = self._comedi.comedi_cancel(self._dev, self._isubdev)
- if rc < 0 :
- self._comedi.comedi_perror("comedi_cancel")
- raise simAioError, "Error cleaning up input command"
- self.state = "Initialized"
-
-
-# define the test suite
-
-def _test_aio_obj(aio=None, start_wait=0, verbose=False) :
- if (verbose) :
- print "_test_aio_obj(start_wait = %g)" % start_wait
- nsamps = 10
- out_data = array([0]*nsamps, dtype=uint16)
- in_data = array([0]*nsamps, dtype=uint16)
- for i in range(nsamps) :
- out_data[i] = int(30000.0+3000.0*sin(2*pi*i/float(nsamps)))
- aio.setup(10, 1000, out_data)
- aio.arm()
- sleep(start_wait)
- aio.start_read(in_data)
- aio.reset()
- if (verbose) :
- print "out_data:\n", out_data
- print "in_data:\n", in_data
- print "residual:\n[",
- for i, o in zip(in_data, out_data) :
- print int(i)-int(o),
- print "]"
- return (out_data, in_data)
-
-def _repeat_aio_test(aio=None, num_tests=100, start_wait=0, verbose=False) :
- print "_repeat_aio_test()"
- grads = array([0]*num_tests, dtype=float32)
- good = 0
- bad = 0
- good_run = 0
- good_run_arr = []
- for i in range(num_tests) :
- out_data, in_data = _test_aio_obj(aio, start_wait)
- gradient, intercept, r_value, p_value, std_err = linregress(out_data, in_data)
- grads[i] = gradient
- if verbose :
- print "wait %2d, run %2d, gradient %g" % (start_wait, i, gradient)
- if gradient < .7 :
- bad += 1
- good_run_arr.append(good_run)
- good_run = 0
- else :
- good += 1
- good_run += 1
- good_run_arr.append(good_run)
- print "failure rate %g%% in %d runs" % ((float(bad)/float(good+bad))*100.0, num_tests)
- call = 'echo "'
- for num in good_run_arr :
- call += "%d " % num
- call += '" | stem_leaf 2'
- print "good run stem and leaf:"
- system(call)
-
-def _test_aio_obj_multi_chan(aio=None, start_wait=0, verbose=False) :
- if (verbose) :
- print "_test_aio_obj_multi_chan(start_wait = %g)" % start_wait
- nsamps = 10
- out_data = array([0]*nsamps*aio.o_nchan, dtype=uint16)
- in_data = array([0]*nsamps*aio.i_nchan, dtype=uint16)
- # set up interleaved data
- for i in range(nsamps) :
- out_data[i*aio.o_nchan] = int(30000.0+3000.0*sin(2*pi*i/float(nsamps)))
- for j in range(1, aio.o_nchan) :
- out_data[i*aio.o_nchan + j] = 0
- aio.setup(10, 1000, out_data)
- aio.arm()
- sleep(start_wait)
- aio.start_read(in_data)
- aio.reset()
- if (verbose) :
- print "#",
- for j in range(aio.o_nchan) :
- print "%s\t" % aio._ochan[j],
- for j in range(aio.i_nchan) :
- print "%s\t" % aio._ichan[j],
- print ""
- for i in range(nsamps) :
- for j in range(aio.o_nchan) :
- print "%s\t" % out_data[i*aio.o_nchan+j],
- for j in range(aio.i_nchan) :
- print "%s\t" % in_data[i*aio.i_nchan+j],
- print ""
- return (out_data, in_data)
-
-
-
-def test() :
- aio = aio_obj()
- _test_aio_obj(aio, start_wait = 0, verbose=True)
- _test_aio_obj(aio, start_wait = 0.5, verbose=True)
- aio.close()
- aio.open()
- #_repeat_aio_test(aio, num_tests=100, start_wait=0, verbose=False)
- aio.close()
-
- aiom = aio_obj(in_chan=(0,1,2,3), out_chan=(0,1))
- _test_aio_obj_multi_chan(aiom, start_wait = 0, verbose=True)
-
-if __name__ == "__main__" :
- test()
+++ /dev/null
-# Use Comedi drivers for single-shot analog input/output
-
-import comedi
-
-VERSION = 0.0
-VERBOSE_DEBUG = True
-
-class sngAioError (Exception) :
- "Single point Analog IO error"
- pass
-
-class ai_obj :
- def __init__(self, filename="/dev/comedi0", subdevice=-1, chan=(0,1,2,3), aref=0, range=0) :
- self.verbose = False
- self.comedi = comedi
- self.filename = filename
- self.state = "Closed"
- self.open()
- if (subdevice < 0) : # autodetect an output device
- self.subdev = self.comedi.comedi_find_subdevice_by_type(self.dev, self.comedi.COMEDI_SUBD_AI, 0) # 0 is starting subdevice
- else :
- self.subdev = subdevice
- type = self.comedi.comedi_get_subdevice_type(self.dev, self.subdev)
- if type != self.comedi.COMEDI_SUBD_AI :
- raise sngAioError, "Comedi subdevice %d has wrong type %d" % (self.subdev, type)
- self.chan = chan
- self.aref = aref
- self.range = range
- subdev_n_chan = self.comedi.comedi_get_n_channels(self.dev, self.subdev)
- self.maxdata = []
- self.comedi_range = []
- for chan in self.chan :
- if int(chan) != chan :
- raise sngAioError, "Channels must be integers, not %s" % str(chan)
- if chan >= subdev_n_chan :
- raise sngAioError, "Channel %d > subdevice %d's largest chan %d" % (chan, self.subdev, subdev_n_chan-1)
- n_range = self.comedi.comedi_get_n_ranges(self.dev, self.subdev, chan)
- if range > n_range :
- raise sngAioError, "Range %d > subdevice %d, chan %d's largest range %d" % (range, subdev, chan, n_range-1)
- maxdata = self.comedi.comedi_get_maxdata(self.dev, self.subdev, chan)
- self.maxdata.append(maxdata)
- comrange = self.comedi.comedi_get_range(self.dev, self.subdev, chan, range)
- # comrange becomes invalid if device is closed, so make a copy...
- comrange_copy = self.comedi.comedi_range()
- comrange_copy.min = comrange.min
- comrange_copy.max = comrange.max
- comrange_copy.unit = comrange.unit
- self.comedi_range.append(comrange_copy)
- def __del__(self) :
- self.close()
- def open(self) :
- if self.state == "Closed" :
- self.dev = self.comedi.comedi_open(self.filename)
- self.state = "Opened"
- def close(self) :
- if self.state != "Closed" :
- rc = self.comedi.comedi_close(self.dev)
- if rc < 0 :
- self.comedi.comedi_perror("comedi_close")
- raise sngAioError, "Cannot close %s" % self.filename
- self.state = "Closed"
- def comedi_to_phys(self, chan_index, comedi) :
- phys = self.comedi.comedi_to_phys(comedi, self.comedi_range[chan_index], self.maxdata[chan_index])
- if self.verbose :
- print "comedi %d = %g Volts on subdev %d, chan %d, range [%g, %g], max %d" % (comedi, phys, self.subdev, self.chan[chan_index], self.comedi_range[chan_index].max, self.comedi_range[chan_index].min, self.maxdata[chan_index])
- return phys
- def phys_to_comedi(self, chan_index, phys) :
- comedi = self.comedi.comedi_from_phys(phys, self.comedi_range[chan_index], self.maxdata[chan_index])
- if self.verbose :
- print "%g Volts = comedi %d on subdev %d, chan %d, range [%g, %g], max %d" % (phys, comedi, self.subdev, self.chan[chan_index], self.comedi_range[chan_index].max, self.comedi_range[chan_index].min, self.maxdata[chan_index])
- return comedi
- def read_chan_index(self, chan_index) :
- rc, data = self.comedi.comedi_data_read(self.dev, self.subdev, self.chan[chan_index], self.range, self.aref);
- if rc != 1 : # the number of samples read
- raise sngAioError, "comedi_data_read returned %d" % rc
- return data
- def read(self) :
- out = range(len(self.chan))
- for i in range(len(self.chan)) :
- out[i] = self.read_chan_index(i)
- #print "Read %s, got %s" % (str(self.chan), str(out))
- return out
-
-def _test_ai_obj() :
- ai = ai_obj()
- print "read ", ai.read()
- print "read ", ai.read()
- print "read ", ai.read()
- print "read ", ai.read()
- ai.close()
- print "ai success"
-
-class ao_obj :
- def __init__(self, filename="/dev/comedi0", subdevice=-1, chan=(0,1), aref=0, range=0) :
- self.verbose = False
- self.comedi = comedi
- self.filename = filename
- self.state = "Closed"
- self.open()
- if (subdevice < 0) : # autodetect an output device
- self.subdev = self.comedi.comedi_find_subdevice_by_type(self.dev, self.comedi.COMEDI_SUBD_AO, 0) # 0 is starting subdevice
- else :
- self.subdev = subdevice
- type = self.comedi.comedi_get_subdevice_type(self.dev, self.subdev)
- if type != self.comedi.COMEDI_SUBD_AO :
- raise sngAioError, "Comedi subdevice %d has wrong type %d" % (self.subdev, type)
- self.chan = chan
- self.aref = aref
- self.range = range
- subdev_n_chan = self.comedi.comedi_get_n_channels(self.dev, self.subdev)
- self.maxdata = []
- self.comedi_range = []
- for chan in self.chan :
- if chan >= subdev_n_chan :
- raise sngAioError, "Channel %d > subdevice %d's largest chan %d" % (chan, self.subdev, subdev_n_chan-1)
- n_range = self.comedi.comedi_get_n_ranges(self.dev, self.subdev, chan)
- if range > n_range :
- raise sngAioError, "Range %d > subdevice %d, chan %d's largest range %d" % (range, subdev, chan, n_range-1)
- maxdata = self.comedi.comedi_get_maxdata(self.dev, self.subdev, chan)
- self.maxdata.append(maxdata)
- comrange = self.comedi.comedi_get_range(self.dev, self.subdev, chan, range)
- # comrange becomes invalid if device is closed, so make a copy...
- comrange_copy = self.comedi.comedi_range()
- comrange_copy.min = comrange.min
- comrange_copy.max = comrange.max
- comrange_copy.unit = comrange.unit
- self.comedi_range.append(comrange_copy)
- def __del__(self) :
- self.close()
- def open(self) :
- if self.state != "Closed" :
- raise sngAioError, "Invalid state %s" % self.state
- self.dev = self.comedi.comedi_open(self.filename)
- self.state = "Opened"
- def close(self) :
- if self.state != "Closed" :
- for i in range(len(self.chan)) :
- self.write_chan_index(i, self.phys_to_comedi(i, 0))
- rc = self.comedi.comedi_close(self.dev)
- if rc < 0 :
- self.comedi.comedi_perror("comedi_close")
- raise sngAioError, "Cannot close %s" % self.filename
- self.state = "Closed"
- def comedi_to_phys(self, chan_index, comedi) :
- phys = self.comedi.comedi_to_phys(int(comedi), self.comedi_range[chan_index], self.maxdata[chan_index])
- if self.verbose :
- print "comedi %d = %g Volts on subdev %d, chan %d, range [%g, %g], max %d" % (comedi, phys, self.subdev, self.chan[chan_index], self.comedi_range[chan_index].max, self.comedi_range[chan_index].min, self.maxdata[chan_index])
- return phys
- def phys_to_comedi(self, chan_index, phys) :
- comedi = self.comedi.comedi_from_phys(phys, self.comedi_range[chan_index], self.maxdata[chan_index])
- if self.verbose :
- print "%g Volts = comedi %d on subdev %d, chan %d, range [%g, %g], max %d" % (phys, comedi, self.subdev, self.chan[chan_index], self.comedi_range[chan_index].max, self.comedi_range[chan_index].min, self.maxdata[chan_index])
- return comedi
- def write_chan_index(self, chan_index, data) :
- #print "set output on chan %d to %d" % (chan_index, data)
- rc = self.comedi.comedi_data_write(self.dev, self.subdev, self.chan[chan_index], self.range, self.aref, int(data));
- if rc != 1 : # the number of samples written
- raise sngAioError, 'comedi_data_write returned %d' % rc
- def write(self, data) :
- if len(data) != len(self.chan) :
- raise sngAioError, "data length %d != the number of channels (%d)" % (len(data), len(self.chan))
- for i in range(len(self.chan)) :
- self.write_chan_index(i, data[i])
-
-def _test_ao_obj() :
- ao = ao_obj()
- ao.write([0,0])
- ao.write([3000,3000])
- ao.write([0,0])
- ao.close()
- print "ao success"
-
-def _fit_with_residual(out_data, in_data) :
- from scipy.stats import linregress
- gradient, intercept, r_value, p_value, std_err = linregress(out_data, in_data)
- print "y = %g + %g x" % (intercept, gradient)
- print "r = ", r_value # correlation coefficient = covariance / (std_dev_x*std_dev_y)
- print "p = ", p_value # probablility of measuring this ?slope? for non-correlated, normally-distruibuted data
- print "err = ", std_err # root mean sqared error of best fit
- if gradient < .7 or p_value > 0.05 :
- raise sngAioError, "Out channel 0 != in channel 0"
- residual = zeros((points,))
- for i in range(points) :
- pred_y = intercept + gradient * out_data[i]
- residual[i] = in_data[i] - pred_y
- return residual
-
-def plot_data(out_data0, in_data0, residual0, out_data1, in_data1, residual1) :
- try :
- from pylab import plot, show, subplot
- subplot(311)
- plot(out_data0, in_data0, 'r.-', out_data1, in_data1, 'b.')
- subplot(312)
- plot(out_data0, residual0, 'r.', residual1, 'b.')
- subplot(313)
- plot(in_data0, 'r.', out_data1, 'b.')
- show() # if interactive mode is off...
- #raw_input("Press enter to continue") # otherwise, pause
- except ImportError :
- pass # ignore plot erros
-
-def _test_aio() :
- from scipy.stats import linregress
- from numpy import linspace, zeros
- ao = ao_obj(chan=(0,1))
- ai = ai_obj(chan=(0,1))
- start = 0.1 * ao.maxdata[0]
- stop = 0.9 * ao.maxdata[0]
- points = 10
- out_data0 = linspace(start, stop, points)
- out_data1 = linspace(stop, start, points)
- in_data0 = zeros((points,))
- in_data1 = zeros((points,))
- for i in range(points) :
- ao.write([out_data0[i], out_data1[i]])
- id = ai.read()
- in_data0[i] = id[0]
- in_data1[i] = id[1]
- ai.close()
- ao.close()
- residual0 = _fit_with_residual(out_data0, in_data0)
- residual1 = _fit_with_residual(out_data1, in_data1)
- if VERBOSE_DEBUG :
- plot_data(out_data0, in_data0, residual0, out_data1, in_data1, residual1)
- for i in range(points) :
- if abs(residual0[i]) > 10 : # HACK, hardcoded maximum nonlinearity
- raise Exception, "Input 0, point %d (x %d), y-value %d has excessive residual %d" % (i, out_data0[i], in_data0[i], residual0[i])
- if abs(residual1[i]) > 10 : # HACK, hardcoded maximum nonlinearity
- raise Exception, "Input 1, point %d (x %d), y-value %d has excessive residual %d" % (i, out_data1[i], in_data1[i], residual1[i])
-
- print "_test_aio success"
-
-def test() :
- _test_ai_obj()
- _test_ao_obj()
- _test_aio()
-
-if __name__ == "__main__" :
- test()
+++ /dev/null
-"""Use Comedi drivers for single-shot digital input/output
-
-Being single-shot implementations, read/writes will be software timed,
-so this module would not be a good choice if you need millisecond
-resolution. However, it does provide a simple and robust way to
-generate/aquire signals at 1 second and greater timescales.
-"""
-
-import comedi as c
-
-VERSION = 0.0
-
-class dioError (Exception) :
- "Digital IO error"
- pass
-
-class dio_obj :
- def __init__(self, filename="/dev/comedi0", subdevice=2, chan=(0,1,2,3), aref=0, range=0, output=True) :
- self.filename = filename
- self.subdev = subdevice
- self.chan = chan
- self.aref = aref
- self.range = range
- self.output = output
- self.dev = c.comedi_open(filename)
- if self.dev < 0 :
- raise dioError, "Cannot open %s" % self.filename
- type = c.comedi_get_subdevice_type(self.dev, self.subdev)
- if type != c.COMEDI_SUBD_DIO :
- raise dioError, "Comedi subdevice %d has wrong type %d" % (self.subdev, type)
- if self.output :
- self.set_to_output()
- else :
- self.set_to_input()
- def set_to_output(self) :
- for chan in self.chan :
- rc = c.comedi_dio_config(self.dev, self.subdev, chan, c.COMEDI_OUTPUT)
- if rc != 1 : # yes, comedi_dio_config returns 1 on success, -1 on failure, as of comedilib-0.8.1
- raise dioError, 'comedi_dio_config("%s", %d, %d, %d) returned %d' % (self.filename, self.subdev, chan, c.COMEDI_OUTPUT, rc)
- self.output = True
- def set_to_input(self) :
- for chan in self.chan :
- rc = c.comedi_dio_config(self.dev, self.subdev, chan, c.COMEDI_INPUT)
- if rc != 1 :
- raise dioError, 'comedi_dio_config("%s", %d, %d, %d) returned %d' % (self.filename, self.subdev, chan, c.COMEDI_INPUT, rc)
- self.output = False
- def write_chan_index(self, chan_index, data) :
- if self.output != True :
- raise dioError, "Must be an output to write"
- rc = c.comedi_data_write(self.dev, self.subdev, self.chan[chan_index], self.range, self.aref, data);
- if rc != 1 : # the number of samples written
- raise dioError, "comedi_data_write returned %d" % rc
- def read_chan_index(self, chan_index) :
- if self.output == True :
- raise dioError, "Must be an input to read"
- rc, data = c.comedi_data_read(self.dev, self.subdev, self.chan[chan_index], self.range, self.aref);
- if rc != 1 : # the number of samples read
- raise dioError, "comedi_data_read returned %d" % rc
- return data
- def write_port(self, data) :
- "Channel significance increases with array index"
- for i in range(len(self.chan)) :
- self.write_chan_index(i, (data >> i) % 2)
- def read_port(self) :
- data = 0
- for i in range(len(self.chan)) :
- data += self.read_chan_index(i) << i
- return data
-
-class write_dig_port (dio_obj) :
- def __call__(self, data) :
- self.write_port(data)
-
-def _test_dio_obj() :
- d = dio_obj()
- d.set_to_output()
- d.write_chan_index(0, 1)
- d.write_chan_index(0, 0)
- d.write_port(7)
- d.set_to_input()
- data = d.read_chan_index(0)
- print "channel %d is %d" % (d.chan[0], data)
- data = d.read_port()
- print "port value is %d" % data
- print "dio_obj success"
-
-def _test_write_dig_port() :
- p = write_dig_port()
- for data in [0, 1, 2, 3, 4, 5, 6, 7] :
- p(data)
- print "write_dig_port success"
-
-def test() :
- _test_dio_obj()
- _test_write_dig_port()
-
-if __name__ == "__main__" :
- test()
projects / pycomedi.git / commitdiff