Merge pull request #21 from lightwave-lab/development

Version 1.0.7

Author: thomaslima

lightlab/equipment/abstract_drivers/TekScopeAbstract.py

@@ -1,5 +1,4 @@
 import numpy as np
-import pyvisa
 
 from lightlab import logger
 from lightlab.util.data import Waveform, FunctionBundle
@@ -180,17 +179,8 @@ def __transferData(self, chan):
         chStr = 'CH' + str(chan)
         self.setConfigParam('DATA:ENCDG', 'ASCII')
         self.setConfigParam('DATA:SOURCE', chStr)
-        self.open()
-        try:
-            voltRaw = self.mbSession.query_ascii_values('CURV?')
-        except pyvisa.VisaIOError as err:
-            logger.error('Problem during query_ascii_values(\'CURV?\')')
-            try:
-                self.close()
-            except pyvisa.VisaIOError:
-                logger.error('Failed to close! %s', self.address)
-            raise err
-        self.close()
+
+        voltRaw = self.query_ascii_values('CURV?')
         return voltRaw
 
     def __scaleData(self, voltRaw):
@@ -217,7 +207,7 @@ def __scaleData(self, voltRaw):
             * get(self._yScaleParam) \
             + get('YZERO')
 
-        timeDivision = float(self.getConfigParam('HORIZONTAL:MAIN:SCALE'))
+        timeDivision = float(self.getConfigParam('HORIZONTAL:MAIN:SCALE', forceHardware=True))
         time = np.linspace(-1, 1, len(voltage)) / 2 * timeDivision * 10
 
         return time, voltage

lightlab/equipment/lab_instruments/HP_8152A_PM.py

@@ -67,10 +67,10 @@ def proccessWeirdRead(readString):
             val *= -1
         return str(val)
 
-    def query(self, *args, **kwargs):  # pylint: disable=arguments-differ
+    def robust_query(self, *args, **kwargs):  # pylint: disable=arguments-differ
         ''' Conditionally check for read character doubling
         '''
-        retRaw = super().query(*args, **kwargs)  # pylint: disable=arguments-differ
+        retRaw = self.query(*args, **kwargs)  # pylint: disable=arguments-differ
         if self.doReadDoubleCheck:
             return self.proccessWeirdRead(retRaw)
         else:
@@ -89,7 +89,7 @@ def powerDbm(self, channel=1):
         trial = 0
         while trial < 10:  # Sometimes it gets out of range, so we have to try a few times
             self.write('CH' + str(channel))
-            powStr = self.query('TRG')
+            powStr = self.robust_query('TRG')
             v = float(powStr)
             if abs(v) < 999:  # check if it's reasonable
                 break

lightlab/equipment/lab_instruments/Tektronix_DPO4032_Oscope.py

@@ -1,5 +1,156 @@
+import numpy as np
+from lightlab.util.data import Waveform
 from .Tektronix_DPO4034_Oscope import Tektronix_DPO4034_Oscope
+import pyvisa
+from lightlab import visalogger as logger
 
 
 class Tektronix_DPO4032_Oscope(Tektronix_DPO4034_Oscope):
+    '''
+    Manual: https://www.imperial.ac.uk/media/imperial-college/research-centres-and-groups/centre-for-bio-inspired-technology/7293027.PDF
+    '''
     totalChans = 2
