# ----------------------------------------------------
# Note
# ----------------------------------------------------
"""
Code is good but still areas for improving and making
more pythonic - A.I. is good at brain storming ideas
to do this.
Noticed that the measurement give spurious outputs sometimes.
I am not sure if this is a program error or not but it is
always resolved by turning the Keithley on and off again.
There may be a need for a reset sequence before executing
large measurements.
There is somethimes a lack of clarity in naming. If Keithley or
Arduino is not specified then the Keithley can be assumed to be
the device of interest as this is a keithley driver.
Need to trouble shoot and ensure parameters are actually being
updated - e.g. nplc.
An idea in the future would be one class that incorporates both
drivers.
"""
# ----------------------------------------------------
# Imports
# ----------------------------------------------------
import math
import os
import time
from dataclasses import dataclass, field
from typing import Any
import numpy as np
import pyvisa
from pySerialTransfer import pySerialTransfer as txfer
from semi_cr.core.lab.drivers.cqcc import CQCC, SwitchPairControlStruct
from semi_cr.setup.juno.cqcc.utils.parser import CQCC_V1
# ----------------------------------------------------
# Dataclasses Classes
# ----------------------------------------------------
# Keithley and arduino
[docs]
@dataclass
class Keithley:
address: str
status: Any = None
[docs]
@dataclass
class Arduino:
connected: bool = False
communication_port: str = "COM4"
status: Any = None
[docs]
@dataclass
class Device:
keithley: Keithley
arduino: Arduino
# Full matrix parameters
[docs]
@dataclass
class FilePaths:
cqcc_v1: str = r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml"
keithley_script: str = r"src\semi_cr\setup\juno\cqcc\conf\keitheley_scripts\keithley_script.lua"
data: str = r"src\semi_cr\setup\juno\cqcc\tests"
# Channel to check before executing large measurement
[docs]
@dataclass
class Check:
ch1: int = 12
ch2: int = 6
repeats: int = 5
tolerance: float = 0.01
theoretical_resistance: float = 100_000 # Ohms
err_message: str = "High deviation from the theoretical resistance detected: power Keithley on and off and try again."
[docs]
@dataclass
class FullMatrixParams:
path: FilePaths = field(default_factory=FilePaths)
check: Check = field(default_factory=Check)
full_scan: bool = False
chunk_size: int = 20
# Settings for source and measurement
[docs]
@dataclass
class Source:
channel: str = "a"
delay: float = 0.1
iv: str = "v"
current_level: float = 0.01
voltage_level: float = 0.01 # V
[docs]
@dataclass
class Measurement:
filter: str = "FILTER_REPEAT_AVG"
filter_count: int = 3
filter_status: bool = True
nplc: int = 10
[docs]
@dataclass
class Settings:
source: Source
measurement: Measurement
# ----------------------------------------------------
# Helper Classes
# ----------------------------------------------------
# Raises connection error
[docs]
class ConnectionError(Exception):
pass
# If there is an issue if the Keithley set-up
[docs]
class KeithleySetupError(Exception):
pass
# ----------------------------------------------------
# Driver Class
# ----------------------------------------------------
[docs]
class Keithley_2636b_driver:
# ----------------------------------------------------
# Initialise, Connect, Disconnect, Checking Connection
# ----------------------------------------------------
def __init__(self, device):
# Import keithley, arduino, and pyvisa
self._device = device
self.rm = pyvisa.ResourceManager()
# Set Base Parameters
self._settings = Settings(source=Source(), measurement=Measurement())
[docs]
def connect(self):
try:
# Connect
self._device.keithley.status = self.rm.open_resource(self._device.keithley.address)
self._device.keithley.status.read_termination = "\n"
self._device.keithley.status.timeout = 10000 # Set timeout to 5 seconds
except pyvisa.VisaIOError as e:
raise ConnectionError(f"Connection error: {e}")
[docs]
def disconnect(self):
if self._device.keithley.status is not None:
self._device.keithley.status.close()
self._device.keithley.status = None
[docs]
def connect_arduino(self):
try:
self._device.arduino.status = txfer.SerialTransfer(self._device.arduino.communication_port)
self._device.arduino.status.open()
time.sleep(2) # allow some time for the Arduino to completely reset
self._device.arduino.connected = True
except Exception:
import traceback # noqa: PLC0415
traceback.print_exc()
try:
self._device.arduino.status.close()
except Exception:
pass
self._device.arduino.connected = False
[docs]
def check_connection(self):
if self._device.keithley.status is None:
raise ConnectionError("Not connected to the device. Please call connect() first.")
# ----------------------------------------------------
# Updating Measurement Parameters
# ----------------------------------------------------
# Used to update dataclasses
[docs]
def update_settings(self, **kwargs):
# Channel
if "channel" in kwargs:
if kwargs["channel"] not in {"a", "b"}:
raise ValueError("Channel must be 'a' or 'b'.")
else:
self._settings.source.channel = kwargs["channel"]
# Delay
if "delay" in kwargs:
if not isinstance(kwargs["delay"], (int, float)):
raise ValueError("Delay must be a number")
elif kwargs["delay"] <= 0:
raise ValueError("Delay must be positive")
else:
self._settings.source.delay = kwargs["delay"]
# Voltage
if "voltage_level" in kwargs:
if not isinstance(kwargs["voltage"], (int, float)):
raise ValueError("Voltage must be a number")
elif kwargs["voltage"] <= 0:
raise ValueError("Voltage must be positive")
else:
self._settings.source.voltage_level = kwargs["voltage"]
# Current
if "current_level" in kwargs:
if not isinstance(kwargs["current_level"], (int, float)):
raise ValueError("Current must be a number")
elif kwargs["current_level"] <= 0:
raise ValueError("Current level must be positive")
else:
self._settings.source.current_level = kwargs["current_level"]
# iv
if "iv" in kwargs:
if kwargs["iv"] not in {"i", "v"}:
raise ValueError("iv must be 'i' or 'v'")
else:
self._settings.source.iv = kwargs["iv"]
# Filter
if "filter" in kwargs:
if kwargs["filter"] not in {"FILTER_REPEAT_AVG"}: # TODO: Update this with more parameters ...
raise ValueError("Delay must be a number")
else:
self._settings.measurement.filter = kwargs["filter"]
# Filter count
if "filter_count" in kwargs:
if type(kwargs["filter_count"]) is not int:
raise ValueError("Filter count must be an interger")
elif kwargs["filter_count"] <= 0:
raise ValueError("Filter count must be positive interger")
else:
self._settings.measurement.filter_count = kwargs["filter_count"]
# Filter status
if "filter_status" in kwargs:
if kwargs["filter_status"] not in {False, True}:
raise ValueError("Filter status must be a boolean")
else:
self._settings.measurement.filter = kwargs["filter_status"]
# Nplc
if "nplc" in kwargs:
if type(kwargs["nplc"]) is not int:
raise ValueError("Nplc must be an interger")
elif kwargs["nplc"] <= 0:
raise ValueError("Nplc must be positive interger")
else:
self._settings.measurement.nplc = kwargs["nplc"]
# ----------------------------------------------------
# Applying Settings Parameters
# ----------------------------------------------------
# Used to apply settings to Keithley
def _apply_settings(self):
try:
# Update source settings
if self._settings.source.iv == "v":
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.func = smu{self._settings.source.channel}.OUTPUT_DCVOLTS")
self._device.keithley.status.write(
f"smu{self._settings.source.channel}.source.level{self._settings.source.iv} = {self._settings.source.voltage_level}"
)
else:
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.func = smu{self._settings.source.channel}.OUTPUT_DCCURRENT")
self._device.keithley.status.write(
f"smu{self._settings.source.channel}.source.level{self._settings.source.iv} = {self._settings.source.current_level}"
)
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.delay = {self._settings.source.delay}")
# Update measurement _settings
self._device.keithley.status.write(f"smu{self._settings.source.channel}.measure.nplc = {self._settings.measurement.nplc}")
if self._settings.measurement.filter_status:
self._device.keithley.status.write(
f"smu{self._settings.source.channel}.measure.filter.type = smu{self._settings.source.channel}.{self._settings.measurement.filter}"
)
self._device.keithley.status.write(f"smu{self._settings.source.channel}.measure.filter.count = {self._settings.measurement.filter_count}")
self._device.keithley.status.write(f"smu{self._settings.source.channel}.measure.filter.enable = smu{self._settings.source.channel}.FILTER_ON")
else:
self._device.keithley.status.write(f"smu{self._settings.source.channel}.measure.filter.enable = smu{self._settings.source.channel}.FILTER_OFF")
except pyvisa.VisaIOError as e:
raise ConnectionError(f"Connection error: {e}")
except ValueError as e:
raise ValueError(f"Value error: {e}")
# ----------------------------------------------------
# Manual Measurement of I and V
# ----------------------------------------------------
[docs]
def measure_resistance(self):
self.check_connection()
self._apply_settings()
try:
# Measure resistance
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.output = smu{self._settings.source.channel}.OUTPUT_ON")
resistance = self._device.keithley.status.query(f"print(smu{self._settings.source.channel}.measure.r())")
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.output = smu{self._settings.source.channel}.OUTPUT_OFF")
return float(resistance)
except pyvisa.VisaIOError as e:
raise ConnectionError(f"Measurement error: {e}")
[docs]
def measure_current(self):
self.check_connection()
self._apply_settings()
try:
# Measure current
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.output = smu{self._settings.source.channel}.OUTPUT_ON")
current = self._device.keithley.status.query(f"print(smu{self._settings.source.channel}.measure.i())")
self._device.keithley.status.write(f"smu{self._settings.source.channel}.source.output = smu{self._settings.source.channel}.OUTPUT_OFF")
return float(current)
# Checking for measurement error
except pyvisa.VisaIOError as e:
raise ConnectionError(f"Measurement error: {e}")
# ----------------------------------------------------
# Full Matrix Measurement
# ----------------------------------------------------
[docs]
def measure_full_resistance_matrix(self): # Hand fnc. a dataclass with all necessary parameters
_ = time.time()
self._fmps = FullMatrixParams()
# Connect to Keithley and check readings
self.connect()
# Check readings before executing a large measurement
self.check_keithley_readings()
# Reset and apply settings
self._device.keithley.status.clear() # clear the buffer after uploading the script
self._device.keithley.status.write("reset()")
self._apply_settings()
print("Settings applied")
self.print_settings()
print("#####################################")
# Connect to arduino
self.connect_arduino()
# Implement the half / full matrix
try:
count = 1
if self._fmps.full_scan:
# Perform a full scan
# generate command among all the channels
num_comb = 96**2 # calculate the number of combinations # TODO: Integrate this into the dataclass
# upload Lua script to Keithley
self.upload_keithley_script_full_matrix(
self._fmps.path.keithley_script,
num_comb,
)
self._device.keithley.status.write("script.anonymous.run()") # call the measurement function in the lua script
time.sleep(2) # add delay to ensure Keithley is ready before sending commands
for i in range(1, 97):
for j in range(1, 97):
path = {"input": f"dc{i}", "output": f"dc{j}"}
arduino_signal = CQCC_V1(self._fmps.path.cqcc_v1).path_to_signal(path)
# construct data
command = SwitchPairControlStruct()
command.exp_type = "K"
command.pin_list = [pin for pin, _ in arduino_signal.items()]
command.level_list = [value for _, value in arduino_signal.items()]
command.num_pairs = len(command.pin_list)
if count % self._fmps.chunk_size == 1:
command.tail = False
self.send_switchPairControlStruct_to_arduino(command, header=True) # for the first command within the chunk, send the header
elif (count % self._fmps.chunk_size == 0) or (count == num_comb):
# if is last command of a chunk or it is last commend in the loop, set tail true and wait for execution
command.tail = True
self.send_switchPairControlStruct_to_arduino(command, header=False) # for the others, don't need to
# for every 20 commands, wait for execution
ack = self.wait_for_ack()
print(f"ACK {count}/{num_comb}: {ack}")
else:
# otherwise just send the command without header and tail false
command.tail = False
self.send_switchPairControlStruct_to_arduino(command, header=False)
count += 1
else:
# generate command among all the channels
num_comb = math.comb(96, 2) + 96 # calculate the number of combinations
# upload Lua script to Keithley
self.upload_keithley_script_full_matrix(
self._fmps.path.keithley_script,
num_comb,
)
self._device.keithley.status.write("script.anonymous.run()") # call the measurement function in the lua script
time.sleep(2) # add delay to ensure Keithley is ready before sending commands
# if not full scan, measure the diagonal channel and lower triangle
for i in range(1, 97):
for j in range(i, 97):
path = {"input": f"dc{i}", "output": f"dc{j}"}
arduino_signal = CQCC_V1(self._fmps.path.cqcc_v1).path_to_signal(path)
# construct data
command = SwitchPairControlStruct()
command.exp_type = "K"
command.pin_list = [pin for pin, _ in arduino_signal.items()]
command.level_list = [value for _, value in arduino_signal.items()]
command.num_pairs = len(command.pin_list)
print(count) # Debugging
if count % self._fmps.chunk_size == 1:
command.tail = False
self.send_switchPairControlStruct_to_arduino(command, header=True) # for the first command within the chunk, send the header
elif (count % self._fmps.chunk_size == 0) or (count == num_comb):
# if is last command of a chunk or it is last commend in the loop, set tail true and wait for execution
command.tail = True
self.send_switchPairControlStruct_to_arduino(command, header=False) # for the others, don't need to
# for every 20 commands, wait for execution
ack = self.wait_for_ack()
print(f"ACK {count}/{num_comb}: {ack}")
else:
# otherwise just send the command without header and tail false
command.tail = False
self.send_switchPairControlStruct_to_arduino(command, header=False)
count += 1
time.sleep(2)
print("All measurements done, retrieving data...")
data = self.read_keithley_buffer() # read the measurement data from Keithley
self._device.keithley.status.close()
# TODO: Think about deleting this as moving towards workflow as the data would be saved in qharbour anyway ...
print(data)
np.save(
os.path.join(
self._fmps.path.data,
f"keithley_measurement_data_fullscan_X002002C1D1_{self._fmps.full_scan}_{time.strftime('%Y-%m-%d_%H-%M-%S')}.npy",
),
data,
) # save raw data for debugging
return data
except Exception as e:
print(f"Error occurred: {e}")
self._device.arduino.connected = False
# ----------------------------------------------------
# Helper Fncs.
# ----------------------------------------------------
# Ensure correct operation before running
[docs]
def check_keithley_readings(self):
# Define upper and lower bounds
lb = (1 - self._fmps.check.tolerance) * self._fmps.check.tolerance
ub = (1 + self._fmps.check.tolerance) * self._fmps.check.tolerance
# Connect to arduino
cqcc = CQCC()
cqcc.connect(self._device.arduino.communication_port)
# Notify the user and make measurements
print("Checking Keithley Readings")
cqcc.set_channel("Positive", self._fmps.check.ch1, state=True)
cqcc.set_channel("GND", self._fmps.check.ch2, state=True)
for i in range(self._fmps.check.repeats):
resistance = self.measure_resistance()
print(f"Resistance ({i + 1}) = {resistance} Ohms")
if (resistance < lb) or (resistance > ub):
cqcc.disconnect()
raise KeithleySetupError(self._fmps.check.err_message)
print("No errors detcted in Keithley set-up.")
print("#####################################")
cqcc.disconnect()
[docs]
def upload_keithley_script_full_matrix(self, path_to_script, limit=None):
print("Uploading Lua script...")
time.sleep(1)
with open(path_to_script, encoding="utf-8") as f:
for line in f:
line_split = line.rstrip()
print(line_split)
if line:
if limit is not None:
line_split = line_split.replace("$num_of_trigger", str(limit))
self._device.keithley.status.write(line_split)
time.sleep(0.01)
print("Upload complete")
[docs]
def send_switchPairControlStruct_to_arduino(self, data: SwitchPairControlStruct, header=True):
sendSize = 0
if header:
sendSize = self._device.arduino.status.tx_obj(data.exp_type, start_pos=sendSize)
self._device.arduino.status.send(sendSize) # send header first to indicate the type of command
sendSize = 0
sendSize = self._device.arduino.status.tx_obj(data.pin_list, start_pos=sendSize)
sendSize = self._device.arduino.status.tx_obj(data.level_list, start_pos=sendSize)
sendSize = self._device.arduino.status.tx_obj(data.num_pairs, start_pos=sendSize)
sendSize = self._device.arduino.status.tx_obj(data.tail, start_pos=sendSize)
self._device.arduino.status.send(sendSize)
[docs]
def wait_for_ack(self):
if not self._device.arduino.connected or not self._device.arduino.status:
return None
try:
while True:
if self._device.arduino.status.available():
recSize = 0
ack = self._device.arduino.status.rx_obj(obj_type="c", start_pos=recSize)
ack = int.from_bytes(ack)
print(f"Received ACK: {ack}")
if ack == 1:
return ack
elif self._device.arduino.status.status.value <= 0:
if self._device.arduino.status.status == txfer.Status.CRC_ERROR:
print("ERROR: CRC_ERROR")
elif self._device.arduino.status.status == txfer.Status.PAYLOAD_ERROR:
print("ERROR: PAYLOAD_ERROR")
elif self._device.arduino.status.status == txfer.Status.STOP_BYTE_ERROR:
print("ERROR: STOP_BYTE_ERROR")
else:
print(f"ERROR: {txfer.Status.name}")
return None
except Exception:
raise Exception("Data is empty or invalid")
[docs]
def read_keithley_buffer(self):
n = int(float(self._device.keithley.status.query("print(smua.nvbuffer1.n)")))
print(f"Number of readings in buffer: {n}")
data = []
for i in range(int(n)):
reading = self._device.keithley.status.query(f"print(smua.nvbuffer1.readings[{i + 1}])")
data.append(float(reading))
time.sleep(0.01) # add a small delay to avoid overwhelming the instrument
return data
# ----------------------------------------------------
# Debugging Fnc.
# ----------------------------------------------------
[docs]
def print_settings(self):
self.check_connection()
inst = self._device.keithley.status
print("NPLC :", inst.query("print(smua.measure.nplc)").strip())
print("Delay :", inst.query("print(smua.source.delay)").strip())
print("Voltage :", inst.query("print(smua.source.levelv)").strip())
print("Filter :", inst.query("print(smua.measure.filter.enable)").strip())
print("Filter Count:", inst.query("print(smua.measure.filter.count)").strip())