Ticket #47971: patch-DTT-src-awg-awgbase.py.in.diff

DTT/src/awg/awgbase.py.in

@@ +1 @@
+# -*- mode: python -*-
+# Version $Id$
+"""
+ctypes wrapping of AWG API
+
+Author: Christopher Wipf, Massachusetts Institute of Technology
+  
+This module calls the libawg library for the standard set of
+waveforms, the libtestpoint library for testpoint handling, and the
+libSIStr library (from awgstream) for arbitrary loops and streams.
+
+"""
+import os
+import platform
+from ctypes import CDLL, Structure, c_int, c_uint, c_longlong, c_ulonglong, \
+    c_float, c_double, c_char, c_char_p, POINTER
+from numpy import array
+from numpy.ctypeslib import ndpointer
+
+__docformat__ = 'restructuredtext'
+
+sys = platform.system( )
+ext = '.so'
+awglibdir = '__libdir__'
+if ( len(awglibdir) > 0 ):
+    awglibdir += os.sep
+if ( sys == 'Darwin' ):
+    ext = '.dylib'
+libawg = CDLL(awglibdir + 'libawg' + ext)
+libtestpoint = CDLL(awglibdir + 'libtestpoint' + ext)
+libSIStr = None
+try:
+    libSIStr = CDLL(awglibdir + 'libSIStr' + ext)
+except:
+    pass
+
+####
+#### tconv.h
+####
+
+# One epoch expressed in ns
+_EPOCH = 1e9 / 16.0
+
+libawg.TAInow.argtypes = []
+libawg.TAInow.restype = c_ulonglong
+TAInow = libawg.TAInow
+GPSnow = TAInow
+
+
+####
+#### awgtype.h
+####
+
+# AWG_WaveType enum values
+awgNone = 0
+awgSine = 1
+awgSquare = 2
+awgRamp = 3
+awgTriangle = 4
+awgImpulse = 5
+awgConst = 6
+awgNoiseN = 7
+awgNoiseU = 8
+awgArb = 9
+awgStream = 10
+
+# Phasing types
+AWG_PHASING_STEP = 0
+AWG_PHASING_LINEAR = 1
+AWG_PHASING_QUADRATIC = 2
+AWG_PHASING_LOG = 3
+
+class AWG_Component(Structure):
+    _fields_ = [("wtype", c_uint),
+                ("_par_ctypes", c_double * 4),
+                ("start", c_ulonglong),
+                ("duration", c_ulonglong),
+                ("restart", c_ulonglong),
+                ("ramptype", c_int),
+                ("_ramptime_ctypes", c_ulonglong * 2),
+                ("_ramppar_ctypes", c_double * 4)]
+    @property
+    def par(self):
+        return tuple(self._par_ctypes)
+
+    @par.setter
+    def par(self, val):
+        self._par_ctypes = (c_double * 4)(*val)
+
+    @property
+    def ramptime(self):
+        return tuple(self._ramptime_ctypes)
+
+    @ramptime.setter
+    def ramptime(self, val):
+        self._ramptime_ctypes = (c_ulonglong * 2)(*val)
+
+    @property
+    def ramppar(self):
+        return tuple(self._ramppar_ctypes)
+
+    @ramppar.setter
+    def ramppar(self, val):
+        self._ramppar_ctypes = (c_double * 4)(*val)
+
+
+####
+#### awgapi.h
+####
+
+libawg.awgSetChannel.argtypes = [c_char_p,]
+libawg.awgSetChannel.restype = c_int
+awgSetChannel = libawg.awgSetChannel
+
+libawg.awgAddWaveform.argtypes = [c_int, POINTER(AWG_Component), c_int]
+libawg.awgAddWaveform.restype = c_int
+def awgAddWaveform(slot, comp):
+    comp_array = (AWG_Component * len(comp))(*comp)
+    return libawg.awgAddWaveform(slot, comp_array, len(comp))
+
+libawg.awgStopWaveform.argtypes = [c_int, c_int, c_ulonglong]
+libawg.awgStopWaveform.restype = c_int
+awgStopWaveform = libawg.awgStopWaveform
+
+libawg.awgClearWaveforms.argtypes = [c_int,]
+libawg.awgClearWaveforms.restype = c_int
+awgClearWaveforms = libawg.awgClearWaveforms
+
+libawg.awgRemoveChannel.argtypes = [c_int,]
+libawg.awgRemoveChannel.restype = c_int
+awgRemoveChannel = libawg.awgRemoveChannel
+
+libawg.awgSetGain.argtypes = [c_int, c_double, c_ulonglong]
+libawg.awgSetGain.restype = c_int
+awgSetGain = libawg.awgSetGain
+
+libawg.awgSetFilter.argtypes = [c_int, POINTER(c_double), c_int]
+libawg.awgSetFilter.restype = c_int
+def awgSetFilter(slot, coeffs):
+    coeffs_array = (c_double * len(coeffs))(*coeffs)
+    return libawg.awgSetFilter(slot, coeffs_array, len(coeffs))
+
+libawg.awg_cleanup.argtypes = []
+libawg.awg_cleanup.restype = None
+awg_cleanup = libawg.awg_cleanup
+
+
+####
+#### testpoint.h
+####
+
+libtestpoint.tpRequestName.argtypes = [c_char_p, c_ulonglong, 
+                                       POINTER(c_ulonglong), POINTER(c_int)]
+libtestpoint.tpRequestName.restype = c_int
+def tpRequestName(tpNames, timeout, time, epoch):
+    if time is not None:
+        time_ptr = POINTER(c_ulonglong)(time)
+    else:
+        time_ptr = POINTER(c_ulonglong)()
+    if epoch is not None:
+        epoch_ptr = POINTER(c_int)(epoch)
+    else:
+        epoch_ptr = POINTER(c_int)()
+    return libtestpoint.tpRequestName(tpNames, timeout, time_ptr, epoch_ptr)
+
+libtestpoint.tpClearName.argtypes = [c_char_p,]
+libtestpoint.tpClearName.restype = c_int
+tpClearName = libtestpoint.tpClearName
+
+libtestpoint.testpoint_cleanup.argtypes = []
+libtestpoint.testpoint_cleanup.restype = None
+testpoint_cleanup = libtestpoint.testpoint_cleanup
+
+
+if libSIStr:
+  ####
+  #### SIStr.h
+  ####
+
+  # Compile-time parameters
+  SIStr_MAXCHANNAMELENGTH = 64
+  # Target "lead time" for sending waveform data, in NANOseconds
+  SIStr_LEADTIME = 7 * 10**9
+  # Status codes
+  SIStr_OK = 0
+
+  class SIStrBuf(Structure):
+    pass
+
+  SIStrBuf._fields_ = [('gpstime', c_int),
+                       ('epoch', c_int),
+                       ('iblock', c_int),
+                       ('size', c_int),
+                       ('ndata', c_int),
+                       ('next', POINTER(SIStrBuf)),
+                       ('data', POINTER(c_float))]
+
+  class SIStream(Structure):
+    _fields_ = [('magic', c_int),
+                ('id', c_int),
+                ('channel', c_char * SIStr_MAXCHANNAMELENGTH),
+                ('samprate', c_int),
+                ('starttime', c_double),
+                ('slot', c_int),
+                ('tp', c_int),
+                ('comp', c_int),
+                ('blocksize', c_int),
+                ('nblocks', c_int),
+                ('curgps', c_int),
+                ('curepoch', c_int),
+                ('sentgps', c_int),
+                ('sentepoch', c_int),
+                ('nbufs', c_int),
+                ('curbuf', POINTER(SIStrBuf)),
+                ('firstbuf', POINTER(SIStrBuf)),
+                ('lastbuf', POINTER(SIStrBuf)),
+                ('lastsend', c_longlong),
+                ('minwait', c_longlong),
+                ('aborted', c_int)]
+
+  libSIStr.SIStrAppInfo.argtypes = [c_char_p,]
+  libSIStr.SIStrAppInfo.restype = None
+  SIStrAppInfo = libSIStr.SIStrAppInfo
+  
+  libSIStr.SIStrOpen.argtypes = [POINTER(SIStream), c_char_p, c_int, c_double]
+  libSIStr.SIStrOpen.restype = c_int
+  SIStrOpen = libSIStr.SIStrOpen
+
+  libSIStr.SIStrErrorMsg.argtypes = [c_int,]
+  libSIStr.SIStrErrorMsg.restype = c_char_p
+  SIStrErrorMsg = libSIStr.SIStrErrorMsg
+
+  libSIStr.SIStrAppend.argtypes = [POINTER(SIStream), ndpointer(c_float),
+                                 c_int, c_float]
+  libSIStr.SIStrAppend.restype = c_int
+  def SIStrAppend(sis, data, scale):
+      data_array = array(data).astype('f')
+      return libSIStr.SIStrAppend(sis, data_array, data_array.size, scale)
+
+  libSIStr.SIStrClose.argtypes = [POINTER(SIStream),]
+  libSIStr.SIStrClose.restype = c_int
+  SIStrClose = libSIStr.SIStrClose
+
+  libSIStr.SIStrAbort.argtypes = [POINTER(SIStream),]
+  libSIStr.SIStrAbort.restype = c_int
+  SIStrAbort = libSIStr.SIStrAbort