+    _recLenParam = 'HORIZONTAL:RECORDLENGTH'
+
+    def timebaseConfig(self, avgCnt=None, duration=None):
+        ''' Timebase and acquisition configure
+
+            Args:
+                avgCnt (int): averaging done by the scope
+                duration (float): time, in seconds, for data to be acquired
+
+            Returns:
+                (dict) The present values of all settings above
+        '''
+        self.setConfigParam('HORIZONTAL:MAIN:SAMPLERATE', 2.5e9)
+
+        if avgCnt is not None and avgCnt > 1:
+            self.setConfigParam('ACQUIRE:NUMAVG', avgCnt, forceHardware=True)
+        if duration is not None:
+            self.setConfigParam('HORIZONTAL:MAIN:SCALE', duration / 10)
+            self.setConfigParam(self._recLenParam, 10 * int(duration * 2.5e9))
+            self.setConfigParam('DATA:START', 1)
+            self.setConfigParam('DATA:STOP', int(duration * 2.5e9))
+
+        presentSettings = dict()
+        presentSettings['avgCnt'] = self.getConfigParam('ACQUIRE:NUMAVG', forceHardware=True)
+        presentSettings['duration'] = self.getConfigParam(
+            'HORIZONTAL:MAIN:SCALE', forceHardware=True)
+        # presentSettings['position'] = self.getConfigParam('HORIZONTAL:MAIN:POSITION', forceHardware=True)
+        presentSettings['nPts'] = self.getConfigParam(self._recLenParam, forceHardware=True)
+        return presentSettings
+
+    def __scaleData(self, voltRaw):
+        ''' Scale to second and voltage units.
+
+            DSA and DPO are very annoying about treating ymult and yscale differently.
+            TDS uses ymult not yscale
+
+            Args:
+                voltRaw (ndarray): what is returned from ``__transferData``
+
+            Returns:
+                (ndarray): time in seconds, centered at t=0 regardless of timebase position
+                (ndarray): voltage in volts
+
+            Notes:
+                The formula for real voltage should be (Y - YOFF) * YSCALE + YZERO.
+                The Y represents the position of the sampled point on-screen,
+                YZERO, the reference voltage, YOFF, the offset position, and
+                YSCALE, the conversion factor between position and voltage.
+        '''
+        get = lambda param: float(self.getConfigParam('WFMOUTPRE:' + param, forceHardware=True))
+        voltage = (np.array(voltRaw) - get('YOFF')) \
+            * get(self._yScaleParam) \
+            + get('YZERO')
+
+        sample_rate = float(self.getConfigParam('HORIZONTAL:MAIN:SAMPLERATE', forceHardware=True))
+        # time = np.linspace(-1, 1, len(voltage)) / 2 * timeDivision * 10
+        time = np.arange(len(voltage)) / sample_rate
+        time -= np.mean(time)
+
+        return time, voltage
+
+    def acquire(self, chans=None, timeout=None, **kwargs):
+        ''' Get waveforms from the scope.
+
+            If chans is None, it won't actually trigger, but it will configure.
+
+            If unspecified, the kwargs will be derived from the previous state of the scope.
+            This is useful if you want to play with it in lab while working with this code too.
+
+            Args:
+                chans (list): which channels to record at the same time and return
+                avgCnt (int): number of averages. special behavior when it is 1
+                duration (float): window width in seconds
+                position (float): trigger delay
+                nPts (int): number of sample points
+                timeout (float): time to wait for averaging to complete in seconds
+                    If it is more than a minute, it will do a test first
+
+
+            Returns:
+                list[Waveform]: recorded signals
+        '''
+        self.timebaseConfig(**kwargs)
+        if chans is None:
+            return
+
+        for c in chans:
+            if c > self.totalChans:
+                raise Exception('Received channel: ' + str(c) +
+                                '. Max channels of this scope is ' + str(self.totalChans))
+
+        # Channel select
+        for ich in range(1, 1 + self.totalChans):
+            thisState = 1 if ich in chans else 0
+            self.setConfigParam('SELECT:CH' + str(ich), thisState)
+
+        isSampling = kwargs.get('avgCnt', 0) == 1
+        self._setupSingleShot(isSampling)
+        self._triggerAcquire(timeout=timeout)
+        wfms = [None] * len(chans)
+        for i, c in enumerate(chans):
+            vRaw = self.__transferData(c)
+            t, v = self.__scaleData(vRaw)
+            # Optical modules might produce 'W' instead of 'V'
+            unit = self.__getUnit()
+            wfms[i] = Waveform(t, v, unit=unit)
+
+        return wfms
+
+    def __getUnit(self):
+        ''' Gets the unit of the waveform as a string.
+
+            Normally, this will be '"V"', which can be converted to 'V'
+        '''
+
+        yunit_query = self.getConfigParam('WFMOUTPRE:YUNIT', forceHardware=True)
+        return yunit_query.replace('"', '')
+
+    def __transferData(self, chan):
+        ''' Returns the raw data pulled from the scope as time (seconds) and voltage (Volts)
+            Args:
+                chan (int): one channel at a time
+
+            Returns:
+                :mod:`data.Waveform`: a time, voltage paired signal
+
+            Todo:
+                Make this binary transfer to go even faster
+        '''
+        chStr = 'CH' + str(chan)
+        self.setConfigParam('DATA:ENCDG', 'ASCII')
+        self.setConfigParam('DATA:SOURCE', chStr)
+        self.open()
+        try:
+            voltRaw = self.mbSession.query_ascii_values('CURV?')
+        except pyvisa.VisaIOError as err:
+            logger.error('Problem during query_ascii_values(\'CURV?\')')
+            try:
+                self.close()
+            except pyvisa.VisaIOError:
+                logger.error('Failed to close! %s', self.address)
+            raise err
+        self.close()
+        return voltRaw

