QuEL-3 Deploy Check¶

This notebook validates QuEL-3 instrument deployment against your current quelware/clientware endpoint.

Choose system_id = "144Q-LF-Q3" for real deploy checks. You can also load 64Q-HF-Q1 or 144Q-LF-Q1 to inspect shared configuration loading, but deploy/preflight cells will skip because instrument deployment is QuEL-3-specific.

It supports two connection modes:

• use_standalone_client = False: full quelware server mode (ListUnits is called)
• use_standalone_client = True: standalone worker/clientware mode (ListUnits is bypassed)

In [ ]:

from pathlib import Path

from qubex.backend.backend_controller import BACKEND_KIND_QUEL3
from qubex.backend.quel3 import Quel3BackendController
from qubex.core.async_bridge import DEFAULT_TIMEOUT_SECONDS, get_shared_async_bridge
from qubex.system.config_loader import ConfigLoader
from qubex.system.quel3 import Quel3TargetDeployPlanner

from pathlib import Path from qubex.backend.backend_controller import BACKEND_KIND_QUEL3 from qubex.backend.quel3 import Quel3BackendController from qubex.core.async_bridge import DEFAULT_TIMEOUT_SECONDS, get_shared_async_bridge from qubex.system.config_loader import ConfigLoader from qubex.system.quel3 import Quel3TargetDeployPlanner

In [ ]:

# Example selection
system_id = "144Q-LF-Q3"
supported_system_ids = {"64Q-HF-Q1", "144Q-LF-Q1", "144Q-LF-Q3"}
if system_id not in supported_system_ids:
    raise ValueError(f"Unsupported system_id: {system_id}")

candidate_root_dirs = [
    Path.cwd(),
    Path.cwd() / "docs/examples/system",
    Path.cwd().parent,
    Path.cwd().parent.parent,
]
example_root = next(
    (
        path
        for path in candidate_root_dirs
        if (path / "config").is_dir() and (path / "params").is_dir()
    ),
    None,
)
if example_root is None:
    raise FileNotFoundError(
        "Could not find `docs/examples/system/{config,params}` from the current working directory."
    )

config_dir = example_root / "config"
params_dir = example_root / "params" / system_id

server_host = "localhost"
server_port = 50051
use_standalone_client = True
client_mode = "standalone" if use_standalone_client else "server"
standalone_unit_label: str | None = "quel3-02-a01" if use_standalone_client else None
deploy_roles: set[str] | None = None
box_ids: list[str] = ["QT1"] if use_standalone_client else []

print(f"example_root: {example_root.resolve()}")
print(f"system_id: {system_id}")
print(f"server: {server_host}:{server_port}")
print(f"client_mode: {client_mode}")
print(f"standalone_unit_label: {standalone_unit_label}")
print(f"deploy_roles: {sorted(deploy_roles) if deploy_roles is not None else 'ALL'}")
print(f"config_dir: {config_dir.resolve()}")
print(f"params_dir: {params_dir.resolve()}")

# Example selection system_id = "144Q-LF-Q3" supported_system_ids = {"64Q-HF-Q1", "144Q-LF-Q1", "144Q-LF-Q3"} if system_id not in supported_system_ids: raise ValueError(f"Unsupported system_id: {system_id}") candidate_root_dirs = [ Path.cwd(), Path.cwd() / "docs/examples/system", Path.cwd().parent, Path.cwd().parent.parent, ] example_root = next( ( path for path in candidate_root_dirs if (path / "config").is_dir() and (path / "params").is_dir() ), None, ) if example_root is None: raise FileNotFoundError( "Could not find `docs/examples/system/{config,params}` from the current working directory." ) config_dir = example_root / "config" params_dir = example_root / "params" / system_id server_host = "localhost" server_port = 50051 use_standalone_client = True client_mode = "standalone" if use_standalone_client else "server" standalone_unit_label: str | None = "quel3-02-a01" if use_standalone_client else None deploy_roles: set[str] | None = None box_ids: list[str] = ["QT1"] if use_standalone_client else [] print(f"example_root: {example_root.resolve()}") print(f"system_id: {system_id}") print(f"server: {server_host}:{server_port}") print(f"client_mode: {client_mode}") print(f"standalone_unit_label: {standalone_unit_label}") print(f"deploy_roles: {sorted(deploy_roles) if deploy_roles is not None else 'ALL'}") print(f"config_dir: {config_dir.resolve()}") print(f"params_dir: {params_dir.resolve()}")

