init

src/main.py

@@ -0,0 +1,30 @@
+from rigol_py_rig.rigol import RigolOsc
+
+import matplotlib.pyplot as plt
+
+
+# uv
+def main():
+    osc = RigolOsc("TCPIP::10.112.1.9::INSTR")
+    print(osc.getTrigStatus())
+    osc.single()
+
+    depth = osc.getMemoryDepth()
+    # osc.setPoints(depth)
+
+    ret = osc.channels[0].getWaveform()
+
+    print(osc.channels[0].getOffset())
+
+    print(f"Y inc {osc.channels[0].getYincrement()}")
+    print(f"X inc {osc.channels[0].getXincrement()}")
+
+    print(f"Y orogin {osc.channels[0].getYOrigin()}")
+    print(f"Y reference {osc.channels[0].getYReference()}")
+
+    plt.plot(ret)
+    plt.show()
+
+
+if __name__ == "__main__":
+    main()

src/rigol_py_rig/rigol.py

@@ -0,0 +1,145 @@
+import wave
+from pyvisa.resources.resource import Resource
+import pyvisa
+from enum import Enum
+
+import numpy as np
+
+CHANNEL_COUNT = 4
+
+
+class TrigState(Enum):
+    TD = 0
+    WAIT = (1,)
+    RUN = (2,)
+    AUTO = (3,)
+    STOP = (4,)
+
+
+class Channel:
+    def __init__(self, index, instr):
+        self.index = index
+        self.instr = instr
+        self.enabled = False
+        self.probe = 1
+        self.scale_V = 1e-1
+
+        self.setProbe(self.probe)
+        self.setVScale(self.scale_V)
+
+    def setProbe(self, probe: float):
+        self.instr.write(f":CHANnel{self.index}:PROBe {probe}")
+
+    def setVScale(self, scale: float):
+        self.instr.write(f":CHANnel{self.index}:SCALe {scale}")
+
+    def getWaveformRaw(self):
+        self.instr.write(f":WAVeform:SOURce CHANnel{self.index}")
+        self.instr.write(":WAV:DATA?")
+        response = self.instr.read_raw()
+        print(f"total resp len: {len(response)}")
+        data_start_offset = int(response[1:2].decode("ASCII")) + 2
+        data_len = (len(response) - data_start_offset) // 2 * 2
+
+        print(f"data len{data_len}")
+
+        print(f" raw data byte count:{len(response[data_start_offset:])}")
+        print(f"data offset: {data_start_offset}")
+
+        dt = np.dtype("<u2")
+        datapoints = np.frombuffer(
+            response[(data_start_offset) : (data_start_offset + data_len)], dtype=dt
+        )
+
+        return datapoints
+
+    def getOffset(self):
+
+        ret = self.instr.query(f":CHANnel{self.index}:OFFSet?")
+        return float(ret)
+
+    def setOffset(self, offset: float):
+
+        self.instr.write(f":CHANnel{self.index}:OFFSet {offset}")
+
+    def getWaveform(self):
+
+        waveform = self.getWaveformRaw()
+        offset = self.getYReference()
+
+        waveform = (waveform.astype("int") - offset) * self.getYincrement()
+
+        return waveform
+
+    #
+    def getYincrement(self):
+        self.instr.write(f":WAVeform:SOURce CHANnel{self.index}")
+
+        ret = self.instr.query(":WAVeform:YINCrement?")
+        return float(ret)
+
+    def getXincrement(self):
+
+        self.instr.write(f":WAVeform:SOURce CHANnel{self.index}")
+
+        ret = self.instr.query(":WAVeform:XINCrement?")
+        return float(ret)
+
+    def getYOrigin(self):
+
+        self.instr.write(f":WAVeform:SOURce CHANnel{self.index}")
+
+        ret = self.instr.query(":WAVeform:YORigin?")
+        return int(ret)
+
+    def getYReference(self):
+
+        self.instr.write(f":WAVeform:SOURce CHANnel{self.index}")
+
+        ret = self.instr.query(":WAVeform:YREFerence?")
+        return int(ret)
+
+
+class RigolOsc:
+    def __init__(self, visa_address: str):
+        rm = pyvisa.ResourceManager()
+        self.instr: Resource = rm.open_resource(visa_address)
+
+        self.channels = [Channel(i + 1, self.instr) for i in range(CHANNEL_COUNT)]
+        self.initialize()
+
+    def single(self):  # arm single mode
+        self.instr.write(":SINGle")
+        while self.getTrigStatus() != TrigState.STOP:
+            pass
+
+    def getTrigStatus(self):
+        ret = self.instr.query(":TRIGger:STATus?")
+        if "TD" in ret:
+            return TrigState.TD
+        elif "WAIT" in ret:
+            return TrigState.WAIT
+        elif "RUN" in ret:
+            return TrigState.RUN
+        elif "AUTO" in ret:
+            return TrigState.AUTO
+        elif "STOP" in ret:
+            return TrigState.STOP
+        else:
+            print("error")
+            print(ret)
+            raise (Exception("Unknown state"))
+
+    def setPoints(self, points):
+        self.instr.write(f":WAVeform:POINts {points}")
+
+    def initialize(self):
+        self.instr.write(":WAVeform:FORMat WORD")
+        self.instr.write(":WAVeform:MODE RAW")
+
+    def getMemoryDepth(self):
+
+        ret = self.instr.query(":ACQuire:MDEPth?")
+        depth = int(float(ret))
+        print(depth)
+        return depth