Add a public generate_cutting_experiments function

circuit_knitting/cutting/__init__.py

@@ -27,6 +27,7 @@
     partition_problem
     cut_gates
     decompose_gates
+    generate_cutting_experiments
     execute_experiments
     reconstruct_expectation_values
 
@@ -86,7 +87,11 @@
     decompose_gates,
     PartitionedCuttingProblem,
 )
-from .cutting_evaluation import execute_experiments, CuttingExperimentResults
+from .cutting_evaluation import (
+    execute_experiments,
+    CuttingExperimentResults,
+    generate_cutting_experiments,
+)
 from .cutting_reconstruction import reconstruct_expectation_values
 from .wire_cutting_transforms import cut_wires, expand_observables
 
@@ -95,6 +100,7 @@
     "partition_problem",
     "cut_gates",
     "decompose_gates",
+    "generate_cutting_experiments",
     "execute_experiments",
     "reconstruct_expectation_values",
     "PartitionedCuttingProblem",

circuit_knitting/cutting/cutting_evaluation.py

@@ -13,7 +13,8 @@
 
 from __future__ import annotations
 
-from typing import Any, NamedTuple
+from typing import NamedTuple
+from collections import defaultdict
 from collections.abc import Sequence
 from itertools import chain
 
@@ -119,10 +120,10 @@ def execute_experiments(
 
     # Generate the sub-experiments to run on backend
     (
-        subexperiments,
+        _,
         coefficients,
-        sampled_frequencies,
-    ) = _generate_cutting_experiments(
+        subexperiments,
+    ) = generate_cutting_experiments(
         circuits,
         subobservables,
         num_samples,
@@ -238,12 +239,56 @@ def _append_measurement_circuit(
     return qc
 
 
-def _generate_cutting_experiments(
+def generate_cutting_experiments(
     circuits: QuantumCircuit | dict[str | int, QuantumCircuit],
     observables: PauliList | dict[str | int, PauliList],
-    num_samples: int,
-) -> tuple[list[list[list[QuantumCircuit]]], list[tuple[Any, WeightType]], list[float]]:
-    """Generate all the experiments to run on the backend and their associated coefficients."""
+    num_samples: int | float,
+) -> tuple[
+    list[QuantumCircuit] | dict[str | int, list[QuantumCircuit]],
+    list[tuple[float, WeightType]],
+    list[list[list[QuantumCircuit]]],
+]:
+    """
+    Generate cutting subexperiments and their associated weights.
+
+    If the input circuit and observables are not split into multiple partitions, the output
+    subexperiments will be contained within a 1D array.
+
+    If the input circuit and observables is split into multiple partitions, the output
+    subexperiments will be returned as a dictionary which maps a partition label to to
+    a 1D array containing the subexperiments associated with that partition.
+
+    In both cases, the subexperiment lists are ordered as follows:
+        :math:`[sample_{0}observable_{0}, sample_{0}observable_{1}, ..., sample_{0}observable_{N}, ..., sample_{M}observable_{N}]`
+
+    The weights will always be returned as a 1D array -- one weight for each unique sample.
+
+    Args:
+        circuits: The circuit(s) to partition and separate
+        observables: The observable(s) to evaluate for each unique sample
+        num_samples: The number of samples to draw from the quasi-probability distribution. If set
+            to infinity, the weights will be generated rigorously rather than by sampling from
+            the distribution.
+    Returns:
+        A tuple containing the cutting experiments and their associated weights.
+        If the input circuits is a :class:`QuantumCircuit` instance, the output subexperiments
+        will be a sequence of circuits -- one for every unique sample and observable. If the
+        input circuits are represented as a dictionary keyed by partition labels, the output
+        subexperiments will also be a dictionary keyed by partition labels and containing
+        the subexperiments for each partition.
+        The weights are always a sequence of length-2 tuples, where each tuple contains the
+        weight and the :class:`WeightType`. Each weight corresponds to one unique sample.
+
+    Raises:
+        ValueError: ``num_samples`` must either be an integer or infinity.
+        ValueError: :class:`SingleQubitQPDGate` instances must have the cut number
+            appended to the gate label.
+        ValueError: :class:`SingleQubitQPDGate` instances are not allowed in unseparated circuits.
+    """
+    if isinstance(num_samples, float):
+        if num_samples != np.inf:
+            raise ValueError("num_samples must either be an integer or infinity.")
+
     # Retrieving the unique bases, QPD gates, and decomposed observables is slightly different
     # depending on the format of the execute_experiments input args, but the 2nd half of this function
     # can be shared between both cases.
@@ -285,18 +330,34 @@ def _generate_cutting_experiments(
     # Sort samples in descending order of frequency
     sorted_samples = sorted(random_samples.items(), key=lambda x: x[1][0], reverse=True)
 
-    # Generate the outputs -- sub-experiments, coefficients, and frequencies
-    subexperiments: list[list[list[QuantumCircuit]]] = []
-    coefficients = []
-    sampled_frequencies = []
+    # Generate the output experiments and weights
+    subexperiments_dict: dict[str | int, list[QuantumCircuit]] = defaultdict(list)
+    weights: list[tuple[float, WeightType]] = []
+    for i, (subcircuit, label) in enumerate(
+        strict_zip(subcircuit_list, sorted(subsystem_observables.keys()))
+    ):
+        for z, (map_ids, (redundancy, weight_type)) in enumerate(sorted_samples):
+            actual_coeff = np.prod(
+                [basis.coeffs[map_id] for basis, map_id in strict_zip(bases, map_ids)]
+            )
+            sampled_coeff = (redundancy / num_samples) * (kappa * np.sign(actual_coeff))
+            if i == 0:
+                weights.append((sampled_coeff, weight_type))
+            map_ids_tmp = map_ids
+            if is_separated:
+                map_ids_tmp = tuple(map_ids[j] for j in subcirc_map_ids[i])
+            decomp_qc = decompose_qpd_instructions(
+                subcircuit, subcirc_qpd_gate_ids[i], map_ids_tmp
+            )
+            so = subsystem_observables[label]
+            for j, cog in enumerate(so.groups):
+                meas_qc = _append_measurement_circuit(decomp_qc, cog)
+                subexperiments_dict[label].append(meas_qc)
+
+    # Generate legacy subexperiments list
+    subexperiments_legacy: list[list[list[QuantumCircuit]]] = []
     for z, (map_ids, (redundancy, weight_type)) in enumerate(sorted_samples):
-        subexperiments.append([])
-        actual_coeff = np.prod(
-            [basis.coeffs[map_id] for basis, map_id in strict_zip(bases, map_ids)]
-        )
-        sampled_coeff = (redundancy / num_samples) * (kappa * np.sign(actual_coeff))
-        coefficients.append((sampled_coeff, weight_type))
-        sampled_frequencies.append(redundancy)
+        subexperiments_legacy.append([])
         for i, (subcircuit, label) in enumerate(
             strict_zip(subcircuit_list, sorted(subsystem_observables.keys()))
         ):
@@ -306,13 +367,21 @@ def _generate_cutting_experiments(
             decomp_qc = decompose_qpd_instructions(
                 subcircuit, subcirc_qpd_gate_ids[i], map_ids_tmp
             )
-            subexperiments[-1].append([])
+            subexperiments_legacy[-1].append([])
             so = subsystem_observables[label]
             for j, cog in enumerate(so.groups):
                 meas_qc = _append_measurement_circuit(decomp_qc, cog)
-                subexperiments[-1][-1].append(meas_qc)
+                subexperiments_legacy[-1][-1].append(meas_qc)
 
-    return subexperiments, coefficients, sampled_frequencies
+    # If the circuit wasn't separated, return the subexperiments as a list
+    subexperiments_out: list[QuantumCircuit] | dict[
+        str | int, list[QuantumCircuit]
+    ] = subexperiments_dict
+    assert isinstance(subexperiments_out, dict)
+    if len(subexperiments_out.keys()) == 1:
+        subexperiments_out = subexperiments_dict[list(subexperiments_dict.keys())[0]]
+
+    return subexperiments_out, weights, subexperiments_legacy
 
 
 def _run_experiments_batch(
@@ -377,14 +446,25 @@ def _get_mapping_ids_by_partition(
         subcirc_map_ids.append([])
         for i, inst in enumerate(circ.data):
             if isinstance(inst.operation, SingleQubitQPDGate):
-                decomp_id = int(inst.operation.label.split("_")[-1])
+                try:
+                    decomp_id = int(inst.operation.label.split("_")[-1])
+                except (AttributeError, ValueError):
+                    _raise_bad_qpd_gate_labels()
                 decomp_ids.add(decomp_id)
                 subcirc_qpd_gate_ids[-1].append([i])
                 subcirc_map_ids[-1].append(decomp_id)
 
     return subcirc_qpd_gate_ids, subcirc_map_ids
 
 
+def _raise_bad_qpd_gate_labels() -> None:
+    raise ValueError(
+        "BaseQPDGate instances in input circuit(s) should have their "
+        'labels suffixed with "_<cut_#>" so that sibling SingleQubitQPDGate '
+        "instances may be grouped and sampled together."
+    )
+
+
 def _get_bases_by_partition(
     circuits: Sequence[QuantumCircuit], subcirc_qpd_gate_ids: list[list[list[int]]]
 ) -> list[QPDBasis]:
@@ -393,9 +473,13 @@ def _get_bases_by_partition(
     bases_dict = {}
     for i, subcirc in enumerate(subcirc_qpd_gate_ids):
         for basis_id in subcirc:
-            decomp_id = int(
-                circuits[i].data[basis_id[0]].operation.label.split("_")[-1]
-            )
+            try:
+                decomp_id = int(
+                    circuits[i].data[basis_id[0]].operation.label.split("_")[-1]
+                )
+            except (AttributeError, ValueError):  # pragma: no cover
+                _raise_bad_qpd_gate_labels()  # pragma: no cover
+
             bases_dict[decomp_id] = circuits[i].data[basis_id[0]].operation.basis
     bases = [bases_dict[key] for key in sorted(bases_dict.keys())]
 

test/cutting/test_cutting_decomposition.py

@@ -23,14 +23,17 @@
 
 from circuit_knitting.cutting import (
     partition_circuit_qubits,
+    generate_cutting_experiments,
     partition_problem,
     cut_gates,
 )
 from circuit_knitting.cutting.instructions import Move
 from circuit_knitting.cutting.qpd import (
     QPDBasis,
+    SingleQubitQPDGate,
     TwoQubitQPDGate,
     BaseQPDGate,
+    WeightType,
 )
 
 
@@ -284,3 +287,105 @@ def test_unused_qubits(self):
         )
         assert subcircuits.keys() == {"A", "B"}
         assert subobservables.keys() == {"A", "B"}
+
+    def test_generate_cutting_experiments(self):
+        with self.subTest("simple circuit and observable"):
+            qc = QuantumCircuit(2)
+            qc.append(
+                TwoQubitQPDGate(QPDBasis.from_gate(CXGate()), label="cut_cx"),
+                qargs=[0, 1],
+            )
+            comp_weights = [
+                (0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (-0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (-0.5, WeightType.EXACT),
+            ]
+            subexperiments, weights, _ = generate_cutting_experiments(
+                qc, PauliList(["ZZ"]), np.inf
+            )
+            assert weights == comp_weights
+            assert len(weights) == len(subexperiments)
+            for exp in subexperiments:
+                assert isinstance(exp, QuantumCircuit)
+
+        with self.subTest("simple circuit and observable as dict"):
+            qc = QuantumCircuit(2)
+            qc.append(
+                SingleQubitQPDGate(
+                    QPDBasis.from_gate(CXGate()), label="cut_cx_0", qubit_id=0
+                ),
+                qargs=[0],
+            )
+            qc.append(
+                SingleQubitQPDGate(
+                    QPDBasis.from_gate(CXGate()), label="cut_cx_0", qubit_id=1
+                ),
+                qargs=[1],
+            )
+            comp_weights = [
+                (0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (-0.5, WeightType.EXACT),
+                (0.5, WeightType.EXACT),
+                (-0.5, WeightType.EXACT),
+            ]
+            subexperiments, weights, _ = generate_cutting_experiments(
+                {"A": qc}, {"A": PauliList(["ZY"])}, np.inf
+            )
+            assert weights == comp_weights
+            assert len(weights) == len(subexperiments)
+            for exp in subexperiments:
+                assert isinstance(exp, QuantumCircuit)
+
+        with self.subTest("test bad num_samples"):
+            qc = QuantumCircuit(4)
+            with pytest.raises(ValueError) as e_info:
+                generate_cutting_experiments(qc, PauliList(["ZZZZ"]), 4.5)
+            assert (
+                e_info.value.args[0]
+                == "num_samples must either be an integer or infinity."
+            )
+        with self.subTest("test bad label"):
+            qc = QuantumCircuit(2)
+            qc.append(
+                TwoQubitQPDGate(QPDBasis.from_gate(CXGate()), label="cut_cx"),
+                qargs=[0, 1],
+            )
+            partitioned_problem = partition_problem(
+                qc, "AB", observables=PauliList(["ZZ"])
+            )
+            partitioned_problem.subcircuits["A"].data[0].operation.label = "newlabel"
+            with pytest.raises(ValueError) as e_info:
+                generate_cutting_experiments(
+                    partitioned_problem.subcircuits,
+                    partitioned_problem.subobservables,
+                    np.inf,
+                )
+            assert e_info.value.args[0] == (
+                "BaseQPDGate instances in input circuit(s) should have their labels suffixed with "
+                '"_<cut_#>" so that sibling SingleQubitQPDGate instances may be grouped and sampled together.'
+            )
+        with self.subTest("test bad observable size"):
+            qc = QuantumCircuit(4)
+            with pytest.raises(ValueError) as e_info:
+                generate_cutting_experiments(qc, PauliList(["ZZ"]), np.inf)
+            assert e_info.value.args[0] == (
+                "Quantum circuit qubit count (4) does not match qubit count of observable(s) (2)."
+                "  Try providing `qubit_locations` explicitly."
+            )
+        with self.subTest("test single qubit qpd gate in unseparated circuit"):
+            qc = QuantumCircuit(2)
+            qc.append(
+                SingleQubitQPDGate(QPDBasis.from_gate(CXGate()), 0, label="cut_cx_0"),
+                qargs=[0],
+            )
+            with pytest.raises(ValueError) as e_info:
+                generate_cutting_experiments(qc, PauliList(["ZZ"]), np.inf)
+            assert (
+                e_info.value.args[0]
+                == "SingleQubitQPDGates are not supported in unseparable circuits."
+            )