pycomedi/channel.pyx

   1 # Copyright
   2
   3 "Wrap channel-wide Comedi functions in `Channel` and related classes"
   4
   5 cimport numpy as _numpy
   6 import numpy as _numpy
   7
   8 cimport _comedi_h
   9 cimport _comedilib_h
  10 from calibration cimport CalibratedConverter as _CalibratedConverter
  11 from range cimport Range as _Range
  12 from subdevice cimport Subdevice as _Subdevice
  13
  14 from pycomedi import LOG as _LOG
  15 from chanspec import ChanSpec as _ChanSpec
  16 from pycomedi import PyComediError as _PyComediError
  17 import _error
  18 import constant as _constant
  19
  20
  21 cdef class Channel (object):
  22     """Class bundling channel-related functions
  23
  24     >>> from .device import Device
  25     >>> from . import constant
  26
  27     >>> d = Device('/dev/comedi0')
  28     >>> d.open()
  29     >>> s = d.get_read_subdevice()
  30     >>> c = s.channel(0)
  31
  32     >>> c.get_maxdata()
  33     65535L
  34     >>> c.get_n_ranges()
  35     16
  36     >>> c.get_range(0)
  37     <Range unit:volt min:-10.0 max:10.0>
  38     >>> c.find_range(constant.UNIT.volt, 0, 5)
  39     <Range unit:volt min:0.0 max:5.0>
  40
  41     >>> d.close()
  42     """
  43     cdef public _Subdevice subdevice
  44     cdef public int index
  45
  46     def __cinit__(self):
  47         self.index = -1
  48
  49     def __init__(self, subdevice, index):
  50         super(Channel, self).__init__()
  51         self.subdevice = subdevice
  52         self.index = index
  53
  54     def get_maxdata(self):
  55         ret = _comedilib_h.comedi_get_maxdata(
  56             self.subdevice.device.device,
  57             self.subdevice.index, self.index)
  58         if ret < 0:
  59             _error.raise_error(function_name='comedi_get_maxdata', ret=ret)
  60         return ret
  61
  62     def get_n_ranges(self):
  63         ret = _comedilib_h.comedi_get_n_ranges(
  64             self.subdevice.device.device,
  65             self.subdevice.index, self.index)
  66         if ret < 0:
  67             _error.raise_error(function_name='comedi_get_n_ranges', ret=ret)
  68         return ret
  69
  70     cdef _get_range(self, index):
  71         cdef _comedilib_h.comedi_range *rng
  72         cdef _Range ret
  73         # Memory pointed to by the return value is freed on Device.close().
  74         rng = _comedilib_h.comedi_get_range(
  75             self.subdevice.device.device,
  76             self.subdevice.index, self.index, index)
  77         if rng is NULL:
  78             _error.raise_error(function_name='comedi_get_range')
  79         ret = _Range(value=index)
  80         ret.set_comedi_range(rng[0])
  81         # rng[0] is a sneaky way to dereference rng, since Cython
  82         # doesn't support *rng.
  83         return ret
  84
  85     def get_range(self, index):
  86         return self._get_range(index)
  87
  88     def _find_range(self, unit, min, max):
  89         "Search for range"
  90         ret = _comedilib_h.comedi_find_range(
  91             self.subdevice.device.device,
  92             self.subdevice.index, self.index,
  93             _constant.bitwise_value(unit), min, max)
  94         if ret < 0:
  95             _error.raise_error(function_name='comedi_find_range', ret=ret)
  96         return ret
  97
  98     def find_range(self, unit, min, max):
  99         """Search for range
 100
 101         `unit` should be an item from `constants.UNIT`.
 102         """
 103         return self.get_range(self._find_range(unit, min, max))
 104
 105
 106 cdef class DigitalChannel (Channel):
 107     """Channel configured for reading or writing digital data.
 108
 109     >>> from .device import Device
 110     >>> from . import constant
 111
 112     >>> d = Device('/dev/comedi0')
 113     >>> d.open()
 114     >>> s = d.find_subdevice_by_type(constant.SUBDEVICE_TYPE.dio)
 115     >>> c = s.channel(0, factory=DigitalChannel)
 116
 117     >>> c.get_maxdata()
 118     1L
 119     >>> c.get_n_ranges()
 120     1
 121     >>> c.get_range(0)
 122     <Range unit:volt min:0.0 max:5.0>
 123
 124     >>> direction = c.dio_get_config()
 125     >>> direction  # doctest: +SKIP
 126     <_NamedInt input>
 127
 128     >>> c.dio_config(_constant.IO_DIRECTION.input)
 129     >>> data = c.dio_read()
 130     >>> data
 131     1
 132
 133     >>> c.dio_config(_constant.IO_DIRECTION.output)
 134     >>> c.dio_write(1)
 135
 136     >>> c.dio_config(direction)
 137
 138     >>> d.close()
 139     """
 140     def dio_config(self, dir):
 141         """Change input/output properties
 142
 143         `dir` should be an item from `constants.IO_DIRECTION`.
 144         """
 145         ret = _comedilib_h.comedi_dio_config(
 146             self.subdevice.device.device,
 147             self.subdevice.index, self.index,
 148             _constant.bitwise_value(dir))
 149         if ret < 0:
 150             _error.raise_error(function_name='comedi_dio_config', ret=ret)
 151
 152     def dio_get_config(self):
 153         """Query input/output properties
 154
 155         Return an item from `constant.IO_DIRECTION`.
 156         """
 157         cpdef unsigned int dir
 158         ret = _comedilib_h.comedi_dio_get_config(
 159             self.subdevice.device.device,
 160            self.subdevice.index, self.index, &dir)
 161         if ret < 0:
 162             _error.raise_error(function_name='comedi_dio_get_config', ret=ret)
 163         return _constant.IO_DIRECTION.index_by_value(dir)
 164
 165     def dio_read(self):
 166         "Read a single bit"
 167         cpdef unsigned int bit
 168         ret = _comedilib_h.comedi_dio_read(
 169             self.subdevice.device.device,
 170             self.subdevice.index, self.index, &bit)
 171         if ret < 0:
 172             _error.raise_error(function_name='comedi_dio_read', ret=ret)
 173         return int(bit)
 174
 175     def dio_write(self, bit):
 176         "Write a single bit"
 177         ret = _comedilib_h.comedi_dio_write(
 178             self.subdevice.device.device,
 179             self.subdevice.index, self.index, bit)
 180         if ret < 0:
 181             _error.raise_error(function_name='comedi_dio_write', ret=ret)
 182
 183
 184 cdef class AnalogChannel (Channel):
 185     """Channel configured for reading or writing analog data.
 186
 187     `range` should be a `Range` instance, `aref` should be an
 188     `constants.AREF` instance.  If not specified, defaults are chosen
 189     based on the capabilities of the subdevice.
 190
 191     >>> from .device import Device
 192     >>> from . import constant
 193
 194     >>> d = Device('/dev/comedi0')
 195     >>> d.open()
 196     >>> s = d.get_read_subdevice()
 197     >>> c = s.channel(0, factory=AnalogChannel)
 198
 199     >>> c.range
 200     <Range unit:volt min:-10.0 max:10.0>
 201     >>> c.aref
 202     <_NamedInt ground>
 203
 204     >>> data = c.data_read()
 205     >>> data  # doctest: +SKIP
 206     32670L
 207     >>> converter = c.get_converter()
 208     >>> converter  # doctest: +NORMALIZE_WHITESPACE
 209     <CalibratedConverter
 210      to_physical:{coefficients:[-10.0, 0.00030518043793392844] origin:0.0}
 211      from_physical:{coefficients:[0.0, 3276.75] origin:-10.0}>
 212     >>> physical_data = converter.to_physical(data)
 213     >>> physical_data  # doctest: +SKIP
 214     -0.029755092698558021
 215     >>> converter.from_physical(physical_data) == data
 216     True
 217
 218     >>> data = c.data_read_n(5)
 219     >>> data  # doctest: +SKIP
 220     array([32674, 32673, 32674, 32672, 32675], dtype=uint32)
 221
 222     >>> c.data_read_hint()
 223     >>> c.data_read()  # doctest: +SKIP
 224     32672L
 225
 226     >>> data = c.data_read_delayed(nano_sec=1e3)
 227     >>> data  # doctest: +SKIP
 228     32672L
 229
 230     >>> s = d.get_write_subdevice()
 231     >>> c = s.channel(0, factory=AnalogChannel)
 232
 233     >>> converter = c.get_converter()
 234     >>> converter  # doctest: +NORMALIZE_WHITESPACE
 235     <CalibratedConverter
 236      to_physical:{coefficients:[-10.0, 0.00030518043793392844] origin:0.0}
 237      from_physical:{coefficients:[0.0, 3276.75] origin:-10.0}>
 238
 239     >>> c.data_write(converter.from_physical(0))
 240
 241     >>> d.close()
 242
 243     Even after the device is closed, the range information is
 244     retained.
 245
 246     >>> c.range
 247     <Range unit:volt min:-10.0 max:10.0>
 248     """
 249     cdef public _Range range
 250     cdef public object aref
 251
 252     def __init__(self, range=None, aref=None, **kwargs):
 253         super(AnalogChannel, self).__init__(**kwargs)
 254         if range == None:
 255             range = self.get_range(0)
 256         self.range = range
 257         if aref == None:
 258             flags = self.subdevice.get_flags()
 259             for ar in _constant.AREF:
 260                 if getattr(flags, ar.name):
 261                     aref = ar
 262                     break
 263                 raise _PyComediError(
 264                     '%s does not support any known analog reference type (%s)'
 265                     % (self.subdevice, flags))
 266         self.aref = aref
 267
 268     # syncronous stuff
 269
 270     def data_read(self):
 271         "Read one sample"
 272         cdef _comedi_h.lsampl_t data
 273         ret = _comedilib_h.comedi_data_read(
 274             self.subdevice.device.device,
 275             self.subdevice.index, self.index,
 276             _constant.bitwise_value(self.range),
 277             _constant.bitwise_value(self.aref),
 278             &data)
 279         if ret < 0:
 280             _error.raise_error(function_name='comedi_data_read', ret=ret)
 281         return data
 282
 283     def data_read_n(self, n):
 284         "Read `n` samples (timing between samples is undefined)."
 285         data = _numpy.ndarray(shape=(n,), dtype=_numpy.uint32)
 286         ret = _comedilib_h.comedi_data_read_n(
 287             self.subdevice.device.device,
 288             self.subdevice.index, self.index,
 289             _constant.bitwise_value(self.range),
 290             _constant.bitwise_value(self.aref),
 291             <_comedilib_h.lsampl_t *>_numpy.PyArray_DATA(data), n)
 292         if ret < 0:
 293             _error.raise_error(function_name='comedi_data_read_n', ret=ret)
 294         return data
 295
 296     def data_read_hint(self):
 297         """Tell driver which channel/range/aref you will read next
 298
 299         Used to prepare an analog input for a subsequent call to
 300         comedi_data_read.  It is not necessary to use this function,
 301         but it can be useful for eliminating inaccuaracies caused by
 302         insufficient settling times when switching the channel or gain
 303         on an analog input.  This function sets an analog input to the
 304         channel, range, and aref specified but does not perform an
 305         actual analog to digital conversion.
 306
 307         Alternatively, one can simply use `.data_read_delayed()`,
 308         which sets up the input, pauses to allow settling, then
 309         performs a conversion.
 310         """
 311         ret = _comedilib_h.comedi_data_read_hint(
 312             self.subdevice.device.device,
 313             self.subdevice.index, self.index,
 314             _constant.bitwise_value(self.range),
 315             _constant.bitwise_value(self.aref))
 316         if ret < 0:
 317             _error.raise_error(function_name='comedi_data_read_hint', ret=ret)
 318
 319     def data_read_delayed(self, nano_sec=0):
 320         """Read single sample after delaying specified settling time.
 321
 322         Although the settling time is specified in integer
 323         nanoseconds, the actual settling time will be rounded up to
 324         the nearest microsecond.
 325         """
 326         cdef _comedi_h.lsampl_t data
 327         ret = _comedilib_h.comedi_data_read_delayed(
 328             self.subdevice.device.device,
 329             self.subdevice.index, self.index,
 330             _constant.bitwise_value(self.range),
 331             _constant.bitwise_value(self.aref),
 332             &data, int(nano_sec))
 333         if ret < 0:
 334             _error.raise_error(function_name='comedi_data_read_delayed',
 335                                ret=ret)
 336         return data
 337
 338     def data_write(self, data):
 339         """Write one sample
 340
 341         Returns 1 (the number of data samples written).
 342         """
 343         ret = _comedilib_h.comedi_data_write(
 344             self.subdevice.device.device,
 345             self.subdevice.index, self.index,
 346             _constant.bitwise_value(self.range),
 347             _constant.bitwise_value(self.aref),
 348             int(data))
 349         if ret != 1:
 350             _error.raise_error(function_name='comedi_data_write', ret=ret)
 351
 352     def chanspec(self):
 353         return _ChanSpec(chan=self.index, range=self.range, aref=self.aref)
 354
 355
 356     cdef _comedilib_h.comedi_polynomial_t get_softcal_converter(
 357         self, direction, calibration):
 358         """
 359
 360         `direction` should be a value from `constant.CONVERSION_DIRECTION`.
 361         """
 362         cdef _comedilib_h.comedi_polynomial_t poly
 363         #rc = _comedilib_h.comedi_get_softcal_converter(
 364         #    self.subdevice.device.device,
 365         #    self.subdevice.index, self.index,
 366         #    _constant.bitwise_value(self.range),
 367         #    _constant.bitwise_value(direction),
 368         #    calibration, &poly)
 369         #if rc < 0:
 370         #    _error.raise_error(function_name='comedi_get_softcal_converter',
 371         #                       ret=rc)
 372         return poly
 373
 374     cdef _comedilib_h.comedi_polynomial_t get_hardcal_converter(
 375         self, direction):
 376         """
 377
 378         `direction` should be a value from `constant.CONVERSION_DIRECTION`.
 379         """
 380         cdef _comedilib_h.comedi_polynomial_t poly
 381         rc = _comedilib_h.comedi_get_hardcal_converter(
 382             self.subdevice.device.device,
 383             self.subdevice.index, self.index,
 384             _constant.bitwise_value(self.range),
 385             _constant.bitwise_value(direction), &poly)
 386         if rc < 0:
 387             _error.raise_error(function_name='comedi_get_hardcal_converter',
 388                                ret=rc)
 389         return poly
 390
 391     cdef _get_converter(self, calibration):
 392         cdef _comedilib_h.comedi_polynomial_t to_physical, from_physical
 393         cdef _CalibratedConverter ret
 394         flags = self.subdevice.get_flags()
 395         if flags.soft_calibrated:
 396             #if calibration is None:
 397             #    calibration = self.subdevice.device.parse_calibration()
 398             raise NotImplementedError()
 399         else:
 400             to_physical = self.get_hardcal_converter(
 401                 _constant.CONVERSION_DIRECTION.to_physical)
 402             from_physical = self.get_hardcal_converter(
 403                 _constant.CONVERSION_DIRECTION.from_physical)
 404         ret = _CalibratedConverter()
 405         ret._to_physical = to_physical
 406         ret._from_physical = from_physical
 407         return ret
 408
 409     def get_converter(self, calibration=None):
 410         return self._get_converter(calibration)