pycomedi/utility.py

   1 # Copyright (C) 2010  W. Trevor King
   2 #
   3 # This program is free software: you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation, either version 3 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  15
  16 "Useful utility functions and classes."
  17
  18 import array as _array
  19 import mmap as _mmap
  20 import os as _os
  21 import threading as _threading
  22 import time as _time
  23
  24 import comedi as _comedi
  25 import numpy as _numpy
  26
  27 from . import LOG as _LOG
  28 from . import classes as _classes
  29 from . import constants as _constants
  30
  31
  32 # types from comedi.h
  33 sampl = _numpy.uint16
  34 lsampl = _numpy.uint32
  35 sampl_typecode = 'H'
  36 lsampl_typecode = 'L'
  37
  38 def subdevice_dtype(subdevice):
  39     "Return the appropriate `numpy.dtype` based on subdevice flags"
  40     if subdevice.get_flags().lsampl:
  41         return lsampl
  42     return sampl
  43
  44 def subdevice_typecode(subdevice):
  45     "Return the appropriate `array` type based on subdevice flags"
  46     if subdevice.get_flags().lsampl:
  47         return lsampl_typecode
  48     return sampl_typecode
  49
  50 def sampl_array(list):
  51     "Convert a Python list/tuple into a `sampl` array"
  52     ret = _comedi.sampl_array(len(list))
  53     for i,x in enumerate(list):
  54         ret[i] = x
  55     return ret
  56     #return _array.array(sampl_typecode, list)
  57
  58 def lsampl_array(list):
  59     "Convert a Python list/tuple into an `lsampl` array"
  60     ret = _comedi.lsampl_array(len(list))
  61     for i,x in enumerate(list):
  62         ret[i] = x
  63     return ret
  64     #return _array.array(lsampl_typecode, list)
  65
  66 def set_insn_chanspec(insn, channels):
  67     "Configure `insn.chanspec` from the list `channels`"
  68     chanlist = _classes.Chanlist(channels)
  69     insn.chanspec = chanlist.chanlist()
  70
  71 def set_insn_data(insn, data):
  72     "Configure `insn.data` and `insn.n` from the iterable `data`"
  73     insn.n = len(data)
  74     insn.data = lsampl_array(data)
  75
  76 def set_cmd_chanlist(cmd, channels):
  77     "Configure `cmd.chanlist` and `cmd.chanlist_len` from the list `channels`"
  78     chanlist = _classes.Chanlist(channels)
  79     cmd.chanlist = chanlist.chanlist()
  80     cmd.chanlist_len = len(chanlist)
  81
  82 def set_cmd_data(cmd, data):
  83     "Configure `cmd.data` and `cmd.data_len` from the iterable `data`"
  84     cmd.data = sampl_array(data)
  85     cmd.data_len = len(data)
  86
  87 def inttrig_insn(subdevice):
  88     """Setup an internal trigger for a given `subdevice`
  89
  90     From the Comedi docs `section 4.4`_ (Instruction for internal
  91     triggering):
  92
  93       This special instruction has `INSN_INTTRIG` as the insn flag in
  94       its instruction data structure. Its execution causes an internal
  95       triggering event. This event can, for example, cause the device
  96       driver to start a conversion, or to stop an ongoing
  97       acquisition. The exact meaning of the triggering depends on the
  98       card and its particular driver.
  99
 100       The `data[0]` field of the `INSN_INTTRIG` instruction is
 101       reserved for future use, and should be set to `0`.
 102
 103     From the comedi source (`comedi.comedi_fops.c:parse_insn()`), we
 104     see that the `chanspec` attribute is ignored for `INSN_INTTRIG`,
 105     so we don't bother setting it here.
 106
 107     .. _section 4.4: http://www.comedi.org/doc/x621.html
 108     """
 109     insn = subdevice.insn()
 110     insn.insn = _constants.INSN.inttrig.value
 111     set_insn_data(insn, [0])
 112     return insn
 113
 114
 115 def _builtin_array(array):
 116     "`array` is an array from the builtin :mod:`array` module"
 117     return isinstance(array, _array.array)
 118
 119
 120 class _ReadWriteThread (_threading.Thread):
 121     "Base class for all reader/writer threads"
 122     def __init__(self, subdevice, buffer, name=None):
 123         if name == None:
 124             name = '<%s subdevice %d>' % (
 125                 self.__class__.__name__, subdevice._index)
 126         self.subdevice = subdevice
 127         self.buffer = buffer
 128         super(_ReadWriteThread, self).__init__(name=name)
 129
 130     def _file(self):
 131         """File for reading/writing data to `.subdevice`
 132
 133         This file may use the internal comedi fileno, so do not close
 134         it when you are finished.  The file will eventually be closed
 135         when the backing `Device` instance is closed.
 136         """
 137         return self.subdevice._device.file
 138
 139
 140 class Reader (_ReadWriteThread):
 141     """`read()`-based reader
 142
 143     Examples
 144     --------
 145
 146     Setup a temporary data file for testing.
 147
 148     >>> from os import close, remove
 149     >>> from tempfile import mkstemp
 150     >>> fd,t = mkstemp(suffix='.dat', prefix='pycomedi-')
 151     >>> f = _os.fdopen(fd, 'r+')
 152     >>> buf = _numpy.array([[0,10],[1,11],[2,12]], dtype=_numpy.uint16)
 153     >>> buf.tofile(t)
 154
 155     Override the default `Reader` methods for our dummy subdevice.
 156
 157     >>> class TestReader (Reader):
 158     ...     def _file(self):
 159     ...         return f
 160
 161     Run the test reader.
 162
 163     >>> rbuf = 0*buf
 164     >>> r = TestReader(subdevice=None, buffer=rbuf, name='Reader-doctest')
 165     >>> r.start()
 166     >>> r.join()
 167
 168     The input buffer is updated in place, and is also available as the
 169     reader's `buffer` attribute.
 170
 171     >>> rbuf
 172     array([[ 0, 10],
 173            [ 1, 11],
 174            [ 2, 12]], dtype=uint16)
 175     >>> r.buffer
 176     array([[ 0, 10],
 177            [ 1, 11],
 178            [ 2, 12]], dtype=uint16)
 179
 180     While `numpy` arrays make multi-channel indexing easy, they do
 181     require an external library.  For single-channel input, the
 182     `array` module is sufficient.
 183
 184     >>> f.seek(0)
 185     >>> rbuf = _array.array('H', [0]*buf.size)
 186     >>> r = TestReader(subdevice=None, buffer=rbuf, name='Reader-doctest')
 187     >>> r.start()
 188     >>> r.join()
 189     >>> rbuf
 190     array('H', [0, 10, 1, 11, 2, 12])
 191     >>> r.buffer
 192     array('H', [0, 10, 1, 11, 2, 12])
 193
 194     Cleanup the temporary data file.
 195
 196     >>> f.close()  # no need for `close(fd)`
 197     >>> remove(t)
 198     """
 199     def run(self):
 200         builtin_array = _builtin_array(self.buffer)
 201         f = self._file()
 202         if builtin_array:
 203             # TODO: read into already allocated memory (somehow)
 204             size = len(self.buffer)
 205             a = _array.array(self.buffer.typecode)
 206             a.fromfile(f, size)
 207             self.buffer[:] = a
 208         else:  # numpy.ndarray
 209             # TODO: read into already allocated memory (somehow)
 210             buf = _numpy.fromfile(
 211                 f, dtype=self.buffer.dtype, count=self.buffer.size)
 212             a = _numpy.ndarray(
 213                 shape=self.buffer.shape, dtype=self.buffer.dtype,
 214                 buffer=buf)
 215             self.buffer[:] = a
 216
 217
 218 class Writer (_ReadWriteThread):
 219     """`write()`-based writer
 220
 221     Examples
 222     --------
 223
 224     Setup a temporary data file for testing.
 225
 226     >>> from os import close, remove
 227     >>> from tempfile import mkstemp
 228     >>> fd,t = mkstemp(suffix='.dat', prefix='pycomedi-')
 229     >>> f = _os.fdopen(fd, 'r+')
 230     >>> buf = _numpy.array([[0,10],[1,11],[2,12]], dtype=_numpy.uint16)
 231
 232     Override the default `Writer` methods for our dummy subdevice.
 233
 234     >>> class TestWriter (Writer):
 235     ...     def _file(self):
 236     ...         return f
 237
 238     Run the test writer.
 239
 240     >>> r = TestWriter(subdevice=None, buffer=buf, name='Writer-doctest')
 241     >>> r.start()
 242     >>> r.join()
 243     >>> a = _array.array('H')
 244     >>> a.fromfile(open(t, 'rb'), buf.size)
 245     >>> a
 246     array('H', [0, 10, 1, 11, 2, 12])
 247
 248     While `numpy` arrays make multi-channel indexing easy, they do
 249     require an external library.  For single-channel input, the
 250     `array` module is sufficient.
 251
 252     >>> f.seek(0)
 253     >>> buf = _array.array('H', [2*x for x in buf.flat])
 254     >>> r = TestWriter(subdevice=None, buffer=buf, name='Writer-doctest')
 255     >>> r.start()
 256     >>> r.join()
 257     >>> a = _array.array('H')
 258     >>> a.fromfile(open(t, 'rb'), len(buf))
 259     >>> a
 260     array('H', [0, 20, 2, 22, 4, 24])
 261
 262     Cleanup the temporary data file.
 263
 264     >>> f.close()  # no need for `close(fd)`
 265     >>> remove(t)
 266     """
 267     def run(self):
 268         f = self._file()
 269         self.buffer.tofile(f)
 270         f.flush()
 271
 272
 273 class _MMapReadWriteThread (_ReadWriteThread):
 274     "`mmap()`-based reader/wrtier"
 275     def __init__(self, *args, **kwargs):
 276         super(_MMapReadWriteThread, self).__init__(*args, **kwargs)
 277         self._prot = None
 278
 279     def _sleep_time(self, mmap_size):
 280         "Expected seconds needed to write a tenth of the mmap buffer"
 281         return 0
 282
 283     def run(self):
 284         # all sizes measured in bytes
 285         builtin_array = _builtin_array(self.buffer)
 286         mmap_size = self._mmap_size()
 287         sleep_time = self._sleep_time(mmap_size)
 288         mmap = _mmap.mmap(
 289             self._fileno(), mmap_size, self._prot, _mmap.MAP_SHARED)
 290         mmap_offset = 0
 291         buffer_offset = 0
 292         if builtin_array:
 293             remaining = len(self.buffer)
 294         else:  # numpy.ndarray
 295             remaining = self.buffer.size
 296         remaining *= self.buffer.itemsize
 297         action = self._initial_action(
 298             mmap, buffer_offset, remaining,
 299             mmap_size, action_bytes=mmap_size)
 300         buffer_offset += action
 301         remaining -= action
 302
 303         while remaining > 0:
 304             action_bytes = self._action_bytes()
 305             if action_bytes > 0:
 306                 action,mmap_offset = self._act(
 307                     mmap, mmap_offset, buffer_offset, remaining,
 308                     mmap_size, action_bytes=action_bytes,
 309                     builtin_array=builtin_array)
 310                 buffer_offset += action
 311                 remaining -= action
 312             else:
 313                 _time.sleep(sleep_time)
 314
 315     def _act(self, mmap, mmap_offset, buffer_offset, remaining, mmap_size,
 316              action_bytes=None, builtin_array=None):
 317         if action_bytes == None:
 318             action_bytes = self.subdevice.get_buffer_contents()
 319         if mmap_offset + action_bytes >= mmap_size - 1:
 320             action_bytes = mmap_size - mmap_offset
 321             wrap = True
 322         else:
 323             wrap = False
 324         action_size = min(action_bytes, remaining)
 325         self._mmap_action(mmap, buffer_offset, action_size, builtin_array)
 326         mmap.flush()  # (offset, size),  necessary?  calls msync?
 327         self._mark_action(action_size)
 328         if wrap:
 329             mmap.seek(0)
 330             mmap_offset = 0
 331         return action_size, mmap_offset
 332
 333     # pull out subdevice calls for easier testing
 334
 335     def _mmap_size(self):
 336         return self.subdevice.get_buffer_size()
 337
 338     def _fileno(self):
 339         return self.subdevice._device.fileno()
 340
 341     def _action_bytes(self):
 342         return self.subdevice.get_buffer_contents()
 343
 344     # hooks for subclasses
 345
 346     def _initial_action(self, mmap, buffer_offset, remaining, mmap_size,
 347                         action_bytes=None):
 348         return 0
 349
 350     def _mmap_action(self, mmap, offset, size):
 351         raise NotImplementedError()
 352
 353     def _mark_action(self, size):
 354         raise NotImplementedError()
 355
 356
 357 # MMap classes have more subdevice-based methods to override
 358 _mmap_docstring_overrides = '\n    ...     '.join([
 359         'def _mmap_size(self):',
 360         '    from os.path import getsize',
 361         '    return getsize(t)',
 362         'def _fileno(self):',
 363         '    return fd',
 364         'def _action_bytes(self):',
 365         '    return 4',
 366         'def _mark_action(self, size):',
 367         '    pass',
 368         'def _file',
 369         ])
 370
 371
 372 class MMapReader (_MMapReadWriteThread):
 373     __doc__ = Reader.__doc__
 374     for _from,_to in [
 375         # convert class and function names
 376         ('`read()`', '`mmap()`'),
 377         ('Writer', 'MMapWriter'),
 378         ('def _file', _mmap_docstring_overrides)]:
 379         __doc__ = __doc__.replace(_from, _to)
 380
 381     def __init__(self, *args, **kwargs):
 382         super(MMapReader, self).__init__(*args, **kwargs)
 383         self._prot = _mmap.PROT_READ
 384
 385     def _mmap_action(self, mmap, offset, size, builtin_array):
 386         offset /= self.buffer.itemsize
 387         s = size / self.buffer.itemsize
 388         if builtin_array:
 389             # TODO: read into already allocated memory (somehow)
 390             a = _array.array(self.buffer.typecode)
 391             a.fromstring(mmap.read(size))
 392             self.buffer[offset:offset+s] = a
 393         else:  # numpy.ndarray
 394             # TODO: read into already allocated memory (somehow)
 395             a = _numpy.fromstring(mmap.read(size), dtype=self.buffer.dtype)
 396             self.buffer.flat[offset:offset+s] = a
 397
 398     def _mark_action(self, size):
 399         self.subdevice.mark_buffer_read(size)
 400
 401
 402 class MMapWriter (_MMapReadWriteThread):
 403     __doc__ = Writer.__doc__
 404     for _from,_to in [
 405         ('`write()`', '`mmap()`'),
 406         ('Reader', 'MMapReader'),
 407         ('def _file', _mmap_docstring_overrides)]:
 408         __doc__ = __doc__.replace(_from, _to)
 409
 410     def __init__(self, *args, **kwargs):
 411         super(MMapWriter, self).__init__(*args, **kwargs)
 412         self._prot = _mmap.PROT_WRITE
 413         self.buffer = buffer(self.buffer)
 414
 415     def _mmap_action(self, mmap, offset, size, builtin_array):
 416         mmap.write(self.buffer[offset:offset+size])
 417
 418     def _mark_action(self, size):
 419         self.subdevice.mark_buffer_written(size)
 420
 421
 422 del _mmap_docstring_overrides