doc/demo/cmd.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright
   4
   5 """Use comedi commands for asyncronous input.
   6
   7 An example for directly using Comedi commands.  Comedi commands
   8 are used for asynchronous acquisition, with the timing controlled
   9 by on-board timers or external events.
  10
  11 Based on David A. Schleef's `comedilib/demo/cmd.c`.
  12 """
  13
  14 import sys as _sys
  15 import time as _time
  16
  17 import numpy as _numpy
  18
  19 from pycomedi import LOG as _LOG
  20 import pycomedi.constant as _constant
  21 from pycomedi.device import Device as _Device
  22 from pycomedi.subdevice import StreamingSubdevice as _StreamingSubdevice
  23 from pycomedi.channel import AnalogChannel as _AnalogChannel
  24 from pycomedi.chanspec import ChanSpec as _ChanSpec
  25 import pycomedi.utility as _utility
  26
  27
  28 def open_channels(device, subdevice, channels, range, aref):
  29     """Subdevice index and list of channel indexes
  30     to ``Subdevice`` instance and list of ``AnalogChannel`` instances
  31     """
  32     if subdevice >= 0:
  33         subdevice = device.subdevice(subdevice, factory=_StreamingSubdevice)
  34     else:
  35         subdevice = device.find_subdevice_by_type(
  36             _constant.SUBDEVICE_TYPE.ai, factory=_StreamingSubdevice)
  37     channels = [subdevice.channel(
  38             index=i, factory=_AnalogChannel, range=range, aref=aref)
  39                 for i in channels]
  40     return(subdevice, channels)
  41
  42 def prepare_command(subdevice, channels, period, num_scans):
  43     """Create a periodic sampling command.
  44
  45     Ask comedilib to create a generic sampling command and then modify
  46     the parts we want.
  47     """
  48     command = subdevice.get_cmd_generic_timed(
  49         len(channels), scan_period_ns=period)
  50     command.chanlist = channels
  51     command.stop_src = _constant.TRIG_SRC.count
  52     command.stop_arg = num_scans
  53     return command
  54
  55 def test_command(subdevice, num_tests=2):
  56     """Adjust a command as necessary to get valid arguments.
  57     """
  58     _LOG.info('command before testing:\n{}'.format(subdevice.cmd))
  59     for i in range(2):
  60         rc = subdevice.command_test()
  61         if rc is None:
  62             _LOG.info('command is valid')
  63             return
  64         _LOG.info('test {} returned {}\n{}'.format(i, rc, subdevice.cmd))
  65     _LOG.error('error preparing command: {}'.format(rc))
  66     _sys.exit(1)
  67 test_command.__test__ = False  # test_command is not a Nose test
  68
  69 def write_data(stream, channels, data, physical=False):
  70     if physical:
  71         converters = [c.get_converter() for c in channels]
  72         physical_data = _numpy.zeros(data.shape, dtype=float32)
  73         for i,c in enumerate(converters):
  74             physical_data[:,i] = c.to_physical(data[:,i])
  75         data = physical_data
  76     for row in range(data.shape[0]):
  77         stream.write('\t'.join(str(x) for x in data[row,:]))
  78         stream.write('\n')
  79
  80 def read(device, subdevice=None, channels=[0], range=0, aref=0, period=0,
  81          num_scans=2, reader=_utility.Reader, physical=False,
  82          stream=_sys.stdout):
  83     """Read ``num_scans`` samples from each specified channel.
  84     """
  85     subdevice,channels = open_channels(
  86         device=device, subdevice=subdevice, channels=channels, range=range,
  87         aref=aref)
  88     subdevice.cmd = prepare_command(
  89         subdevice=subdevice, channels=channels, period=period,
  90         num_scans=num_scans)
  91     rc = test_command(subdevice=subdevice)
  92     read_buffer = _numpy.zeros(
  93         (num_scans, len(channels)),
  94         dtype=subdevice.get_dtype())
  95     reader = reader(subdevice=subdevice, buffer=read_buffer, name='Reader')
  96     start = _time.time()
  97     _LOG.info('start time: {}'.format(start))
  98     subdevice.command()
  99     reader.start()
 100     reader.join()
 101     stop = _time.time()
 102     _LOG.info('stop time: {}'.format(stop))
 103     _LOG.info('time: {}'.format(stop - start))
 104     write_data(
 105         stream=stream, channels=channels, data=read_buffer, physical=physical)
 106
 107
 108 def run(filename, **kwargs):
 109     """
 110     >>> import StringIO
 111     >>> stdout = StringIO.StringIO()
 112     >>> run(filename='/dev/comedi0', stream=stdout)
 113     >>> print(stdout.getvalue())  # doctest: +SKIP
 114     29694
 115     29693
 116     <BLANKLINE>
 117     >>> stdout.truncate(0)
 118     >>> run(filename='/dev/comedi0', reader=_utility.MMapReader, stream=stdout)
 119     >>> print(stdout.getvalue())  # doctest: +SKIP
 120     29691
 121     29691
 122     <BLANKLINE>
 123     """
 124     device = _Device(filename=filename)
 125     device.open()
 126     try:
 127         read(device=device, **kwargs)
 128     finally:
 129         device.close()
 130
 131
 132 if __name__ == '__main__':
 133     import pycomedi_demo_args
 134
 135     args = pycomedi_demo_args.parse_args(
 136         description=__doc__,
 137         argnames=['filename', 'subdevice', 'channels', 'aref', 'range',
 138                   'num-scans', 'mmap', 'frequency', 'physical', 'verbose'])
 139
 140     _LOG.info(('measuring device={0.filename} subdevice={0.subdevice} '
 141                'channels={0.channels} range={0.range} '
 142                'analog-reference={0.aref}'
 143                ).format(args))
 144
 145     if args.mmap:
 146         reader = _utility.MMapReader
 147     else:
 148         reader = _utility.Reader
 149
 150     run(filename=args.filename, subdevice=args.subdevice,
 151         channels=args.channels, aref=args.aref, range=args.range,
 152         num_scans=args.num_scans, reader=reader, period=args.period,
 153         physical=args.physical)