Transpilation and routing of measurement circuits (#13)

* deal with routing of transpiled circuits

* add docstring and explanations

* simplify routing of measurement circuit

* "post-composition pass manager" is now optional

* add composed circuits to povm pub metadata

* do not override already existing classical registers

* update TODOs

* remove final measurement of input circuit

* format notebook

* add unittest

* do not print full circuit in metadata

* return JobStatus

.gitignore

@@ -134,5 +134,6 @@ dmypy.json
 
 # untracked local files
 local_files/
+.vscode/
 
 .DS_STORE

povm_toolbox/library/metadata/povm_metadata.py

@@ -12,8 +12,12 @@
 
 from __future__ import annotations
 
+import dataclasses
 from dataclasses import dataclass
 
+import numpy as np
+from qiskit.circuit import QuantumCircuit
+
 from ..povm_implementation import POVMImplementation
 
 
@@ -23,3 +27,32 @@ class POVMMetadata:
 
     povm_implementation: POVMImplementation
     """The POVM implementation which produced the results to which this metadata belongs."""
+
+    composed_circuit: QuantumCircuit
+    """The quantum circuit which produced the results to which this metadata belongs.
+
+    This circuit is the composition of the quantum circuit specified by a PUB and the
+    measurement circuit generated by the POVM implementation used. This is the quantum
+    circuit that is eventually sent to the internal :class:`.BaseSamplerV2`.
+    """
+
+    def __repr__(self):
+        """Implement the default ``__repr__`` method to avoid printing large objects.
+
+        E.g., the attribute ``composed_circuit`` is a quantum circuit. With the default
+        ``dataclass.__repr__``, it would be entirely drawn. As this is recursive,
+        the full circuit would be printed when printing the :class:`.PrimitiveResult`
+        object returned by the :meth:`.POVMSampler.run` method. The redefinition
+        here avoids this.
+        """
+        lst_fields = []
+        for field in dataclasses.fields(self):
+            f_name = field.name
+            f_val = getattr(self, field.name)
+            if isinstance(f_val, np.ndarray):
+                f_val = f'np.ndarray<{",".join(map(str, f_val.shape))}>'
+            elif isinstance(f_val, QuantumCircuit):
+                f_val = f_val.__repr__()
+            lst_fields.append((f_name, f_val))
+        f_repr = ", ".join(f"{name}={value}" for name, value in lst_fields)
+        return f"{self.__class__.__name__}({f_repr})"

povm_toolbox/library/metadata/rpm_metadata.py

@@ -12,7 +12,6 @@
 
 from __future__ import annotations
 
-import dataclasses
 from dataclasses import dataclass
 
 import numpy as np
@@ -31,24 +30,3 @@ class RPMMetadata(POVMMetadata):
     where ``pv`` is the bindings array provided by the user to run with the
     parametrized quantum circuit supplied in the :meth:`.POVMSampler.run` method.
     """
-
-    def __repr__(self):
-        """Implement the default ``__repr__`` method to avoid printing large ``numpy.array``.
-
-        The attribute ``pvm_keys`` will typically be a large ``numpy.ndarray`` object.
-        With the default ``dataclass.__repr__``, it would be entirely printed. As this
-        is recursive, the full array would be printed when printing the :class:`.PrimitiveResult`
-        object returned by the :meth:`.POVMSampler.run` method. The redefinition here avoids this.
-        """
-        lst_fields = []
-        for field in dataclasses.fields(self):
-            f_name = field.name
-            f_val = getattr(self, field.name)
-            f_val = (
-                f_val
-                if not isinstance(f_val, np.ndarray)
-                else f'np.ndarray<{",".join(map(str, f_val.shape))}>'
-            )
-            lst_fields.append((f_name, f_val))
-        f_repr = ", ".join(f"{name}={value}" for name, value in lst_fields)
-        return f"{self.__class__.__name__}({f_repr})"

povm_toolbox/library/povm_implementation.py

@@ -18,11 +18,12 @@
 
 import numpy as np
 from qiskit.circuit import QuantumCircuit
+from qiskit.circuit.exceptions import CircuitError
 from qiskit.primitives.containers import DataBin
 from qiskit.primitives.containers.bindings_array import BindingsArray
 from qiskit.primitives.containers.bit_array import BitArray
 from qiskit.primitives.containers.sampler_pub import SamplerPub
-from qiskit.transpiler import StagedPassManager
+from qiskit.transpiler import StagedPassManager, TranspileLayout
 
 from povm_toolbox.quantum_info.base_povm import BasePOVM
 
@@ -64,7 +65,7 @@ def to_sampler_pub(
         circuit: QuantumCircuit,
         circuit_binding: BindingsArray,
         shots: int,
-        pass_manager: StagedPassManager,
+        pass_manager: StagedPassManager | None = None,
     ) -> tuple[SamplerPub, MetadataT]:
         """Append the measurement circuit(s) to the supplied circuit.
 
@@ -77,6 +78,9 @@ def to_sampler_pub(
             circuit: A quantum circuit.
             parameter_values: A bindings array.
             shots: A specific number of shots to run with.
+            pass_manager: An optional pass manager. After the supplied circuit has
+                been composed with the measurement circuit, the pass manager will
+                transpile the composed circuit.
 
         Returns:
             A tuple of a sampler pub and a dictionary of metadata which include
@@ -90,6 +94,53 @@ def to_sampler_pub(
         # TODO: is it the right place to coerce the ``SamplerPub`` ? Or should
         # just return a ``SamplerPubLike`` object that the SamplerV2 will coerce?
 
+    def compose_circuits(self, circuit: QuantumCircuit) -> QuantumCircuit:
+        """Compose the circuit to sample from, with the measurement circuit.
+
+        Args:
+            circuit: Quantum circuit to be sampled from.
+
+        Returns:
+            The composition of the supplied quantum circuit with the measurement
+            circuit of this POVM implementation.
+        """
+        # Create a copy of the circuit and remove final measurements:
+        dest_circuit = circuit.remove_final_measurements(inplace=False)
+
+        if dest_circuit.layout is None:
+            # Basic one-to-one layout
+            index_layout = list(range(dest_circuit.num_qubits))
+
+        elif isinstance(dest_circuit.layout, TranspileLayout):
+            # Extract the final layout of the transpiled circuit (ancillas are filtered).
+            index_layout = dest_circuit.layout.final_index_layout(filter_ancillas=True)
+        else:
+            raise NotImplementedError
+
+        # Check that the number of qubits of the circuit (before the transpilation, if
+        # applicable) matches the number of qubits of the POVM implementation.
+        if self.n_qubit != len(index_layout):
+            raise ValueError(
+                f"The supplied circuit (acting on {len(index_layout)} qubits)"
+                " does not match this POVM implementation which acts on"
+                f" {self.n_qubit} qubits."
+            )
+
+        try:
+            dest_circuit.add_register(*self.msmt_qc.cregs)
+        except CircuitError as exc:
+            raise CircuitError(
+                f"{exc}\nNote: the supplied quantum circuit should not have a classical register"
+                " which has the same name as the classical register that the POVM"
+                " implementation uses to store measurement outcomes (creg name: "
+                f"'{self.classical_register_name}').\nTo fix it, either delete this register or "
+                " change the name of the register of the supplied circuit or of the"
+                " POVM implementation."
+            ) from exc
+
+        # Compose the two circuits with the correct routing.
+        return dest_circuit.compose(self.msmt_qc, qubits=index_layout, clbits=self.msmt_qc.clbits)
+
     @abstractmethod
     def reshape_data_bin(self, data: DataBin) -> DataBin:
         """TODO."""

povm_toolbox/library/randomized_projective_measurements.py

@@ -131,7 +131,7 @@ def to_sampler_pub(
         circuit: QuantumCircuit,
         circuit_binding: BindingsArray,
         shots: int,
-        pass_manager: StagedPassManager,
+        pass_manager: StagedPassManager | None = None,
     ) -> tuple[SamplerPub, RPMMetadata]:
         """Append the measurement circuit(s) to the supplied circuit.
 
@@ -147,6 +147,9 @@ def to_sampler_pub(
             circuit: A quantum circuit.
             circuit_binding: A bindings array.
             shots: A specific number of shots to run with.
+            pass_manager: An optional pass manager. After the supplied circuit has
+                been composed with the measurement circuit, the pass manager will
+                transpile the composed circuit.
 
         Returns:
             A tuple of a sampler pub and a dictionary of metadata which include
@@ -219,21 +222,20 @@ def to_sampler_pub(
 
         combined_binding = BindingsArray.coerce(binding_data)
 
-        # TODO: assert circuit qubit routing and stuff
-        # TODO: assert both circuits are compatible, in particular no measurements at the end of ``circuits``
-        # TODO: how to compose classical registers ? CR used for POVM measurements should remain separate
-        # TODO: how to deal with transpilation ?
+        composed_circuit = self.compose_circuits(circuit)
 
-        composed_circuit = circuit.compose(self.msmt_qc)
-        composed_isa_circuit = pass_manager.run(composed_circuit)
+        if pass_manager is not None:
+            composed_circuit = pass_manager.run(composed_circuit)
 
         pub = SamplerPub(
-            circuit=composed_isa_circuit,
+            circuit=composed_circuit,
             parameter_values=combined_binding,
             shots=self.shot_batch_size,
         )
 
-        metadata = RPMMetadata(povm_implementation=self, pvm_keys=pvm_idx)
+        metadata = RPMMetadata(
+            povm_implementation=self, composed_circuit=composed_circuit, pvm_keys=pvm_idx
+        )
 
         return (pub, metadata)
 

povm_toolbox/sampler/povm_sampler.py

@@ -16,7 +16,7 @@
 
 from qiskit.primitives import BaseSamplerV2
 from qiskit.primitives.containers.sampler_pub import SamplerPub
-from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager
+from qiskit.transpiler import StagedPassManager
 
 from povm_toolbox.library.metadata import POVMMetadata
 from povm_toolbox.library.povm_implementation import POVMImplementation
@@ -45,6 +45,7 @@ def run(
         *,
         shots: int | None = None,
         povm: POVMImplementation | None = None,
+        pass_manager: StagedPassManager | None = None,
     ) -> POVMSamplerJob:
         """Run and collect samples from each pub.
 
@@ -57,20 +58,18 @@ def run(
             povm: A POVM implementation that defines the measurement to perform
                 for each pub that does not specify it own POVM. If ``None``, each pub
                 has to specify its own POVM.
+            pass_manager: An optional pass manager. For each pub, its circuit will be
+                composed with the associated measurement circuit. If a pass manager is
+                provided, it will transpile the composed circuits.
 
         Returns:
             The POVM sampler job object.
         """
-        # TODO: we need to revisit this as part part of issue #37
-        pm = generate_preset_pass_manager(
-            optimization_level=1, backend=getattr(self.sampler, "_backend", None)
-        )
-
         coerced_sampler_pubs: list[SamplerPub] = []
         metadata: list[POVMMetadata] = []
         for pub in pubs:
             povm_sampler_pub = POVMSamplerPub.coerce(pub=pub, shots=shots, povm=povm)
-            sampler_pub, pub_metadata = povm_sampler_pub.to_sampler_pub(pass_manager=pm)
+            sampler_pub, pub_metadata = povm_sampler_pub.to_sampler_pub(pass_manager=pass_manager)
             coerced_sampler_pubs.append(sampler_pub)
             metadata.append(pub_metadata)
 

povm_toolbox/sampler/povm_sampler_job.py

@@ -78,7 +78,7 @@ def result(self) -> PrimitiveResult[POVMPubResult]:
 
     def status(self) -> JobStatus:
         """Return the status of the job."""
-        raise NotImplementedError("Subclass of BasePrimitiveJob must implement `status` method.")
+        return self.base_job.status()
 
     def done(self) -> bool:
         """Return whether the job has successfully run."""

povm_toolbox/sampler/povm_sampler_pub.py

@@ -231,7 +231,7 @@ def validate(self):
 
     def to_sampler_pub(
         self,
-        pass_manager: StagedPassManager,
+        pass_manager: StagedPassManager | None = None,
     ) -> tuple[SamplerPub, POVMMetadata]:
         """TODO."""
         return self.povm.to_sampler_pub(