Code: Select all

'''picoscope PicoScope5000 library interface'''

#import sys
#import time
from ctypes import byref, c_long, c_short, c_int, windll, c_void_p, CFUNCTYPE, \
    c_float, c_ulong
from functools import partial
import time

#from ctypes import *  # shouldn't import *, but there are so many items to use...
#from struct import unpack

LIBNAME = 'ps5000.dll'

UnitInfo = {"PICO_DRIVER_VERSION":0,
           "PICO_USB_VERSION":1,
           "PICO_HARDWARE_VERSION":2,
           "PICO_VARIANT_INFO":3,
           "PICO_BATCH_AND_SERIAL":4,
           "PICO_CAL_DATE":5,
           "PICO_KERNEL_VERSION":6}

class PicoConst: # PS5000 Constants

    class SigGen:
        class WaveType:
            PS5000_SINE = 0         #sine wave
            PS5000_SQUARE = 1       #square wave
            PS5000_TRIANGLE = 2     #triangle wave
            PS5000_RAMP_UP = 3      #rising sawtooth
            PS5000_RAMP_DOWN = 4    #falling sawtooth
            PS5000_SINC = 5         #(sin x)/x
            PS5000_GAUSSIAN = 6     #Gaussian
            PS5000_HALF_SINE = 7    #half (full-wave rectified) sine
            PS5000_DC_VOLTAGE = 8   #DC voltage
            PS5000_WHITE_NOISE = 9  #white noise
        class TriggerTypes:
            SIGGEN_RISING = 0     #trigger on rising edge
            SIGGEN_FALLING = 1     #trigger on falling edge
            SIGGEN_GATE_HIGH = 2     #run while trigger is high
            SIGGEN_GATE_LOW = 3     #run while trigger is low
        class TriggerSource:
            SIGGEN_NONE = 0     #run without waiting for trigger
            SIGGEN_SCOPE_TRIG = 1     #use scope trigger
            SIGGEN_AUX_IN = 2     #use AUXIO input
            SIGGEN_EXT_IN = 3     #use EXT input
            SIGGEN_SOFT_TRIG = 4     #wait for software trigger provided by ps5000SigGenSoftwareControl

    class Trigger:
        class Directions:
            ABOVE = 0               #for gated triggers: above a threshold
            BELOW = 1               #for gated triggers: below a threshold
            RISING = 2              #for threshold triggers: rising edge
            FALLING = 3             #for threshold triggers: falling edge
            RISING_OR_FALLING = 4   #for threshold triggers: either edge
            INSIDE = 5              #for window - qualified triggers: inside window
            OUTSIDE = 6             #for window - qualified triggers: outside window
            ENTER = 7               #for window triggers: entering the window
            EXIT = 8                #for window triggers: leaving the window
            ENTER_OR_EXIT = 9       #for window triggers: either entering or leaving the window
            NONE = 10               #no trigger

    class Ranges:# channel range values/codes
    #    RANGE_20MV = 1  # 20 mV
    #    RANGE_50MV = 2  # 50 mV
        LowRange = 3
        RANGE_100MV = 3  # 100 mV
        RANGE_200MV = 4  # 200 mV
        RANGE_500MV = 5  # 500 mV
        RANGE_1V = 6  # 1 V
        RANGE_2V = 7  # 2 V
        RANGE_5V = 8  # 5 V
        RANGE_10V = 9  # 10 V
        RANGE_20V = 10 # 20 V


    # channels
    class Channels:
        CHANNEL_A = 0
        CHANNEL_B = 1
        CHANNEL_NONE = 5
        CHANNELS = [CHANNEL_A, CHANNEL_B, CHANNEL_NONE]
    class Couplings:
        # coupling
        COUPLING_DC = 1
        COUPLING_AC = 0
        COUPLINGS = [COUPLING_DC, COUPLING_AC]

    class RatioModes:
        RATIO_MODE_NONE = 0
        RATIO_MODE_AGGREGATE = 1

