Added measurement methods that use overlappedY()

k-teng/k_teng_interactive.py

@@ -1,6 +1,6 @@
 """
 run this before using this library:
-ipython -i keithley_interactive.py
+ipython -i k_teng_interactive.py
 
 always records iv-t curves
     i-data -> smua.nvbuffer1
@@ -29,10 +29,10 @@ if __name__ == "__main__":
 
 
 from .keithley import keithley as _keithley
+from .keithley.measure import measure_count as _measure_count, measure as _measure
 from .utility import data as _data
 from .utility.data import load_dataframe
 from .utility import file_io
-from .utility import testing
 
 _runtime_vars = {
     "last-measurement": ""
@@ -41,7 +41,7 @@ _runtime_vars = {
 settings = {
     "datadir":      path.expanduser("~/data"),
     "name":         "measurement",
-    "interval":     0.01,
+    "interval":     0.02,
     "beep":         True,
 }
 config_path = path.expanduser("~/.config/k-teng.json")
@@ -52,85 +52,113 @@ test = False
 k = None
 
 
-def _measure(max_measurements=None, max_points_shown=None, monitor=False):
+
+def _update_print(i, ival, vval):
+    print(f"{i:5d} - {ival:.12f} A - {vval:.5f} V" + " "*10, end='\r')
+
+class _Monitor:
     """
-    Monitor the voltage with matplotlib.
+    Monitor v and i data
+    """
+    def __init__(self, max_points_shown=None, use_print=False):
+        self.max_points_shown = max_points_shown
+        self.use_print = use_print
+        self.index = []
+        self.vdata = []
+        self.idata = []
+
+        plt.ion()
+        self.fig1, (self.vax, self.iax) = plt.subplots(2, 1)
+
+        self.vline,  = self.vax.plot(self.index, self.vdata, color="g")
+        self.vax.set_ylabel("Voltage [V]")
+        self.vax.grid(True)
+
+        self.iline,  = self.iax.plot(self.index, self.idata, color="m")
+        self.iax.set_ylabel("Current [A]")
+        self.iax.grid(True)
+
+    def update(self, i, ival, vval):
+        if self.use_print:
+            _update_print(i, ival, vval)
+        self.index.append(i)
+        self.idata.append(ival)
+        self.vdata.append(vval)
+        # update data
+        self.iline.set_xdata(self.index)
+        self.iline.set_ydata(self.idata)
+        self.vline.set_xdata(self.index)
+        self.vline.set_ydata(self.vdata)
+        # recalculate limits and set them for the view
+        self.iax.relim()
+        self.vax.relim()
+        if self.max_points_shown and i > self.max_points_shown:
+            self.iax.set_xlim(i - self.max_points_shown, i)
+            self.vax.set_xlim(i - self.max_points_shown, i)
+        self.iax.autoscale_view()
+        self.vax.autoscale_view()
+        # update plot
+        self.fig1.canvas.draw()
+        self.fig1.canvas.flush_events()
+
+    def __del__(self):
+        plt.close(self.fig1)
+
+
+def monitor_count(count=5000, interval=settings["interval"], max_points_shown=160):
+    """
+    Take <count> measurements in <interval> and monitor live with matplotlib.
 
     @details:
         - Resets the buffers
         - Opens a matplotlib window and takes measurements depending on settings["interval"]
-        - Waits for the user to press a key
+        Uses the device internal overlappedY measurement method, which allows for greater precision
-    @param max_points_shown : how many points should be shown at once. None means infinite
+        You can take the data from the buffer afterwards, using save_csv
-    @param max_measurements : maximum number of measurements. None means infinite
+    @param count: count
-    You can take the data from the buffer afterwards, using save_csv """
+    @param interval: interval, defaults to settings["interval"]
-    global k, settings, test, _runtime_vars
+    @param max_points_shown: how many points should be shown at once. None means infinite
-    print(f"Starting measurement with:\n\tinterval = {settings['interval']}s\nUse <C-c> to stop. Save the data using 'save_csv()' afterwards.")
+    """
-    _runtime_vars["last_measurement"] = dtime.now().isoformat()
+    plt_monitor = _Monitor(max_points_shown, use_print=True)
-    if not test:
+    update_func = plt_monitor.update
-        _keithley.reset(k, verbose=True)
-        k.write("smua.source.output = 1")
-        k.write("format.data = format.ASCII\nformat.asciiprecision = 12")
-    # jupyter:
-    # clear_output(wait=True)
-    # plt.plot(data)
-    # plt.show()
-    index = []
-    vdata = []
-    idata = []
-    if monitor:
-        plt.ion()
-        fig1, (vax, iax) = plt.subplots(2, 1)
 
