Drivers

1
@routine
2
def fast_charge_diagram(ctx, gate_x, gate_y, contact, session=None):
3
    """Fast charge diagram with OPX current measurement."""
4
    device = ctx.resources.device
5
6
    vx = np.linspace(-2.0, -0.5, 200)
7
    vy = np.linspace(-2.0, -0.5, 200)
8
9
    # OPX measures DC current at the specified contact
10
    v_data, i_data = device.sweep_2d(
11
        gate_x, vx.tolist(),
12
        gate_y, vy.tolist(),
13
        contact,  # Must be a contact configured with OPX measure_channel
14
        session=session
15
    )
16
17
    return {"voltages": v_data, "currents": i_data}

1
@routine
2
def initialize_breakout(ctx):
3
    device = ctx.resources.device
4
5
    # Ground all channels for safety
6
    device.ground_breakout_lines()
7
8
    # Unground channels to enable control
9
    device.unground_breakout_lines()
10
11
    # Connect all pads to their configured instruments
12
    device.connect_breakout_lines()
13
14
    # Disconnect when done
15
    device.disconnect_breakout_lines()

1
from stanza.utils import device_from_yaml
2
3
# Load device and get QSwitch instance
4
device = device_from_yaml("device.yaml")
5
qswitch = device.breakout_box_instrument
6
7
# Single channel operations
8
qswitch.set_grounded("G1")              # Connect G1 to ground (relay 0)
9
qswitch.set_ungrounded("G1")            # Disconnect G1 from ground
10
qswitch.set_connected("G1", 9)          # Connect G1 to relay 9 (QDAC)
11
qswitch.set_disconnected("G1", 9)       # Disconnect G1 from relay 9
12
13
# Check state
14
is_grounded = qswitch.get_grounded("G1")
15
is_connected = qswitch.get_connected("G1", 9)
16
17
# Multi-channel operations (faster than looping)
18
qswitch.set_grounded(["G1", "G2", "DRAIN"])
19
qswitch.set_connected(["G1", "G2"], 9)
20
21
# Query multiple channels - Get the connection status of G1 and G2 to line 9
22
states = qswitch.get_connected(["G1", "G2"], 9)  # Returns [True, False]

1
@routine
2
def safe_measurement(ctx, gate, voltages, contact):
3
    device = ctx.resources.device
4
    qswitch = device.breakout_box_instrument
5
6
    try:
7
        # Connect measurement path
8
        qswitch.set_ungrounded([gate, contact])
9
        device.connect_breakout_lines()
10
11
        # Run measurement
12
        v_data, i_data = device.sweep_1d(gate, voltages, contact)
13
14
        return {"voltages": v_data, "currents": i_data}
15
16
    finally:
17
        # Set all pads to zero volts
18
        device.zero()
19
        # Always ground channels when done
20
        device.disconnect_breakout_lines()
21
        device.ground_breakout_lines()

1
from stanza.base.instruments import BaseInstrument
2
from stanza.base.channels import ControlChannel, MeasurementChannel, ChannelConfig
3
import socket
4
5
class MyCustomDriver(BaseInstrument):
6
    """Driver for my custom instrument."""
7
8
    def __init__(self, name: str, ip_addr: str, port: int = 5025,
9
                 slew_rate: float = 1.0):
10
        super().__init__(name=name)
11
12
        self.ip_addr = ip_addr
13
        self.port = port
14
        self.default_slew_rate = slew_rate
15
16
        # Connect to instrument
17
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
18
        self.sock.connect((ip_addr, port))
19
20
    def set_voltage(self, pad: str, voltage: float) -> None:
21
        """Set voltage on a pad."""
22
        channel = self.get_channel(pad)
23
24
        # Validate bounds
25
        v_min, v_max = channel.config.voltage_range
26
        if not (v_min <= voltage <= v_max):
27
            raise ValueError(f"Voltage {voltage} out of bounds [{v_min}, {v_max}]")
28
29
        # Send to hardware
30
        cmd = f"SOUR{channel.control_channel}:VOLT {voltage}\n"
31
        self.sock.sendall(cmd.encode())
32
33
        # Update parameter
34
        channel.set_parameter("voltage", voltage)
35
36
    def get_voltage(self, pad: str) -> float:
37
        """Get current voltage on a pad."""
38
        channel = self.get_channel(pad)
39
40
        # Query hardware
41
        cmd = f"SOUR{channel.control_channel}:VOLT?\n"
42
        self.sock.sendall(cmd.encode())
43
        response = self.sock.recv(1024).decode().strip()
44
45
        voltage = float(response)
46
        channel.set_parameter("voltage", voltage)
47
48
        return voltage
49
50
    def get_slew_rate(self, pad: str) -> float:
51
        """Get slew rate for a pad."""
52
        channel = self.get_channel(pad)
53
        return channel.get_parameter_value("slew_rate")
54
55
    def measure(self, pad: str) -> float:
56
        """Measure current at a pad."""
57
        channel = self.get_channel(pad)
58
59
        # Trigger measurement
60
        cmd = f"MEAS{channel.measure_channel}:CURR\n"
61
        self.sock.sendall(cmd.encode())
62
        response = self.sock.recv(1024).decode().strip()
63
64
        return float(response)
65
66
    def close(self) -> None:
67
        """Close connection."""
68
        self.sock.close()

1
from stanza.base.instruments import BaseControlInstrument
2
import serial
3
4
class SerialDAC(BaseControlInstrument):
5
    """Driver for a simple serial DAC."""