class PicoStatus:# PS5000 Driver Return codes
    import copy
    PICO_OK = 0x00000000
    PICO_MAX_UNITS_OPENED = 0x00000001
    PICO_MEMORY_FAIL = 0x00000002
    PICO_NOT_FOUND = 0x00000003
    PICO_FW_FAIL = 0x00000004
    PICO_OPEN_OPERATION_IN_PROGRESS = 0x00000005
    PICO_OPERATION_FAILED = 0x00000006
    PICO_NOT_RESPONDING = 0x00000007
    PICO_CONFIG_FAIL = 0x00000008
    PICO_KERNEL_DRIVER_TOO_OLD = 0x00000009
    PICO_EEPROM_CORRUPT = 0x0000000A
    PICO_OS_NOT_SUPPORTED = 0x0000000B
    PICO_INVALID_HANDLE = 0x0000000C
    PICO_INVALID_PARAMETER = 0x0000000D
    PICO_INVALID_TIMEBASE = 0x0000000E
    PICO_INVALID_VOLTAGE_RANGE = 0x0000000F
    PICO_INVALID_CHANNEL = 0x00000010
    PICO_INVALID_TRIGGER_CHANNEL = 0x00000011
    PICO_INVALID_CONDITION_CHANNEL = 0x00000012
    PICO_NO_SIGNAL_GENERATOR = 0x00000013
    PICO_STREAMING_FAILED = 0x00000014
    PICO_BLOCK_MODE_FAILED = 0x00000015
    PICO_NULL_PARAMETER = 0x00000016
    PICO_ETS_MODE_SET = 0x00000017
    PICO_DATA_NOT_AVAILABLE = 0x00000018
    PICO_STRING_BUFFER_TOO_SMALL = 0x00000019
    PICO_ETS_NOT_SUPPORTED = 0x0000001A
    PICO_AUTO_TRIGGER_TIME_TOO_SHORT = 0x0000001B
    PICO_BUFFER_STALL = 0x0000001C
    PICO_TOO_MANY_SAMPLES = 0x0000001D
    PICO_TOO_MANY_SEGMENTS = 0x0000001E
    PICO_PULSE_WIDTH_QUALIFIER = 0x0000001F
    PICO_DELAY = 0x00000020
    PICO_SOURCE_DETAILS = 0x00000021
    PICO_CONDITIONS = 0x00000022
    PICO_USER_CALLBACK = 0x00000023
    PICO_DEVICE_SAMPLING = 0x00000024
    PICO_NO_SAMPLES_AVAILABLE = 0x00000025
    PICO_SEGMENT_OUT_OF_RANGE = 0x00000026
    PICO_BUSY = 0x00000027
    PICO_STARTINDEX_INVALID = 0x00000028
    PICO_INVALID_INFO = 0x00000029
    PICO_INFO_UNAVAILABLE = 0x0000002A
    PICO_INVALID_SAMPLE_INTERVAL = 0x0000002B
    PICO_TRIGGER_ERROR = 0x0000002C
    PICO_MEMORY = 0x0000002D
    PICO_SIG_GEN_PARAM = 0x0000002E
    PICO_SHOTS_SWEEPS_WARNING = 0x0000002F
    PICO_SIGGEN_TRIGGER_SOURCE = 0x00000030
    PICO_AUX_OUTPUT_CONFLICT = 0x00000031
    PICO_AUX_OUTPUT_ETS_CONFLICT = 0x00000032
    PICO_WARNING_EXT_THRESHOLD_CONFLICT = 0x00000033
    PICO_WARNING_AUX_OUTPUT_CONFLICT = 0x00000034
    PICO_SIGGEN_OUTPUT_OVER_VOLTAGE = 0x00000035
    PICO_DELAY_NULL = 0x00000036
    PICO_INVALID_BUFFER = 0x00000037
    PICO_SIGGEN_OFFSET_VOLTAGE = 0x00000038
    PICO_SIGGEN_PK_TO_PK = 0x00000039
    PICO_CANCELLED = 0x0000003A
    PICO_SEGMENT_NOT_USED = 0x0000003B
    PICO_INVALID_CALL = 0x0000003C
    PICO_NOT_USED = 0x0000003F
    PICO_INVALID_SAMPLERATIO = 0x00000040
    PICO_INVALID_STATE = 0x00000041
    PICO_NOT_ENOUGH_SEGMENTS = 0x00000042
    PICO_DRIVER_FUNCTION = 0x00000043
    PICO_RESERVED = 0x00000044
    PICO_INVALID_COUPLING = 0x00000045
    PICO_BUFFERS_NOT_SET = 0x00000046
    PICO_RATIO_MODE_NOT_SUPPORTED = 0x00000047
    PICO_RAPID_NOT_SUPPORT_AGGREGATION = 0x00000048
    PICO_INVALID_TRIGGER_PROPERTY = 0x00000049
    # this next bit gives me a way to debug and get a name from a result number
    # oh the joys of Python...
    loc = copy.copy(locals())
    NumToName = {}
    for i in loc:
        if i[0:4] == "PICO":
            NumToName[loc[i]] = i