-        vline,  = vax.plot(index, vdata, color="g")
+    print(f"Starting measurement with:\n\tinterval = {interval}s\nSave the data using 'save_csv()' afterwards.")
-        vax.set_ylabel("Voltage [V]")
-        vax.grid(True)
-
-        iline,  = iax.plot(index, idata, color="m")
-        iax.set_ylabel("Current [A]")
-        iax.grid(True)
     try:
-        i = 0
+        _measure_count(k, V=True, I=True, count=count, interval=interval, beep_done=False, verbose=False, update_func=update_func, update_interval=0.05, testing=test)
-        while max_measurements is None or i < max_measurements:
-            index.append(i)
-            if test:
-                idata.append(testing.testcurve(i, peak_width=1, amplitude=5e-8))
-                vdata.append(-testing.testcurve(i, peak_width=2, amplitude=15))
-                # data.append(np.random.rand())
-            else:
-                # data.append(float(k.query("print(smua.measure.v(smua.nvbuffer1))").strip('\n')))
-                i_val, v_val = tuple(float(v) for v in k.query("print(smua.measure.iv(smua.nvbuffer1, smua.nvbuffer2))").strip('\n').split('\t'))
-                idata.append(i_val)
-                vdata.append(v_val)
-            print(f"{i:5d} - {idata[-1]:.12f} A - {vdata[-1]:.5f} V", end='\r')
-            if monitor:
-                # update data
-                iline.set_xdata(index)
-                iline.set_ydata(idata)
-                vline.set_xdata(index)
-                vline.set_ydata(vdata)
-                # recalculate limits and set them for the view
-                iax.relim()
-                vax.relim()
-                if max_points_shown and i > max_points_shown:
-                    iax.set_xlim(i - max_points_shown, i)
-                    vax.set_xlim(i - max_points_shown, i)
-                iax.autoscale_view()
-                vax.autoscale_view()
-                # update plot
-                fig1.canvas.draw()
-                fig1.canvas.flush_events()
-            sleep(settings["interval"])
-            i += 1
     except KeyboardInterrupt:
         if not test:
-            k.write("smua.source.output = 0")
+            k.write(f"smua.source.output = smua.OUTPUT_OFF")
-        if monitor:
+        print("Monitoring cancelled, measurement might still continue" + " "*50)
-            plt.close(fig1)
+    else:
-        print("Measurement stopped" + " "*50)
+        print("Measurement finished" + " "*50)
-    return vdata, idata
+
+def measure_count(count=5000, interval=settings["interval"]):
+    """
+    Take <count> measurements in <interval>
+
+    @details:
+        - Resets the buffers
+        - Takes <count> measurements depending on settings["interval"]
+        Uses the device internal overlappedY measurement method, which allows for greater precision
+        You can take the data from the buffer afterwards, using save_csv
+    @param count: count
+    @param interval: interval, defaults to settings["interval"]
+    """
+    update_func = _update_print
+
+    print(f"Starting measurement with:\n\tinterval = {interval}s\nSave the data using 'save_csv()' afterwards.")
+    try:
+        _measure_count(k, V=True, I=True, count=count, interval=interval, beep_done=False, verbose=False, update_func=update_func, update_interval=0.05, testing=test)
+    except KeyboardInterrupt:
+        if not test:
+            k.write(f"smua.source.output = smua.OUTPUT_OFF")
+        print("Monitoring cancelled, measurement might still continue" + " "*50)
+    else:
+        print("Measurement finished" + " "*50)
 
 
-def monitor(max_measurements=None, max_points_shown=160):
+
+
+def monitor(interval=settings["interval"], max_measurements=None, max_points_shown=160):
     """
     Monitor the voltage with matplotlib.
 
@@ -138,13 +166,20 @@ def monitor(max_measurements=None, max_points_shown=160):
         - Resets the buffers
         - Opens a matplotlib window and takes measurements depending on settings["interval"]
         - Waits for the user to press a key
+        Uses python's time.sleep() for waiting the interval, which is not very precise. Use measure_count for better precision.
+        You can take the data from the buffer afterwards, using save_csv.
     @param max_points_shown : how many points should be shown at once. None means infinite
     @param max_measurements : maximum number of measurements. None means infinite
-    You can take the data from the buffer afterwards, using save_csv """
+    """
-    _measure(max_measurements=max_measurements, max_points_shown=max_points_shown, monitor=True)
+    global _runtime_vars
+    _runtime_vars["last_measurement"] = dtime.now().isoformat()
+    print(f"Starting measurement with:\n\tinterval = {interval}s\nUse <C-c> to stop. Save the data using 'save_csv()' afterwards.")
+    plt_monitor = _Monitor(use_print=True, max_points_shown=max_points_shown)
+    update_func = plt_monitor.update
+    _measure(k, interval=interval, max_measurements=max_measurements, update_func=update_func, testing=test)
 
 
-def measure(max_measurements=None):
+def measure(interval=settings["interval"], max_measurements=None):
     """
     Measure voltages
 
