import time
from typing import Any
from pySerialTransfer import pySerialTransfer as txfer
from qcodes.instrument import Instrument, InstrumentChannel
from semi_cr.setup.juno.cqcc.utils.parser import CQCC_V1
import networkx as nx
from semi_cr.core.lab.station.routing_v1.models import ConnectionRoutable, Routable
from semi_cr.core.lab.station.graphing.ids import (
instrument_node_id,
qcodes_node_id,
terminal_node_id,
)
from semi_cr.core.lab.station.graphing.models import Node, Edge
from semi_cr.core.lab.station.graphing.operations import build_nx_graph
[docs]
def terminal_id_from_endpoint(
endpoint: str,
) -> str:
owner, terminal = (
endpoint[:-1].split(
"[",
maxsplit=1,
)
)
return terminal_node_id(
owner,
terminal,
)
[docs]
def add_physical_connection(
G: nx.MultiDiGraph,
a: str,
b: str,
modality: str,
) -> None:
attrs = {
"kind": "connection",
"modality": modality,
}
G.add_edge(
a,
b,
**attrs,
)
G.add_edge(
b,
a,
**attrs,
)
[docs]
class CQCCChannel(InstrumentChannel, Routable):
def __init__(
self,
parent: "QcodesCQCC",
name: str,
channel: int,
enabled: bool = True,
) -> None:
super().__init__(
parent=parent,
name=name,
)
self._channel_number = channel
self._is_enabled = enabled
@property
def channel_number(self) -> int:
return self._channel_number
@property
def is_enabled(self) -> bool:
return self._is_enabled
@property
def capabilities(self) -> frozenset[str]:
return frozenset({
"routing",
})
@property
def graph(self) -> nx.MultiDiGraph:
channel_id = qcodes_node_id(self)
terminal_id = self.parent.terminal_id(
self.short_name
)
attrs = {
"capabilities": {
"routing",
},
"modality": "dc",
"channel": self.channel_number,
"enabled": self.is_enabled,
}
nodes = [
Node(
id=channel_id,
kind="instrument_channel",
name=self.short_name,
obj=self,
attrs=attrs,
),
Node(
id=terminal_id,
kind="terminal",
name=(
f"{self.parent.owner_name}"
f"[{self.short_name}]"
),
obj=self,
attrs=attrs,
),
]
edges = [
Edge(
source=channel_id,
target=terminal_id,
kind="represents",
)
]
return build_nx_graph(
nodes,
edges,
)
[docs]
class ConnectionError(Exception):
pass
[docs]
class SwitchSingleControlStruct:
"""
Class for constructing data between PC and MCU
This class is used for open single channel control
"""
exp_type: str = "M"
pin_list: list[int]
level_list: list[int]
num_pairs: int = 9 # this is always 9 for single control
[docs]
class SwitchPairControlStruct:
"""
Class for constructing data between PC and MCU
This class is used for open pair channel control
"""
exp_type: str = "K"
pin_list: list[int]
level_list: list[int]
num_pairs: int = 18 # this is always 18 for pair control (9 for positive, 9 for GND
tail: bool = False
[docs]
class CQCC:
def __init__(self, settings_path=r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml"):
"""
Initialize cqcc driver
Parameters:
Settings_path: stores the hardware mapping
"""
self.cqcc_parser = CQCC_V1(settings_path)
self.arduino = None
@property
def connected(self) -> bool:
return (
self.arduino is not None
and self.arduino.connection is not None
and self.arduino.connection.is_open
)
[docs]
def connect(self, COM):
if self.connected:
return
try:
self.arduino = txfer.SerialTransfer(COM) # initialize the serial transfer
self.arduino.open()
time.sleep(2) # wait for the connection to be established
if not self.connected:
raise ConnectionError(
f"Serial port {COM} did not open successfully."
)
except txfer.InvalidSerialPort as e:
raise ConnectionError(f"Connection error: Invalid serial port {COM}: {e}")
def _check_connected(self) -> None:
"""Raise if the CQCC is not connected to the Arduino."""
if not self.connected:
raise ConnectionError(
"CQCC is not connected to the Arduino. "
"Please call connect() first."
)
def _send_single_signal(
self,
signal: dict[int, int],
) -> None:
command = SwitchSingleControlStruct()
command.exp_type = "M"
command.pin_list = list(signal.keys())
command.level_list = list(signal.values())
command.num_pairs = len(command.pin_list)
self.send_switchSingleControlStruct_to_arduino(
command
)
ack = self.wait_for_ack()
if not ack:
raise ConnectionError(
"CQCC command failed. "
"No ACK received from the Arduino."
)
def _select_channel(
self,
polarity: str,
channel: int,
) -> None:
signal = self.cqcc_parser.port_to_signal(
f"dc{channel}",
polarity,
)
self._send_single_signal(signal)
@staticmethod
def _validate_polarity(polarity: str) -> None:
"""Validate the CQCC routing bank."""
valid_polarities = {"Positive", "GND"}
if polarity not in valid_polarities:
raise ValueError(
f"Invalid polarity {polarity!r}. "
f"Expected one of {sorted(valid_polarities)}."
)
@staticmethod
def _validate_channel(channel: int) -> None:
"""Validate a CQCC DC channel number."""
if isinstance(channel, bool) or not isinstance(channel, int):
raise TypeError(
f"Channel must be an integer, "
f"got {type(channel).__name__}."
)
if not 1 <= channel <= 96:
raise ValueError(
f"Channel must be between 1 and 96, "
f"got {channel}."
)
def _clear_bank(
self,
polarity: str,
) -> None:
signal = self.cqcc_parser.bank_off_signal(
polarity
)
self._send_single_signal(signal)
[docs]
def all_off(self) -> None:
self._check_connected()
signal = self.cqcc_parser.all_off_signal
command = SwitchSingleControlStruct()
command.exp_type = "M"
command.pin_list = list(signal.keys())
command.level_list = list(signal.values())
command.num_pairs = len(command.pin_list)
self.send_switchSingleControlStruct_to_arduino(
command
)
ack = self.wait_for_ack()
if not ack:
raise ConnectionError(
"Failed to switch all CQCC channels off."
)
[docs]
def set_channel(
self,
polarity: str,
channel: int,
state: bool
)-> None:
self._check_connected()
self._validate_polarity(polarity)
self._validate_channel(channel)
if state:
self._select_channel(
polarity=polarity,
channel=channel,
)
else:
self._clear_bank(
polarity=polarity,
)
# if not self.arduino.connection.is_open:
# raise ConnectionError(
# "Not connected to the Arduino. Please call connect() first."
# )
# try:
# if polarity not in ["Positive", "GND"]:
# raise ValueError("Polarity must be either 'Positive' or 'GND'.")
# if (channel > 0) and (channel < 97):
# signal = self.cqcc_parser.port_to_signal(f"dc{channel}", polarity)
# command = SwitchSingleControlStruct()
# command.exp_type = (
# "M" # let controller know now is set a single channel
# )
# command.pin_list = [pin for pin, _ in signal.items()]
# command.level_list = [value for _, value in signal.items()]
# command.num_pairs = len(command.pin_list)
# self.send_switchSingleControlStruct_to_arduino(command)
# ack = self.wait_for_ack()
# if ack:
# print(
# f"Channel {channel} set to {'HIGH' if state else 'LOW'} successfully."
# )
# else:
# raise ConnectionError(
# f"Failed to set channel {channel}. No ACK received from the Arduino."
# )
# else:
# raise ValueError("Channel number must be between 1 and 96.")
# except ValueError as e:
# raise ValueError(f"Value error: {e}")
# except ConnectionError as e:
# raise ConnectionError(f"Connection error: {e}")
[docs]
def disconnect(self):
if self.arduino is None:
return
try:
if self.arduino.connection.is_open:
self.arduino.close()
finally:
self.arduino = None
# except Exception as e:
# raise ConnectionError(
# f"Connection error: Error closing the connection: {e}"
# )
# function for waiting for ACK from MCU
[docs]
def wait_for_ack(self):
if not self.arduino.connection.is_open:
raise ConnectionError(
"Not connected to the Arduino. Please call connect() first."
)
try:
while True:
if self.arduino.available():
recSize = 0
ack = self.arduino.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.arduino.status.value <= 0:
if self.arduino.status == txfer.Status.CRC_ERROR:
print("ERROR: CRC_ERROR")
elif self.arduino.status == txfer.Status.PAYLOAD_ERROR:
print("ERROR: PAYLOAD_ERROR")
elif self.arduino.status == txfer.Status.STOP_BYTE_ERROR:
print("ERROR: STOP_BYTE_ERROR")
else:
print("ERROR: {}".format(txfer.Status.name))
return None
except Exception as e:
raise ConnectionError(f"Connection error: Error waiting for ACK: {e}")
[docs]
def send_switchSingleControlStruct_to_arduino(
self, data: SwitchSingleControlStruct
):
sendSize = 0
sendSize = self.arduino.tx_obj(data.exp_type, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.pin_list, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.level_list, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.num_pairs, start_pos=sendSize)
self.arduino.send(sendSize)
[docs]
def send_switchPairControlStruct_to_arduino(
self, data: SwitchPairControlStruct, header=True
):
sendSize = 0
if header:
sendSize = self.arduino.tx_obj(data.exp_type, start_pos=sendSize)
self.arduino.send(
sendSize
) # send header first to indicate the type of command
sendSize = 0
sendSize = self.arduino.tx_obj(data.pin_list, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.level_list, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.num_pairs, start_pos=sendSize)
sendSize = self.arduino.tx_obj(data.tail, start_pos=sendSize)
self.arduino.send(sendSize)
[docs]
class QcodesCQCC(Instrument, ConnectionRoutable):
def __init__(
self,
name: str,
port: str,
channels: list[dict[str, Any]] | None = None,
connections: list[dict[str, Any]] | None = None,
) -> None:
super().__init__(name)
self.port = port
self.cqcc = CQCC(
settings_path=(
r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml"
)
)
self._channel_configs = list(
channels or []
)
self._connection_configs = list(connections or [])
self._channels: dict[str, CQCCChannel] = {}
for cfg in self._channel_configs:
if not cfg.get("enabled", True):
continue
channel_name = str(cfg["name"])
channel_number = int(cfg["channel"])
channel = CQCCChannel(
parent=self,
name=channel_name,
channel=channel_number,
enabled=cfg.get(
"enabled",
True,
),
)
self.add_submodule(channel_name,channel)
self._channels[channel_name] = channel
@property
def owner_name(self) -> str:
return getattr(
self,
"config_name",
self.name,
)
[docs]
def terminal_id(
self,
terminal_name: str,
) -> str:
return terminal_node_id(
self.owner_name,
terminal_name,
)
[docs]
def get_terminal(
self,
terminal_name: str,
) -> Any:
if terminal_name in self._channels:
return self._channels[terminal_name]
if terminal_name in {"Positive", "GND"}:
return self
raise KeyError(
f"Unknown terminal "
f"{terminal_name!r} "
f"on CQCC "
f"{self.owner_name!r}."
)
[docs]
def connect(self) -> None:
if self.cqcc.connected:
return
self.cqcc.connect(self.port)
[docs]
def disconnect(self) -> None:
self.cqcc.disconnect()
[docs]
def close(self):
try:
self.disconnect()
finally:
super().close()
[docs]
def apply_route_configuration(
self,
channel: int,
polarity: str,
) -> None:
self.cqcc.set_channel(
polarity=polarity,
channel=channel,
state=True,
)
[docs]
def deactivate_route_configuration(
self,
channel: int,
polarity: str,
) -> None:
self.cqcc.set_channel(
polarity=polarity,
channel=channel,
state=False,
)
[docs]
def get_idn(self) -> dict[str, str | None]:
return {
"vendor": "NQCP_Spin",
"model": self.name,
"serial": None,
"firmware": None,
}
@property
def capabilities(self) -> frozenset[str]:
return frozenset({
"routing",
})
@property
def graph(self) -> nx.MultiDiGraph:
cqcc_id = instrument_node_id(
self.owner_name
)
G = build_nx_graph(
nodes=[
Node(
id=cqcc_id,
kind="instrument",
name=self.owner_name,
obj=self,
)
],
edges=[],
)
# -----------------------------------------
# DC1 ... DC96
# -----------------------------------------
for channel in self._channels.values():
G = nx.compose(
G,
channel.graph,
)
channel_id = qcodes_node_id(
channel
)
G.add_edge(
cqcc_id,
channel_id,
kind="contains",
)
# -----------------------------------------
# CQCC common terminals: V+ and GND
# -----------------------------------------
common_terminals = {
"Positive": "V+",
"GND": "GND",
}
for terminal_name, label in common_terminals.items():
terminal_id = self.terminal_id(
terminal_name
)
G.add_node(
terminal_id,
kind="terminal",
name=f"{self.owner_name}[{label}]",
terminal_path=terminal_name,
label=label,
obj=self,
capabilities={"routing"},
modality="dc",
)
G.add_edge(
cqcc_id,
terminal_id,
kind="contains",
)
# -----------------------------------------
# Configurable connections
# -----------------------------------------
for channel in self._channels.values():
channel_number = channel.channel_number
dc_terminal_id = self.terminal_id(
channel.short_name
)
for polarity in ("Positive", "GND"):
bank_terminal_id = self.terminal_id(
polarity
)
attrs = {
"kind": "configurable",
"instrument_name": self.owner_name,
"channel": channel_number,
"polarity": polarity,
"active": False,
"modality": "dc",
}
G.add_edge(
dc_terminal_id,
bank_terminal_id,
**attrs,
)
G.add_edge(
bank_terminal_id,
dc_terminal_id,
**attrs,
)
# -----------------------------------------
# External fixed connections
# -----------------------------------------
for connection_cfg in self._connection_configs:
local_name = str(
connection_cfg["name"]
)
local_terminal_id = self.terminal_id(
local_name
)
for endpoint in connection_cfg.get(
"endpoints",
[],
):
remote_terminal_id = (
terminal_id_from_endpoint(
endpoint
)
)
G.add_node(
remote_terminal_id,
kind="terminal",
)
add_physical_connection(
G,
remote_terminal_id,
local_terminal_id,
modality="dc",
)
return G