In [ ]:

loader = ConfigLoader(
    system_id=system_id,
    config_dir=config_dir,
    params_dir=params_dir,
    autoload=False,
)
loader.load()
experiment_system = loader.get_experiment_system()

configured_box_ids = [box.id for box in experiment_system.boxes]
if len(box_ids) == 0:
    box_ids = configured_box_ids

is_quel3_backend = loader.backend_kind == BACKEND_KIND_QUEL3
if client_mode == "standalone" and standalone_unit_label is None:
    raise ValueError("Set `standalone_unit_label` when `use_standalone_client = True`.")

controller = (
    Quel3BackendController(
        quelware_endpoint=server_host,
        quelware_port=server_port,
        client_mode=client_mode,
        standalone_unit_label=standalone_unit_label,
    )
    if is_quel3_backend
    else None
)

readout_targets_by_box: dict[str, list[str]] = {}
for label, target in sorted(experiment_system.gen_targets.items()):
    if target.type.value != "READ":
        continue
    readout_targets_by_box.setdefault(target.channel.port.box_id, []).append(label)

print("backend_kind:", loader.backend_kind)
print("system_id:", loader.system_id)
print("chip_id:", loader.chip_id)
print("boxes in system:", configured_box_ids)
print("boxes to deploy:", box_ids)
print("#gen_targets:", len(experiment_system.gen_targets))
print("deploy enabled:", is_quel3_backend)
if controller is not None:
    print("controller client_mode:", controller.client_mode)
print("readout targets by box:")
for box_id, labels in sorted(readout_targets_by_box.items()):
    print(f"- {box_id}: {tuple(labels)}")

if "QT1" in readout_targets_by_box:
    t1_readout_targets = tuple(readout_targets_by_box["QT1"])
    print(
        "QT1 readout target check:",
        t1_readout_targets == ("RQ000", "RQ001", "RQ002", "RQ003"),
    )
else:
    t1_readout_targets = ()
    print("QT1 not present in current example.")

loader = ConfigLoader( system_id=system_id, config_dir=config_dir, params_dir=params_dir, autoload=False, ) loader.load() experiment_system = loader.get_experiment_system() configured_box_ids = [box.id for box in experiment_system.boxes] if len(box_ids) == 0: box_ids = configured_box_ids is_quel3_backend = loader.backend_kind == BACKEND_KIND_QUEL3 if client_mode == "standalone" and standalone_unit_label is None: raise ValueError("Set `standalone_unit_label` when `use_standalone_client = True`.") controller = ( Quel3BackendController( quelware_endpoint=server_host, quelware_port=server_port, client_mode=client_mode, standalone_unit_label=standalone_unit_label, ) if is_quel3_backend else None ) readout_targets_by_box: dict[str, list[str]] = {} for label, target in sorted(experiment_system.gen_targets.items()): if target.type.value != "READ": continue readout_targets_by_box.setdefault(target.channel.port.box_id, []).append(label) print("backend_kind:", loader.backend_kind) print("system_id:", loader.system_id) print("chip_id:", loader.chip_id) print("boxes in system:", configured_box_ids) print("boxes to deploy:", box_ids) print("#gen_targets:", len(experiment_system.gen_targets)) print("deploy enabled:", is_quel3_backend) if controller is not None: print("controller client_mode:", controller.client_mode) print("readout targets by box:") for box_id, labels in sorted(readout_targets_by_box.items()): print(f"- {box_id}: {tuple(labels)}") if "QT1" in readout_targets_by_box: t1_readout_targets = tuple(readout_targets_by_box["QT1"]) print( "QT1 readout target check:", t1_readout_targets == ("RQ000", "RQ001", "RQ002", "RQ003"), ) else: t1_readout_targets = () print("QT1 not present in current example.")