class PicoError(Exception):
    '''pico scope error'''

# TODO: add checks in methods to ensure lib, handle are valid
class PicoWrapper(object):
    '''picoscope PicoScope5000 interface'''

    def __init__(self):
        self.handle = None
        self.blockReady = False

        self.setupScopeReadyCallback()

        # load the library
        self.lib = windll.LoadLibrary(LIBNAME)
        if not self.lib:
            raise PicoError('could not open library: %s' % LIBNAME)


    ################## low level methods #####################
    def __getattr__(self, name):
        '''this method will call methods starting with ps5000 via the library'''

        # get a handle to the requested method if name looks like a ps5000 method
        if name.lower().startswith('ps5000'):
            try:
                func = getattr(self.lib, name)
            except AttributeError:
                raise PicoError('Library "%s" does not support method "%s"' % (LIBNAME, name))
            else:
                # return a partial function of the library method with handle passed in
                return partial(func, self.handle)

        # not a ps5000 request, defer
        else:
            raise AttributeError


    def ps5000CloseUnit(self):
        '''low-level close the interface to the unit'''
        res = self.lib.ps5000CloseUnit(self.handle)
        self.handle = None
        return res


    def ps5000OpenUnit(self):
        '''low-level open interface to unit'''
        self.handle = c_short()
        self.returnStatus = self.lib.ps5000OpenUnit(byref(self.handle))
        return self.returnStatus


    ################## higher level methods #####################
    def allocateBuffer(self, samples):
        buffer = (c_short * samples)()
        return buffer

    def closeUnit(self):
        '''close the unit'''
        res = self.ps5000CloseUnit()
        #print res
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Close Unit: ' + PicoStatus.NumToName[res])


    def flashLed(self, start):
        '''flash led on front of unit'''
        '''< 0 Flash LED Indefinitely
        0 Stop Flashing
        > 0 Flash *start* times'''
        res = self.ps5000FlashLed(start)
        #print "LED:" + PicoStatus.PicoStat[res]
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Flash LED: ' + PicoStatus.NumToName[res])

    def getUnitInfo(self):
        mystr = " " * 20 # String, 20 chars long
        r = c_short() # short integer, for the function to pass string length back to us
        unitRetInfo = {}
        for i in UnitInfo:
            res = scope.ps5000GetUnitInfo(mystr, len(mystr), byref(r), UnitInfo[i])
            if res == PicoStatus.PICO_OK:
                    unitRetInfo[i] = mystr[:r.value - 1] # we do a -1 because the last byte in the string is always a null byte (0x00)
            else:
                raise PicoError('GetInfo: ' + PicoStatus.NumToName[res])
        return unitRetInfo #return a dictionary that has all the details of the scope.


    def getTimebase(self, timebase, noSamples, oversample = 0, segIndex = 0):
        '''return the (time_interval, max_samples) for the given parameters'''

        timeIntervalNanoseconds = c_long()
        maxSamples = c_long()
        res = self.ps5000GetTimebase(timebase, noSamples, byref(timeIntervalNanoseconds),
                                       oversample, byref(maxSamples), segIndex)
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Get Timebase: ' + PicoStatus.NumToName[res])
        else:
            nanoSecondsPerSample = timeIntervalNanoseconds.value
            maxSamplesAvailable = maxSamples.value
            return nanoSecondsPerSample, maxSamplesAvailable

    def getValues(self, numPoints, startIdx = 0):
        '''return requested amount of data '''
        myBuffer = self.allocateBuffer(numPoints) # allocate a buffer of the size requested.
        #print myBuffer
        self.setBuffer(myBuffer) # point the picoscope at our buffer

        retSamples = c_long(numPoints)
        overflow = c_short(0)
        res = self.ps5000GetValues(startIdx, byref(retSamples), 0, PicoConst.RatioModes.RATIO_MODE_NONE, 0, byref(overflow))

        if res <> PicoStatus.PICO_OK:
            raise PicoError('Get Values: ' + PicoStatus.NumToName[res])
        else:
            return myBuffer

    def openUnit(self):
        '''open interface to unit'''
        res = self.ps5000OpenUnit()
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Open Unit: ' + PicoStatus.NumToName[res])

