from typing import Any
from qcodes.instrument.channel import ChannelList
from qcodes_contrib_drivers.drivers.QDevil.QDAC2 import QDac2, QDac2Channel
import networkx as nx
from semi_cr.core.lab.station.routing_v1.models import Routable, ConnectionRoutable
from semi_cr.core.lab.station.graphing.models import (
Node,
Edge,
)
from semi_cr.core.lab.station.graphing.ids import qcodes_node_id, terminal_node_id, instrument_node_id
from semi_cr.core.lab.station.graphing.operations import build_nx_graph
[docs]
class QdaciiChannel(QDac2Channel, Routable):
def __init__(
self,
parent: "Qdacii",
name: str,
channel: int,
role: str | None = None,
enabled: bool = True,
):
super().__init__(parent, name, channel)
self._channel_number = channel
self._role = role
self._is_enabled = enabled
self.add_parameter(
"channel_role",
initial_cache_value=self._role,
get_cmd=None,
set_cmd=False,
)
self.add_parameter(
"channel_enabled",
initial_cache_value=self._is_enabled,
get_cmd=None,
set_cmd=False,
)
self.add_parameter(
"channel_number",
initial_cache_value=self._channel_number,
get_cmd=None,
set_cmd=False,
)
self.add_parameter(
"voltage",
unit="V",
label=f"QDAC channel {self.channel_number} voltage",
get_cmd=self.get_voltage,
set_cmd=self.set_voltage,
get_parser=float,
set_parser=float,
)
# self.add_parameter(
# "voltage_readback",
# unit="V",
# label=f"QDAC channel {self.channel_number} voltage readback",
# get_cmd=self.get_voltage,
# set_cmd=False,
# get_parser=float,
# )
self.add_parameter(
"current",
unit="A",
label=f"QDAC channel {self.channel_number} current",
get_cmd=self.get_current,
set_cmd=False,
get_parser=float,
)
[docs]
def set_voltage(self, voltage: float) -> None:
self.dc_constant_V(voltage)
[docs]
def get_voltage(self) -> float:
return float(self.dc_constant_V())
[docs]
def get_current(self) -> float:
return float(self.read_current_A()[0])
@property
def capabilities(self) -> frozenset[str]:
return frozenset({
"voltage_source",
# "voltage_readback",
"current_meter",
})
@property
def role(self):
return self._role
@property
def is_enabled(self):
return self._is_enabled
@property
def channel_number(self):
return self._channel_number
@property
def graph(self) -> nx.MultiDiGraph:
terminal_id = self.parent.terminal_id(
self.short_name
)
channel_id = qcodes_node_id(self)
attrs = {
"capabilities": {
"voltage_source",
"current_meter",
},
"quantities": {
"voltage",
"current",
},
"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}[{self.short_name}]",
obj=self,
attrs=attrs,
),
]
# edges = [
# Edge(
# source=channel_id,
# target=terminal_id,
# kind="represents",
# )
# ]
edges = []
return build_nx_graph(nodes, edges)
[docs]
class Qdacii(QDac2, ConnectionRoutable):
def __init__(
self,
name: str,
ip_address: str,
port: int,
default_range: str = "high",
default_filter: str = "med",
default_slew_rate: float = 1000,
channels: dict[str, Any] | None = None,
trigger_inputs: list[dict[str, Any]] | None = None,
**kwargs,
) -> None:
address = f"TCPIP::{ip_address}::{port}::SOCKET"
super().__init__(
name=name,
address=address,
visalib="@py",
device_clear=False,
**kwargs,
)
self.ip_address = ip_address
self.port = port
self.default_range = default_range
self.default_filter = default_filter
self.default_slew_rate = default_slew_rate
channel_cfgs = {
int(cfg["channel"]): cfg
for cfg in channels or []
if cfg.get("enabled", True)
}
for channel_number, cfg in channel_cfgs.items():
old_channel = self.channel(channel_number)
old_name = old_channel.short_name
new_channel = QdaciiChannel(
parent=self,
name=old_name,
channel=channel_number,
role=cfg.get("role"),
enabled=cfg.get("enabled", True),
)
# Replace in submodules
self.submodules[old_name] = new_channel
# Replace attribute
setattr(self, old_name, new_channel)
# ---------------------------------------------
# Trigger inputs
# ---------------------------------------------
self._trigger_input_configs = list(trigger_inputs or [])
self.trigger_inputs: dict[str, Any] = {}
for trigger_cfg in self._trigger_input_configs:
trigger_name = str(
trigger_cfg["name"]
)
try:
trigger = self.submodules[
trigger_name
]
except KeyError as exc:
raise KeyError(
f"Configured trigger input "
f"{trigger_name!r} does not exist "
"as a QDAC submodule."
) from exc
self.trigger_inputs[
trigger_name
] = trigger
@property
def owner_name(self) -> str:
return 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:
# Trigger terminal
if terminal_name in self.trigger_inputs:
return self.trigger_inputs[
terminal_name
]
# DC voltage channels
for ch_number in range(1, 25):
ch = self.channel(
ch_number
)
if ch.short_name == terminal_name:
return ch
raise KeyError(
f"Unknown terminal "
f"{terminal_name!r} on QDAC "
f"{self.owner_name!r}."
)
[docs]
def channel_list(self, channels: list[int]) -> ChannelList:
"""
Return a locked QCoDeS ChannelList containing selected QDAC channels.
"""
qdac_channels = [self.channel(ch) for ch in channels]
return ChannelList(
parent=self,
name="selected_channels",
chan_type=type(qdac_channels[0]),
chan_list=qdac_channels,
)
@property
def graph(self) -> nx.MultiDiGraph:
qdac_id = instrument_node_id(
self.owner_name
)
nodes = [
Node(
id=qdac_id,
kind="instrument",
name=self.owner_name,
obj=self,
)
]
G = build_nx_graph(
nodes,
[],
)
# -------------------------------------------------
# DC voltage channels
# -------------------------------------------------
for ch_number in range(1, 25):
ch = self.channel(
ch_number
)
if not isinstance(
ch,
QdaciiChannel,
):
continue
channel_graph = ch.graph
G = nx.compose(
G,
channel_graph,
)
channel_id = qcodes_node_id(
ch
)
if channel_id in G:
G.add_edge(
qdac_id,
channel_id,
kind="contains",
)
# -------------------------------------------------
# External trigger inputs
# -------------------------------------------------
for trigger_cfg in self._trigger_input_configs:
trigger_name = str(
trigger_cfg["name"]
)
trigger = self.trigger_inputs[
trigger_name
]
local_terminal_id = self.terminal_id(
trigger_name
)
G.add_node(
local_terminal_id,
kind="terminal",
name=(
f"{self.owner_name}"
f"[{trigger_name}]"
),
obj=trigger,
role=trigger_cfg.get(
"role",
"trigger_input",
),
modality="digital_trigger",
direction="input",
)
G.add_edge(
qdac_id,
local_terminal_id,
kind="contains",
)
# ---------------------------------------------
# External physical connections
# ---------------------------------------------
for endpoint in trigger_cfg.get("endpoints", []):
remote_terminal_id = terminal_id_from_endpoint(endpoint)
G.add_node(
remote_terminal_id,
kind="terminal",
)
G.add_edge(
remote_terminal_id,
local_terminal_id,
kind="connection",
modality="digital_trigger",
)
return G
[docs]
def terminal_id_from_endpoint(
endpoint: str,
) -> str:
owner, terminal = (
endpoint[:-1].split(
"[",
maxsplit=1,
)
)
return terminal_node_id(
owner,
terminal,
)