Sweeps with Callback Functions¶

This notebook shows you how the call-back functionality of LabOne Q can be used to do a very general 2D sweep. The two axes are swept with call-back functions, and at each point of the sweep some data is acquired with the MFLI DAQ module

0. General Imports¶

In [ ]:

import matplotlib.pyplot as plt
import numpy as np
import time

from laboneq.simple import *

import matplotlib.pyplot as plt import numpy as np import time from laboneq.simple import *

1. Device Setup¶

1.1 Create device setup¶

In [ ]:

descriptor = """\
instruments:
  MFLI:
  - address: DEV5534
    uid: device_mfli
"""

device_setup = DeviceSetup.from_descriptor(
    descriptor,
    server_host="your_ip_address",
    server_port=8004,
    setup_name="MySetup",
)

descriptor = """\ instruments: MFLI: - address: DEV5534 uid: device_mfli """ device_setup = DeviceSetup.from_descriptor( descriptor, server_host="your_ip_address", server_port=8004, setup_name="MySetup", )

2. MFLI example¶

2.1 Connect to session¶

In [ ]:

# create and connect to session
session = Session(device_setup=device_setup)
session.connect(do_emulation=True)

# create and connect to session session = Session(device_setup=device_setup) session.connect(do_emulation=True)

2.2 Experiment definition¶

In [ ]:

## constant definition
INT_TIME = 30e-3

# Define sweep parameter
outer_sweep_parameter = LinearSweepParameter(
    uid="outer_sweep_parameter", start=1, stop=5, count=3
)
inner_sweep_parameter = LinearSweepParameter(
    uid="inner_sweep_parameter", start=0, stop=4, count=3
)

## Create Experiment
exp = Experiment("Generic experiment")

# define experiment
with exp.sweep(uid="outer_sweep", parameter=outer_sweep_parameter):
    # use near-time callback
    exp.call("setSlowAxis", param=outer_sweep_parameter)
    with exp.sweep(uid="inner_sweep", parameter=inner_sweep_parameter):
        # use near-time callback
        exp.call("setFastAxis", param=inner_sweep_parameter)
        exp.call("readMFLI", settling_time=0.1)
        with exp.acquire_loop_rt(uid="RT_shots", count=1):
            pass

## constant definition INT_TIME = 30e-3 # Define sweep parameter outer_sweep_parameter = LinearSweepParameter( uid="outer_sweep_parameter", start=1, stop=5, count=3 ) inner_sweep_parameter = LinearSweepParameter( uid="inner_sweep_parameter", start=0, stop=4, count=3 ) ## Create Experiment exp = Experiment("Generic experiment") # define experiment with exp.sweep(uid="outer_sweep", parameter=outer_sweep_parameter): # use near-time callback exp.call("setSlowAxis", param=outer_sweep_parameter) with exp.sweep(uid="inner_sweep", parameter=inner_sweep_parameter): # use near-time callback exp.call("setFastAxis", param=inner_sweep_parameter) exp.call("readMFLI", settling_time=0.1) with exp.acquire_loop_rt(uid="RT_shots", count=1): pass

2.3 Configure MFLI and DAQ module¶

In [ ]:

# shortcut for the used MFLI in the setup
mfli = session.devices["device_mfli"]

# configure MFLI
demod = mfli.demods[0]  # which demodulator to use (depends on MF option)

with mfli.set_transaction():
    mfli.demods["*"].enable(False)
    mfli.oscs[0].freq(1e6)
    mfli.sigouts[0].enable(True)
    demod.order(1)
    demod.rate(1e3)
    demod.trigger("continuous")
    demod.timeconstant(10e-3)
    demod.enable(True)

# Parameters
DEMOD_RATE_MFLI = demod.rate()  # read the value from the instrument
NUM_COLS = int(
    np.ceil(DEMOD_RATE_MFLI * INT_TIME)
)  # Number of samples per burst. Corresponds to length of time trace in units of sampling rate.

# Module creation
daq_module = mfli._session.modules.daq  # Create DAQ module
daq_module.device(mfli)  # Assign DAQ module to instrument
daq_module.type(0)  # Continuous acquisition
daq_module.endless(False)  # Single acquisition/trace

# Shape of my grid
daq_module.grid.mode(
    4
)  # Specify how the acquired data is sampled onto the matrix’s horizontal axis (4='exact')
daq_module.count(1)  # Number of grids to be acquired
daq_module.grid.cols(
    NUM_COLS
)  # Length of acquired trace (in units of demodulator sample)
daq_module.grid.rows(1)  # Number of rows per acquisition run
daq_module.grid.rowrepetition(
    False
)  # Averaging mode of rows (irrevelant for grid.rows(1))
# True: First average each row, then fill the next row -> sequential averaging
# False: First fill each row, then average the rows -> cyclic averaging

# Subscribe to the values that should be measured
# Nodes to read
sample_nodes = [
    demod.sample.r.avg,
    demod.sample.theta.avg,
]
for node in sample_nodes:
    daq_module.subscribe(node)

# Print relevant settings if needed
# print(f"Columns: {daq_module.grid.cols()}")
# print(f"Rows: {daq_module.grid.rows()}")
# print(f"Repetitions: {daq_module.grid.repetitions()}")
# print(f"Holdoff: {daq_module.holdoff.time()}")

