working interaction graph representation

.pre-commit-config.yaml

@@ -67,9 +67,9 @@ repos:
     hooks:
       - id: ruff
         args: ["--fix", "--show-fixes"]
-        types_or: [python, pyi, jupyter]
+        types_or: [python, pyi]
       - id: ruff-format
-        types_or: [python, pyi, jupyter]
+        types_or: [python, pyi]
 
   # Also run Black on examples in the documentation
   - repo: https://github.com/adamchainz/blacken-docs

pyproject.toml

@@ -66,7 +66,6 @@ docs = [
     "qiskit[visualization]",
     "torch",
     "torch_geometric",
-    "pyzx",
 ]
 
 [project.urls]

src/mqt/predictor/evaluation.py

@@ -119,7 +119,7 @@ def create_mqtpredictor_result(qc: QuantumCircuit, figure_of_merit: reward.figur
             )
     else:
         try:
-            qc_compiled = ml.qcompile(qc, figure_of_merit=figure_of_merit, devices=devices)
+            qc_compiled = ml.qcompile(qc, figure_of_merit=figure_of_merit)
             if qc_compiled:
                 assert isinstance(qc_compiled, tuple)
                 return Result(
@@ -137,10 +137,9 @@ def create_mqtpredictor_result(qc: QuantumCircuit, figure_of_merit: reward.figur
 def evaluate_all_sample_circuits() -> None:
     """Evaluates all sample circuits and saves the results to a csv file."""
     res_csv = []
-    devices = get_available_devices()
 
     results = Parallel(n_jobs=-1, verbose=3, backend="threading")(
-        delayed(evaluate_sample_circuit)(str(file), devices)
+        delayed(evaluate_sample_circuit)(str(file))
         for file in list(ml.helper.get_path_training_circuits().glob("*.qasm"))
     )
     res_csv.append(list(results[0].keys()))
@@ -156,11 +155,10 @@ def evaluate_all_sample_circuits() -> None:
 def evaluate_ghz_circuits() -> None:
     """Evaluates all GHZ circuits and saves the results to a csv file."""
     res_csv = []
-    devices = get_available_devices()
 
     path = Path(str(resources.files("mqt.predictor"))) / "ml" / "training_data" / "ghz"
     results = Parallel(n_jobs=-1, verbose=3, backend="threading")(
-        delayed(evaluate_sample_circuit)(str(file), devices) for file in list(path.glob("*.qasm"))
+        delayed(evaluate_sample_circuit)(str(file)) for file in list(path.glob("*.qasm"))
     )
     res_csv.append(list(results[0].keys()))
     res_csv.extend([list(res.values()) for res in results])
@@ -172,7 +170,7 @@ def evaluate_ghz_circuits() -> None:
     )
 
 
-def evaluate_sample_circuit(filename: str, devices: list[Device]) -> dict[str, Any]:
+def evaluate_sample_circuit(filename: str) -> dict[str, Any]:
     """Evaluates a given sample circuit and returns the results as a dictionary.
 
     Args:
@@ -190,8 +188,8 @@ def evaluate_sample_circuit(filename: str, devices: list[Device]) -> dict[str, A
         "num_qubits": str(Path(filename).stem).replace("_", " ").split(" ")[-1],
     }
     qc = QuantumCircuit.from_qasm_file(filename)
-    results.update(create_mqtpredictor_result(qc, "expected_fidelity", filename=filename, devices=devices).get_dict())
-    results.update(create_mqtpredictor_result(qc, "critical_depth", filename=filename, devices=devices).get_dict())
+    results.update(create_mqtpredictor_result(qc, "expected_fidelity", filename=filename).get_dict())
+    results.update(create_mqtpredictor_result(qc, "critical_depth", filename=filename).get_dict())
 
     for dev in get_available_devices():
         results.update(create_qiskit_result(qc, dev).get_dict())

src/mqt/predictor/ml/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from mqt.predictor.ml.GNNautoencoder import GNNAutoencoder
-from mqt.predictor.ml.GNNclassifier import GNNClassifier, MultiGNNClassifier
+from mqt.predictor.ml.gnn_autoencoder import GNNAutoencoder
+from mqt.predictor.ml.gnn_classifier import GNNClassifier, MultiGNNClassifier
 from mqt.predictor.ml.helper import qcompile
 from mqt.predictor.ml.predictor import Predictor
 

src/mqt/predictor/ml/GNNautoencoder.py → src/mqt/predictor/ml/gnn_autoencoder.py

@@ -6,7 +6,7 @@
 from torch_geometric.loader import DataLoader
 from torch_geometric.nn import VGAE
 
-from mqt.predictor.ml.GNN import Net
+from mqt.predictor.ml.gnn import Net
 
 if TYPE_CHECKING:
     from torch_geometric.data import Dataset

src/mqt/predictor/ml/GNNclassifier.py → src/mqt/predictor/ml/gnn_classifier.py

@@ -6,7 +6,7 @@
 from torch import nn
 from torch_geometric.loader import DataLoader
 
-from mqt.predictor.ml.GNN import Net
+from mqt.predictor.ml.gnn import Net
 
 if TYPE_CHECKING:
     from torch_geometric.data import Dataset

src/mqt/predictor/ml/helper.py

@@ -7,10 +7,9 @@
 
 import networkx as nx  # type: ignore[import-untyped]
 import numpy as np
-import pyzx as zx
 from joblib import dump
 from qiskit import QuantumCircuit
-from qiskit.circuit import Clbit
+from qiskit.circuit import Clbit, Qubit
 from qiskit.converters import circuit_to_dag
 from qiskit.dagcircuit import DAGInNode, DAGOutNode
 from qiskit.transpiler.passes import RemoveBarriers
@@ -24,7 +23,6 @@
     import rustworkx as rx
     import torch_geometric
     from numpy._typing import NDArray
-    from qiskit.circuit import Qubit
     from sklearn.ensemble import RandomForestClassifier
 
     from mqt.bench.devices import Device
@@ -200,10 +198,6 @@ def create_feature_dict(qc: str | QuantumCircuit, graph_features: bool = False)
             feature_dict["graph"] = circuit_to_graph(qc, ops_list_encoding)
         except Exception:
             feature_dict["graph"] = None
-        try:
-            feature_dict["zx_graph"] = qasm_to_zx(qc.qasm())
-        except Exception:  # e.g. zx-calculus not supported for all circuits
-            feature_dict["zx_graph"] = None
 
     for key in ops_list_dict:
         feature_dict[key] = float(ops_list_dict[key])
@@ -219,8 +213,8 @@ def create_feature_dict(qc: str | QuantumCircuit, graph_features: bool = False)
     feature_dict["liveness"] = supermarq_features.liveness
     feature_dict["directed_program_communication"] = supermarq_features.directed_program_communication
     feature_dict["gate_coverage"] = supermarq_features.directed_critical_depth
-    feature_dict["singleQ_gates_per_layer"] = supermarq_features.singleQ_gates_per_layer
-    feature_dict["multiQ_gates_per_layer"] = supermarq_features.multiQ_gates_per_layer
+    feature_dict["single_qubit_gates_per_layer"] = supermarq_features.single_qubit_gates_per_layer
+    feature_dict["multi_qubit_gates_per_layer"] = supermarq_features.multi_qubit_gates_per_layer
     feature_dict["my_critical_depth"] = supermarq_features.my_critical_depth
     return feature_dict
 
@@ -331,105 +325,6 @@ def rustworkx_to_networkx(graph: rx.PyDAG[Any, Any], ops_list_encoding: dict[str
     return nx_graph
 
 
-def zx_to_networkx(zx_graph: zx.graph.BaseGraph) -> nx.DiGraph:
-    # create a networkx graph
-    nx_graph = nx.DiGraph()
-
-    # Add nodes to the NetworkX graph
-    for idx in zx_graph.vertices():
-        t = zx_graph.type(idx)
-        phase = float(zx_graph.phase(idx))
-
-        nx_graph.add_node(idx, gate=t, phase=phase)
-
-    # Add edges to the NetworkX graph
-    for e in zx_graph.edges():
-        source, target = e
-        t = zx_graph.edge_type(e)
-
-        nx_graph.add_edge(source, target, wire=t)
-
-    return nx_graph
-
-
-def substitute_cp_and_cu1_gate(qasm: str) -> str:
-    """
-    Substitute all occurrences of the cp gate in a qasm string with a custom gate definition.
-    """
-
-    # Function to replace cp gate with custom gate definition
-    def replace_cp_gate(match: re.Match[str]) -> str:
-        phase, qubit1, qubit2 = match.groups()
-        if "/" in phase:
-            numerator, denominator = phase.split("/")
-            new_denominator = str(int(denominator) * 2)
-            phase_2 = numerator + "/" + new_denominator
-        else:
-            phase_2 = phase + "/2"
-
-        # rz is same as u1 up to a global phase
-        return (
-            f"rz({phase_2}) {qubit2};\n"
-            f"cx {qubit1},{qubit2};\n"
-            f"rz(-{phase_2}) {qubit2};\n"
-            f"cx {qubit1},{qubit2};\n"
-            f"rz({phase_2}) {qubit2};\n"
-        )
-
-    # Replace all occurrences of the cp gate with the custom gate definition
-    # cp is same as cu1
-    qasm = re.sub(r"cp\((.+?)\) (.+?),(.+?);", replace_cp_gate, qasm)
-    # Replace all occurrences of the cp gate with the custom gate definition
-    qasm = re.sub(r"cu1\((.+?)\) (.+?),(.+?);", replace_cp_gate, qasm)
-
-    return qasm.replace("--", "")
-
-
-def format_u1_gate(qasm: str) -> str:
-    def format_u1(match: re.Match[str]) -> str:
-        phase = match.group(1)
-        return f"u1({phase})"
-
-    return re.sub(r"u\(0,0,(.+?)\)", format_u1, qasm)
-
-
-def replace_swap_gate(qasm: str) -> str:
-    def format_swap(match: re.Match[str]) -> str:
-        qubit1 = match.group(1)
-        qubit2 = match.group(2)
-        return f"cx {qubit1},{qubit2}; cx {qubit2},{qubit1}; cx {qubit1},{qubit2};"
-
-    return re.sub(r"swap (.+?),(.+?);", format_swap, qasm)
-
-
-def qasm_to_zx(qasm: str) -> zx.Circuit:
-    """
-    Convert a qasm string to a zx-calculus string.
-    """
-    qasm = substitute_cp_and_cu1_gate(qasm)
-    qasm = format_u1_gate(qasm)
-    qasm = replace_swap_gate(qasm)
-    qasm = qasm.replace("u1(", "rz(")
-    qasm = qasm.replace("p(", "rz(")
-
-    try:
-        zx_circ = zx.Circuit.from_qasm(qasm)
-        zx_graph = zx_circ.to_graph()
-
-        nx_graph = zx_to_networkx(zx_graph)
-
-        #### Postprocessing ###
-        # Remove root and leaf nodes (in and out nodes)
-        nodes_to_remove = [node for node, degree in nx_graph.degree() if degree < 1]
-        nx_graph.remove_nodes_from(nodes_to_remove)
-
-        # Convert to torch_geometric data
-        return from_networkx(nx_graph, group_node_attrs=all, group_edge_attrs=all)
-    except Exception as e:
-        msg = f"Error in qasm_to_zx: {e}"
-        raise ValueError(msg) from e
-
-
 def circuit_to_graph(qc: QuantumCircuit, ops_list_encoding: dict[str, int]) -> torch_geometric.data.Data:
     """
     Convert a quantum circuit to a torch_geometric graph.
@@ -452,6 +347,59 @@ def circuit_to_graph(qc: QuantumCircuit, ops_list_encoding: dict[str, int]) -> t
     return from_networkx(nx_graph, group_node_attrs=all, group_edge_attrs=all)
 
 
+def circuit_to_interaction_graph(qc: QuantumCircuit, ops_list_encoding: dict[str, int]) -> torch_geometric.data.Data:
+    """
+    Convert a quantum circuit to a torch_geometric graph.
+    """
+    ### Preprocessing ###
+    circ = RemoveBarriers()(qc)
+
+    # Convert to a rustworkx DAG
+    dag_circuit = circuit_to_dag(circ, copy_operations=False)
+
+    # Create an empty interaction nx graph
+    nx_interaction_graph = nx.DiGraph()
+
+    qubit_idx: dict[Qubit, int] = {}  # map qubit to index
+    op_count: dict[tuple[int, int], dict[str, float]] = {}  # count of operations on edge
+    for wire in dag_circuit.wires:
+        if isinstance(wire, Qubit):
+            for node in dag_circuit.nodes_on_wire(wire):
+                if type(node) in [DAGInNode, DAGOutNode]:
+                    continue
+
+                for i, qarg in enumerate(node.qargs):
+                    if qarg not in qubit_idx:
+                        qubit_idx[qarg] = len(qubit_idx)
+                    if i == 0:
+                        u = v = qubit_idx[qarg]
+                        count_increase = 1.0
+                    elif i == 1:
+                        v = qubit_idx[qarg]
+                        # prevent double counting
+                        count_increase = 0.5
+
+                if nx_interaction_graph.has_edge(u, v):
+                    # update gate count on existing edge
+                    op_count[(u, v)][node.op.name] += count_increase
+                    nx_interaction_graph[u][v]["gate_count"] = op_count[(u, v)]
+                else:  # add new edge
+                    op_count[(u, v)] = dict.fromkeys(ops_list_encoding.keys(), 0)
+                    op_count[(u, v)][node.op.name] += count_increase
+                    nx_interaction_graph.add_edge(u, v, gate_count=op_count[(u, v)])
+
+    # transform all edge features to integer numpy arrays
+    for _u, _v, data in nx_interaction_graph.edges(data=True):
+        data["gate_count"] = np.array(list(data["gate_count"].values()), dtype=np.int16)
+
+    # add initial 1.0 feature for all nodes
+    for node in nx_interaction_graph.nodes():
+        nx_interaction_graph.nodes[node]["feature"] = np.ones(1, dtype=np.int16)
+
+    # Convert to torch_geometric data
+    return from_networkx(nx_interaction_graph, group_node_attrs="all", group_edge_attrs="all")
+
+
 @dataclass
 class TrainingData:
     X_train: NDArray[np.float64]

src/mqt/predictor/ml/predictor.py

@@ -13,7 +13,7 @@
 from sklearn.ensemble import RandomForestClassifier
 from sklearn.model_selection import GridSearchCV, train_test_split
 
-from mqt.bench.devices import Device, get_available_devices, get_device_by_name
+from mqt.bench.devices import Device, get_available_device_names, get_available_devices, get_device_by_name
 from mqt.predictor import ml, reward, rl, utils
 
 if TYPE_CHECKING:
@@ -386,28 +386,28 @@ def get_prepared_training_data(
             ml.helper.TrainingData: The prepared training data.
         """
         training_data, names_list, raw_scores_list = ml.helper.load_training_data(figure_of_merit)
-        _unzipped_training_data_x, _unzipped_training_data_y = zip(*training_data, strict=False)
+        unzipped_training_data_x, unzipped_training_data_y = zip(*training_data, strict=False)
         scores_list: list[list[float]] = [[] for _ in range(len(raw_scores_list))]
-        X_raw = list(unzipped_training_data_X)
-        X_list: list[list[float]] = [[] for _ in range(len(X_raw))]
-        X_graph: list[list[Data]] = [[] for _ in range(len(X_raw))]
-        y_list = list(unzipped_training_data_Y)
-        for i in range(len(X_raw)):
+        x_raw = list(unzipped_training_data_x)
+        x_list: list[list[float]] = [[] for _ in range(len(x_raw))]
+        x_graph: list[list[Data]] = [[] for _ in range(len(x_raw))]
+        y_list = list(unzipped_training_data_y)
+        for i in range(len(x_raw)):
             if graph_only:
-                X_graph[i] = list(X_raw[i][:2])  # only graphs
+                x_graph[i] = list(x_raw[i][:2])  # only graphs
             else:
-                X_list[i] = list(X_raw[i][2:])  # all but graphs
+                x_list[i] = list(x_raw[i][2:])  # all but graphs
             scores_list[i] = list(raw_scores_list[i])
 
         y, indices = np.array(y_list), np.array(range(len(y_list)))
 
         if graph_only:
-            X: Any = X_graph
+            x: Any = x_graph
         else:
-            X = np.array(X_list)
+            x = np.array(x_list)
             # Store all non zero feature indices
-            non_zero_indices = [i for i in range(len(X[0])) if sum(X[:, i]) > 0]
-            X = X[:, non_zero_indices]
+            non_zero_indices = [i for i in range(len(x[0])) if sum(x[:, i]) > 0]
+            x = x[:, non_zero_indices]
 
             if save_non_zero_indices:
                 data = np.asarray(non_zero_indices)

src/mqt/predictor/reward.py

@@ -65,15 +65,3 @@ def expected_fidelity(qc: QuantumCircuit, device: Device, precision: int = 10) -
             res *= specific_fidelity
 
     return cast(float, np.round(res, precision))
-
-
-def calc_qubit_index(qargs: list[Qubit], qregs: list[QuantumRegister], index: int) -> int:
-    offset = 0
-    for reg in qregs:
-        if qargs[index] not in reg:
-            offset += reg.size
-        else:
-            qubit_index: int = offset + reg.index(qargs[index])
-            return qubit_index
-    error_msg = f"Global qubit index for local qubit {index} index not found."
-    raise ValueError(error_msg)