lightlab/equipment/lab_instruments/Tektronix_PPG3202.py

@@ -0,0 +1,132 @@
+from . import VISAInstrumentDriver
+from lightlab.equipment.abstract_drivers import Configurable
+from lightlab.laboratory.instruments import PatternGenerator
+
+import numpy as np
+import time
+from lightlab import visalogger as logger
+
+
+class Tektronix_PPG3202(VISAInstrumentDriver, Configurable):
+    ''' Python driver for Tektronix PPG 3202.
+
+        Basic functionality includes setting all parameters on the main pannel and specifying data rate.
+        Other functionality includes setting output data pattern on specifies channel.
+
+        `Manual <https://www.tek.com/bit-error-rate-tester/patternpro-ppg-series-pattern-generator-manual/ppg1600-ppg3000-ppg3200-0>`
+    '''
+
+    instrument_category = PatternGenerator
+    __ClockDivider = np.array([1, 2, 4, 8, 16])
+    __channels = np.array([1, 2])
+    __patternType = np.array(['PRBS', 'DATA'])
+    __waitTime = 0        # May be needed to prevent Timeout error
+
+    def __init__(self, name='Pattern Generator', address=None, **kwargs):
+        # Address should be something like 'USB0::0xXXXX::0xXXXX::XXXXXXX::INSTR' when using USB connection
+        VISAInstrumentDriver.__init__(self, name=name, address=address, **kwargs)
+        Configurable.__init__(self)
+
+    def __setVoltage(self, chan=None, amp=None, offset=None):
+        ''' Set the voltage on the specified channel
+        '''
+        if amp is not None and chan in self.__channels:
+            time.sleep(self.__waitTime)
+            cmd = str(':VOLT' + str(chan) + ':POS ' + str(amp) + 'V')
+            self.setConfigParam(cmd, None, True)
+        if offset is not None:
+            time.sleep(self.__waitTime)
+            cmd = str(':VOLT' + str(chan) + ':POS:OFFS ' + str(offset) + 'V')
+            self.setConfigParam(cmd, None, True)
+
+    def __setPatternType(self, chan=None, ptype=None):
+        ''' Set the data pattern on the specified channel. Type can only be 'PRBS'
+            or 'DATA'
+        '''
+        if ptype is not None and chan in self.__channels:
+            if ptype not in self.__patternType:
+                logger.exception('Wrong Pattern Type!')
+            else:
+                time.sleep(self.__waitTime)
+                cmd = str(':DIG' + str(chan) + ':PATT:TYPE ' + str(ptype))
+                self.setConfigParam(cmd, None, True)
+
+    def setDataRate(self, rate=None):
+        ''' Set the data rate of the PPG. Data rate can only be in the range of
+            1.5 Gb/s to 32 Gb/s
+        '''
+        if rate is not None:
+            if rate < 1.5 or rate > 32:
+                logger.exception('Invalid Data Rate!')
+            else:
+                time.sleep(self.__waitTime)
+                cmd = str(':FREQ ' + str(rate) + 'e9')
+                self.setConfigParam(cmd, None, True)
+
+    def setMainParam(self, chan=None, amp=None, offset=None, ptype=None):
+        ''' One function to set all parameters on the main window
+        '''
+        if chan is None:
+            logger.exception('Please Specify Channel Number!')
+        else:
+            self.__setVoltage(chan, amp, offset)
+            self.__setPatternType(chan, ptype)
+
+    def setClockDivider(self, div=None):
+        if div is not None:
+            if (div in self.__ClockDivider):
+                time.sleep(self.__waitTime)
+                cmd = str(':OUTP:CLOC:DIV ' + str(div))
+                self.setConfigParam(cmd, None, True)
+            else:
+                logger.exception('Wrong Clock Divider Value!')
+
+    def setDataMemory(self, chan=None, startAddr=None, bit=None, data=None):
+        if chan is not None and chan in self.__channels:
+            time.sleep(self.__waitTime)
+            cmd = str(':DIG' + str(chan) + ':PATT:DATA ' + str(startAddr) + ',' + str(bit) + ',' + str(data))
+            self.setConfigParam(cmd, None, True)
+        else:
+            logger.exception('Please choose Channel 1 or 2!')
+
+    def setHexDataMemory(self, chan=None, startAddr=None, bit=None, Hdata=None):
+        if chan is not None and chan in self.__channels:
+            time.sleep(self.__waitTime)
+            cmd = str(':DIG' + str(chan) + ':PATT:HDAT ' + str(startAddr) + ',' + str(bit) + ',' + str(Hdata))
+            self.setConfigParam(cmd, None, True)
+        else:
+            logger.exception('Please choose Channel 1 or 2!')
+
+    def channelOn(self, chan=None):
+        if chan is not None and chan in self.__channels:
+            time.sleep(self.__waitTime)
+            cmd = str(':OUTP' + str(chan) + ' ON')
+            self.setConfigParam(cmd, None, True)
+        else:
+            logger.exception('Please choose Channel 1 or 2!')
+
+    def channelOff(self, chan=None):
+        if chan is not None and chan in self.__channels:
+            time.sleep(self.__waitTime)
+            cmd = str(':OUTP' + str(chan) + ' OFF')
+            self.setConfigParam(cmd, None, True)
+        else:
+            logger.exception('Please choose Channel 1 or 2!')
+
+    def getAmplitude(self, chan=None):
+        if chan is not None and chan in self.__channels:
+            return self.query(':VOLT' + str(chan) + ':POS?')
+
+    def getOffset(self, chan=None):
+        if chan is not None and chan in self.__channels:
+            return self.query(':VOLT' + str(chan) + ':POS:OFFS?')
+
+    def getDataRate(self):
+        return self.query(':FREQ?')
+
+    def getPatternType(self, chan=None):
+        if chan is not None and chan in self.__channels:
+            return self.query(':DIG' + str(chan) + ':PATT:TYPE?')
+
+    def getClockDivider(self):
+        return self.query(':OUTP:CLOC:DIV?')

lightlab/equipment/visa_bases/driver_base.py

@@ -3,6 +3,7 @@
 import socket
 import time
 from lightlab import visalogger as logger
+from pyvisa.util import from_ascii_block
 
 
 class InstrumentSessionBase(ABC):
@@ -37,6 +38,10 @@ def write(self):
     def query(self):
         pass
 
+    @abstractmethod
+    def wait(self):
+        pass
+
     @abstractmethod
     def clear(self):
         pass
@@ -45,6 +50,13 @@ def clear(self):
     def query_raw_binary(self):
         pass
 
+    def query_ascii_values(self, message, converter='f', separator=',',
+                           container=list):
+        ''' Taken from pvisa.'''
+
+        block = self.query(message)
+        return from_ascii_block(block, converter, separator, container)
+
     def instrID(self):
         r"""Returns the \*IDN? string"""
         return self.query('*IDN?')
@@ -56,7 +68,7 @@ def timeout(self):
 
     @timeout.setter
     @abstractmethod
-    def termination(self, newTimeout):
+    def timeout(self, newTimeout):
         pass