Experiment run hook methods

qiskit_experiments/framework/base_analysis.py

@@ -101,6 +101,10 @@ def run(
         if figures:
             experiment_data.add_figures(figures)
 
+        # Run post analysis
+        if experiment_data.experiment is not None:
+            experiment_data.experiment._post_analysis_action(experiment_data)
+
         return experiment_data
 
     def _format_analysis_result(self, data, experiment_id, experiment_components=None):

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,24 @@ 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)
+            experiment_data.add_data(job, post_processing_callback=self.run_analysis)
+        else:
+            experiment_data.add_data(job)
 
-        # Return the ExperimentData future
         return experiment_data
 
     def _initialize_experiment_data(
@@ -167,11 +143,106 @@ def _initialize_experiment_data(
 
         return experiment_data._copy_metadata()
 
-    def run_analysis(self, experiment_data, **options) -> ExperimentData:
+    def _pre_transpile_action(self, backend: Backend):
+        """An extra subroutine executed before transpilation.
+
+        Note:
+            This method may be implemented by a subclass that requires to update the
+            transpiler configuration based on the given backend instance,
+            otherwise the transpiler configuration should be updated with the
+            :py:meth:`_default_transpile_options` method.
+
+            For example, some specific transpiler options might change depending on the real
+            hardware execution or circuit simulator execution.
+            By default, this method does nothing.
+
+        Args:
+            backend: Target backend.
+        """
+        pass
+
+    # pylint: disable = unused-argument
+    def _post_transpile_action(
+        self, circuits: List[QuantumCircuit], backend: Backend
+    ) -> List[QuantumCircuit]:
+        """An extra subroutine executed after transpilation.
+
+        Note:
+            This method may be implemented by a subclass that requires to update the
+            circuit or its metadata after transpilation.
+            Without this method, the transpiled circuit will be immediately executed on the backend.
+            This method enables the experiment to modify the circuit with pulse gates,
+            or some extra metadata regarding the transpiled sequence of instructions.
+
+            By default, this method just passes transpiled circuits to the execution chain.
+
+        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 return transpiled circuits.
+
+        Args:
+            backend: Target backend.
+            options: User provided runtime options.
+
+        Returns:
+            Transpiled circuit to execute.
+        """
+        # Run pre transpile if implemented by subclasses.
+        self._pre_transpile_action(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_action(circuits, backend)
+
+        return circuits
+
+    def _post_analysis_action(self, experiment_data: ExperimentData):
+        """An extra subroutine executed after analysis.
+
+        Note:
+            This method may be implemented by a subclass that requires to perform
+            extra data processing based on the analyzed experimental result.
+
+            Note that the analysis routine will not complete until the backend job
+            is executed, and this method will be called after the analysis routine
+            is completed though a handler of the experiment result will be immediately
+            returned to users (a future object). This method is automatically triggered
+            when the analysis is finished, and will be processed in background.
+
+            If this method updates some other (mutable) objects, you may need manage
+            synchronization of update of the object data. Otherwise you may want to
+            call :meth:`block_for_results` method of the ``experiment_data`` here
+            to freeze processing chain until the job result is returned.
+
+            By default, this method does nothing.
+
+        Args:
+            experiment_data: A future object of the experimental result.
+        """
+        pass
+
+    def run_analysis(self, experiment_data: ExperimentData, **options) -> ExperimentData:
         """Run analysis and update ExperimentData with analysis result.
 
         Args:
-            experiment_data (ExperimentData): the experiment data to analyze.
+            experiment_data: 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.
@@ -180,7 +251,7 @@ def run_analysis(self, experiment_data, **options) -> ExperimentData:
             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.
         """
         # Get analysis options
         analysis_options = copy.copy(self.analysis_options)
@@ -335,10 +406,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/common_operations.py

@@ -0,0 +1,47 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""A collection of common operation callback in execution chain."""
+
+
+from qiskit.providers import Backend
+from qiskit.test.mock import FakeBackend
+
+from .base_experiment import BaseExperiment
+
+
+def apply_delay_validation(experiment: BaseExperiment, backend: Backend):
+    """Enable delay duration validation to conform to backend alignment constraints.
+
+    Args:
+        experiment: Experiment instance to run.
+        backend: Target backend.
+    """
+    is_simulator = getattr(backend.configuration(), "simulator", False)
+
+    if not is_simulator and not isinstance(backend, FakeBackend):
+        timing_constraints = getattr(
+            experiment.transpile_options.__dict__, "timing_constraints", {}
+        )
+
+        # alignment=16 is IBM standard. Will be soon provided by IBM providers.
+        # Then, this configuration can be removed.
+        timing_constraints["acquire_alignment"] = getattr(
+            timing_constraints, "acquire_alignment", 16
+        )
+
+        scheduling_method = getattr(
+            experiment.transpile_options.__dict__, "scheduling_method", "alap"
+        )
+        experiment.set_transpile_options(
+            timing_constraints=timing_constraints, scheduling_method=scheduling_method
+        )

qiskit_experiments/framework/composite/batch_experiment.py

@@ -13,62 +13,87 @@
 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 circuits for
+    each nested experiment. This nested circuit list is flattened to a single long list of
+    all circuits to execute, and the 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
+        qubits = list()
         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())
+            for qubit in expr.physical_qubits:
+                if qubit not in qubits:
+                    qubits.append(qubit)
+
         super().__init__(experiments, qubits)
 
-    def circuits(self, backend=None):
+    # pylint: disable=unused-argument
+    def _flatten_circuits(
+        self,
+        circuits: List[List[QuantumCircuit]],
+        num_qubits: int,
+    ) -> List[QuantumCircuit]:
+        """Flatten circuits.
 
+        Note:
+            This experiment just flattens a list of list of circuit to a single long list of
+            circuits. The structure of experiment is kept in a metadata.
+        """
         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