from typing import Any
from urllib.parse import quote
from qcodes.instrument import Instrument, InstrumentChannel, InstrumentModule
from qcodes.parameters import Parameter
from semi_cr.core.lab.station.graphing.parsing import normalize_channel
# Routing depends on these functions producing exactly the same ID everywhere.
# Generic ID utilities
[docs]
def safe_id_part(value: Any) -> str:
return quote(str(value), safe="")
[docs]
def lab_node_id(kind: str, *parts: Any) -> str:
encoded = "/".join(safe_id_part(part) for part in parts)
return f"lab://{safe_id_part(kind)}/{encoded}"
# Static/domain object IDs
[docs]
def terminal_node_id(owner: str, connection: str) -> str:
owner_parts = [safe_id_part(part) for part in owner.split("/")]
return "lab://terminal/" + "/".join(owner_parts) + f"/{safe_id_part(connection)}"
[docs]
def connector_node_id(connector_name: str) -> str:
return lab_node_id("connector", connector_name)
[docs]
def connector_connection_node_id(
connector_name: str,
connection_name: str,
) -> str:
return lab_node_id("connector_connection", connector_name, connection_name)
[docs]
def chip_node_id(chip_name: str) -> str:
return lab_node_id("chip", chip_name)
[docs]
def device_node_id(chip_name: str, device_name: str) -> str:
return lab_node_id("device", chip_name, device_name)
[docs]
def pin_node_id(chip_name: str, device_name: str, pin_name: str) -> str:
return lab_node_id("pin", chip_name, device_name, pin_name)
[docs]
def pad_node_id(chip_name: str, pad_name: str) -> str:
return lab_node_id("pad", chip_name, pad_name)
[docs]
def instrument_node_id(instrument_name: str) -> str:
return lab_node_id(
"instrument",
instrument_name,
)
[docs]
def instrument_module_node_id(parent_name: str, module_name: str) -> str:
parent_parts = [safe_id_part(part) for part in parent_name.split("/")]
module_parts = [safe_id_part(part) for part in module_name.split("/")]
return "lab://instrument_module/" + "/".join(parent_parts + module_parts)
[docs]
def instrument_connection_node_id(owner: str, connection: str) -> str:
owner_parts = [safe_id_part(part) for part in owner.split("/")]
return "lab://instrument_connection/" + "/".join(owner_parts) + f"/{safe_id_part(connection)}"
[docs]
def terminal_id(
instrument: str,
channel: Any,
) -> str:
"""
Canonical ID for a physical instrument terminal.
Examples
--------
QboardII[RF1]
-> lab://terminal/QboardII/RF1
MiniCoax_1[1]
-> lab://terminal/MiniCoax_1/1
qdacii1[ch01]
-> lab://terminal/qdacii1/ch01
opx1000_2/LF_1[out1]
-> lab://terminal/opx1000_2/LF_1/out1
"""
return terminal_node_id(
normalize_channel(instrument),
normalize_channel(channel),
)
# Runtime object IDs
[docs]
def qcodes_node_id(obj: Any) -> str:
if isinstance(obj, Parameter):
full_name = getattr(obj, "full_name", None)
if full_name:
return f"qcodes://param/{safe_id_part(full_name)}"
inst = getattr(obj, "instrument", None)
inst_name = getattr(inst, "name", "unknown")
return f"qcodes://param/{safe_id_part(inst_name)}/{safe_id_part(obj.name)}"
if isinstance(obj, (InstrumentChannel, InstrumentModule)):
parts: list[str] = []
current: Any = obj
while isinstance(current, (InstrumentChannel, InstrumentModule)):
parts.append(current.name)
current = getattr(current, "parent", None)
root = current if isinstance(current, Instrument) else getattr(obj, "root_instrument", None)
root_name = getattr(root, "name", "unknown")
parts.reverse()
path = "/".join(safe_id_part(part) for part in parts)
return f"qcodes://sub/{safe_id_part(root_name)}/{path}"
if isinstance(obj, Instrument):
return f"qcodes://inst/{safe_id_part(obj.name)}"
raise TypeError(
f"qcodes_node_id only accepts QCoDeS objects, got {type(obj).__name__}"
)
# Naming helpers, not IDs
[docs]
def local_device_name(device: Any) -> str:
parent_name = getattr(getattr(device, "parent", None), "name", "")
name = getattr(device, "short_name", None) or getattr(device, "name")
prefix = f"{parent_name}"
if name.startswith(prefix):
name = name.removeprefix(prefix)
return name.lstrip("_")
[docs]
def qualified_device_name(
chip_name: str,
device_name: str,
) -> str:
"""
Compact canonical device identity.
Compatible with NQCP DEVICE NAME FORMAT
Examples
--------
X001008C1 + D1 -> X001008C1D1
B083004D01 + 0033 -> B083004D010033
"""
return (
f"{normalize_channel(chip_name)}"
f"{normalize_channel(device_name)}"
)
# Resolution helpers
[docs]
def resolve_terminal_id(
instrument_name: str,
channel: Any,
device_registry: dict[str, tuple[str, str]],
) -> str:
"""
Resolve an endpoint to either a device-terminal ID or an ordinary
instrument-terminal ID.
"""
if instrument_name in device_registry:
chip_name, local_device_name = device_registry[instrument_name]
return pin_node_id(
chip_name=chip_name,
device_name=local_device_name,
pin_name=str(channel),
)
return terminal_id(
instrument=instrument_name,
channel=channel,
)