# 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.
"""
Base Experiment class.
"""
from abc import ABC, abstractmethod
import copy
from collections import OrderedDict
from typing import Sequence, Optional, Tuple, List, Dict, Union
from qiskit import transpile, QuantumCircuit
from qiskit.providers import Job, Backend
from qiskit.exceptions import QiskitError
from qiskit.qobj.utils import MeasLevel
from qiskit.providers.options import Options
from qiskit.primitives.base import BaseSamplerV2
from qiskit_ibm_runtime import SamplerV2 as Sampler
from qiskit_experiments.framework import BackendData
from qiskit_experiments.framework.store_init_args import StoreInitArgs
from qiskit_experiments.framework.base_analysis import BaseAnalysis
from qiskit_experiments.framework.experiment_data import ExperimentData
from qiskit_experiments.framework.configs import ExperimentConfig
from qiskit_experiments.database_service import Qubit
[docs]
class BaseExperiment(ABC, StoreInitArgs):
"""Abstract base class for experiments."""
def __init__(
self,
physical_qubits: Sequence[int],
analysis: Optional[BaseAnalysis] = None,
backend: Optional[Backend] = None,
experiment_type: Optional[str] = None,
backend_run: Options[bool] = False,
):
"""Initialize the experiment object.
Args:
physical_qubits: list of physical qubits for the experiment.
analysis: Optional, the analysis to use for the experiment.
backend: Optional, the backend to run the experiment on.
experiment_type: Optional, the experiment type string.
backend_run: Optional, use backend run vs the sampler (temporary)
Raises:
QiskitError: If qubits contains duplicates.
"""
# Experiment identification metadata
self.experiment_type = experiment_type
# Circuit parameters
self._num_qubits = len(physical_qubits)
self._physical_qubits = tuple(physical_qubits)
if self._num_qubits != len(set(self._physical_qubits)):
raise QiskitError("Duplicate qubits in physical qubits list.")
# Experiment options
self._experiment_options = self._default_experiment_options()
self._transpile_options = self._default_transpile_options()
self._run_options = self._default_run_options()
# Store keys of non-default options
self._set_experiment_options = set()
self._set_transpile_options = set()
self._set_run_options = set()
self._set_analysis_options = set()
# Set analysis
self._analysis = None
if analysis:
self.analysis = analysis
# Set backend
# This should be called last in case `_set_backend` access any of the
# attributes created during initialization
self._backend = None
self._backend_data = None
self._backend_run = backend_run
if isinstance(backend, Backend):
self._set_backend(backend)
@property
def experiment_type(self) -> str:
"""Return experiment type."""
return self._type
@experiment_type.setter
def experiment_type(self, exp_type: str) -> None:
"""Set the type for the experiment."""
if exp_type is None:
self._type = type(self).__name__
else:
self._type = exp_type
@property
def physical_qubits(self) -> Tuple[int, ...]:
"""Return the device qubits for the experiment."""
return self._physical_qubits
@property
def num_qubits(self) -> int:
"""Return the number of qubits for the experiment."""
return self._num_qubits
@property
def analysis(self) -> Union[BaseAnalysis, None]:
"""Return the analysis instance for the experiment"""
return self._analysis
@analysis.setter
def analysis(self, analysis: Union[BaseAnalysis, None]) -> None:
"""Set the analysis instance for the experiment"""
if analysis is not None and not isinstance(analysis, BaseAnalysis):
raise TypeError("Input is not a None or a BaseAnalysis subclass.")
self._analysis = analysis
@property
def backend(self) -> Union[Backend, None]:
"""Return the backend for the experiment"""
return self._backend
@backend.setter
def backend(self, backend: Union[Backend, None]) -> None:
"""Set the backend for the experiment"""
if not isinstance(backend, Backend):
raise TypeError("Input is not a backend.")
self._set_backend(backend)
def _set_backend(self, backend: Backend):
"""Set the backend for the experiment.
Subclasses can override this method to extract additional
properties from the supplied backend if required.
"""
self._backend = backend
self._backend_data = BackendData(backend)
[docs]
def copy(self) -> "BaseExperiment":
"""Return a copy of the experiment"""
# We want to avoid a deep copy be default for performance so we
# need to also copy the Options structures so that if they are
# updated on the copy they don't effect the original.
ret = copy.copy(self)
if self.analysis:
ret.analysis = self.analysis.copy()
ret._experiment_options = copy.copy(self._experiment_options)
ret._run_options = copy.copy(self._run_options)
ret._transpile_options = copy.copy(self._transpile_options)
ret._set_experiment_options = copy.copy(self._set_experiment_options)
ret._set_transpile_options = copy.copy(self._set_transpile_options)
ret._set_run_options = copy.copy(self._set_run_options)
return ret
[docs]
def config(self) -> ExperimentConfig:
"""Return the config dataclass for this experiment"""
args = tuple(getattr(self, "__init_args__", OrderedDict()).values())
kwargs = dict(getattr(self, "__init_kwargs__", OrderedDict()))
# Only store non-default valued options
experiment_options = dict(
(key, getattr(self._experiment_options, key)) for key in self._set_experiment_options
)
transpile_options = dict(
(key, getattr(self._transpile_options, key)) for key in self._set_transpile_options
)
run_options = dict((key, getattr(self._run_options, key)) for key in self._set_run_options)
return ExperimentConfig(
cls=type(self),
args=args,
kwargs=kwargs,
experiment_options=experiment_options,
transpile_options=transpile_options,
run_options=run_options,
)
[docs]
@classmethod
def from_config(cls, config: Union[ExperimentConfig, Dict]) -> "BaseExperiment":
"""Initialize an experiment from experiment config"""
if isinstance(config, dict):
config = ExperimentConfig(**dict)
ret = cls(*config.args, **config.kwargs)
if config.experiment_options:
ret.set_experiment_options(**config.experiment_options)
if config.transpile_options:
ret.set_transpile_options(**config.transpile_options)
if config.run_options:
ret.set_run_options(**config.run_options)
return ret
[docs]
def run(
self,
backend: Optional[Backend] = None,
sampler: Optional[BaseSamplerV2] = None,
analysis: Optional[Union[BaseAnalysis, None]] = "default",
timeout: Optional[float] = None,
backend_run: Optional[bool] = None,
**run_options,
) -> ExperimentData:
"""Run an experiment and perform analysis.
Args:
backend: Optional, the backend to run on. Will override existing backend settings.
sampler: Optional, the sampler to run the experiment on.
If None then a sampler will be invoked from previously
set backend
analysis: Optional, a custom analysis instance to use for performing
analysis. If None analysis will not be run. If ``"default"``
the experiments :meth:`analysis` instance will be used if
it contains one.
timeout: Time to wait for experiment jobs to finish running before
cancelling.
backend_run: Use backend run (temp option for testing)
run_options: backend runtime options used for circuit execution.
Returns:
The experiment data object.
Raises:
QiskitError: If experiment is run with an incompatible existing
ExperimentData container.
"""
if (
(backend is not None)
or (sampler is not None)
or analysis != "default"
or run_options
or (backend_run is not None)
):
# Make a copy to update analysis or backend if one is provided at runtime
experiment = self.copy()
if backend_run is not None:
experiment._backend_run = backend_run
# we specified a backend OR a sampler
if (backend is not None) or (sampler is not None):
if sampler is None:
# backend only specified
experiment._set_backend(backend)
elif backend is None:
# sampler only specifid
experiment._set_backend(sampler._backend)
else:
# we specified both a sampler and a backend
if self._backend_run:
experiment._set_backend(backend)
else:
experiment._set_backend(sampler._backend)
if isinstance(analysis, BaseAnalysis):
experiment.analysis = analysis
if run_options:
experiment.set_run_options(**run_options)
else:
experiment = self
if experiment.backend is None:
raise QiskitError("Cannot run experiment, no backend has been set.")
# Finalize experiment before executions
experiment._finalize()
# Generate and transpile circuits
transpiled_circuits = experiment._transpiled_circuits()
# Initialize result container
experiment_data = experiment._initialize_experiment_data()
# Run options
run_opts = experiment.run_options.__dict__
# Run jobs
jobs = experiment._run_jobs(transpiled_circuits, sampler=sampler, **run_opts)
experiment_data.add_jobs(jobs, timeout=timeout)
# Optionally run analysis
if analysis and experiment.analysis:
return experiment.analysis.run(experiment_data)
else:
return experiment_data
def _initialize_experiment_data(self) -> ExperimentData:
"""Initialize the return data container for the experiment run"""
return ExperimentData(experiment=self)
def _finalize(self):
"""Finalize experiment object before running jobs.
Subclasses can override this method to set any final option
values derived from other options or attributes of the
experiment before `_run` is called.
"""
pass
def _max_circuits(self, backend: Backend = None):
"""
Calculate the maximum number of circuits per job for the experiment.
"""
# set backend
if backend is None:
if self.backend is None:
raise QiskitError("A backend must be provided.")
backend = self.backend
# Get max circuits for job splitting
max_circuits_option = getattr(self.experiment_options, "max_circuits", None)
max_circuits_backend = BackendData(backend).max_circuits
if max_circuits_option and max_circuits_backend:
return min(max_circuits_option, max_circuits_backend)
elif max_circuits_option:
return max_circuits_option
else:
return max_circuits_backend
[docs]
def job_info(self, backend: Backend = None):
"""
Get information about job distribution for the experiment on a specific
backend.
Args:
backend: Optional, the backend for which to get job distribution
information. If not specified, the experiment must already have a
set backend.
Returns:
dict: A dictionary containing information about job distribution.
- "Total number of circuits in the experiment": Total number of
circuits in the experiment.
- "Maximum number of circuits per job": Maximum number of
circuits in one job based on backend and experiment settings.
- "Total number of jobs": Number of jobs needed to run this
experiment on the currently set backend.
Raises:
QiskitError: if backend is not specified.
"""
max_circuits = self._max_circuits(backend)
total_circuits = len(self.circuits())
if max_circuits is None:
num_jobs = 1
else:
num_jobs = (total_circuits + max_circuits - 1) // max_circuits
return {
"Total number of circuits in the experiment": total_circuits,
"Maximum number of circuits per job": max_circuits,
"Total number of jobs": num_jobs,
}
def _run_jobs(
self, circuits: List[QuantumCircuit], sampler: BaseSamplerV2 = None, **run_options
) -> List[Job]:
"""Run circuits on backend as 1 or more jobs."""
max_circuits = self._max_circuits(self.backend)
# Run experiment jobs
if max_circuits and (len(circuits) > max_circuits):
# Split jobs for backends that have a maximum job size
job_circuits = [
circuits[i : i + max_circuits] for i in range(0, len(circuits), max_circuits)
]
else:
# Run as single job
job_circuits = [circuits]
# Run jobs
if not self._backend_run:
if sampler is None:
# instantiate a sampler from the backend
sampler = Sampler(self.backend)
# have to hand set some of these options
# see https://docs.quantum.ibm.com/api/qiskit-ibm-runtime
# /qiskit_ibm_runtime.options.SamplerExecutionOptionsV2
if "init_qubits" in run_options:
sampler.options.execution.init_qubits = run_options["init_qubits"]
if "rep_delay" in run_options:
sampler.options.execution.rep_delay = run_options["rep_delay"]
if "meas_level" in run_options:
if run_options["meas_level"] == 2:
sampler.options.execution.meas_type = "classified"
elif run_options["meas_level"] == 1:
if "meas_return" in run_options:
if run_options["meas_return"] == "avg":
sampler.options.execution.meas_type = "avg_kerneled"
else:
sampler.options.execution.meas_type = "kerneled"
else:
# assume this is what is wanted if no meas return specified
sampler.options.execution.meas_type = "kerneled"
else:
raise QiskitError("Only meas level 1 + 2 supported by sampler")
if "noise_model" in run_options:
sampler.options.simulator.noise_model = run_options["noise_model"]
if "seed_simulator" in run_options:
sampler.options.simulator.seed_simulator = run_options["seed_simulator"]
if run_options.get("shots") is not None:
sampler.options.default_shots = run_options.get("shots")
jobs = [sampler.run(circs) for circs in job_circuits]
else:
jobs = [self.backend.run(circs, **run_options) for circs in job_circuits]
return jobs
[docs]
@abstractmethod
def circuits(self) -> List[QuantumCircuit]:
"""Return a list of experiment circuits.
Returns:
A list of :class:`~qiskit.circuit.QuantumCircuit`.
.. note::
These circuits should be on qubits ``[0, .., N-1]`` for an
*N*-qubit experiment. The circuits mapped to physical qubits
are obtained via the internal :meth:`_transpiled_circuits` method.
"""
# NOTE: Subclasses should override this method using the `options`
# values for any explicit experiment options that affect circuit
# generation
def _transpiled_circuits(self) -> List[QuantumCircuit]:
"""Return a list of experiment circuits, transpiled.
This function can be overridden to define custom transpilation.
"""
transpile_opts = copy.copy(self.transpile_options.__dict__)
transpile_opts["initial_layout"] = list(self.physical_qubits)
transpiled = transpile(self.circuits(), self.backend, **transpile_opts)
return transpiled
@classmethod
def _default_experiment_options(cls) -> Options:
"""Default experiment options.
Experiment Options:
max_circuits (Optional[int]): The maximum number of circuits per job when
running an experiment on a backend.
"""
# Experiment subclasses should override this method to return
# an `Options` object containing all the supported options for
# that experiment and their default values. Only options listed
# here can be modified later by the different methods for
# setting options.
return Options(max_circuits=None)
@property
def experiment_options(self) -> Options:
"""Return the options for the experiment."""
return self._experiment_options
[docs]
def set_experiment_options(self, **fields):
"""Set the experiment options.
Args:
fields: The fields to update the options
Raises:
AttributeError: If the field passed in is not a supported options
"""
for field in fields:
if not hasattr(self._experiment_options, field):
raise AttributeError(
f"Options field {field} is not valid for {type(self).__name__}"
)
self._experiment_options.update_options(**fields)
self._set_experiment_options = self._set_experiment_options.union(fields)
@classmethod
def _default_transpile_options(cls) -> Options:
"""Default transpiler options for transpilation of circuits"""
# Experiment subclasses can override this method if they need
# to set specific default transpiler options to transpile the
# experiment circuits.
return Options(optimization_level=0)
@property
def transpile_options(self) -> Options:
"""Return the transpiler options for the :meth:`run` method."""
return self._transpile_options
[docs]
def set_transpile_options(self, **fields):
"""Set the transpiler options for :meth:`run` method.
Args:
fields: The fields to update the options
Raises:
QiskitError: If `initial_layout` is one of the fields.
.. seealso:: The :ref:`guide_setting_options` guide for code example.
"""
if "initial_layout" in fields:
raise QiskitError(
"Initial layout cannot be specified as a transpile option"
" as it is determined by the experiment physical qubits."
)
self._transpile_options.update_options(**fields)
self._set_transpile_options = self._set_transpile_options.union(fields)
@classmethod
def _default_run_options(cls) -> Options:
"""Default options values for the experiment :meth:`run` method."""
return Options(meas_level=MeasLevel.CLASSIFIED)
@property
def run_options(self) -> Options:
"""Return options values for the experiment :meth:`run` method."""
return self._run_options
[docs]
def set_run_options(self, **fields):
"""Set options values for the experiment :meth:`run` method.
Args:
fields: The fields to update the options
.. seealso:: The :ref:`guide_setting_options` guide for code example.
"""
self._run_options.update_options(**fields)
self._set_run_options = self._set_run_options.union(fields)
def _metadata(self) -> Dict[str, any]:
"""Return experiment metadata for ExperimentData.
By default, this assumes the experiment is running on qubits only. Subclasses can override
this method to add custom experiment metadata to the returned experiment result data.
"""
metadata = {
"physical_qubits": list(self.physical_qubits),
"device_components": list(map(Qubit, self.physical_qubits)),
}
return metadata
def __json_encode__(self):
"""Convert to format that can be JSON serialized"""
return self.config()
@classmethod
def __json_decode__(cls, value):
"""Load from JSON compatible format"""
return cls.from_config(value)