@@ -152,23 +187,45 @@ def measure(max_measurements=None):
         - Resets the buffers
         - Measure voltages
         - Waits for the user to press a key
+        Uses python's time.sleep() for waiting the interval, which is not very precise. Use measure_count for better precision.
+        You can take the data from the buffer afterwards, using save_csv.
     @param max_measurements : maximum number of measurements. None means infinite
-    You can take the data from the buffer afterwards, using save_csv """
+    """
-    _measure(max_measurements=max_measurements, monitor=False)
+    global _runtime_vars
+    _runtime_vars["last_measurement"] = dtime.now().isoformat()
+    print(f"Starting measurement with:\n\tinterval = {interval}s\nUse <C-c> to stop. Save the data using 'save_csv()' afterwards.")
+    update_func = _update_print
+    _measure(k, interval=interval, max_measurements=max_measurements, update_func=update_func, testing=test)
 
-def automeasure(repeat, repeat_delay=0, max_measurements=None, max_points_shown=120, monitor=True):
+
+def repeat(measure_func: callable, count: int, repeat_delay=0):
     """
     Measure and save to csv multiple times
+
+    @details
+        Repeat count times:
+        - call measure_func
+        - call save_csv
+        - sleep for repeat_delay
+
+    @param measure_func: The measurement function to use. Use a lambda to bind your parameters!
+    @param count: Repeat count times
+
+    Example: Repeat 10 times:
+        repeat(lambda : monitor_count(count=6000, interval=0.02, max_points_shown=200), 10)
     """
-    for i in range(repeat):
+    try:
-        _measure(max_measurements=max_measurements, max_points_shown=max_points_shown, monitor=monitor)
+        for _ in range(count):
-        save_csv()
+            measure_func()
-        sleep(repeat_delay)
+            save_csv()
+            sleep(repeat_delay)
+    except KeyboardInterrupt:
+        pass
 
 
 def get_dataframe():
     """
-    Get a pandas dataframe from the data in smua.nvbuffer1
+    Get a pandas dataframe from the data in smua.nvbuffer1 and smua.nvbuffer2
     """
     global k, settings, _runtime_vars
     if test:
@@ -241,16 +298,18 @@ def load_settings():
     global settings, config_path
     with open(config_path, "r") as file:
         settings = json.load(file)
-    settings["datadir"] = path.expanduser(settings["datadir"])
+    settings["datadir"] = path.expanduser(settings["datadir"])  # replace ~
 
 def help(topic=None):
     if topic == None:
         print("""
 Functions:
-    measure         - measure the voltage
+    measure         - take measurements
-    monitor         - measure the voltage with live monitoring in a matplotlib window
+    monitor         - take measurements with live monitoring in a matplotlib window
-    automeasure     - measure and save to csv multiple times
+    measure_count   - take a fixed number of measurements
-    get_dataframe   - return smua.nvbuffer1 as pandas dataframe
+    monitor_count   - take a fixed number of measurements with live monitoring in a matplotlib window
+    repeat          - measure and save to csv multiple times
+    get_dataframe   - return smua.nvbuffer 1 and 2 as pandas dataframe
     save_csv        - save the last measurement as csv file
     save_pickle     - save the last measurement as pickled pandas dataframe
     load_dataframe  - load a pandas dataframe from csv or pickle
@@ -303,7 +362,7 @@ def init():
 |      <    ______   |    |    |    __)_  /   |   \ /   \  ___
 |    |  \  /_____/   |    |    |        \/    |    \\    \_\  \
 |____|__ \           |____|   /_______  /\____|__  / \______  /
-        \/                            \/         \/         \/      1.0
+        \/                            \/         \/         \/      1.1
 Interactive Shell for TENG measurements with Keithley 2600B
 ---
 Enter 'help()' for a list of commands""")

