# Copyright (c) 2023 - 2026 Chair for Design Automation, TUM
# All rights reserved.
#
# SPDX-License-Identifier: MIT
#
# Licensed under the MIT License
from __future__ import annotations
import math
import re
from collections import defaultdict, deque
from typing import TYPE_CHECKING
from qiskit import QuantumCircuit
from qiskit.converters import circuit_to_dagdependency
from qiskit.dagcircuit import DAGDependency
from qiskit.transpiler.passes import RemoveBarriers, RemoveFinalMeasurements
from ..circuit_parsing import extract_qubits_from_gate, is_qasm_file, parse_qasm_circuit
from .cycles import get_state_idxs
from .graph_utils import create_dist_dict, update_distance_map
from .scheduling import assign_gate_to_pz
__all__ = [
"build_dag_gate_id_lookup",
"create_dag",
"create_initial_circuit",
"create_initial_sequence",
"create_updated_sequence_destructive",
"extract_qubits_from_gate",
"get_all_first_gates_and_update_sequence_non_destructive",
"get_front_layer",
"is_qasm_file",
"manual_copy_dag",
"map_front_gates_to_pzs",
"parse_qasm",
"remove_node",
"remove_processed_gates",
]
if TYPE_CHECKING:
from pathlib import Path
from qiskit.circuit import Qubit
from qiskit.dagcircuit import DAGDepNode
from ..circuit_types import GateInfo, ParsedCircuit
from .graph import Graph
[docs]
def parse_qasm(filename: Path) -> list[tuple[int, ...]]:
"""Parse a QASM file and return qubits used for each gate
preserving their order."""
return parse_qasm_circuit(filename, normalize_registers=False).qubit_sequence
def compile_qasm_file(filename: Path) -> list[tuple[int, ...]]:
"""Compile a QASM file and return the compiled sequence of qubits."""
# Check if the file is a valid QASM file
if not is_qasm_file(filename):
msg = "Invalid QASM file format"
raise ValueError(msg)
# Parse the QASM file to extract the qubits used for each gate
return parse_qasm(filename)
[docs]
def get_front_layer(dag: DAGDependency) -> list[DAGDepNode]:
"""Get the front layer of the DAG."""
front_layer = []
for node in dag.get_nodes():
# If a node has no predecessors, it's in the front layer
if not dag.direct_predecessors(node.node_id):
front_layer.append(node)
return front_layer
[docs]
def remove_node(dag: DAGDependency, node: DAGDepNode) -> None:
"""Execute a node and update the DAG (remove the node and its edges)."""
# if dag.direct_successors(node.node_id):
# for successor in dag.direct_successors(node.node_id):
# dag._multi_graph.remove_edge(node.node_id, successor)
dag._multi_graph.remove_node(node.node_id)
def _build_qubit_to_global_index(dag: DAGDependency) -> dict[Qubit, int]:
"""Map DAG qubits to their canonical wire-order indices."""
return {qubit: index for index, qubit in enumerate(dag.qubits)}
def find_best_gate(
graph: Graph,
front_layer: list[DAGDepNode],
dist_map: dict[int, dict[str, int]],
gate_info_map: dict[DAGDepNode, str],
qubit_to_global: dict[Qubit, int],
) -> DAGDepNode:
"""Find the best gate to execute based on distance."""
min_gate_cost = math.inf
for _, gate_node in enumerate(front_layer):
qubit_indices = [qubit_to_global[q] for q in gate_node.qargs]
pz_of_node = gate_info_map[gate_node]
pz = graph.pzs_name_map[pz_of_node]
if gate_node in pz.getting_processed:
return gate_node
gate_cost = max(dist_map[qs][pz_of_node] for qs in qubit_indices)
# if both ions of 2-qubit gate are in pz execute 2-qubit gate
if len(qubit_indices) == 2 and gate_cost == 0:
return gate_node
if gate_cost < min_gate_cost:
min_gate_cost = gate_cost
best_gate = gate_node
return best_gate
[docs]
def manual_copy_dag(dag: DAGDependency) -> DAGDependency:
new_dag = DAGDependency()
# Recreate quantum registers in the new DAG
for qreg in dag.qregs.values():
new_dag.add_qreg(qreg)
# Iterate over all operation nodes in the original DAG and copy them
for node in dag.get_nodes():
new_dag.add_op_node(node.op, node.qargs, node.cargs)
return new_dag
[docs]
def build_dag_gate_id_lookup(dag: DAGDependency, gate_info: dict[int, GateInfo]) -> dict[int, int]:
"""Match DAG nodes to parsed gate ids by qubits and operation name."""
gate_buckets: dict[tuple[tuple[int, ...], str], deque[int]] = defaultdict(deque)
for gate_id, metadata in sorted(gate_info.items()):
gate_name = re.split(r"[\(\s\[]", metadata.qasm)[0]
gate_buckets[metadata.qubits, gate_name].append(gate_id)
lookup: dict[int, int] = {}
qubit_to_global = _build_qubit_to_global_index(dag)
for node in dag.topological_nodes():
if getattr(node, "type", None) != "op":
continue
qubits = tuple(qubit_to_global[q] for q in node.qargs)
gate_name = getattr(getattr(node, "op", None), "name", None) or getattr(node, "name", None) or ""
key = (qubits, gate_name)
if key not in gate_buckets or not gate_buckets[key]:
msg = f"No parsed gate metadata matches DAG node {node.node_id}."
raise ValueError(msg)
lookup[node.node_id] = gate_buckets[key].popleft()
return lookup
[docs]
def create_dag(filename: Path) -> DAGDependency:
qc = QuantumCircuit.from_qasm_file(filename)
# Remove barriers
qc = RemoveBarriers()(qc)
# Remove measurement operations
qc = RemoveFinalMeasurements()(qc)
return circuit_to_dagdependency(qc)
[docs]
def create_initial_circuit(filename: Path) -> ParsedCircuit:
"""Return a canonicalized parsed circuit with stable gate ids."""
return parse_qasm_circuit(filename)
[docs]
def create_initial_sequence(filename: Path) -> list[tuple[int, ...]]:
# assert file is a qasm file
assert is_qasm_file(filename), "The file is not a valid QASM file."
return parse_qasm(filename)
[docs]
def create_updated_sequence_destructive(
graph: Graph,
filename: Path,
dag_dep: DAGDependency | None,
use_dag: bool,
) -> tuple[list[int], list[int], DAGDependency | None]:
# assert file is a qasm file
assert is_qasm_file(filename), "The file is not a valid QASM file."
seq: list[int]
# generate sequence
if use_dag is False:
seq = list(graph.sequence)
dag_dep = None
else:
assert dag_dep is not None
working_dag = manual_copy_dag(dag_dep)
seq = []
graph.dag_gate_id_lookup = build_dag_gate_id_lookup(working_dag, graph.gate_info)
qubit_to_global = _build_qubit_to_global_index(working_dag)
graph.dist_dict = create_dist_dict(graph)
state = get_state_idxs(graph)
dist_map = update_distance_map(graph, state)
# first_flag = True
while True:
first_gates = get_front_layer(working_dag)
if not first_gates:
break
# get all pzs of this front layer gates (in this logic have to reverse the dict)
pz_info_map = map_front_gates_to_pzs(graph, front_layer_nodes=first_gates)
# reverse it
gate_info_map = {value: key for key, values in pz_info_map.items() for value in values}
for pz_name in pz_info_map:
# only include pzs that can process a gate of front layer
if pz_info_map[pz_name]:
first_gate_to_execute = find_best_gate(
graph,
pz_info_map[pz_name],
dist_map,
gate_info_map,
qubit_to_global,
)
# if first_flag == True:
# next_node = first_gate_to_execute
# first_flag = False
remove_node(working_dag, first_gate_to_execute)
seq.append(graph.dag_gate_id_lookup[first_gate_to_execute.node_id])
flat_seq = [qubit for gate in seq for qubit in graph.get_gate_qubits(gate)]
return seq, flat_seq, dag_dep
def get_front_layer_non_destructive(dag: DAGDependency, virtually_processed_nodes: set[int]) -> list[DAGDepNode]:
"""Get the front layer of the DAG without modifying it."""
front_layer = []
for node in dag.get_nodes():
# Skip nodes we've already processed
if node.node_id in virtually_processed_nodes:
continue
# Check if all predecessors have been processed
predecessors = dag.direct_predecessors(node.node_id)
if not predecessors or all(pred in virtually_processed_nodes for pred in predecessors):
front_layer.append(node)
return front_layer
[docs]
def map_front_gates_to_pzs(graph: Graph, front_layer_nodes: list[DAGDepNode]) -> dict[str, list[DAGDepNode]]:
"""Create list of all front layer gates at each processing zone."""
gates_of_pz_info: dict[str, list[DAGDepNode]] = {pz.name: [] for pz in graph.pzs}
for seq_node in front_layer_nodes:
gate_id = graph.dag_gate_id_lookup[seq_node.node_id]
pz = assign_gate_to_pz(graph, gate_id)
gates_of_pz_info[pz].append(seq_node)
return gates_of_pz_info
[docs]
def remove_processed_gates(graph: Graph, dag: DAGDependency, removed_nodes: dict[str, DAGDepNode]) -> None:
"""
Remove the processed gates of each processing zone from both the DAG and sequence.
Args:
graph: Graph object containing the gate sequence
dag: DAG representing dependencies between gates
first_gates_by_pz: Dictionary mapping processing zones to their first gates
"""
# Track which gates are removed
removed_gates = []
# Process each processing zone's first gate
for _pz_name, first_gate in removed_nodes.items():
# Remove the gate from the sequence
# gate_indices = tuple(first_gate.qindices)
gate_id = graph.dag_gate_id_lookup[first_gate.node_id]
if gate_id in graph.sequence:
graph.sequence.remove(gate_id)
removed_gates.append(first_gate)
# Remove the gate from the DAG
node_id = first_gate.node_id
if dag.get_node(node_id):
dag._multi_graph.remove_node(node_id)
[docs]
def get_all_first_gates_and_update_sequence_non_destructive(
graph: Graph,
dag: DAGDependency,
max_rounds: int = 5,
) -> dict[str, DAGDepNode]:
"""Get the first gates from the DAG for each processing zone (only first round, so they are simultaneously processable).
Continue finding the subsequent "first gates" and update the sequence accordingly.
Creates a compiled list of gates (ordered) for each pz from the DAG Dependency."""
ordered_sequence: list[int] = []
processed_nodes: set[int] = set() # Track nodes we've "virtually removed"
# Dictionary to store the first gate for each processing zone
first_nodes_by_pz = {}
# update dist map
state = get_state_idxs(graph)
dist_map = update_distance_map(graph, state)
qubit_to_global = _build_qubit_to_global_index(dag)
for round_recalc_fl in range(max_rounds):
# Get front layer excluding already processed nodes
front_layer_nodes = get_front_layer_non_destructive(dag, processed_nodes)
# If front layer is empty, done
if not front_layer_nodes:
break
pz_info_map = map_front_gates_to_pzs(graph, front_layer_nodes)
gate_info_map = {value: key for key, values in pz_info_map.items() for value in values}
# Track gates processed in this round to ensure maximum parallelism
round_processed_gates = []
# Process one gate for each processing zone that has available gates
for pz_name in pz_info_map:
if pz_info_map[pz_name]:
# Find the best gate for this processing zone
best_gate = find_best_gate(graph, pz_info_map[pz_name], dist_map, gate_info_map, qubit_to_global)
# Save the first gate that can be processed for each pz (only of first round, since otherwise can not be simultaneously processed)
if round_recalc_fl == 0 and pz_name not in first_nodes_by_pz:
first_nodes_by_pz[pz_name] = best_gate
# Add to the processed list for this round
round_processed_gates.append(best_gate)
# Update the ordered sequence
ordered_sequence.append(graph.dag_gate_id_lookup[best_gate.node_id])
# Mark as processed
processed_nodes.add(best_gate.node_id)
# Remove all processed gates from the original sequence
for gate in round_processed_gates:
gate_id = graph.dag_gate_id_lookup[gate.node_id]
if gate_id in graph.sequence:
graph.sequence.remove(gate_id)
# Update the final sequence
graph.sequence = ordered_sequence + graph.sequence
return first_nodes_by_pz