In [ ]:

planner = Quel3TargetDeployPlanner()

if is_quel3_backend:
    all_requests = planner.build_deploy_requests(
        experiment_system=experiment_system,
        box_ids=box_ids,
    )
    if deploy_roles is None:
        requests = all_requests
    else:
        requests = tuple(
            request for request in all_requests if request.role in deploy_roles
        )

    unit_labels = sorted({request.port_id.split(":", 1)[0] for request in requests})
    if client_mode == "standalone" and len(unit_labels) != 1:
        raise ValueError(
            "Standalone mode expects deploy requests for exactly one upstream unit. "
            f"Got {unit_labels}. Narrow `box_ids`."
        )

    role_counts: dict[str, int] = {}
    for request in requests:
        role_counts[request.role] = role_counts.get(request.role, 0) + 1

    print("all deploy requests:", len(all_requests))
    print("selected deploy requests:", len(requests))
    print("request unit labels:", unit_labels)
    print("role counts:", role_counts)
    for request in requests:
        print(
            f"- port_id={request.port_id}, role={request.role}, "
            f"range=[{request.frequency_range_min_hz:.3e}, {request.frequency_range_max_hz:.3e}], "
            f"targets={request.target_labels}"
        )

    t1_requests = tuple(
        request for request in requests if request.port_id.startswith("quel3-02-a01:")
    )
    if t1_requests:
        print("QT1 deploy request summary:")
        for request in t1_requests:
            print(
                f"- {request.port_id}: role={request.role}, targets={request.target_labels}"
            )
        t1_role_set = {request.role for request in t1_requests}
        print(
            "QT1 role coverage check:",
            t1_role_set == {"TRANSMITTER", "TRANSCEIVER"},
        )
else:
    requests = ()
    print("Loaded non-QuEL-3 example set. Deploy request preview is skipped.")

planner = Quel3TargetDeployPlanner() if is_quel3_backend: all_requests = planner.build_deploy_requests( experiment_system=experiment_system, box_ids=box_ids, ) if deploy_roles is None: requests = all_requests else: requests = tuple( request for request in all_requests if request.role in deploy_roles ) unit_labels = sorted({request.port_id.split(":", 1)[0] for request in requests}) if client_mode == "standalone" and len(unit_labels) != 1: raise ValueError( "Standalone mode expects deploy requests for exactly one upstream unit. " f"Got {unit_labels}. Narrow `box_ids`." ) role_counts: dict[str, int] = {} for request in requests: role_counts[request.role] = role_counts.get(request.role, 0) + 1 print("all deploy requests:", len(all_requests)) print("selected deploy requests:", len(requests)) print("request unit labels:", unit_labels) print("role counts:", role_counts) for request in requests: print( f"- port_id={request.port_id}, role={request.role}, " f"range=[{request.frequency_range_min_hz:.3e}, {request.frequency_range_max_hz:.3e}], " f"targets={request.target_labels}" ) t1_requests = tuple( request for request in requests if request.port_id.startswith("quel3-02-a01:") ) if t1_requests: print("QT1 deploy request summary:") for request in t1_requests: print( f"- {request.port_id}: role={request.role}, targets={request.target_labels}" ) t1_role_set = {request.role for request in t1_requests} print( "QT1 role coverage check:", t1_role_set == {"TRANSMITTER", "TRANSCEIVER"}, ) else: requests = () print("Loaded non-QuEL-3 example set. Deploy request preview is skipped.")

If you see GRPCError(..., "Received :status = '400'", ...) in full-server mode, the endpoint usually does not expose SystemConfigurationService/ListUnits.

In that case, switch use_standalone_client = True and re-run from the cells below.

If you see GRPCError(..., "Received :status = '204'", ...), inspect the returned grpc-status / grpc-message headers. In a standalone setup this often indicates that the selected unit_label does not match a reachable upstream unit.

Standalone mode can deploy both control (TRANSMITTER) and readout (TRANSCEIVER) requests as long as the selected requests belong to one upstream unit.

If system_id points at a QuEL-1 example, the cells below will skip because deploy is QuEL-3-specific.