k-teng/keithley/measure.py

@@ -1,115 +1,106 @@
 from time import sleep
 import numpy as np
 from matplotlib import pyplot as plt
+import pyvisa
 
-smua_settings = """
+from .keithley import reset
-display.clear()
+from ..utility import testing as _testing
-display.settext('starting')
-smua.reset()
-smua.measure.autorangev = smua.AUTORANGE_ON
-smua.measure.autozero = smua.AUTOZERO_ONCE
-smua.source.output = smua.OUTPUT_OFF
--- max 20 V expected
-smua.measure.rangev = 20
 
-"""
+def measure_count(instr, V=True, I=True, count=100, interval=0.05, update_func=None, update_interval=0.5, beep_done=True, verbose=True, testing=False):
-
-script_dir = "scripts/"
-scripts = {
-    "buffer_reset":     "buffer_reset.lua",
-    "smua_reset":       "smua_reset.lua",
-}
-for key,val in scripts.items():
-    scripts[key] = script_dir + scripts[key]
-
-
-def run_lua(instr, script_path, verbose=False):
     """
-    Run a lua script from the host on the instrument
+    Take <count> measurements with <interval> inbetween
-    @param instr : pyvisa instrument
-    @param script_path : full path to the script
-    """
-    with open(script_path, "r") as file:
-        script = file.read()
-    if verbose: print(f"Running script: {script_path}")
-    instr.write(script)
 
-
+    @details
-def measure_V(instr, count=100, interval=0.05):
+        Uses the devices overlappedY function to make the measurements asynchronosly
+        The update_func is optional and only used when I == True and V == True
+        The update_func does not necessarily get all the values that are measured. To obtain the whole measurement, get them from the device buffers (smua.nvbufferX)
+    @param instr: pyvisa instrument
+    @param update_func: Callable that processes the measurements: (index, ival, vval) -> None
+    @param update_interval: interval at which the update_func is called
     """
-    """
-    data = []
-    for _ in range(1000):
-        data.append(tuple(float(v) for v in instr.query("print(smua.measure.v())").strip('\n').split('\t')))
-        # print(i, data[-1])
-        # clear_output(wait=True)
-        plt.plot(data)
-        plt.show()
-        sleep(0.05)
-
-def reset(instr, verbose=False):
-    """
-    Reset smua and its buffers
-    @param instr : pyvisa instrument
-    """
-    run_lua(instr, scripts["smua_reset"], verbose=verbose)
-    run_lua(instr, scripts["buffer_reset"], verbose=verbose)
-
-
-def measure_count(instr, V=True, I=True, count=100, interval=0.05, beep_done=True, verbose=True):
-    """
-    take n measurements at dt interval
-    @param instr : pyvisa instrument
-    """
-    reset(instr, verbose=verbose)
     f_meas = None
     if V and I:
-        f_meas = "smua.measure.iv(smua.nvbuffer1, smua.nvbuffer2)"
+        f_meas = "smua.measure.overlappediv(smua.nvbuffer1, smua.nvbuffer2)"
     elif V:
-        f_meas = "smua.measure.v(smua.nvbuffer1)"
+        f_meas = "smua.measure.overlappedv(smua.nvbuffer1)"
     elif I:
-        f_meas = "smua.measure.i(smua.nvbuffer1)"
+        f_meas = "smua.measure.overlappedi(smua.nvbuffer1)"
     else:
         print("I and/or V needs to be set to True")
         return
 
