From 1832e84cef5cccfcc94212795aaa8f396cc6cf0b Mon Sep 17 00:00:00 2001 From: Se-young Yu Date: Thu, 5 Mar 2026 14:15:13 -0600 Subject: [PATCH 1/5] feat: Implement Distributed Quantum Computing (DQC) plugin with new schema, logic, and tests --- plugins/dqc/dqc/__init__.py | 164 ++++++++++++++++++++++++++++++++++++ plugins/dqc/dqc/logic.py | 138 ++++++++++++++++++++++++++++++ plugins/dqc/test_dqc_rpc.py | 50 +++++++++++ plugins/schema/dqc.yaml | 102 ++++++++++++++++++++++ 4 files changed, 454 insertions(+) create mode 100644 plugins/dqc/dqc/__init__.py create mode 100644 plugins/dqc/dqc/logic.py create mode 100644 plugins/dqc/test_dqc_rpc.py create mode 100644 plugins/schema/dqc.yaml diff --git a/plugins/dqc/dqc/__init__.py b/plugins/dqc/dqc/__init__.py new file mode 100644 index 0000000..d742841 --- /dev/null +++ b/plugins/dqc/dqc/__init__.py @@ -0,0 +1,164 @@ +import logging +from quantnet_controller.common.plugin import ProtocolPlugin, PluginType +from quantnet_controller.common.request import RequestManager, RequestType, RequestParameter +from quantnet_mq import Code +from quantnet_mq.schema.models import dqc, Status as responseStatus +from logic import DQCLogic + +logger = logging.getLogger(__name__) + + +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 + + async def handle_dqc_request(self, request): + """Handle incoming DQC request from partitioner.""" + # Deserialize request + dqc_req = dqc.dqcRequest(**request) + logger.info(f"Received DQC request: {dqc_req.serialize()}") + + try: + # 1. Process commands and agent IDs + # Convert to a plain list of dicts immediately – the raw value is a schema + # proxy type that doesn't support list concatenation. + commands = [dict(c) for c in request.get('payload', {}).get('commands', [])] + agent_ids = sorted(list(set(str(c.get('qpu_id')) for c in commands))) + + # 1b. Find network routes between all cross-QPU pairs (mirrors EGP pattern). + # Cross-QPU commands are identified by having >= 2 entries in qpus_involved. + qpu_pairs = self.logic.extract_qpu_pairs(commands) + routes = {} + # Map route_key -> Path object so we can later inspect hops for BSM nodes. + _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" + ) + + # 1c. Extract BSM nodes from each route and build synthetic BSM commands + # so they are included in the experiment and get scheduled. + node_types = {} # agent_id -> node type string + bsm_commands = [] # synthetic commands injected for BSM agents + _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 + + # Determine which cross-QPU commands belong to this route pair. + 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) + # Create a synthetic command for each cross-QPU command slot + for cmd in pair_cmds: + bsm_cmd = dict(cmd) # shallow copy + 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 node {bsm_id} on route {route_key}") + + # Merge BSM commands into the full command list and refresh agent_ids. + if bsm_commands: + commands = commands + bsm_commands + + # Rebuild agent_ids to include BSM nodes. + all_agent_ids = sorted(list(set(str(c.get('qpu_id')) for c in commands))) + agent_ids = all_agent_ids + + # Safely retrieve the rid from either the dqc_req payload if available, or the raw + # request dictionary. Use a generated UUID as fallback instead of 'unknown'. + 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() + + # 2. 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") + + # 3. Register dynamic experiment with translator + self.request_manager.translator.exp_defs.append(DynamicExp) + + try: + # 4. Get startTime from RequestManager's translator + # This call is locked and calculates the same time the schedule() call will use + start_time, slots = await self.request_manager.translator.get_slots_to_allocate( + agent_ids, DynamicExp + ) + + # 5. Create and schedule request + parameters = RequestParameter(exp_name=exp_name, path=agent_ids) + req_obj = self.request_manager.new_request( + payload=dqc_req, parameters=parameters, rid=rid + ) + + # Blocking schedule (executes the experiment via translator) + await self.request_manager.schedule(req_obj, blocking=True) + + return dqc.dqcResponse( + status=responseStatus( + code=Code.OK.value, value=Code.OK.name, message="DQC execution completed" + ), + rid=rid, + data={"startTime": start_time, "allocations": slots, "routes": routes}, + ) + finally: + # cleanup + 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" + ) diff --git a/plugins/dqc/dqc/logic.py b/plugins/dqc/dqc/logic.py new file mode 100644 index 0000000..66c7abc --- /dev/null +++ b/plugins/dqc/dqc/logic.py @@ -0,0 +1,138 @@ +import logging +from quantnet_controller.common.request import RequestManager, RequestType +from quantnet_controller.common.request_translator import RequestTranslator +from quantnet_controller.common.experimentdefinitions import Experiment, AgentSequences, Sequence +from datetime import timedelta +from collections import defaultdict + +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) + + @staticmethod + def extract_bsm_nodes_from_path(path): + """Extract BSMNode IDs from a router Path object. + + Iterates over *path.hops* and collects every node whose + ``systemSettings.type`` equals ``"BSMNode"``. + + :param path: Path object returned by ``router.find_path``. + :type path: quantnet_controller.common.plugin.Path + :returns: List of BSMNode ID strings found on the path. + :rtype: list[str] + """ + bsm_ids = [] + if path is None or path.hops is None: + return bsm_ids + for hop in path.hops: + if hasattr(hop, 'systemSettings') and hop.systemSettings.type == 'BSMNode': + bsm_ids.append(str(hop.systemSettings.ID)) + return bsm_ids + + def build_dynamic_experiment(self, exp_name, commands_list, node_types=None): + """Build a dynamic Experiment class from the commands list. + + :param exp_name: Unique name for the generated experiment class. + :param commands_list: Flat list of command dicts (may include injected BSM commands). + :param node_types: Optional dict mapping agent/QPU IDs to their node type string + (e.g. ``{"BSM-1": "BSMNode", "QPU-A": "QNode"}``). Defaults to + ``"QNode"`` for any agent not present in the dict. + """ + 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 = [] + + 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 = [] + + agent_sequences_list = [] + current_block = [] + prev_qpus_involved = None + block_index = 0 + + def create_sequence_from_block(block, b_idx, seq_list): + first_cmd_raw = block[0].get('command', '') + first_cmd = str(first_cmd_raw) if first_cmd_raw is not None else "" + first_op = first_cmd.split(' ')[0] if first_cmd else "Unknown" + seq_name = f"Block_{b_idx}_{first_op}" + total_duration = timedelta(microseconds=1000 * len(block)) + deps = [] + if len(seq_list) > 0: + deps.append(seq_list[-1].name) + + class BlockSequence(Sequence): + name = seq_name + class_name = seq_name + duration = total_duration + dependency = deps + + return BlockSequence + + for cmd in cmds: + qpus_inv = cmd.get('qpus_involved', []) + if qpus_inv is None: qpus_inv = [] + current_qpus_involved = set(str(q) for q in qpus_inv) if qpus_inv else set() + + is_boundary = False + if prev_qpus_involved is not None and current_qpus_involved != prev_qpus_involved: + is_boundary = True + + if is_boundary and current_block: + seq_class = create_sequence_from_block(current_block, block_index, agent_sequences_list) + agent_sequences_list.append(seq_class) + block_index += 1 + current_block = [] + + current_block.append(cmd) + prev_qpus_involved = current_qpus_involved + + if current_block: + seq_class = create_sequence_from_block(current_block, block_index, agent_sequences_list) + agent_sequences_list.append(seq_class) + + DynamicAgentSeq.sequences = agent_sequences_list + DynamicExperiment.agent_sequences.append(DynamicAgentSeq) + + return DynamicExperiment diff --git a/plugins/dqc/test_dqc_rpc.py b/plugins/dqc/test_dqc_rpc.py new file mode 100644 index 0000000..550b755 --- /dev/null +++ b/plugins/dqc/test_dqc_rpc.py @@ -0,0 +1,50 @@ +import os +import sys +import json +import asyncio +from quantnet_mq.rpcclient import RPCClient +from quantnet_mq.schema.models import Schema + + +class MyDQC(): + def __init__(self, schedule_file): + self._schedule_file = schedule_file + + async def do_dqc(self): + with open(self._schedule_file, 'r') as f: + commands = json.load(f) + + # Structure the payload as expected by DQCRequest + msg = { + "commands": commands + } + + print(f"Sending DQC request with {len(commands)} commands...") + return json.loads(await self._client.call("dqcRequest", msg, timeout=30.0)) + + async def main(self): + # Adjusted path to match directory structure + 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__": + schedule_file = sys.argv[1] if len(sys.argv) > 1 else "../timeslot_schedule.json" + + if not os.path.exists(schedule_file): + print(f"Error: Schedule file not found at {schedule_file}") + sys.exit(1) + + asyncio.run(MyDQC(schedule_file).main()) diff --git a/plugins/schema/dqc.yaml b/plugins/schema/dqc.yaml new file mode 100644 index 0000000..051abfa --- /dev/null +++ b/plugins/schema/dqc.yaml @@ -0,0 +1,102 @@ +--- +asyncapi: "2.6.0" +id: "urn:gov:quant-net" +info: + description: "Quant-Net RPC Distributed Quantum Computing (DQC) Protocol" + title: "RPC DQC Endpoint" + version: "1.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: + - commands + properties: + commands: + type: array # Flat array of commands + items: + type: object + properties: + timeslot: + type: integer + qpu_id: + type: string + qpu_name: + type: string + command: + type: string + params: + type: array + items: + type: number + qubits: + type: array + items: + type: integer + original_qubits: + type: array + items: + type: string + qpus_involved: + type: array + items: + type: string + qubit_info: + type: array + items: + type: array + items: + type: string + DQCResponse: + title: dqcResponse + type: object + required: + - status + properties: + status: + $ref: "qn-schema:objects/objects.yaml#/components/schemas/Status" + rid: + type: string + data: + type: object # Allocation details + route information + 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 +defaultContentType: application/json From 35627e9ed271b81b9d389639a41d99e80e7489e4 Mon Sep 17 00:00:00 2001 From: Se-young Yu Date: Fri, 13 Mar 2026 16:02:22 -0500 Subject: [PATCH 2/5] feat: Enhance DQC allocation reporting with enriched command blocks --- plugins/dqc/dqc/__init__.py | 10 ++- plugins/dqc/dqc/logic.py | 42 ++++++++-- plugins/dqc/test_dqc.py | 155 ++++++++++++++++++++++++++++++++++++ 3 files changed, 196 insertions(+), 11 deletions(-) create mode 100644 plugins/dqc/test_dqc.py diff --git a/plugins/dqc/dqc/__init__.py b/plugins/dqc/dqc/__init__.py index d742841..21d22d3 100644 --- a/plugins/dqc/dqc/__init__.py +++ b/plugins/dqc/dqc/__init__.py @@ -1,10 +1,11 @@ -import logging 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 +import logging logger = logging.getLogger(__name__) @@ -127,12 +128,13 @@ async def handle_dqc_request(self, request): self.request_manager.translator.exp_defs.append(DynamicExp) try: - # 4. Get startTime from RequestManager's translator - # This call is locked and calculates the same time the schedule() call will use start_time, slots = await self.request_manager.translator.get_slots_to_allocate( agent_ids, DynamicExp ) + # 4. Enrich allocations using the experiment's own self-description + allocations = DynamicExp.get_allocations(slots, Constants.SLOTSIZE.total_seconds()) + # 5. Create and schedule request parameters = RequestParameter(exp_name=exp_name, path=agent_ids) req_obj = self.request_manager.new_request( @@ -147,7 +149,7 @@ async def handle_dqc_request(self, request): code=Code.OK.value, value=Code.OK.name, message="DQC execution completed" ), rid=rid, - data={"startTime": start_time, "allocations": slots, "routes": routes}, + data={"startTime": start_time, "allocations": allocations, "routes": routes}, ) finally: # cleanup diff --git a/plugins/dqc/dqc/logic.py b/plugins/dqc/dqc/logic.py index 66c7abc..aeb2394 100644 --- a/plugins/dqc/dqc/logic.py +++ b/plugins/dqc/dqc/logic.py @@ -1,7 +1,8 @@ import logging from quantnet_controller.common.request import RequestManager, RequestType from quantnet_controller.common.request_translator import RequestTranslator -from quantnet_controller.common.experimentdefinitions import Experiment, AgentSequences, Sequence +from quantnet_controller.common.experimentdefinitions import Experiment, AgentSequences, Sequence, get_num_timeslot +from quantnet_controller.common.constants import Constants from datetime import timedelta from collections import defaultdict @@ -54,11 +55,13 @@ def extract_bsm_nodes_from_path(path): 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 (may include injected BSM commands). - :param node_types: Optional dict mapping agent/QPU IDs to their node type string - (e.g. ``{"BSM-1": "BSMNode", "QPU-A": "QNode"}``). Defaults to - ``"QNode"`` for any agent not present in the dict. + :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 @@ -77,6 +80,30 @@ 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)) @@ -97,7 +124,7 @@ def create_sequence_from_block(block, b_idx, seq_list): first_cmd = str(first_cmd_raw) if first_cmd_raw is not None else "" first_op = first_cmd.split(' ')[0] if first_cmd else "Unknown" seq_name = f"Block_{b_idx}_{first_op}" - total_duration = timedelta(microseconds=1000 * len(block)) + total_duration = Constants.SLOTSIZE * len(block) deps = [] if len(seq_list) > 0: deps.append(seq_list[-1].name) @@ -107,7 +134,8 @@ class BlockSequence(Sequence): class_name = seq_name duration = total_duration dependency = deps - + command_list = [c.get('command') for c in block] + return BlockSequence for cmd in cmds: diff --git a/plugins/dqc/test_dqc.py b/plugins/dqc/test_dqc.py new file mode 100644 index 0000000..35649a2 --- /dev/null +++ b/plugins/dqc/test_dqc.py @@ -0,0 +1,155 @@ +import unittest +import asyncio +import os +import sys +import json +from unittest.mock import MagicMock, AsyncMock, patch + +# --- 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 + +from datetime import timedelta +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")) + +# Import logic +from dqc.logic import DQCLogic + +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() From 47e985f18319e14675ec40ddd77d0e228d59f25e Mon Sep 17 00:00:00 2001 From: Se-young Yu Date: Fri, 24 Apr 2026 15:26:20 -0500 Subject: [PATCH 3/5] added fcfs to dqc --- plugins/dqc/dqc/logic.py | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/plugins/dqc/dqc/logic.py b/plugins/dqc/dqc/logic.py index aeb2394..104b04c 100644 --- a/plugins/dqc/dqc/logic.py +++ b/plugins/dqc/dqc/logic.py @@ -32,25 +32,30 @@ def extract_qpu_pairs(commands_list): pairs.add(tuple(sorted([src, dst]))) return list(pairs) - @staticmethod - def extract_bsm_nodes_from_path(path): - """Extract BSMNode IDs from a router Path object. + def extract_bsm_nodes_from_path(self, path): + """Return the first available BSMNode ID from a router Path object. - Iterates over *path.hops* and collects every node whose - ``systemSettings.type`` equals ``"BSMNode"``. + 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 of BSMNode ID strings found on the path. + :returns: List with at most one BSMNode ID string. :rtype: list[str] """ - bsm_ids = [] if path is None or path.hops is None: - return bsm_ids + return [] for hop in path.hops: - if hasattr(hop, 'systemSettings') and hop.systemSettings.type == 'BSMNode': - bsm_ids.append(str(hop.systemSettings.ID)) - return bsm_ids + 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. From 9e1f2dfc090c0dc6f56cf813910e53ed13341f74 Mon Sep 17 00:00:00 2001 From: Se-young Yu Date: Fri, 5 Jun 2026 15:47:04 -0500 Subject: [PATCH 4/5] feat: Refactor DQC logic and tests for improved command handling and structure --- plugins/dqc/dqc/logic.py | 138 +++++++++++++++++++++--------------- plugins/dqc/test_dqc.py | 80 ++++++++++++++------- plugins/dqc/test_dqc_rpc.py | 26 ++++--- 3 files changed, 145 insertions(+), 99 deletions(-) diff --git a/plugins/dqc/dqc/logic.py b/plugins/dqc/dqc/logic.py index 104b04c..4cb5b5f 100644 --- a/plugins/dqc/dqc/logic.py +++ b/plugins/dqc/dqc/logic.py @@ -1,13 +1,17 @@ +import math import logging -from quantnet_controller.common.request import RequestManager, RequestType -from quantnet_controller.common.request_translator import RequestTranslator +from datetime import timedelta from quantnet_controller.common.experimentdefinitions import Experiment, AgentSequences, Sequence, get_num_timeslot from quantnet_controller.common.constants import Constants -from datetime import timedelta 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 @@ -26,7 +30,7 @@ def extract_qpu_pairs(commands_list): """ pairs = set() for cmd in commands_list: - qpus = cmd.get('qpus_involved') or [] + qpus = cmd.get("qpus_involved") or [] if len(qpus) >= 2: src, dst = str(qpus[0]), str(qpus[1]) pairs.add(tuple(sorted([src, dst]))) @@ -47,11 +51,11 @@ def extract_bsm_nodes_from_path(self, path): 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'): + 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': + 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'})") @@ -60,7 +64,7 @@ def extract_bsm_nodes_from_path(self, path): 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 + 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. @@ -76,11 +80,11 @@ def build_dynamic_experiment(self, exp_name, commands_list, node_types=None): # Group commands by agent/qpu commands_by_agent = defaultdict(list) for cmd in commands_list: - qpu_id = str(cmd.get('qpu_id')) + 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 = [] @@ -96,75 +100,93 @@ def get_allocations(cls, slots_to_allocate, slot_size_sec): blocks = [] for seq in agent_seq.sequences: num = get_num_timeslot(seq) - block_slots = agent_slots[slot_ptr : slot_ptr + num] + 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', [])) - ] - }) + 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)) - + 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 = [] - - agent_sequences_list = [] + + # First pass: collect raw blocks (list of command dicts) + raw_blocks = [] current_block = [] prev_qpus_involved = None - block_index = 0 - - def create_sequence_from_block(block, b_idx, seq_list): - first_cmd_raw = block[0].get('command', '') - first_cmd = str(first_cmd_raw) if first_cmd_raw is not None else "" - first_op = first_cmd.split(' ')[0] if first_cmd else "Unknown" - seq_name = f"Block_{b_idx}_{first_op}" - total_duration = Constants.SLOTSIZE * len(block) - deps = [] - if len(seq_list) > 0: - deps.append(seq_list[-1].name) - - class BlockSequence(Sequence): - name = seq_name - class_name = seq_name - duration = total_duration - dependency = deps - command_list = [c.get('command') for c in block] - - return BlockSequence - for cmd in cmds: - qpus_inv = cmd.get('qpus_involved', []) - if qpus_inv is None: qpus_inv = [] - current_qpus_involved = set(str(q) for q in qpus_inv) if qpus_inv else set() - - is_boundary = False + 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: - is_boundary = True - - if is_boundary and current_block: - seq_class = create_sequence_from_block(current_block, block_index, agent_sequences_list) - agent_sequences_list.append(seq_class) - block_index += 1 + if current_block: + raw_blocks.append(current_block) current_block = [] - current_block.append(cmd) prev_qpus_involved = current_qpus_involved - if current_block: - seq_class = create_sequence_from_block(current_block, block_index, agent_sequences_list) - agent_sequences_list.append(seq_class) - + 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) diff --git a/plugins/dqc/test_dqc.py b/plugins/dqc/test_dqc.py index 35649a2..4b4e93f 100644 --- a/plugins/dqc/test_dqc.py +++ b/plugins/dqc/test_dqc.py @@ -1,45 +1,55 @@ import unittest -import asyncio import os import sys -import json -from unittest.mock import MagicMock, AsyncMock, patch +from unittest.mock import MagicMock +from datetime import timedelta # --- MOCKING MODULES BEFORE IMPORT --- mock_exp_defs = MagicMock() -class MockExperiment: pass -class MockAgentSequences: pass -class MockSequence: pass + + +class MockExperiment: + pass + + +class MockAgentSequences: + pass + + +class MockSequence: + pass + + mock_exp_defs.Experiment = MockExperiment mock_exp_defs.AgentSequences = MockAgentSequences mock_exp_defs.Sequence = MockSequence -from datetime import timedelta + mock_exp_defs.Sequence.duration = timedelta(seconds=0) -sys.modules['quantnet_controller.common.experimentdefinitions'] = mock_exp_defs +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 +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 +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 +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() +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. @@ -48,31 +58,47 @@ class MockSequence: pass # Import logic from dqc.logic import DQCLogic + 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": 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"]}, + { + "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, + # 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) @@ -80,26 +106,26 @@ def test_build_dynamic_experiment(self): 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") diff --git a/plugins/dqc/test_dqc_rpc.py b/plugins/dqc/test_dqc_rpc.py index 550b755..19a4a56 100644 --- a/plugins/dqc/test_dqc_rpc.py +++ b/plugins/dqc/test_dqc_rpc.py @@ -6,32 +6,30 @@ from quantnet_mq.schema.models import Schema -class MyDQC(): +class MyDQC: def __init__(self, schedule_file): self._schedule_file = schedule_file async def do_dqc(self): - with open(self._schedule_file, 'r') as f: + with open(self._schedule_file, "r") as f: commands = json.load(f) - + # Structure the payload as expected by DQCRequest - msg = { - "commands": commands - } - + msg = {"commands": commands} + print(f"Sending DQC request with {len(commands)} commands...") - return json.loads(await self._client.call("dqcRequest", msg, timeout=30.0)) + return json.loads(await self._client.call("dqcRequest", msg, timeout=60.0)) async def main(self): # Adjusted path to match directory structure 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:") @@ -41,10 +39,10 @@ async def main(self): if __name__ == "__main__": - schedule_file = sys.argv[1] if len(sys.argv) > 1 else "../timeslot_schedule.json" - + schedule_file = sys.argv[1] if len(sys.argv) > 1 else "../qnpack/qnpack/dqc_sim/timeslot_schedule.json" + if not os.path.exists(schedule_file): print(f"Error: Schedule file not found at {schedule_file}") sys.exit(1) - + asyncio.run(MyDQC(schedule_file).main()) From 54cfc3dab31386d593485d5b1c89aac44e7d4978 Mon Sep 17 00:00:00 2001 From: Se-young Yu Date: Tue, 7 Jul 2026 17:42:40 -0500 Subject: [PATCH 5/5] feat: Add IR converter and topology adapter for DQC plugin - Implemented ir_converter.py to convert labeled per-QPU IR commands into a flat timeslot record format for DQCLogic. - Added topology_adapter.py to convert live controller topology into data structures compatible with qnpack frontends and DQC scheduling. - Updated test_dqc.py to reflect changes in DQCLogic and removed unused agent_ids. - Enhanced test_dqc_rpc.py to support both tket and Cisco modes for DQC requests, including synthetic test circuits. - Updated dqc.yaml schema to version 2.0.0, adding support for circuit_mode and partitioned_commands. --- plugins/dqc/dqc/__init__.py | 308 +++++++++++++++++++++------ plugins/dqc/dqc/ir_converter.py | 318 ++++++++++++++++++++++++++++ plugins/dqc/dqc/topology_adapter.py | 109 ++++++++++ plugins/dqc/test_dqc.py | 8 +- plugins/dqc/test_dqc_rpc.py | 191 +++++++++++++++-- plugins/schema/dqc.yaml | 83 ++++---- 6 files changed, 897 insertions(+), 120 deletions(-) create mode 100644 plugins/dqc/dqc/ir_converter.py create mode 100644 plugins/dqc/dqc/topology_adapter.py diff --git a/plugins/dqc/dqc/__init__.py b/plugins/dqc/dqc/__init__.py index 21d22d3..ed62431 100644 --- a/plugins/dqc/dqc/__init__.py +++ b/plugins/dqc/dqc/__init__.py @@ -1,14 +1,64 @@ +""" +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 -import logging 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) @@ -34,27 +84,41 @@ def destroy(self): def reset(self): pass + # ── Public request handler ──────────────────────────────────────────────── + async def handle_dqc_request(self, request): - """Handle incoming DQC request from partitioner.""" - # Deserialize 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: - # 1. Process commands and agent IDs - # Convert to a plain list of dicts immediately – the raw value is a schema - # proxy type that doesn't support list concatenation. - commands = [dict(c) for c in request.get('payload', {}).get('commands', [])] - agent_ids = sorted(list(set(str(c.get('qpu_id')) for c in commands))) - - # 1b. Find network routes between all cross-QPU pairs (mirrors EGP pattern). - # Cross-QPU commands are identified by having >= 2 entries in qpus_involved. + # 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 = {} - # Map route_key -> Path object so we can later inspect hops for BSM nodes. _path_objects = {} - if qpu_pairs and hasattr(self.ctx, 'router') and self.ctx.router: - for (src, dst) in qpu_pairs: + 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) @@ -66,14 +130,11 @@ async def handle_dqc_request(self, request): routes[route_key] = None _path_objects[route_key] = None else: - logger.debug( - "No router plugin available or no cross-QPU pairs; skipping route discovery" - ) + logger.debug("No router plugin available or no cross-QPU pairs; skipping route discovery") - # 1c. Extract BSM nodes from each route and build synthetic BSM commands - # so they are included in the experiment and get scheduled. - node_types = {} # agent_id -> node type string - bsm_commands = [] # synthetic commands injected for BSM agents + # ── 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(): @@ -81,86 +142,215 @@ async def handle_dqc_request(self, request): if not bsm_ids: continue - # Determine which cross-QPU commands belong to this route pair. 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', [])] + 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' + node_types[bsm_id] = "BSMNode" if bsm_id in _seen_bsm_ids: continue _seen_bsm_ids.add(bsm_id) - # Create a synthetic command for each cross-QPU command slot for cmd in pair_cmds: - bsm_cmd = dict(cmd) # shallow copy - bsm_cmd['qpu_id'] = bsm_id - bsm_cmd['node_type'] = 'BSMNode' + 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 node {bsm_id} on route {route_key}") + logger.info(f"Injected {len(pair_cmds)} BSM command(s) for " f"node {bsm_id} on route {route_key}") - # Merge BSM commands into the full command list and refresh agent_ids. if bsm_commands: commands = commands + bsm_commands - # Rebuild agent_ids to include BSM nodes. - all_agent_ids = sorted(list(set(str(c.get('qpu_id')) for c in commands))) + all_agent_ids = sorted(list(set(str(c.get("qpu_id")) for c in commands))) agent_ids = all_agent_ids - # Safely retrieve the rid from either the dqc_req payload if available, or the raw - # request dictionary. Use a generated UUID as fallback instead of 'unknown'. - rid = getattr(dqc_req.payload, 'rid', None) or request.get('id') - if not rid or rid == 'unknown': + # ── 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() - # 2. Build dynamic experiment structure + # ── 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") - # 3. Register dynamic experiment with translator + # ── 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 - ) + start_time, slots = await self.request_manager.translator.get_slots_to_allocate(agent_ids, DynamicExp) - # 4. Enrich allocations using the experiment's own self-description allocations = DynamicExp.get_allocations(slots, Constants.SLOTSIZE.total_seconds()) - # 5. Create and schedule request parameters = RequestParameter(exp_name=exp_name, path=agent_ids) - req_obj = self.request_manager.new_request( - payload=dqc_req, parameters=parameters, rid=rid - ) + req_obj = self.request_manager.new_request(payload=dqc_req, parameters=parameters, rid=rid) - # Blocking schedule (executes the experiment via translator) 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" - ), + status=responseStatus(code=Code.OK.value, value=Code.OK.name, message="DQC execution completed"), rid=rid, - data={"startTime": start_time, "allocations": allocations, "routes": routes}, + data={ + "startTime": start_time, + "allocations": allocations, + "routes": routes, + "simulation_payload": sim_payload, + }, ) finally: - # cleanup 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" + 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/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 index 4b4e93f..e495b5f 100644 --- a/plugins/dqc/test_dqc.py +++ b/plugins/dqc/test_dqc.py @@ -3,7 +3,8 @@ 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() @@ -55,9 +56,6 @@ class MockSequence: # 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")) -# Import logic -from dqc.logic import DQCLogic - class TestDQCLogic(unittest.TestCase): def setUp(self): @@ -95,7 +93,7 @@ def test_build_dynamic_experiment(self): dynamic_experiment = logic.build_dynamic_experiment("TestExp", commands_list) self.assertEqual(dynamic_experiment.name, "TestExp") - agent_ids = ["LBNL-A", "LBNL-B"] + # 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, diff --git a/plugins/dqc/test_dqc_rpc.py b/plugins/dqc/test_dqc_rpc.py index 19a4a56..4d2efeb 100644 --- a/plugins/dqc/test_dqc_rpc.py +++ b/plugins/dqc/test_dqc_rpc.py @@ -1,3 +1,28 @@ +""" +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 @@ -5,23 +30,138 @@ from quantnet_mq.rpcclient import RPCClient from quantnet_mq.schema.models import Schema +# ── Minimal synthetic test circuit (tket mode) ──────────────────────────────── -class MyDQC: - def __init__(self, schedule_file): - self._schedule_file = schedule_file +# 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, + }, + ], +} - async def do_dqc(self): - with open(self._schedule_file, "r") as f: - commands = json.load(f) - # Structure the payload as expected by DQCRequest - msg = {"commands": commands} +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 - print(f"Sending DQC request with {len(commands)} commands...") - return json.loads(await self._client.call("dqcRequest", msg, timeout=60.0)) + 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): - # Adjusted path to match directory structure schema_path = os.path.join(os.path.dirname(__file__), "../schema/dqc.yaml") Schema.load_schema(schema_path, ns="dqc") @@ -39,10 +179,27 @@ async def main(self): if __name__ == "__main__": - schedule_file = sys.argv[1] if len(sys.argv) > 1 else "../qnpack/qnpack/dqc_sim/timeslot_schedule.json" - - if not os.path.exists(schedule_file): - print(f"Error: Schedule file not found at {schedule_file}") - sys.exit(1) + 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(MyDQC(schedule_file).main()) + asyncio.run(client.main()) diff --git a/plugins/schema/dqc.yaml b/plugins/schema/dqc.yaml index 051abfa..470485d 100644 --- a/plugins/schema/dqc.yaml +++ b/plugins/schema/dqc.yaml @@ -4,7 +4,7 @@ id: "urn:gov:quant-net" info: description: "Quant-Net RPC Distributed Quantum Computing (DQC) Protocol" title: "RPC DQC Endpoint" - version: "1.0.0" + version: "2.0.0" components: messages: RequestMessage: @@ -33,45 +33,44 @@ components: payload: type: object required: - - commands + - circuit_mode properties: - commands: - type: array # Flat array of commands - items: - type: object - properties: - timeslot: - type: integer - qpu_id: - type: string - qpu_name: - type: string - command: - type: string - params: - type: array - items: - type: number - qubits: - type: array - items: - type: integer - original_qubits: - type: array - items: - type: string - qpus_involved: - type: array - items: - type: string - qubit_info: - type: array - items: - type: array - items: - type: string + 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 + title: dqcResponse type: object required: - status @@ -81,7 +80,8 @@ components: rid: type: string data: - type: object # Allocation details + route information + type: object + description: Scheduling allocation details, route information, and simulation results. properties: startTime: type: number @@ -99,4 +99,9 @@ components: 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