# shortcut for the used MFLI in the setup mfli = session.devices["device_mfli"] # configure MFLI demod = mfli.demods[0] # which demodulator to use (depends on MF option) with mfli.set_transaction(): mfli.demods["*"].enable(False) mfli.oscs[0].freq(1e6) mfli.sigouts[0].enable(True) demod.order(1) demod.rate(1e3) demod.trigger("continuous") demod.timeconstant(10e-3) demod.enable(True) # Parameters DEMOD_RATE_MFLI = demod.rate() # read the value from the instrument NUM_COLS = int( np.ceil(DEMOD_RATE_MFLI * INT_TIME) ) # Number of samples per burst. Corresponds to length of time trace in units of sampling rate. # Module creation daq_module = mfli._session.modules.daq # Create DAQ module daq_module.device(mfli) # Assign DAQ module to instrument daq_module.type(0) # Continuous acquisition daq_module.endless(False) # Single acquisition/trace # Shape of my grid daq_module.grid.mode( 4 ) # Specify how the acquired data is sampled onto the matrix’s horizontal axis (4='exact') daq_module.count(1) # Number of grids to be acquired daq_module.grid.cols( NUM_COLS ) # Length of acquired trace (in units of demodulator sample) daq_module.grid.rows(1) # Number of rows per acquisition run daq_module.grid.rowrepetition( False ) # Averaging mode of rows (irrevelant for grid.rows(1)) # True: First average each row, then fill the next row -> sequential averaging # False: First fill each row, then average the rows -> cyclic averaging # Subscribe to the values that should be measured # Nodes to read sample_nodes = [ demod.sample.r.avg, demod.sample.theta.avg, ] for node in sample_nodes: daq_module.subscribe(node) # Print relevant settings if needed # print(f"Columns: {daq_module.grid.cols()}") # print(f"Rows: {daq_module.grid.rows()}") # print(f"Repetitions: {daq_module.grid.repetitions()}") # print(f"Holdoff: {daq_module.holdoff.time()}")

2.4 Define near-time callbacks for arming MFLI and reading results¶

In [ ]:

def readMFLI(session, settling_time):
    if session.connection_state.emulated:
        return "Running emulation - no real data"

    clockbase = mfli.clockbase()
    timeout = 10  # s

    time.sleep(settling_time)
    daq_module.execute()

    # Retrieve data from UHFLI DAQ module
    start_time = time.time()
    while time.time() - start_time < timeout:
        time.sleep(INT_TIME)

        if daq_module.raw_module.finished() is True:
            progress = daq_module.raw_module.finished()
            print(f"Progress of data acquisition: {100 * progress:.2f}%.")
            break

        progress = daq_module.raw_module.finished()

    if not (time.time() - start_time < timeout):
        print(
            "Data acquisition timed out. Not all results collected, data is corrupted."
        )

    # Get data
    daq_data = daq_module.read(raw=False, clk_rate=clockbase)

    return daq_data


def setSlowAxis(session, param):
    print(f"outer: {int(param)}")


def setFastAxis(session, param):
    print(f"inner: {int(param)}")

def readMFLI(session, settling_time): if session.connection_state.emulated: return "Running emulation - no real data" clockbase = mfli.clockbase() timeout = 10 # s time.sleep(settling_time) daq_module.execute() # Retrieve data from UHFLI DAQ module start_time = time.time() while time.time() - start_time < timeout: time.sleep(INT_TIME) if daq_module.raw_module.finished() is True: progress = daq_module.raw_module.finished() print(f"Progress of data acquisition: {100 * progress:.2f}%.") break progress = daq_module.raw_module.finished() if not (time.time() - start_time < timeout): print( "Data acquisition timed out. Not all results collected, data is corrupted." ) # Get data daq_data = daq_module.read(raw=False, clk_rate=clockbase) return daq_data def setSlowAxis(session, param): print(f"outer: {int(param)}") def setFastAxis(session, param): print(f"inner: {int(param)}")

2.5 Register near-time callbacks¶

In [ ]:

session.register_neartime_callback(setSlowAxis, "setSlowAxis")
session.register_neartime_callback(setFastAxis, "setFastAxis")
session.register_neartime_callback(readMFLI, "readMFLI")

session.register_neartime_callback(setSlowAxis, "setSlowAxis") session.register_neartime_callback(setFastAxis, "setFastAxis") session.register_neartime_callback(readMFLI, "readMFLI")

2.6 Run experiment¶

In [ ]:

my_results = session.run(exp)

my_results = session.run(exp)

3. Plot results¶

In [ ]:

clockbase = mfli.clockbase()

if not session.connection_state.emulated:
    for node in sample_nodes:
        plt.figure()
        for idx in range(my_results.neartime_callback_results["readMFLI"].__len__()):
            results = my_results.neartime_callback_results["readMFLI"][idx][node][
                0
            ]  # Results
            plt.plot(results.time, results.value[0], label=f"readout step {int(idx+1)}")
        plt.xlabel("Time [s]")
        plt.ylabel(str(node))
        plt.legend(loc="best", fontsize=8)
        plt.title("MFLI time traces of demodulated data")
else:
    print("Emulation - nothing to plot")

clockbase = mfli.clockbase() if not session.connection_state.emulated: for node in sample_nodes: plt.figure() for idx in range(my_results.neartime_callback_results["readMFLI"].__len__()): results = my_results.neartime_callback_results["readMFLI"][idx][node][ 0 ] # Results plt.plot(results.time, results.value[0], label=f"readout step {int(idx+1)}") plt.xlabel("Time [s]") plt.ylabel(str(node)) plt.legend(loc="best", fontsize=8) plt.title("MFLI time traces of demodulated data") else: print("Emulation - nothing to plot")