6
7
    def __init__(self, name: str, port: str, baudrate: int = 9600,
8
                 slew_rate: float = 1.0):
9
        super().__init__(name=name)
10
11
        self.default_slew_rate = slew_rate
12
13
        # Open serial port
14
        self.serial = serial.Serial(
15
            port=port,
16
            baudrate=baudrate,
17
            timeout=1.0
18
        )
19
20
    def set_voltage(self, pad: str, voltage: float) -> None:
21
        """Set voltage on a channel."""
22
        channel = self.get_channel(pad)
23
24
        # Validate
25
        v_min, v_max = channel.config.voltage_range
26
        if not (v_min <= voltage <= v_max):
27
            raise ValueError(f"Voltage out of bounds")
28
29
        # Send command (example protocol)
30
        cmd = f"SET {channel.control_channel} {voltage}\n"
31
        self.serial.write(cmd.encode())
32
33
        # Wait for acknowledgment
34
        response = self.serial.readline().decode().strip()
35
        if response != "OK":
36
            raise RuntimeError(f"DAC error: {response}")
37
38
        channel.set_parameter("voltage", voltage)
39
40
    def get_voltage(self, pad: str) -> float:
41
        """Read voltage from a channel."""
42
        channel = self.get_channel(pad)
43
44
        cmd = f"GET {channel.control_channel}\n"
45
        self.serial.write(cmd.encode())
46
47
        response = self.serial.readline().decode().strip()
48
        voltage = float(response)
49
50
        channel.set_parameter("voltage", voltage)
51
        return voltage
52
53
    def get_slew_rate(self, pad: str) -> float:
54
        """Get slew rate."""
55
        return self.default_slew_rate
56
57
    def close(self) -> None:
58
        """Close serial port."""
59
        self.serial.close()

1
def set_voltage(self, pad: str, voltage: float) -> None:
2
    channel = self.get_channel(pad)
3
    v_min, v_max = channel.config.voltage_range
4
5
    if not (v_min <= voltage <= v_max):
6
        raise ValueError(f"Voltage {voltage} out of range [{v_min}, {v_max}]")
7
8
    # Now safe to send to hardware
9
    ...

1
from stanza.exceptions import InstrumentError
2
3
def measure(self, pad: str) -> float:
4
    try:
5
        # Talk to hardware
6
        ...
7
    except socket.timeout:
8
        raise InstrumentError(f"Timeout measuring {pad}")
9
    except Exception as e:
10
        raise InstrumentError(f"Error measuring {pad}: {e}")

1
def close(self) -> None:
2
    """Clean up connections."""
3
    if hasattr(self, 'sock') and self.sock:
4
        self.sock.close()
5
    if hasattr(self, 'serial') and self.serial:
6
        self.serial.close()

1
def __init__(self, name: str, ip_addr: str, is_simulation: bool = False):
2
    super().__init__(name=name)
3
4
    self.is_simulation = is_simulation
5
6
    if not is_simulation:
7
        # Connect to real hardware
8
        self.sock = socket.socket(...)
9
    else:
10
        # Simulation mode
11
        self.simulated_voltages = {}
12
13
def set_voltage(self, pad: str, voltage: float) -> None:
14
    if self.is_simulation:
15
        self.simulated_voltages[pad] = voltage
16
    else:
17
        # Real hardware command
18
        ...

1
import logging
2
3
logger = logging.getLogger(__name__)
4
5
def set_voltage(self, pad: str, voltage: float) -> None:
6
    logger.debug(f"Setting {pad} to {voltage:.3f} V")
7
    ...
8
    logger.debug(f"Set complete")

1
import pytest
2
from stanza.drivers.mycustom import MyCustomDriver
3
from stanza.base.channels import ControlChannel, ChannelConfig
4
5
def test_set_get_voltage():
6
    """Test voltage setting and reading."""
7
    driver = MyCustomDriver(
8
        name="test",
9
        ip_addr="127.0.0.1",
10
        is_simulation=True
11
    )
12
13
    # Add a channel
14
    config = ChannelConfig(
15
        name="G1",
16
        voltage_range=(-3.0, 3.0),
17
        control_channel=1
18
    )
19
    channel = ControlChannel(name="G1", config=config, control_channel=1)
20
    driver.add_channel("G1", channel)
21
22
    # Test
23
    driver.set_voltage("G1", -1.5)
24
    voltage = driver.get_voltage("G1")
25
26
    assert voltage == pytest.approx(-1.5)
27
    driver.close()

1
from stanza.utils import device_from_yaml
2
3
def test_driver_in_device():
4
    """Test driver with full device configuration."""
5
    device = device_from_yaml("test_device.yaml")
6
7
    # Test jump
8
    device.jump({"G1": -1.5, "G2": -0.8})
9
10
    # Test check
11
    v = device.check("G1")
12
    assert v == pytest.approx(-1.5)
13
14
    # Test measure
15
    i = device.measure("DRAIN")
16
    assert isinstance(i, float)
17
18
    # Test sweep
19
    voltages = [-2.0, -1.0, 0.0]
20
    v_data, i_data = device.sweep_1d("G1", voltages, "DRAIN")
21
    assert len(v_data) == 3
22
    assert len(i_data) == 3

1
import socket
2
3
# Test if you can reach the instrument
4
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
sock.settimeout(2.0)
6
try:
7
    sock.connect(("192.168.1.100", 5025))
8
    print("Connection successful")
9
except socket.timeout:
10
    print("Connection timeout - check IP address")
11
except socket.error as e:
12
    print(f"Connection failed: {e}")
13
finally:
14
    sock.close()