Controlling the Rigol DG1022 with Python on Linux (using the usbtmc driver) – Proof of Concept

arbitrary-waveform-test.py

#!/usr/bin/env python2

import usbtmc
import time
from math import sin

listOfDevices = usbtmc.getDeviceList()
dn = listOfDevices[0]
d = usbtmc.UsbTmcDriver(dn)
print d.getName()

voltage = 8
voltage_high = 4
voltage_low = -1
offset = -2
num = 200
frequency = 100000

def rescale(seq, low = 0, high = 16383):
    cur_low = min(seq)
    # shift the sequence to positive values
    if cur_low < 0.0: seq = [val - cur_low for val in seq]
    cur_low = min(seq)
    cur_high = max(seq)
    seq = [int(val*(high-low)*0.999/(cur_high-cur_low)) for val in seq]
    if min(seq) < low or max(seq) > high:
        print seq
        raise NameError("Something went wrong when rescaling values: min: %d, max: %d." % (min(seq), max(seq)))
    return seq

def getSin(samples, periods = 1):
    ## create a list containing  0, 1, 2, ... , num-1
    sequence = range(0,num)
    ## rescale the list to values from 0 to 1
    sequence = [x/float(num) for x in sequence]
    ## create a sine function
    sequence = [sin(x*2*3.14*periods) for x in sequence]
    return sequence

def getSinc(num, periods = 10):
    ## create a list containing  -num/2, -num/2+1, ..., -1, 0, 1, ... , num/2-2, num/2-1
    sequence = range(-num/2,num/2)
    ## rescale the list to values from -0.5 to .5
    sequence = [x/float(num) for x in sequence]
    ## protect against division by 0 and scale to periods:
    sequence = [(x+0.0001/float(num))*2*3.14*periods for x in sequence]
    ## calculate sin(x)/x
    sequence = [sin(x)/x for x in sequence]
    return sequence

def w(command):
    if len(command) < 30: print command
    d.write(command)
    time.sleep(0.2)

#sequence = getSin(num)
sequence = getSinc(num)
sequence = rescale(sequence)

w("OUTP OFF")
w("FUNC USER")
w("FREQ %d" % frequency)
w("VOLT:UNIT VPP")
#w("VOLT %.3f" % voltage)
#w("VOLT:OFFS %.3f" % offset)
w("VOLT:HIGH %.1f" % voltage_high)
w("VOLTage:LOW %.1f" % voltage_low)
w("DATA:DELete VOLATILE")
w("DATA:DAC VOLATILE,%s" % ",".join([str(item) for item in sequence]))
#w("DATA:DAC VOLATILE,8192,16383,8192,0")
time.sleep(.8)
w("FUNC:USER VOLATILE")
time.sleep(.5)
w("FUNCtion:USER?")
print d.read()
w("OUTP ON")

rigolFG.py

#!/usr/bin/env python2

import usbtmc

# Read more on http://blog.philippklaus.de/2012/05/rigol-dg1022-arbitrary-waveform-function-generator/
# 
# This file is similar to https://github.com/sbrinkmann/PyOscilloskop/blob/master/src/rigolScope.py

class RigolFunctionGenerator:
    """Class to control a Rigol DS1000 series oscilloscope"""
    def __init__(self, device = None):
        if(device == None):
            listOfDevices = usbtmc.getDeviceList()
            if(len(listOfDevices) == 0):
                raise ValueError("There is no device to access")
    
            self.device = listOfDevices[0]
        else:
            self.device = device

        self.meas = usbtmc.UsbTmcDriver(self.device)
 
        self.name = self.meas.getName()
        print self.name
 
    def write(self, command):
        """Send an arbitrary command directly to the scope"""
        self.meas.write(command)
 
    def read(self, command):
        """Read an arbitrary amount of data directly from the scope"""
        return self.meas.read(command)
 
    def reset(self):
        """Reset the instrument"""
        self.meas.sendReset()

test.py

#!/usr/bin/env python2

import rigolFG
fg = rigolFG.RigolFunctionGenerator()

usbtmc.py

#!/usr/bin/env python2

# This file is a copy of https://github.com/sbrinkmann/PyOscilloskop/blob/master/src/usbtmc.py

import os

class UsbTmcDriver:
    """Simple implementation of a USBTMC device driver, in the style of visa.h"""

    def __init__(self, device):
        self.device = device
        self.FILE = os.open(device, os.O_RDWR)

        # TODO: Test that the file opened

    def write(self, command):
        os.write(self.FILE, command);

    def read(self, length = 4000):
        return os.read(self.FILE, length)

    def getName(self):
        self.write("*IDN?")
        return self.read(300)

    def sendReset(self):
        self.write("*RST")

def getDeviceList():
    dirList=os.listdir("/dev")
    result=list()

    for fname in dirList:
        if(fname.startswith("usbtmc")):
            result.append("/dev/" + fname)

    return result