diff --git a/plugins/dqc/dqc/__init__.py b/plugins/dqc/dqc/__init__.py new file mode 100644 index 0000000..ed62431 --- /dev/null +++ b/plugins/dqc/dqc/__init__.py @@ -0,0 +1,356 @@ +""" +DQC Plugin — Distributed Quantum Computing request handler. + +Accepts a partitioned circuit (per-QPU command dicts), labels it using +qnpack's labeling pipeline, schedules execution on hardware agents, and +sends the labeled commands to qnpack for NetSquid simulation. + +Input circuit can be provided in two ways: + 1. ``partitioned_commands`` — already parsed per-QPU command dicts + (as produced by a qnpack frontend). The plugin skips frontend.parse() + and goes straight to validate + label. + 2. ``circuit_file`` or ``circuit_content`` — raw circuit source. + The plugin calls the appropriate qnpack frontend to parse it first. + +After scheduling, the labeled commands are serialized to JSON and sent to +qnpack's ``run_from_labeled()`` receiver for simulation. +""" + +import json +import logging +import os +import tempfile + +from munch import Munch + +from quantnet_controller.common.plugin import ProtocolPlugin, PluginType +from quantnet_controller.common.request import RequestManager, RequestType, RequestParameter +from quantnet_controller.common.constants import Constants +from quantnet_mq import Code +from quantnet_mq.schema.models import dqc, Status as responseStatus + +from logic import DQCLogic +from topology_adapter import get_qpu_info_from_topology +from ir_converter import labeled_ir_to_timeslot_schedule + +# qnpack imports — pure Python, no NetSquid dependency +from qnpack.dqc.frontends import load_frontend +from qnpack.dqc.labeling import label_and_build_maps +from qnpack.dqc.validation import validate_commands + +logger = logging.getLogger(__name__) + + +def _to_plain(obj): + """Recursively convert schema proxy objects (Munch, etc.) to plain Python. + + The incoming partitioned_commands come through the schema layer as proxy + objects; their field values may be proxy types that don't have .lower(), + list concatenation, etc. This ensures validate_commands() and the labeler + receive plain str/int/list/dict/bool/None values. + """ + if isinstance(obj, dict): + return {k: _to_plain(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [_to_plain(i) for i in obj] + # Munch and similar proxy types expose items() like a dict + if hasattr(obj, "items") and not isinstance(obj, dict): + return {k: _to_plain(v) for k, v in obj.items()} + return obj + + +class DQC(ProtocolPlugin): + def __init__(self, context): + super().__init__("dqc", PluginType.PROTOCOL, context) + self._server_commands = [ + ("dqcRequest", self.handle_dqc_request, "quantnet_mq.schema.models.dqc.dqcRequest"), + ] + self.ctx = context + + # Initialize logic handler + self.logic = DQCLogic(context) + + # Initialize RequestManager + self.request_manager = RequestManager( + context, plugin_schema=dqc.dqcRequest, request_type=RequestType.EXPERIMENT + ) + + def initialize(self): + pass + + def destroy(self): + pass + + def reset(self): + pass + + # ── Public request handler ──────────────────────────────────────────────── + + async def handle_dqc_request(self, request): + """Handle incoming DQC request. + + Parses (if needed), labels, schedules, and simulates the circuit. + """ + dqc_req = dqc.dqcRequest(**request) + logger.info(f"Received DQC request: {dqc_req.serialize()}") + + try: + # Deserialize via the schema model to get plain Python dicts with + # no proxy objects — request['payload'] contains schema proxies + # whose field values aren't plain str/int, causing .lower() to fail + # inside validate_commands. + payload = json.loads(dqc_req.serialize()).get("payload", {}) + + # ── 1. Get per-QPU command dicts (parse if needed) ──────────────── + partitioned, circuit_mode, frontend_meta = self._get_partitioned_commands(payload) + + # ── 2. Validate + Label via qnpack ──────────────────────────────── + labeled, process_maps = self._label_commands(partitioned) + + # ── 3. Convert labeled IR → flat timeslot schedule ──────────────── + qpu_info, qpu_id_to_label, label_to_qpu_id, bsm_nodes, raw_topology = get_qpu_info_from_topology(self.ctx) + commands = labeled_ir_to_timeslot_schedule(labeled, process_maps, qpu_id_to_label) + + agent_ids = sorted(list(set(str(c.get("qpu_id")) for c in commands))) + + # ── 3b. Find network routes between all cross-QPU pairs ─────────── + qpu_pairs = self.logic.extract_qpu_pairs(commands) + routes = {} + _path_objects = {} + if qpu_pairs and hasattr(self.ctx, "router") and self.ctx.router: + for src, dst in qpu_pairs: + route_key = f"{src}->{dst}" + try: + p = await self.ctx.router.find_path(src, dst) + routes[route_key] = p.to_node_ids() + _path_objects[route_key] = p + logger.info(f"Route found {route_key}: {routes[route_key]}") + except Exception as e: + logger.warning(f"Could not find route {route_key}: {e}") + routes[route_key] = None + _path_objects[route_key] = None + else: + logger.debug("No router plugin available or no cross-QPU pairs; skipping route discovery") + + # ── 3c. Extract BSM nodes and build synthetic BSM commands ───────── + node_types = {} + bsm_commands = [] + _seen_bsm_ids = set() + + for route_key, path_obj in _path_objects.items(): + bsm_ids = self.logic.extract_bsm_nodes_from_path(path_obj) + if not bsm_ids: + continue + + src_qpu, dst_qpu = route_key.split("->") if "->" in route_key else (None, None) + pair_cmds = [ + c + for c in commands + if len(c.get("qpus_involved") or []) >= 2 + and src_qpu in [str(q) for q in c.get("qpus_involved", [])] + and dst_qpu in [str(q) for q in c.get("qpus_involved", [])] + ] + + for bsm_id in bsm_ids: + node_types[bsm_id] = "BSMNode" + if bsm_id in _seen_bsm_ids: + continue + _seen_bsm_ids.add(bsm_id) + for cmd in pair_cmds: + bsm_cmd = dict(cmd) + bsm_cmd["qpu_id"] = bsm_id + bsm_cmd["node_type"] = "BSMNode" + bsm_commands.append(bsm_cmd) + logger.info(f"Injected {len(pair_cmds)} BSM command(s) for " f"node {bsm_id} on route {route_key}") + + if bsm_commands: + commands = commands + bsm_commands + + all_agent_ids = sorted(list(set(str(c.get("qpu_id")) for c in commands))) + agent_ids = all_agent_ids + + # ── Resolve rid ─────────────────────────────────────────────────── + rid = getattr(dqc_req.payload, "rid", None) or request.get("id") + if not rid or rid == "unknown": + from quantnet_controller.common.utils import generate_uuid + + rid = generate_uuid() + + # ── 4. Build dynamic experiment structure ───────────────────────── + exp_name = f"DQC_{rid}" + DynamicExp = self.logic.build_dynamic_experiment(exp_name, commands, node_types=node_types) + if not DynamicExp: + raise Exception("No commands to process") + + # ── 5. Register dynamic experiment with translator ───────────────── + self.request_manager.translator.exp_defs.append(DynamicExp) + + try: + start_time, slots = await self.request_manager.translator.get_slots_to_allocate(agent_ids, DynamicExp) + + allocations = DynamicExp.get_allocations(slots, Constants.SLOTSIZE.total_seconds()) + + parameters = RequestParameter(exp_name=exp_name, path=agent_ids) + req_obj = self.request_manager.new_request(payload=dqc_req, parameters=parameters, rid=rid) + + await self.request_manager.schedule(req_obj, blocking=True) + + # ── 6. Build simulation payload for caller ──────────────────── + sim_payload = self._build_sim_payload(labeled, process_maps, raw_topology, frontend_meta) + + return dqc.dqcResponse( + status=responseStatus(code=Code.OK.value, value=Code.OK.name, message="DQC execution completed"), + rid=rid, + data={ + "startTime": start_time, + "allocations": allocations, + "routes": routes, + "simulation_payload": sim_payload, + }, + ) + finally: + if DynamicExp in self.request_manager.translator.exp_defs: + self.request_manager.translator.exp_defs.remove(DynamicExp) + + except Exception as e: + logger.error(f"DQC processing failed: {e}") + return dqc.dqcResponse( + status=responseStatus(code=Code.FAILED.value, value=Code.FAILED.name, message=f"{e}"), + rid=dqc_req.payload.rid if hasattr(dqc_req.payload, "rid") else "unknown", + ) + + # ── Private helpers ─────────────────────────────────────────────────────── + + def _get_partitioned_commands(self, payload): + """Return per-QPU command dicts and the circuit mode. + + If ``partitioned_commands`` is in the payload it is used directly. + Otherwise the plugin parses the circuit via the appropriate qnpack + frontend using ``circuit_file`` or ``circuit_content``. + + :returns: ``(partitioned, mode)`` where *partitioned* is + ``{int_qpu_id: [cmd_dict, ...]}`` and *mode* is ``"cisco"`` + or ``"tket"``. + :raises ValueError: If neither source is provided. + """ + mode = payload.get("circuit_mode") + if not mode: + raise ValueError("payload.circuit_mode is required") + + if "partitioned_commands" in payload and payload["partitioned_commands"]: + # JSON keys are strings; convert to int (qnpack convention). + # Command lists may come through as schema proxy objects (Munch + # etc.) — recursively convert everything to plain Python dicts/lists + # so that validate_commands and the labeler can call .lower(), etc. + raw = payload["partitioned_commands"] + partitioned = {int(k): [_to_plain(cmd) for cmd in cmds] for k, cmds in raw.items()} + logger.info(f"Using pre-partitioned commands for {len(partitioned)} QPU(s)") + return partitioned, mode, {} + + # No partitioned commands — parse from circuit source + qpu_info, _, _, _, _ = get_qpu_info_from_topology(self.ctx) + circuit_file, tmp_file = self._resolve_circuit_source(payload, mode) + try: + source_field = "qasm_file" if mode == "cisco" else "dist_commands_file" + circuit_cfg = Munch({"mode": mode, source_field: circuit_file}) + frontend, _ = load_frontend(circuit_cfg) + partitioned = frontend.parse(qpu_info) + frontend_meta = { + "num_output_bits": getattr(frontend, "_num_output_bits", None) + or getattr(frontend, "num_output_bits", None), + "output_reg_name": getattr(frontend, "_output_reg_name", None) + or getattr(frontend, "output_reg_name", "m"), + } + logger.info( + f"Parsed circuit via {mode} frontend: " + f"{sum(len(v) for v in partitioned.values())} command(s) " + f"across {len(partitioned)} QPU(s), " + f"output_bits={frontend_meta['num_output_bits']}, " + f"reg={frontend_meta['output_reg_name']}" + ) + return partitioned, mode, frontend_meta + finally: + if tmp_file and os.path.exists(tmp_file): + os.unlink(tmp_file) + + def _resolve_circuit_source(self, payload, mode): + """Return ``(circuit_file_path, tmp_file_path_or_None)``. + + Writes ``circuit_content`` to a temp file if needed. + """ + circuit_file = payload.get("circuit_file") + tmp_file = None + + if not circuit_file: + content = payload.get("circuit_content") + if not content: + raise ValueError( + "Either partitioned_commands, circuit_file, or " + "circuit_content must be provided in the DQC request payload." + ) + suffix = ".qasm" if mode == "cisco" else ".txt" + tmp = tempfile.NamedTemporaryFile(mode="w", suffix=suffix, delete=False, prefix="dqc_circuit_") + tmp.write(content) + tmp.close() + circuit_file = tmp.name + tmp_file = tmp.name + logger.debug(f"Wrote circuit_content to temp file {tmp_file}") + + return circuit_file, tmp_file + + def _label_commands(self, partitioned): + """Validate and label partitioned per-QPU commands via qnpack. + + :param partitioned: ``{int_qpu_id: [cmd_dict, ...]}`` + :returns: ``(labeled_commands, process_maps)`` + :raises ValueError: On hard validation errors. + """ + errors = validate_commands(partitioned) + warnings = [e for e in errors if e.severity != "error"] + hard = [e for e in errors if e.severity == "error"] + for w in warnings: + logger.warning(f"Circuit validation warning: {w}") + if hard: + for e in hard: + logger.error(f"Circuit validation error: {e}") + raise ValueError(f"Circuit validation failed with {len(hard)} error(s). " "See plugin logs for details.") + + labeled, process_maps = label_and_build_maps(partitioned) + logger.info( + f"Labeled {sum(len(v) for v in labeled.values())} command(s); " + f"start_qpus={len(process_maps['start_qpus'])}, " + f"end_qpus={len(process_maps['end_qpus'])}, " + f"ent_labels={len(process_maps['entanglement_gen_labels'])}" + ) + return labeled, process_maps + + def _build_sim_payload(self, labeled, process_maps, topology, frontend_meta=None): + """Build the JSON-serializable payload for qnpack simulation. + + Returns a dict the caller can write to a file and pass to + ``dqc-sim-labeled ``. Sets are converted to lists and + integer QPU-ID keys are stringified for JSON compatibility. + + :param labeled: ``{int_qpu_id: [cmd_dict, ...]}`` + :param process_maps: ``{start_qpus, end_qpus, entanglement_gen_labels}`` + :param topology: Raw topology list from ``get_qpu_info_from_topology`` + :param frontend_meta: Optional dict with ``num_output_bits`` and + ``output_reg_name`` from the frontend, so ``run_from_labeled`` + knows which register columns to read back as the bitstring. + :returns: JSON-serializable dict ready for ``dqc-sim-labeled``. + """ + payload = { + "labeled_commands": {str(k): v for k, v in labeled.items()}, + "process_maps": { + "start_qpus": {str(k): [int(x) for x in v] for k, v in process_maps["start_qpus"].items()}, + "end_qpus": {str(k): [int(x) for x in v] for k, v in process_maps["end_qpus"].items()}, + "entanglement_gen_labels": list(process_maps["entanglement_gen_labels"]), + }, + "topology": topology, + } + if frontend_meta: + if frontend_meta.get("num_output_bits") is not None: + payload["num_output_bits"] = frontend_meta["num_output_bits"] + if frontend_meta.get("output_reg_name"): + payload["output_reg_name"] = frontend_meta["output_reg_name"] + return payload diff --git a/plugins/dqc/dqc/ir_converter.py b/plugins/dqc/dqc/ir_converter.py new file mode 100644 index 0000000..acac255 --- /dev/null +++ b/plugins/dqc/dqc/ir_converter.py @@ -0,0 +1,318 @@ +""" +ir_converter.py +--------------- +Converts labeled per-QPU IR commands into the flat timeslot record format +consumed by DQCLogic.build_dynamic_experiment(). + +The algorithm is adapted from ControllerProtocol.compute_timeslot_schedule() +in qnpack/qnpack/dqc/protocols/controller.py — a cursor-based scheduler that +advances QPUs in lock-step at sync points (entanglement_gen / ejpp start/end) +and independently for local gates. + +Input (from label_and_build_maps): + labeled_commands : {int_qpu_id: [cmd_dict, ...]} + process_maps : {start_qpus, end_qpus, entanglement_gen_labels} + qpu_id_to_label : {int_qpu_id: "LBNL-A", ...} + +Output (consumed by DQCLogic.build_dynamic_experiment): + list[dict] with keys: + timeslot int Monotonic timeslot index (0-based) + qpu_id str Topology label, e.g. "LBNL-A" + command str Text representation, e.g. "H server_1[0]" + command_type str "local_gate" | "starting_process_on_demand" | + "ending_process" | "idle" + qpus_involved list Topology labels of all QPUs in this sync group + params list Numeric parameters (pass-through) + qubits list Qubit indices (pass-through) + original_qubits list Original qubit identifiers (pass-through) +""" + +import logging + +logger = logging.getLogger(__name__) + +# ── Command-type mapping ────────────────────────────────────────────────────── + +_OP_TO_COMMAND_TYPE = { + "entanglement_gen": "starting_process_on_demand", + "ejpp_start": "starting_process_on_demand", + "ejpp_start_link": "starting_process_on_demand", + "ejpp_end": "ending_process", + "ejpp_end_link": "ending_process", + "msg_sender": "ending_process", + "msg_receiver": "ending_process", + "pre_entanglement": "starting_process_on_demand", +} + +_LOCAL_GATE_OPS = frozenset( + [ + "gate", + "h", + "x", + "y", + "z", + "s", + "t", + "sdg", + "tdg", + "cx", + "cy", + "cz", + "swap", + "ccx", + "rx", + "ry", + "rz", + "u1", + "u2", + "u3", + "measure", + "measure_final", + "reset", + "if_gate", # should have been replaced but kept as fallback + ] +) + + +def cmd_to_type(cmd): + """Map a canonical command dict to a command_type string.""" + op = cmd.get("op", "") + if op == "gate": + return "local_gate" + return _OP_TO_COMMAND_TYPE.get(op, "local_gate") + + +def cmd_to_text(cmd): + """Build a human-readable text representation of a command dict. + + Examples: + gate h → "H server_1[0]" + gate cx → "CX server_1[0], server_1[1]" + entanglement_gen emitter → "ENTG server_1[0]->server_2[0]" + ejpp_start → "EJPP_START (sl=0)" + measure → "MEASURE server_1[0]" + """ + op = cmd.get("op", "unknown") + orig = cmd.get("original_qubits", []) + sl = cmd.get("start_label") or cmd.get("target_start_label") + el = cmd.get("end_label") + + if op == "gate": + gate = cmd.get("gate", "gate").upper() + args = ", ".join(orig) if orig else "" + return f"{gate} {args}".strip() + + if op == "entanglement_gen": + role = cmd.get("role", "") + label = cmd.get("entanglement_label", "") + if orig: + return f"ENTG[{label}] {', '.join(orig)} ({role})" + return f"ENTG[{label}] ({role})" + + if op == "pre_entanglement": + desc = f"PRE_ENTG (sl={sl})" if sl is not None else "PRE_ENTG" + return desc + + if op in ("ejpp_start", "ejpp_start_link", "ejpp_end", "ejpp_end_link"): + label_str = cmd.get("label", op) + suffix = f" (sl={sl})" if sl is not None else (f" (el={el})" if el is not None else "") + return f"{label_str.upper()}{suffix}" + + if op in ("measure", "measure_final"): + args = ", ".join(orig) if orig else "" + return f"{op.upper()} {args}".strip() + + if op in ("msg_sender", "msg_receiver"): + exch_label = cmd.get("label", "") + return f"{op.upper()} [{exch_label}]" + + # Fallback: op + original qubits + desc = f"{op.upper()} {', '.join(orig)}" if orig else op.upper() + if sl is not None: + desc += f" (sl={sl})" + elif el is not None: + desc += f" (el={el})" + return desc + + +def _is_sync_cmd(cmd): + """Return True if this command requires cross-QPU synchronisation.""" + if cmd is None: + return False + op = cmd.get("op", "") + if op == "entanglement_gen" and cmd.get("entanglement_label") is not None: + return True + if cmd.get("start_label") is not None: + return True + if cmd.get("end_label") is not None: + return True + return False + + +# ── Main converter ──────────────────────────────────────────────────────────── + + +def labeled_ir_to_timeslot_schedule(labeled_commands, process_maps, qpu_id_to_label): + """Convert labeled per-QPU commands to a flat timeslot schedule. + + The cursor-based algorithm mirrors ControllerProtocol.compute_timeslot_schedule(). + At each iteration one timeslot is produced; the QPUs that advance are those + participating in the highest-priority sync event, or all non-blocked QPUs + for local gate steps. + + Priority order (same as controller): + 1. entanglement_gen labels (requires all emitter/peer QPUs at cursor) + 2. start_labels (ejpp_start + ejpp_start_link must both be at cursor) + 3. end_labels (ejpp_end + ejpp_end_link must both be at cursor) + 4. Local (non-sync) gates — all ready QPUs advance in parallel + 5. Fallback: advance any single blocked QPU + + :param labeled_commands: ``{int_qpu_id: [cmd_dict, ...]}`` from label_and_build_maps + :param process_maps: ``{start_qpus, end_qpus, entanglement_gen_labels}`` + :param qpu_id_to_label: ``{int_qpu_id: "LBNL-A", ...}`` + :returns: Flat list of timeslot record dicts. + :rtype: list[dict] + """ + qpu_ids = sorted(labeled_commands.keys()) + cursors = {qid: 0 for qid in qpu_ids} + + # Pre-compute per-label required QPU sets from process_maps + start_qpus = process_maps.get("start_qpus", {}) # label -> set(qpu_ids) + end_qpus = process_maps.get("end_qpus", {}) # label -> set(qpu_ids) + # entanglement_labels = process_maps.get("entanglement_gen_labels", set()) + + schedule = [] + timeslot = 0 + max_iterations = sum(len(cmds) for cmds in labeled_commands.values()) * 2 + 10 + + def current_cmd(qid): + idx = cursors[qid] + cmds = labeled_commands[qid] + return cmds[idx] if idx < len(cmds) else None + + def make_record(qid, cmd, ts, qpus_involved_labels): + label = qpu_id_to_label.get(qid, str(qid)) + return { + "timeslot": ts, + "qpu_id": label, + "command": cmd_to_text(cmd), + "command_type": cmd_to_type(cmd), + "qpus_involved": qpus_involved_labels, + "params": cmd.get("params", []), + "qubits": cmd.get("qubits", []), + "original_qubits": cmd.get("original_qubits", []), + } + + for _ in range(max_iterations): + # Check if all QPUs have finished + if all(cursors[qid] >= len(labeled_commands[qid]) for qid in qpu_ids): + break + + advanced = set() + records = [] + sync_found = False + + # Collect current label sets at each QPU's cursor + current_ent_labels = {} # ent_label -> set(qpu_ids) at cursor + current_start_labels = {} # start_label -> set(qpu_ids) at cursor + current_end_labels = {} # end_label -> set(qpu_ids) at cursor + + for qid in qpu_ids: + cmd = current_cmd(qid) + if cmd is None: + continue + op = cmd.get("op", "") + ent_l = cmd.get("entanglement_label") + sl = cmd.get("start_label") + el = cmd.get("end_label") + if op == "entanglement_gen" and ent_l is not None: + current_ent_labels.setdefault(ent_l, set()).add(qid) + if sl is not None: + current_start_labels.setdefault(sl, set()).add(qid) + if el is not None: + current_end_labels.setdefault(el, set()).add(qid) + + # Priority 1: entanglement_gen sync + for label in sorted(current_ent_labels.keys(), key=str): + required = start_qpus.get(label, set()) + ready = current_ent_labels.get(label, set()) + if required and required == ready: + involved_labels = [qpu_id_to_label.get(q, str(q)) for q in sorted(required)] + for qid in sorted(required): + cmd = current_cmd(qid) + records.append(make_record(qid, cmd, timeslot, involved_labels)) + advanced.add(qid) + sync_found = True + break + + # Priority 2: start_label sync (ejpp_start + ejpp_start_link) + if not sync_found: + for label in sorted(current_start_labels.keys(), key=str): + required = start_qpus.get(label, set()) + ready = current_start_labels.get(label, set()) + if required and required == ready: + involved_labels = [qpu_id_to_label.get(q, str(q)) for q in sorted(required)] + for qid in sorted(required): + cmd = current_cmd(qid) + records.append(make_record(qid, cmd, timeslot, involved_labels)) + advanced.add(qid) + sync_found = True + break + + # Priority 3: end_label sync (ejpp_end + ejpp_end_link) + if not sync_found: + for label in sorted(current_end_labels.keys(), key=str): + required = end_qpus.get(label, set()) + ready = current_end_labels.get(label, set()) + if required and required == ready: + involved_labels = [qpu_id_to_label.get(q, str(q)) for q in sorted(required)] + for qid in sorted(required): + cmd = current_cmd(qid) + records.append(make_record(qid, cmd, timeslot, involved_labels)) + advanced.add(qid) + sync_found = True + break + + # Priority 4: local (non-sync) commands — all ready QPUs advance + if not sync_found: + for qid in qpu_ids: + cmd = current_cmd(qid) + if cmd is None: + continue + if _is_sync_cmd(cmd): + continue # blocked waiting for peer + own_label = [qpu_id_to_label.get(qid, str(qid))] + records.append(make_record(qid, cmd, timeslot, own_label)) + advanced.add(qid) + + # Priority 5: fallback — advance any single blocked QPU (avoids deadlock) + if not advanced: + for qid in qpu_ids: + cmd = current_cmd(qid) + if cmd is not None: + own_label = [qpu_id_to_label.get(qid, str(qid))] + records.append(make_record(qid, cmd, timeslot, own_label)) + advanced.add(qid) + logger.warning( + f"[ir_converter] Fallback advance on QPU {qid} at " f"timeslot {timeslot}: {cmd_to_text(cmd)}" + ) + break + + if not advanced: + break # nothing left to do + + for qid in advanced: + cursors[qid] += 1 + + schedule.extend(records) + timeslot += 1 + + else: + logger.warning( + f"[ir_converter] Reached max_iterations={max_iterations} without " + f"completing the schedule. Remaining cursors: " + f"{ {qid: (cursors[qid], len(labeled_commands[qid])) for qid in qpu_ids} }" + ) + + logger.debug(f"[ir_converter] Generated {timeslot} timeslots, " f"{len(schedule)} records for {len(qpu_ids)} QPUs") + return schedule diff --git a/plugins/dqc/dqc/logic.py b/plugins/dqc/dqc/logic.py new file mode 100644 index 0000000..4cb5b5f --- /dev/null +++ b/plugins/dqc/dqc/logic.py @@ -0,0 +1,193 @@ +import math +import logging +from datetime import timedelta +from quantnet_controller.common.experimentdefinitions import Experiment, AgentSequences, Sequence, get_num_timeslot +from quantnet_controller.common.constants import Constants +from collections import defaultdict + +# Duration of one DQC timeslot from the partitioner (timeslot_schedule.json). +# Multiple DQC timeslots may fit inside a single agent scheduler slot (SLOTSIZE). +DQC_TIMESLOT_MS = 3 + +logger = logging.getLogger(__name__) + + +class DQCLogic: + def __init__(self, context): + self.context = context + + @staticmethod + def extract_qpu_pairs(commands_list): + """Return unique canonical (src, dst) QPU pairs from cross-QPU commands. + + A cross-QPU command is one whose ``qpus_involved`` list contains two or + more QPU IDs (e.g. an entanglement-generation operation). Pairs are + returned in sorted order so that (A, B) and (B, A) map to the same key. + + :param commands_list: Flat list of command dicts from the DQC request. + :returns: Deduplicated list of ``(qpu_a, qpu_b)`` string tuples. + :rtype: list[tuple[str, str]] + """ + pairs = set() + for cmd in commands_list: + qpus = cmd.get("qpus_involved") or [] + if len(qpus) >= 2: + src, dst = str(qpus[0]), str(qpus[1]) + pairs.add(tuple(sorted([src, dst]))) + return list(pairs) + + def extract_bsm_nodes_from_path(self, path): + """Return the first available BSMNode ID from a router Path object. + + Iterates over *path.hops* in order and returns a single-element list + containing the first BSM node whose latest agent state is ``IN_SPEC``. + If no BSM node is available, returns an empty list. + + :param path: Path object returned by ``router.find_path``. + :type path: quantnet_controller.common.plugin.Path + :returns: List with at most one BSMNode ID string. + :rtype: list[str] + """ + if path is None or path.hops is None: + return [] + for hop in path.hops: + if not (hasattr(hop, "systemSettings") and hop.systemSettings.type == "BSMNode"): + continue + bsm_id = str(hop.systemSettings.ID) + state = self.context.rm.get_node_state(bsm_id) + if state and state.get("value") == "IN_SPEC": + logger.debug(f"Selected BSM node {bsm_id} (IN_SPEC)") + return [bsm_id] + logger.debug(f"Skipping BSM node {bsm_id} (state={state.get('value') if state else 'unknown'})") + return [] + + def build_dynamic_experiment(self, exp_name, commands_list, node_types=None): + """Build a dynamic Experiment class from the commands list. + + Each generated sequence block will store its original commands in a + `command_list` class attribute. + + :param exp_name: Unique name for the generated experiment class. + :param commands_list: Flat list of command dicts. + :param node_types: Optional dict mapping agent/QPU IDs to their node type string. + :returns: The generated Experiment class. + """ + if not commands_list: + return None + if node_types is None: + node_types = {} + + # Group commands by agent/qpu + commands_by_agent = defaultdict(list) + for cmd in commands_list: + qpu_id = str(cmd.get("qpu_id")) + commands_by_agent[qpu_id].append(cmd) + + agent_ids = sorted(list(commands_by_agent.keys())) + + class DynamicExperiment(Experiment): + name = exp_name + agent_sequences = [] + + @classmethod + def get_allocations(cls, slots_to_allocate, slot_size_sec): + """Transform flat allocated slots into enriched command blocks.""" + allocations = {} + # Match agent sequences to allocated slots in order + for agent_id, agent_seq in zip(agent_ids, cls.agent_sequences): + agent_slots = slots_to_allocate[agent_id] + slot_ptr = 0 + blocks = [] + for seq in agent_seq.sequences: + num = get_num_timeslot(seq) + block_slots = agent_slots[slot_ptr: slot_ptr + num] + if block_slots: + blocks.append( + { + "offset": round(block_slots[0] * slot_size_sec, 6), + "commands": [ + {"slot": s, "message": c} + for s, c in zip(block_slots, getattr(seq, "command_list", [])) + ], + } + ) + slot_ptr += num + allocations[agent_id] = blocks + return allocations + + for agent_id in agent_ids: + cmds = sorted(commands_by_agent[agent_id], key=lambda x: x.get("timeslot", 0)) + + _agent_node_type = node_types.get(agent_id, "QNode") + + class DynamicAgentSeq(AgentSequences): + name = f"Seq_{agent_id}" + node_type = _agent_node_type + sequences = [] + + # First pass: collect raw blocks (list of command dicts) + raw_blocks = [] + current_block = [] + prev_qpus_involved = None + for cmd in cmds: + qpus_inv = cmd.get("qpus_involved") or [] + current_qpus_involved = set(str(q) for q in qpus_inv) + if prev_qpus_involved is not None and current_qpus_involved != prev_qpus_involved: + if current_block: + raw_blocks.append(current_block) + current_block = [] + current_block.append(cmd) + prev_qpus_involved = current_qpus_involved + if current_block: + raw_blocks.append(current_block) + + # Second pass: assign agent-slot durations carrying the fractional remainder + # forward so leftover time from one block is absorbed by the next block rather + # than wasting a full agent slot on every block boundary. + # + # Strategy: track total DQC time elapsed and total agent slots allocated. + # For each new block, compute how many new agent slots are needed given what + # has already been allocated. + # + # e.g. SLOTSIZE=100ms, DQC_TIMESLOT=3ms: + # block 0: 19 cmds → 57ms total, ceil(57/100)=1 slot allocated so far=1 + # block 1: 1 cmd → 60ms total, ceil(60/100)=1 slot delta=0 → use 1 (min) + # block 2: 7 cmds → 81ms total, ceil(81/100)=1 slot delta=0 → use 1 (min) + # block 3: 1 cmd → 84ms total, ceil(84/100)=1 slot delta=0 → use 1 (min) + # block 4: 5 cmds → 99ms total, ceil(99/100)=1 slot delta=0 → use 1 (min) + # block 5: 2 cmds → 105ms total, ceil(105/100)=2 slots delta=1 → use 1 + # ...total = 6 agent slots for 6 blocks (vs 6 without packing) + # When commands pack tightly, multiple blocks share a slot until the ceiling + # increments — only then is a new slot consumed. + slotsize_ms = Constants.SLOTSIZE.total_seconds() * 1000 + total_dqc_ms = 0.0 + slots_allocated = 0 + agent_sequences_list = [] + for b_idx, block in enumerate(raw_blocks): + first_cmd = str(block[0].get("command") or "") + first_op = first_cmd.split(" ")[0] if first_cmd else "Unknown" + seq_name = f"Block_{b_idx}_{first_op}" + + total_dqc_ms += len(block) * DQC_TIMESLOT_MS + new_total_slots = math.ceil(total_dqc_ms / slotsize_ms) + agent_slots = max(new_total_slots - slots_allocated, 1) + slots_allocated += agent_slots + + deps = [agent_sequences_list[-1].name] if agent_sequences_list else [] + total_duration = Constants.SLOTSIZE * agent_slots + cmd_list = [c.get("command") for c in block] + + BlockSequence = type(seq_name, (Sequence,), { + "name": seq_name, + "class_name": seq_name, + "duration": total_duration, + "dependency": deps, + "command_list": cmd_list, + }) + + agent_sequences_list.append(BlockSequence) + + DynamicAgentSeq.sequences = agent_sequences_list + DynamicExperiment.agent_sequences.append(DynamicAgentSeq) + + return DynamicExperiment diff --git a/plugins/dqc/dqc/topology_adapter.py b/plugins/dqc/dqc/topology_adapter.py new file mode 100644 index 0000000..deeca73 --- /dev/null +++ b/plugins/dqc/dqc/topology_adapter.py @@ -0,0 +1,109 @@ +""" +topology_adapter.py +------------------- +Converts live controller topology into the data structures expected by +qnpack frontends and the DQC plugin scheduling pipeline. + +The controller's ResourceManager.get_topology(full=True) returns a dict:: + + { + "num_nodes": int, + "num_qubits": int, + "num_channels": int, + "nodes": [ + { + "id": str, + "systemSettings": {"ID": str, "type": "QNode"|"BSMNode", ...}, + "qubitSettings": {"qubits": [...], ...}, + "channels": [...], + ... + }, + ... + ], + "edges": [...], + } + +qnpack frontends expect qpu_info as:: + + {1: {"num_qubits": 40}, 2: {"num_qubits": 40}, ...} + +where the keys are 1-based integer QPU IDs. +""" + +import logging + +logger = logging.getLogger(__name__) + + +def get_qpu_info_from_topology(context): + """Query live topology and build frontend-compatible data structures. + + Separates QNode and BSMNode entries. QNodes receive 1-based integer IDs + (matching qnpack's convention); BSMNodes are returned as-is for use by + route discovery and BSM command injection. + + :param context: Plugin context — must have ``context.rm`` (ResourceManager). + :returns: Four-tuple ``(qpu_info, qpu_id_to_label, label_to_qpu_id, bsm_nodes)``: + + - **qpu_info** ``{int: {"num_qubits": int, ...}}`` — 1-based QPU ID → + qubit count, ready to pass to ``frontend.parse()``. + - **qpu_id_to_label** ``{int: str}`` — 1-based QPU ID → topology label + (e.g. ``{1: "LBNL-A", 2: "LBNL-B"}``). + - **label_to_qpu_id** ``{str: int}`` — reverse mapping. + - **bsm_nodes** ``list[dict]`` — BSMNode dicts from topology, each + containing at least ``id`` and ``systemSettings``. + + :raises RuntimeError: If the topology cannot be fetched or contains no + QNode entries. + """ + topo = context.rm.get_topology(full=True) + # qnpack's create_nodes_from_topology expects a list (topology_data[0]) + # matching the on-disk format [{"nodes": [...], "edges": [...], ...}]. + # The live RPC returns a bare dict, so wrap it. + raw_topology = [dict(topo)] + nodes = topo.get("nodes", []) + + qnodes = [] + bsm_nodes = [] + + for node in nodes: + sys_settings = node.get("systemSettings", {}) + node_type = sys_settings.get("type", "") + node_id = str(sys_settings.get("ID", node.get("id", ""))) + + if node_type == "QNode": + qubit_settings = node.get("qubitSettings", {}) + qubits = qubit_settings.get("qubits", []) + num_qubits = len(qubits) + qnodes.append({"id": node_id, "num_qubits": num_qubits, "_raw": node}) + elif node_type == "BSMNode": + bsm_entry = dict(node) + bsm_entry["id"] = node_id + bsm_nodes.append(bsm_entry) + + if not qnodes: + raise RuntimeError( + "get_qpu_info_from_topology: no QNode entries found in live topology. " + "Ensure all QPU agents have registered with the controller." + ) + + # Sort by node ID for deterministic 1-based assignment + qnodes.sort(key=lambda n: n["id"]) + + qpu_info = {} + qpu_id_to_label = {} + label_to_qpu_id = {} + + for idx, node in enumerate(qnodes, start=1): + label = node["id"] + num_q = node["num_qubits"] + qpu_info[idx] = {"num_qubits": num_q} + qpu_id_to_label[idx] = label + label_to_qpu_id[label] = idx + + logger.debug( + f"[topology_adapter] {len(qnodes)} QNode(s): {qpu_id_to_label}, " + f"{len(bsm_nodes)} BSMNode(s): {[n['id'] for n in bsm_nodes]}" + ) + + return qpu_info, qpu_id_to_label, label_to_qpu_id, bsm_nodes, raw_topology diff --git a/plugins/dqc/test_dqc.py b/plugins/dqc/test_dqc.py new file mode 100644 index 0000000..e495b5f --- /dev/null +++ b/plugins/dqc/test_dqc.py @@ -0,0 +1,179 @@ +import unittest +import os +import sys +from unittest.mock import MagicMock +from datetime import timedelta +# Import logic +from dqc.logic import DQCLogic +# --- MOCKING MODULES BEFORE IMPORT --- + +mock_exp_defs = MagicMock() + + +class MockExperiment: + pass + + +class MockAgentSequences: + pass + + +class MockSequence: + pass + + +mock_exp_defs.Experiment = MockExperiment +mock_exp_defs.AgentSequences = MockAgentSequences +mock_exp_defs.Sequence = MockSequence + + +mock_exp_defs.Sequence.duration = timedelta(seconds=0) + +sys.modules["quantnet_controller.common.experimentdefinitions"] = mock_exp_defs + +mock_request = MagicMock() +mock_request.RequestManager = MagicMock() +mock_request.RequestType = MagicMock() +sys.modules["quantnet_controller.common.request"] = mock_request + +mock_translator_mod = MagicMock() +mock_translator_class = MagicMock() +mock_translator_mod.RequestTranslator = mock_translator_class +sys.modules["quantnet_controller.common.request_translator"] = mock_translator_mod + +mock_plugin = MagicMock() +mock_plugin.ProtocolPlugin = MagicMock() +mock_plugin.PluginType = MagicMock() +sys.modules["quantnet_controller.common.plugin"] = mock_plugin + +mock_mq = MagicMock() +mock_mq.Code = MagicMock() +mock_mq.schema = MagicMock() +sys.modules["quantnet_mq"] = mock_mq +sys.modules["quantnet_mq.schema.models"] = MagicMock() + +# Make `from logic import DQCLogic` inside dqc/__init__.py resolve correctly. +# The plugin runtime adds the plugin folder to sys.path; replicate that here. +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "dqc")) + + +class TestDQCLogic(unittest.TestCase): + def setUp(self): + self.context = MagicMock() + self.context.config = MagicMock() + + def test_build_dynamic_experiment(self): + logic = DQCLogic(self.context) + + # Flat array structure with qpus_involved + commands_list = [ + {"timeslot": 0, "qpu_id": "LBNL-A", "command": "waiting", "qpus_involved": []}, + { + "timeslot": 1, + "qpu_id": "LBNL-A", + "command": "ENTG server_1_link_register[0], server_0_link_register[0]", + "qpus_involved": ["LBNL-A", "LBNL-B"], + }, + { + "timeslot": 2, + "qpu_id": "LBNL-A", + "command": "CU1(1) server_0_link_register[1], server_0_link_register[0]", + "qpus_involved": ["LBNL-A"], + }, + {"timeslot": 3, "qpu_id": "LBNL-A", "command": "H server_1[0]", "qpus_involved": ["LBNL-A"]}, + {"timeslot": 0, "qpu_id": "LBNL-B", "command": "waiting", "qpus_involved": []}, + { + "timeslot": 1, + "qpu_id": "LBNL-B", + "command": "ENTG server_1_link_register[0], server_0_link_register[0]", + "qpus_involved": ["LBNL-A", "LBNL-B"], + }, + ] + + dynamic_experiment = logic.build_dynamic_experiment("TestExp", commands_list) + + self.assertEqual(dynamic_experiment.name, "TestExp") + # agent_ids = ["LBNL-A", "LBNL-B"] + + # Check Agent LBNL-A Sequences + # Note: build_dynamic_experiment doesn't return agent_ids as a separate list anymore, + # but they are in DynamicExperiment.agent_sequences named Seq_ + agent_seq_names = [s.name for s in dynamic_experiment.agent_sequences] + self.assertIn("Seq_LBNL-A", agent_seq_names) + self.assertIn("Seq_LBNL-B", agent_seq_names) + + agent1_seq = [s for s in dynamic_experiment.agent_sequences if s.name == "Seq_LBNL-A"][0] + seqs = agent1_seq.sequences + + self.assertEqual(len(seqs), 3) + self.assertEqual(seqs[0].name, "Block_0_waiting") + self.assertEqual(seqs[0].duration, timedelta(microseconds=1000 * 1)) + + self.assertEqual(seqs[1].name, "Block_1_ENTG") + self.assertEqual(seqs[1].duration, timedelta(microseconds=1000 * 1)) + + self.assertEqual(seqs[2].name, "Block_2_CU1(1)") + self.assertEqual(seqs[2].duration, timedelta(microseconds=1000 * 2)) # Two commands in this block + + # Check dependencies within LBNL-A + self.assertEqual(seqs[0].dependency, []) + self.assertEqual(seqs[1].dependency, ["Block_0_waiting"]) + self.assertEqual(seqs[2].dependency, ["Block_1_ENTG"]) + + # Check Agent LBNL-B Sequences + agent2_seq = [s for s in dynamic_experiment.agent_sequences if s.name == "Seq_LBNL-B"][0] + seqs2 = agent2_seq.sequences + + self.assertEqual(len(seqs2), 2) + self.assertEqual(seqs2[0].name, "Block_0_waiting") + self.assertEqual(seqs2[1].name, "Block_1_ENTG") + + def test_extract_qpu_pairs_basic(self): + """Cross-QPU commands produce canonical, deduplicated pairs.""" + logic = DQCLogic(self.context) + commands = [ + {"timeslot": 0, "qpu_id": "LBNL-A", "command": "waiting", "qpus_involved": []}, + {"timeslot": 1, "qpu_id": "LBNL-A", "command": "ENTG ...", "qpus_involved": ["LBNL-A", "LBNL-B"]}, + {"timeslot": 1, "qpu_id": "LBNL-B", "command": "ENTG ...", "qpus_involved": ["LBNL-A", "LBNL-B"]}, + {"timeslot": 2, "qpu_id": "LBNL-A", "command": "H ...", "qpus_involved": ["LBNL-A"]}, + ] + pairs = logic.extract_qpu_pairs(commands) + # Duplicate cross-QPU entry should appear only once + self.assertEqual(len(pairs), 1) + self.assertIn(("LBNL-A", "LBNL-B"), pairs) + + def test_extract_qpu_pairs_canonical_order(self): + """Pairs are stored in sorted order regardless of which QPU appears first.""" + logic = DQCLogic(self.context) + commands = [ + {"timeslot": 0, "qpu_id": "LBNL-B", "command": "ENTG ...", "qpus_involved": ["LBNL-B", "LBNL-A"]}, + ] + pairs = logic.extract_qpu_pairs(commands) + self.assertEqual(pairs, [("LBNL-A", "LBNL-B")]) + + def test_extract_qpu_pairs_empty(self): + """Commands with no cross-QPU involvement return an empty list.""" + logic = DQCLogic(self.context) + commands = [ + {"timeslot": 0, "qpu_id": "LBNL-A", "command": "H ...", "qpus_involved": ["LBNL-A"]}, + {"timeslot": 1, "qpu_id": "LBNL-A", "command": "waiting", "qpus_involved": None}, + {"timeslot": 2, "qpu_id": "LBNL-A", "command": "waiting", "qpus_involved": []}, + ] + pairs = logic.extract_qpu_pairs(commands) + self.assertEqual(pairs, []) + + def test_extract_qpu_pairs_multiple(self): + """Multiple distinct QPU pairs are all returned.""" + logic = DQCLogic(self.context) + commands = [ + {"timeslot": 0, "qpu_id": "LBNL-A", "command": "ENTG ...", "qpus_involved": ["LBNL-A", "LBNL-B"]}, + {"timeslot": 1, "qpu_id": "LBNL-B", "command": "ENTG ...", "qpus_involved": ["LBNL-B", "LBNL-C"]}, + ] + pairs = logic.extract_qpu_pairs(commands) + self.assertEqual(len(pairs), 2) + self.assertIn(("LBNL-A", "LBNL-B"), pairs) + self.assertIn(("LBNL-B", "LBNL-C"), pairs) + + +if __name__ == "__main__": + unittest.main() diff --git a/plugins/dqc/test_dqc_rpc.py b/plugins/dqc/test_dqc_rpc.py new file mode 100644 index 0000000..4d2efeb --- /dev/null +++ b/plugins/dqc/test_dqc_rpc.py @@ -0,0 +1,205 @@ +""" +test_dqc_rpc.py +--------------- +Integration test client for the DQC plugin. + +Two usage modes: + + Tket mode (pre-partitioned commands): + python test_dqc_rpc.py [partitioned_commands.json] + + Cisco mode (raw QASM circuit content): + python test_dqc_rpc.py --cisco + +If no arguments are provided, a minimal synthetic tket-mode circuit is sent. + +The partitioned_commands JSON file must contain a per-QPU command dict:: + + { + "1": [{"op": "gate", "gate": "h", "qubits": [0], ...}, ...], + "2": [{"op": "ejpp_start_link", ...}, ...] + } + +Keys are string integers (1-based QPU IDs) — matching qnpack's convention. +""" + +import os +import sys +import json +import asyncio +from quantnet_mq.rpcclient import RPCClient +from quantnet_mq.schema.models import Schema + +# ── Minimal synthetic test circuit (tket mode) ──────────────────────────────── + +# Field names match tket_frontend.py exactly so qpu.py handlers find what they need. +_SYNTHETIC_TKET_COMMANDS = { + "1": [ + { + "op": "gate", + "gate": "h", + "qubit": 20, + "qubits": [20], + "params": [], + "original_qubits": ["server_1[0]"], + "is_remote": False, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + { + "op": "ejpp_start", + "qubit": 20, + "qubits": [0], + "params": [], + "original_qubits": ["server_1_link[0]"], + "is_remote": True, + "data_qubit": 20, + "l_local": 0, + "link_qpu_id": 2, + "link_qubit_idx": 0, + "data_qpu_id": 1, + "peer_qpu_id": 2, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + { + "op": "ejpp_end", + "qubits": [20], + "params": [], + "original_qubits": ["server_1[0]"], + "is_remote": True, + "comm_qubit": 0, + "data_qubit": 20, + "l_local": 0, + "link_qpu_id": 2, + "link_qubit_idx": 0, + "data_qpu_id": 1, + "peer_qpu_id": 2, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + ], + "2": [ + { + "op": "ejpp_start_link", + "qubit": 0, + "qubits": [0], + "params": [], + "original_qubits": ["server_2_link[0]"], + "is_remote": True, + "link_qubit_idx": 0, + "l_local": 0, + "data_qpu_id": 1, + "peer_qpu_id": 1, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + { + "op": "ejpp_end_link", + "qubit": 0, + "qubits": [0], + "params": [], + "original_qubits": ["server_2_link[0]"], + "is_remote": True, + "link_qubit_idx": 0, + "data_qpu_id": 1, + "peer_qpu_id": 1, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + { + "op": "gate", + "gate": "x", + "qubit": 20, + "qubits": [20], + "params": [], + "original_qubits": ["server_2[0]"], + "is_remote": False, + "label": None, + "start_label": None, + "end_label": None, + "clbit": None, + }, + ], +} + + +class DQCTestClient: + def __init__(self, partitioned_commands=None, circuit_mode="tket", circuit_content=None): + self._partitioned_commands = partitioned_commands + self._circuit_mode = circuit_mode + self._circuit_content = circuit_content + + async def do_dqc(self): + if self._circuit_content is not None: + msg = { + "circuit_mode": self._circuit_mode, + "circuit_content": self._circuit_content, + } + print( + f"Sending DQC request: mode={self._circuit_mode}, circuit_content ({len(self._circuit_content)} chars)" + ) + else: + msg = { + "circuit_mode": self._circuit_mode, + "partitioned_commands": self._partitioned_commands, + } + total_cmds = sum(len(v) for v in self._partitioned_commands.values()) + print( + f"Sending DQC request: mode={self._circuit_mode}, " + f"{len(self._partitioned_commands)} QPU(s), {total_cmds} command(s) total" + ) + return json.loads(await self._client.call("dqcRequest", msg, timeout=120.0)) + + async def main(self): + schema_path = os.path.join(os.path.dirname(__file__), "../schema/dqc.yaml") + Schema.load_schema(schema_path, ns="dqc") + + self._client = RPCClient("dqc-client", host=os.getenv("HOST", "localhost")) + self._client.set_handler("dqcRequest", None, "quantnet_mq.schema.models.dqc.dqcRequest") + + await self._client.start() + + try: + ret = await self.do_dqc() + print("Response received:") + print(json.dumps(ret, indent=2)) + except Exception as e: + print(f"Error during DQC call: {e}") + + +if __name__ == "__main__": + if len(sys.argv) > 1 and sys.argv[1] == "--cisco": + if len(sys.argv) < 3: + print("Error: --cisco requires a circuit file path") + sys.exit(1) + qasm_file = sys.argv[2] + if not os.path.exists(qasm_file): + print(f"Error: file not found: {qasm_file}") + sys.exit(1) + with open(qasm_file) as f: + circuit_content = f.read() + client = DQCTestClient(circuit_mode="cisco", circuit_content=circuit_content) + elif len(sys.argv) > 1: + cmd_file = sys.argv[1] + if not os.path.exists(cmd_file): + print(f"Error: file not found: {cmd_file}") + sys.exit(1) + with open(cmd_file) as f: + partitioned_commands = json.load(f) + print(f"Loaded partitioned commands from {cmd_file}") + client = DQCTestClient(partitioned_commands=partitioned_commands, circuit_mode="tket") + else: + client = DQCTestClient(partitioned_commands=_SYNTHETIC_TKET_COMMANDS, circuit_mode="tket") + + asyncio.run(client.main()) diff --git a/plugins/schema/dqc.yaml b/plugins/schema/dqc.yaml new file mode 100644 index 0000000..470485d --- /dev/null +++ b/plugins/schema/dqc.yaml @@ -0,0 +1,107 @@ +--- +asyncapi: "2.6.0" +id: "urn:gov:quant-net" +info: + description: "Quant-Net RPC Distributed Quantum Computing (DQC) Protocol" + title: "RPC DQC Endpoint" + version: "2.0.0" +components: + messages: + RequestMessage: + name: dqcRequest + messageId: dqc.request.id + payload: + "$ref": "#/components/schemas/DQCRequest" + ResponseMessage: + name: dqcResponse + messageId: dqc.response.id + payload: + "$ref": "#/components/schemas/DQCResponse" + schemas: + DQCRequest: + title: dqcRequest + type: object + required: + - cmd + - agentId + - payload + properties: + cmd: + type: string + agentId: + type: string + payload: + type: object + required: + - circuit_mode + properties: + circuit_mode: + type: string + enum: [cisco, tket] + description: > + Which frontend produced the partitioned commands. + 'cisco' = QASM3 (QASM3Frontend), 'tket' = tket (TketFrontend). + partitioned_commands: + type: object + description: > + Per-QPU command lists as produced by frontend.parse(qpu_info). + Keys are 1-based integer QPU IDs (as strings); values are + ordered lists of command dicts with fields: op, qubits, params, + original_qubits, is_remote, start_label, end_label, plus + op-specific fields (e.g. role, peer_qpu_id for entanglement_gen; + data_qubit, link_qpu_id for ejpp_start, etc.). + If omitted, circuit_file or circuit_content must be provided + so the plugin can parse the circuit itself. + additionalProperties: + type: array + items: + type: object + circuit_file: + type: string + description: > + Path to the circuit source file on the controller filesystem. + For circuit_mode 'cisco': a .qasm file path. + For circuit_mode 'tket': a dist_commands.txt path. + Used when partitioned_commands is not provided. + circuit_content: + type: string + description: > + Inline circuit source text (alternative to circuit_file). + The plugin writes this to a temp file before parsing. + Used when partitioned_commands is not provided. + DQCResponse: + title: dqcResponse + type: object + required: + - status + properties: + status: + $ref: "qn-schema:objects/objects.yaml#/components/schemas/Status" + rid: + type: string + data: + type: object + description: Scheduling allocation details, route information, and simulation results. + properties: + startTime: + type: number + description: "Scheduled experiment start time" + allocations: + type: object + description: "Timeslot allocation details per agent" + routes: + type: object + description: > + Network routes between cross-QPU pairs found during scheduling. + Keyed by 'src->dst'; value is an ordered list of node IDs on + the path (or null if no route was found). + additionalProperties: + type: array + items: + type: string + simulation: + type: object + description: > + Results from qnpack NetSquid simulation. + Present when labeled commands were forwarded to the simulator. +defaultContentType: application/json