#TODO:
#ps5000OpenUnitAsync
#ps5000OpenUnitProgress

#TODO: FIX THIS!
    def ready(self):
        '''indicate if previous block acquire is complete - returns bool'''
        '''Only useful if scope data is transferred in Async mode!'''
        res = self.ps5000DataReady()
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Set Channel: ' + PicoStatus.NumToName[res])
        else:
            return bool(res)

    def runBlock(self, numPreTrig, numPostTrig, timebase, oversample):
        '''start acquisition of block of data; return expected duration of acquisition in mS'''
        '''self.scopeReady is a pointer to our callback function that tells us when a data block is ready'''
        '''self.scopeReady is setup and defined in setupScopeReadyCallback() and when called by the ps5000'''
        '''driver will set self.blockReady to True so that we know the scope is ready to transfer data'''

        timeIndisposedMs = c_long()
        pParameter = c_void_p()
        self.blockReady = False
        res = self.ps5000RunBlock(numPreTrig, numPostTrig, timebase, oversample, byref(timeIndisposedMs), 0, self.scopeReady, byref(pParameter))
        if res <> PicoStatus.PICO_OK:
            raise PicoError('RunBlock: ' + PicoStatus.NumToName[res])
        else:
            #print pParameter.value
            #print time_indisposed_ms.value
            return timeIndisposedMs.value

    def setBuffer(self, dataBuffer, channel = PicoConst.Channels.CHANNEL_A):
#        register the passed buffer with the ps5000 driver, and pass it the buffer size
        #change  this to set buffer
        res = self.ps5000SetDataBuffer(channel, byref(dataBuffer), len(dataBuffer))
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Get Values: ' + PicoStatus.NumToName[res])

#ps5000RunStreaming

    def setChannel(self, channel, enabled, coupling, range):
        '''select channel and modes'''
        res = self.ps5000SetChannel(channel, enabled, coupling, range)

        if res <> PicoStatus.PICO_OK:
            raise PicoError('Set Channel: ' + PicoStatus.NumToName[res])

#ps5000SetEts

#ps5000SetPulseWidthQualifier
#ps5000SetSigGenArbitrary
#ps5000SetSigGenBuiltIn

    def setTrigger(self, source, threshold, direction, delay, auto_trigger_ms):
        '''set the trigger mode'''
        res = self.ps5000SetTrigger(source, threshold, direction, delay, auto_trigger_ms)

        if res == 0:
            raise PicoError('set_trigger: failed')

    def setupScopeReadyCallback(self):
        def ps5000Callback(retHandle, stat, pointer):
            #This is the callback function that is necessary to tell us when the PS5000 is ready to send data.
            # I suspect it would be better being set up elsewhere...
            if retHandle == self.handle.value:
                self.blockReady = True
                #tell the rest of our class that data is available.
            return 0
            #we have to return 0 as the caller needs a void returned 
        #set up and store the pointer for our callback function, to be passed to the RunBlock Function...
        CallBackDef = CFUNCTYPE(c_int, c_short, c_int, c_void_p)
        self.scopeReady = CallBackDef(ps5000Callback) # pointer for our callback function.

#ps5000_set_trigger2

    def stop(self):
        '''stop scope acquisition'''
        res = self.ps5000Stop()

        if res <> PicoStatus.PICO_OK:
            raise PicoError('Set Channel: ' + PicoStatus.NumToName[res])


class PicoScope5000(PicoWrapper):
    '''PicoScope5000 class utilizing the PicoWrapper with higher level functionality'''

    def __init__(self):
        PicoWrapper.__init__(self)
        self.interval = None
        self.units = None
        self.timebase = None
        self.oversample = None
        self.nsPerSample = None
        self.maxNoSamples = None
        self.CurRange = None

    def autoSetupRange(self, startIdx = 0, numPoints = 32000, Channel = PicoConst.Channels.CHANNEL_A, Coupling = PicoConst.Couplings.COUPLING_AC):
        '''autosetup of V/div to ensure entire signal is captured'''
        myBuffer = self.allocateBuffer(numPoints)
        self.setBuffer(myBuffer)

        self.CurRange = PicoConst.Ranges.LowRange
        retSamples = c_long(numPoints)
        overflow = c_short(0)

        timeIndisposedMs = self.runBlock(0, numPoints, self.timebase, self.oversample)
        waitTime = timeIndisposedMs / 1000
        time.sleep(0.1 + waitTime) # should be long enough for the block to be ready... but just in case...
        while self.blockReady == False:
#            time.sleep(0.1)
            print "PicoWait"
        res = self.ps5000GetValues(startIdx, byref(retSamples), 0, PicoConst.RatioModes.RATIO_MODE_NONE, 0, byref(overflow))
        #self.blockReady = False
        while overflow.value == 1:
            self.CurRange += 1
            #print self.CurRange
            overflow.value = 0
            self.setChannel(Channel, 1, Coupling, self.CurRange)
            time.sleep(0.1)
            self.ps5000Stop()
            self.runBlock(0, numPoints, self.timebase, self.oversample)
            while self.blockReady == False:
                pass
            self.ps5000GetValues(0, byref(retSamples), 0, PicoConst.RatioModes.RATIO_MODE_NONE, 0, byref(overflow))
        myBuffer = None # erase the buffer, we don't care about it here
        if res <> PicoStatus.PICO_OK:
            raise PicoError('Get Values: ' + PicoStatus.NumToName[res])


    def setTimebase(self, maxIntervalNs, minSamples, oversample):
        '''set the timebase specifying...; return actual sample interval'''
        for i in range(256): # ??
            try:
                interval, samples = self.getTimebase(i, minSamples, oversample)
                #maxNoSamples += None
                #units += None
            except PicoError:
                pass
            else:
                # once the sample interval is more than we want, we move back one
                if interval > maxIntervalNs:
                    self.timebase = i - 1
                    self.oversample = oversample
                    self.nsPerSample, self.maxNoSamples = self.getTimebase(self.timebase, minSamples, oversample)
                    print 'selected', self.nsPerSample, self.maxNoSamples
                    return self.nsPerSample

    def getData(self, numSamples):
        '''blocking call to wait for data'''
        self.runBlock(0, numSamples, self.timebase, self.oversample)
        while self.blockReady == False:
            #print self.blockReady
            time.sleep(0.1)
            pass  # TODO: sleep
        "Data ready!"
        return self.getValues(numSamples)
        '''Data is returned as an c_int array'''


    def getTimedData(self, numSamples):
        '''blocking call to wait for data'''
        while not self.ready():
            pass  # TODO: sleep
        return self.getTimesAndValues(self.units, numSamples)


    def acquireBlock(self, num_samples):
        '''acquisition of num_samples at current nsPerSample'''
        self.runBlock(num_samples, self.timebase, 0)
        return self.getData()


    def startBlock(self, numPreTrig = 0, numPostTrig = 10000):
        '''start run_block with current data setup'''
        self.runBlock(numPreTrig, numPostTrig, self.timebase, self.oversample)

#########
#Test Functions

def enableSigGen():
    offset = c_long(-10)
    pkToPk = c_ulong(1000000)
    waveType = c_short(PicoConst.SigGen.WaveType.PS5000_SINE)
    StartFreq = c_float(10000)
    endFreq = c_float(10000)
    freqInc = c_float(0)
    dwellTime = c_float(0)
    sweepType = 0
    whiteNoise = c_short(0)
    shots = c_ulong(0)
    sweeps = c_ulong(0)
    triggerType = PicoConst.SigGen.TriggerTypes.SIGGEN_RISING
    triggerSource = PicoConst.SigGen.TriggerSource.SIGGEN_NONE
    extInThresh = 0
    res = scope.ps5000SetSigGenBuiltIn(offset, pkToPk, waveType, StartFreq, endFreq, freqInc, dwellTime, sweepType, whiteNoise, shots, sweeps, triggerType, triggerSource, extInThresh)
    #print res
    time.sleep(0.5)
    return res

################################################################################
if __name__ == '__main__':

    # example code

    print 'creating'
    scope = PicoScope5000()
    print 'open'
    scope.openUnit()
    print scope.getUnitInfo()
    scope.setChannel(PicoConst.Channels.CHANNEL_A, 1, PicoConst.Couplings.COUPLING_AC, PicoConst.Ranges.LowRange)
    timePerSample = scope.setTimebase(250, 100000, 0)
    print "Time between samples:" , timePerSample, "nS"
    enableSigGen()
    time.sleep(1)
    scope.autoSetupRange()
    scope.ps5000FlashLed(5)
    test = 0
    returnedData = scope.getData(1000) # data is returned as an array of c_shorts
    print tuple(returnedData)
    print "Vols/div Range used: " + str(scope.CurRange)
    print scope.closeUnit()
    print 'done'