# Converts a route into a runtime backend
from __future__ import annotations
import networkx as nx
from typing import TYPE_CHECKING, Any
from semi_cr.core.lab.station.routing_v1.errors import (
BackendResolutionError,
)
from semi_cr.core.lab.station.routing_v1.models import (
Route,
RouteRequest,
)
from semi_cr.core.lab.station.routing_v1.queries import (
find_pin_route_to_role,
find_route,
)
if TYPE_CHECKING:
from semi_cr.core.lab.devices.pinned import PinnedDevice
from qcodes.parameters import Parameter
# def find_backend_parameter_for_pin(
# graph: nx.MultiDiGraph,
# device: PinnedDevice,
# pin_name: str,
# role: str,
# parameter_name: str,
# **filters: Any,
# ) -> Parameter:
# route = find_pin_route_to_role(
# graph=graph,
# device=device,
# pin_name=pin_name,
# role=role,
# **filters,
# )
# # Usually the target role node is the last node in the route.
# # But checking the reversed route makes this robust if the object
# # is attached one node before/after the terminal.
# for node_id in reversed(route):
# data = graph.nodes[node_id]
# obj = data.get("obj")
# if obj is None:
# continue
# parameter = getattr(obj, parameter_name, None)
# if isinstance(parameter, Parameter):
# return parameter
# filter_text = "" if not filters else " with filters " + ", ".join(f"{k}={v!r}" for k, v in filters.items())
# raise LookupError(f"Found route from pin {pin_name!r} to role {role!r}{filter_text}, but no backend parameter {parameter_name!r} was found on the route.")
[docs]
def find_backend_parameter_for_pin(
graph: nx.MultiDiGraph,
device: PinnedDevice,
pin_name: str,
role: str,
parameter_name: str,
quantity: str | None = None,
modality: str | None = None,
) -> Parameter:
"""
Find a route from a device pin and resolve a QCoDeS parameter
from that route.
"""
pin = getattr(device, pin_name)
request = RouteRequest(
pin=pin,
role=role,
quantity=quantity,
modality=modality,
)
route = find_route(
graph,
request,
)
return resolve_parameter(
graph,
route,
parameter_name,
)
# def find_backend_object_for_pin(
# graph: nx.MultiDiGraph,
# device: PinnedDevice,
# pin_name: str,
# role: str,
# modality: str | None = None,
# ):
# route = find_pin_route_to_role(
# graph=graph,
# device=device,
# pin_name=pin_name,
# role=role,
# modality=modality,
# )
# for node_id in route:
# data = graph.nodes[node_id]
# if data.get("role") != role:
# continue
# if modality is not None and data.get("modality") != modality:
# continue
# obj = data.get("obj")
# if obj is not None:
# return obj
# raise LookupError(f"No backend object found for pin {pin_name!r} with role={role!r}, modality={modality!r}.")
[docs]
def find_backend_object_for_pin(
graph: nx.MultiDiGraph,
device: PinnedDevice,
pin_name: str,
role: str,
*,
quantity: str | None = None,
modality: str | None = None,
) -> object:
"""
Find a route from a device pin and return its runtime backend
object.
"""
pin = getattr(device, pin_name)
request = RouteRequest(
pin=pin,
role=role,
quantity=quantity,
modality=modality,
)
route = find_route(
graph,
request,
)
return resolve_backend_object(
graph,
route,
)
# %%
[docs]
def resolve_backend_object(
graph: nx.MultiDiGraph,
route: Route,
) -> object:
"""
Resolve the runtime backend object represented by a route.
The target node is authoritative: route discovery is responsible
for selecting a target node that provides the requested capability.
"""
target_node = route.target_node
if target_node not in graph:
raise BackendResolutionError(
f"Route target {target_node!r} does not exist "
"in the station graph."
)
backend = graph.nodes[target_node].get("obj")
if backend is None:
raise BackendResolutionError(
f"Route target {target_node!r} has no runtime "
f"object for {_format_request(route.request)}."
)
return backend
[docs]
def resolve_parameter(
graph: nx.MultiDiGraph,
route: Route,
parameter_name: str,
) -> Parameter:
"""
Resolve a QCoDeS parameter from objects attached to a route.
Nodes are inspected from the route target back toward the device
pin. This supports graphs where the runtime object is attached to
the backend node itself or to an adjacent represented node.
"""
found_non_parameter = False
for node_id in reversed(route.path):
if node_id not in graph:
continue
obj = graph.nodes[node_id].get("obj")
if obj is None:
continue
parameter = getattr(
obj,
parameter_name,
None,
)
if parameter is None:
continue
if isinstance(parameter, Parameter):
return parameter
found_non_parameter = True
if found_non_parameter:
reason = (
f"An attribute named {parameter_name!r} was found, "
"but it is not a QCoDeS Parameter."
)
else:
reason = (
f"No parameter named {parameter_name!r} was found "
"on the route."
)
raise BackendResolutionError(
f"Could not resolve parameter {parameter_name!r} for "
f"{_format_request(route.request)}. {reason} "
f"Route: {route.path!r}."
)
def _format_request(
request: RouteRequest,
) -> str:
pin_name = getattr(
request.pin,
"short_name",
repr(request.pin),
)
fields = [
f"pin={pin_name!r}",
f"role={request.role!r}",
]
if request.quantity is not None:
fields.append(
f"quantity={request.quantity!r}"
)
if request.modality is not None:
fields.append(
f"modality={request.modality!r}"
)
return ", ".join(fields)