Experiment run hook methods

qiskit_experiments/framework/base_experiment.py

@@ -22,7 +22,6 @@
 from qiskit.providers import BaseJob
 from qiskit.providers.backend import Backend
 from qiskit.providers.basebackend import BaseBackend as LegacyBackend
-from qiskit.test.mock import FakeBackend
 from qiskit.exceptions import QiskitError
 from qiskit.qobj.utils import MeasLevel
 from qiskit_experiments.framework import Options
@@ -96,10 +95,6 @@ def run(
 
         Returns:
             The experiment data object.
-
-        Raises:
-            QiskitError: if experiment is run with an incompatible existing
-                         ExperimentData container.
         """
         # Create experiment data container
         experiment_data = self._initialize_experiment_data(backend, experiment_data)
@@ -109,43 +104,25 @@ def run(
         run_opts.update_options(**run_options)
         run_opts = run_opts.__dict__
 
-        # Scheduling parameters
-        if backend.configuration().simulator is False and isinstance(backend, FakeBackend) is False:
-            timing_constraints = getattr(self.transpile_options.__dict__, "timing_constraints", {})
-            timing_constraints["acquire_alignment"] = getattr(
-                timing_constraints, "acquire_alignment", 16
-            )
-            scheduling_method = getattr(
-                self.transpile_options.__dict__, "scheduling_method", "alap"
-            )
-            self.set_transpile_options(
-                timing_constraints=timing_constraints, scheduling_method=scheduling_method
-            )
-
         # Generate and transpile circuits
-        transpile_opts = copy.copy(self.transpile_options.__dict__)
-        transpile_opts["initial_layout"] = list(self._physical_qubits)
-        circuits = transpile(self.circuits(backend), backend, **transpile_opts)
-        self._postprocess_transpiled_circuits(circuits, backend, **run_options)
+        circuits = self.run_transpile(backend)
 
+        # Execute experiment
         if isinstance(backend, LegacyBackend):
             qobj = assemble(circuits, backend=backend, **run_opts)
             job = backend.run(qobj)
         else:
             job = backend.run(circuits, **run_opts)
 
-        # Add Job to ExperimentData and add analysis for post processing.
-        run_analysis = None
-
         # Add experiment option metadata
         self._add_job_metadata(experiment_data, job, **run_opts)
 
-        if analysis and self.__analysis_class__ is not None:
-            run_analysis = self.run_analysis
-
-        experiment_data.add_data(job, post_processing_callback=run_analysis)
+        # Run analysis
+        if analysis:
+            experiment_data = self.run_analysis(experiment_data, job)
+        else:
+            experiment_data.add_data(job)
 
-        # Return the ExperimentData future
         return experiment_data
 
     def _initialize_experiment_data(
@@ -167,29 +144,114 @@ def _initialize_experiment_data(
 
         return experiment_data._copy_metadata()
 
-    def run_analysis(self, experiment_data, **options) -> ExperimentData:
+    def _pre_transpile_hook(self, backend: Backend):
+        """An extra subroutine executed before transpilation.
+
+        Args:
+            backend: Target backend.
+        """
+        pass
+
+    # pylint: disable = unused-argument
+    def _post_transpile_hook(
+        self, circuits: List[QuantumCircuit], backend: Backend
+    ) -> List[QuantumCircuit]:
+        """An extra subroutine executed after transpilation.
+
+        Args:
+            circuits: List of transpiled circuits.
+            backend: Target backend.
+
+        Returns:
+            List of circuits to execute.
+        """
+        return circuits
+
+    def run_transpile(self, backend: Backend, **options) -> List[QuantumCircuit]:
+        """Run transpile and returns transpiled circuits.
+
+        Args:
+            backend: Target backend.
+            options: User provided runtime options.
+
+        Returns:
+            Transpiled circuit to execute.
+        """
+        # Run pre transpile. This is implemented by each experiment subclass.
+        self._pre_transpile_hook(backend)
+
+        # Get transpile options
+        transpile_options = copy.copy(self.transpile_options)
+        transpile_options.update_options(
+            initial_layout=list(self._physical_qubits),
+            **options,
+        )
+        transpile_options = transpile_options.__dict__
+
+        circuits = transpile(circuits=self.circuits(backend), backend=backend, **transpile_options)
+
+        # Run post transpile. This is implemented by each experiment subclass.
+        circuits = self._post_transpile_hook(circuits, backend)
+
+        return circuits
+
+    def _post_analysis_hook(self, experiment_data: ExperimentData):
+        """An extra subroutine executed after analysis.
+
+        Args:
+            experiment_data: A future object of the experiment result.
+
+        Note:
+            The experiment_data may contain a future object as an experiment result
+            and the previous analysis routine has not completed yet.
+            If the hook should be executed immediately, call :meth:`block_for_results` method
+            before starting the data processing code.
+        """
+        pass
+
+    def run_analysis(
+        self, experiment_data: ExperimentData, job: BaseJob = None, **options
+    ) -> ExperimentData:
         """Run analysis and update ExperimentData with analysis result.
 
         Args:
-            experiment_data (ExperimentData): the experiment data to analyze.
-            options: additional analysis options. Any values set here will
-                     override the value from :meth:`analysis_options`
-                     for the current run.
+            experiment_data: The experiment data to analyze.
+            job: The future object of experiment result which is currently running on the backend.
+            options: Additional analysis options. Any values set here will
+                override the value from :meth:`analysis_options` for the current run.
 
         Returns:
             An experiment data object containing the analysis results and figures.
 
         Raises:
-            QiskitError: if experiment_data container is not valid for analysis.
+            QiskitError: Method is called with an empty experiment result.
         """
+        run_analysis = self.analysis() if self.__analysis_class__ else None
+
         # Get analysis options
         analysis_options = copy.copy(self.analysis_options)
         analysis_options.update_options(**options)
         analysis_options = analysis_options.__dict__
 
-        # Run analysis
-        analysis = self.analysis()
-        analysis.run(experiment_data, **analysis_options)
+        if not job and run_analysis is not None:
+            # Run analysis immediately
+            if not experiment_data.data():
+                raise QiskitError(
+                    "Experiment data seems to be empty and no running job is provided. "
+                    "At least one data entry is required to run analysis."
+                )
+            experiment_data = run_analysis.run(experiment_data, **analysis_options)
+        else:
+            # Run analysis when job is completed
+            experiment_data.add_data(
+                data=job,
+                post_processing_callback=run_analysis.run,
+                **analysis_options,
+            )
+
+        # Run post analysis. This is implemented by each experiment subclass.
+        self._post_analysis_hook(experiment_data)
+
         return experiment_data
 
     @property
@@ -335,10 +397,6 @@ def set_analysis_options(self, **fields):
         """
         self._analysis_options.update_options(**fields)
 
-    def _postprocess_transpiled_circuits(self, circuits, backend, **run_options):
-        """Additional post-processing of transpiled circuits before running on backend"""
-        pass
-
     def _metadata(self) -> Dict[str, any]:
         """Return experiment metadata for ExperimentData.
 

qiskit_experiments/framework/composite/batch_experiment.py

@@ -13,62 +13,80 @@
 Batch Experiment class.
 """
 
-from collections import OrderedDict
+from typing import List
 
 from qiskit import QuantumCircuit
 
+
 from .composite_experiment import CompositeExperiment
 
 
 class BatchExperiment(CompositeExperiment):
-    """Batch experiment class"""
+    """Batch experiment class.
+
+    This experiment takes multiple experiment instances and generates
+    a list of flattened circuit to execute.
+    The experimental circuits are executed ony be one on the target backend as a single job.
+
+    If an experiment analysis needs results of different types of experiments,
+    ``BatchExperiment`` may be convenient to describe the flow of the entire experiment.
+
+    The experimental result of ``i``-th experiment can be accessed by
+
+    .. code-block:: python3
+
+        batch_exp = BatchExperiment([exp1, exp2, exp3])
+        batch_result = batch_exp.run(backend)
+
+        exp1_res = batch_result.component_experiment_data(0)  # data of exp1
+        exp2_res = batch_result.component_experiment_data(1)  # data of exp2
+        exp3_res = batch_result.component_experiment_data(2)  # data of exp3
+
+    One can also create a custom analysis class that estimates some parameters by
+    combining above analysis results. Here the ``exp*_res`` is a single
+    :py:class:`~qiskit_experiments.framework.experiment_data.ExperimentData` class of
+    a standard experiment, and the associated analysis will be performed once the batch job
+    is completed. Thus analyzed parameter value of each experiment can be obtained as usual.
+
+    .. code-block:: python3
+
+        param_x = exp1_res.analysis_results("target_parameter_x")
+        param_y = exp2_res.analysis_results("target_parameter_y")
+        param_z = exp3_res.analysis_results("target_parameter_z")
+
+        param_xyz = param_x + param_y + param_z  # do some computation
+
+    The final parameter ``param_xyz`` can be returned as an outcome of this batch experiment.
+    """
 
     def __init__(self, experiments):
         """Initialize a batch experiment.
 
         Args:
             experiments (List[BaseExperiment]): a list of experiments.
         """
-
-        # Generate qubit map
-        self._qubit_map = OrderedDict()
-        logical_qubit = 0
-        for expr in experiments:
-            for physical_qubit in expr.physical_qubits:
-                if physical_qubit not in self._qubit_map:
-                    self._qubit_map[physical_qubit] = logical_qubit
-                    logical_qubit += 1
-        qubits = tuple(self._qubit_map.keys())
+        qubits = sorted(set(sum([list(expr.physical_qubits) for expr in experiments], [])))
         super().__init__(experiments, qubits)
 
-    def circuits(self, backend=None):
+    def _flatten_circuits(
+        self,
+        circuits: List[List[QuantumCircuit]],
+        num_qubits: int,
+    ) -> List[QuantumCircuit]:
+        """Flatten circuits.
 
+        Note:
+            This experiment concatenates sub experiment circuits.
+        """
         batch_circuits = []
 
-        # Generate data for combination
-        for index, expr in enumerate(self._experiments):
-            if self.physical_qubits == expr.physical_qubits:
-                qubit_mapping = None
-            else:
-                qubit_mapping = [self._qubit_map[qubit] for qubit in expr.physical_qubits]
-            for circuit in expr.circuits(backend):
-                # Update metadata
-                circuit.metadata = {
+        for expr_idx, sub_circs in enumerate(circuits):
+            for sub_circ in sub_circs:
+                sub_circ.metadata = {
                     "experiment_type": self._type,
-                    "composite_metadata": [circuit.metadata],
-                    "composite_index": [index],
+                    "composite_index": [expr_idx],
+                    "composite_metadata": [sub_circ.metadata],
                 }
-                # Remap qubits if required
-                if qubit_mapping:
-                    circuit = self._remap_qubits(circuit, qubit_mapping)
-                batch_circuits.append(circuit)
-        return batch_circuits
+                batch_circuits.append(sub_circ)
 
-    def _remap_qubits(self, circuit, qubit_mapping):
-        """Remap qubits if physical qubit layout is different to batch layout"""
-        num_qubits = self.num_qubits
-        num_clbits = circuit.num_clbits
-        new_circuit = QuantumCircuit(num_qubits, num_clbits, name="batch_" + circuit.name)
-        new_circuit.metadata = circuit.metadata
-        new_circuit.append(circuit, qubit_mapping, list(range(num_clbits)))
-        return new_circuit
+        return batch_circuits