-    instr.write(f"smua.measure.count = {count}")
+    i = 0
-    instr.write(f"smua.measure.interval = {interval}")
+    if not testing:
+        reset(instr, verbose=verbose)
+        instr.write(f"smua.measure.count = {count}")
+        instr.write(f"smua.measure.interval = {interval}")
 
-    instr.write(f"smua.source.output = smua.OUTPUT_ON")
+        # start measurement
-    instr.write(f_meas)
+        instr.write(f"smua.source.output = smua.OUTPUT_ON")
-    instr.write(f"smua.source.output = smua.OUTPUT_OFF")
+        instr.write(f_meas)
 
-    if beep_done:
+        condition = lambda: float(instr.query("print(status.operation.measuring.condition)").strip("\n ")) != 0
-        instr.write("beeper.beep(0.3, 1000)")
+    else:
+        condition = lambda: i < int(float(count) * interval / update_interval)
+
+    sleep(update_interval)
+    # for live viewing
+
+    # will return 2.0 while measruing
+    while condition():
+        if update_func and V and I:
+            try:
+                if not testing:
+                    ival = float(instr.query("print(smua.nvbuffer1.readings[smua.nvbuffer1.n])").strip("\n"))
+                    vval = float(instr.query("print(smua.nvbuffer2.readings[smua.nvbuffer2.n])").strip("\n"))
+                else:
+                    ival = _testing.testcurve(i, peak_width=1, amplitude=5e-8)
+                    vval = -_testing.testcurve(i, peak_width=2, amplitude=15)
+                update_func(i, ival, vval)
+            except ValueError:
+                pass
+        sleep(update_interval)
+        i += 1
+
+    if not testing:
+        instr.write(f"smua.source.output = smua.OUTPUT_OFF")
+
+        if beep_done:
+            instr.write("beeper.beep(0.3, 1000)")
 
 
-
+def measure(instr, interval, update_func=None, max_measurements=None, testing=False):
-def event_test_TODO():
+    """
-        # Type of event we want to be notified about
+    @details:
-    event_type = pyvisa.constants.EventType.service_request
+        - Resets the buffers
-    # Mechanism by which we want to be notified
+        - Until KeyboardInterrupt:
-    event_mech = pyvisa.constants.EventMechanism.queue
+            - Take measurement
-    keithley.enable_event(event_type, event_mech)
+            - Call update_func
-
+            - Wait interval
-    # Instrument specific code to enable service request
+        Uses python's time.sleep() for waiting the interval, which is not very precise. Use measure_count for better precision
-    # (for example on operation complete OPC)
+        You can take the data from the buffer afterwards, using save_csv
-    keithley.write("*SRE 1")
+    @param instr: pyvisa instrument
-    keithley.write("INIT")
+    @param update_func: Callable that processes the measurements: (index, ival, vval) -> None
-
+    @param max_measurements : maximum number of measurements. None means infinite
-
+    """
-    with open("script.lua", "r") as file:
+    if not testing:
-        script = file.read()
+        reset(instr, verbose=True)
-    # for line in script.split('\n'):
+        instr.write("smua.source.output = smua.OUTPUT_ON")
-    #     input(line)
+        instr.write("format.data = format.ASCII\nformat.asciiprecision = 12")
-    #     keithley.write(line)
+    try:
-    keithley.write(script)
+        i = 0
-
+        while max_measurements is None or i < max_measurements:
-    # Wait for the event to occur
+            if testing:
-    response = keithley.wait_on_event(event_type, 1000)
+                ival = _testing.testcurve(i, peak_width=1, amplitude=5e-8)
-    assert response.event.event_type == event_type
+                vval = -_testing.testcurve(i, peak_width=2, amplitude=15)
-    assert response.timed_out == False
+            else:
-
+                ival, vval = tuple(float(v) for v in instr.query("print(smua.measure.iv(smua.nvbuffer1, smua.nvbuffer2))").strip('\n').split('\t'))
-    instr.disable_event(event_type, event_mech)
+            if update_func:
-    keithley.query_ascii_values("printbuffer(1, 10, smua.nvbuffer1)", 6)
+                update_func(i, ival, vval)
-    print(voltages)
+            sleep(interval)
+            i += 1
+    except KeyboardInterrupt:
+        pass
+    if not testing:
+        instr.write("smua.source.output = smua.OUTPUT_OFF")
+    print("Measurement stopped" + " "*50)