In [ ]:

def run_async(factory, *, timeout: float = DEFAULT_TIMEOUT_SECONDS):
    """Run one async factory through the shared notebook bridge."""
    bridge = get_shared_async_bridge(key="quel3-deploy-check")
    return bridge.run(factory, timeout=timeout)


async def preflight_check():
    """Inspect client connectivity before issuing deploy requests."""
    if controller is None:
        print("Skipping preflight: loaded example is not QuEL-3.")
        return
    if len(requests) == 0:
        raise ValueError("No deploy requests were built.")

    client_factory = controller.connection_manager.load_quelware_client_factory()
    async with client_factory(server_host, server_port) as client:
        print("unit_labels:", client.list_unit_labels())
        resources = await client.list_resource_infos()
        print("resource_count:", len(resources))
        port_info = await client.get_port_info(requests[0].port_id)
        print("first_port:", port_info)


run_async(preflight_check)

def run_async(factory, *, timeout: float = DEFAULT_TIMEOUT_SECONDS): """Run one async factory through the shared notebook bridge.""" bridge = get_shared_async_bridge(key="quel3-deploy-check") return bridge.run(factory, timeout=timeout) async def preflight_check(): """Inspect client connectivity before issuing deploy requests.""" if controller is None: print("Skipping preflight: loaded example is not QuEL-3.") return if len(requests) == 0: raise ValueError("No deploy requests were built.") client_factory = controller.connection_manager.load_quelware_client_factory() async with client_factory(server_host, server_port) as client: print("unit_labels:", client.list_unit_labels()) resources = await client.list_resource_infos() print("resource_count:", len(resources)) port_info = await client.get_port_info(requests[0].port_id) print("first_port:", port_info) run_async(preflight_check)

In [ ]:

if controller is None:
    print("Skipping deploy: loaded example is not QuEL-3.")
    deployed = {}
else:
    for request in requests:
        print(
            f"deploying alias={request.alias}, port_id={request.port_id}, targets={request.target_labels}"
        )
    deployed = controller.deploy_instruments(requests=requests)

request_by_alias = {request.alias: request for request in requests}
deployed_roles = sorted({request_by_alias[alias].role for alias in deployed})
control_targets = tuple(
    request.target_labels[0]
    for request in request_by_alias.values()
    if request.role == "TRANSMITTER"
)
readout_targets = tuple(
    request.target_labels[0]
    for request in request_by_alias.values()
    if request.role == "TRANSCEIVER"
)

print("deploy completed. aliases:", len(deployed))
print("deployed roles:", deployed_roles)
print("control target count:", len(control_targets))
print("readout target count:", len(readout_targets))
for alias, infos in deployed.items():
    request = request_by_alias[alias]
    print(
        f"- {alias}: port_id={request.port_id}, targets={request.target_labels}, instrument_count={len(infos)}"
    )

print(
    "full deploy role coverage check:",
    set(deployed_roles) == {"TRANSMITTER", "TRANSCEIVER"},
)

if controller is None: print("Skipping deploy: loaded example is not QuEL-3.") deployed = {} else: for request in requests: print( f"deploying alias={request.alias}, port_id={request.port_id}, targets={request.target_labels}" ) deployed = controller.deploy_instruments(requests=requests) request_by_alias = {request.alias: request for request in requests} deployed_roles = sorted({request_by_alias[alias].role for alias in deployed}) control_targets = tuple( request.target_labels[0] for request in request_by_alias.values() if request.role == "TRANSMITTER" ) readout_targets = tuple( request.target_labels[0] for request in request_by_alias.values() if request.role == "TRANSCEIVER" ) print("deploy completed. aliases:", len(deployed)) print("deployed roles:", deployed_roles) print("control target count:", len(control_targets)) print("readout target count:", len(readout_targets)) for alias, infos in deployed.items(): request = request_by_alias[alias] print( f"- {alias}: port_id={request.port_id}, targets={request.target_labels}, instrument_count={len(infos)}" ) print( "full deploy role coverage check:", set(deployed_roles) == {"TRANSMITTER", "TRANSCEIVER"}, )