Source code for qiskit_experiments.framework.base_analysis

# 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
# 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 analysis class.
from abc import ABC, abstractmethod
import copy
from collections import OrderedDict
from datetime import datetime
from typing import List, Tuple, Union, Dict
import warnings

from dateutil import tz

from qiskit_experiments.database_service.device_component import Qubit
from qiskit_experiments.framework import Options
from qiskit_experiments.framework.containers.artifact_data import ArtifactData
from qiskit_experiments.framework.store_init_args import StoreInitArgs
from qiskit_experiments.framework.experiment_data import ExperimentData
from qiskit_experiments.framework.containers import FigureData, FigureType
from qiskit_experiments.framework.configs import AnalysisConfig
from qiskit_experiments.framework.analysis_result_data import AnalysisResultData, as_table_element

[docs] class BaseAnalysis(ABC, StoreInitArgs): """Abstract base class for analyzing Experiment data. The data produced by experiments (i.e. subclasses of BaseExperiment) are analyzed with subclasses of BaseAnalysis. The analysis is typically run after the data has been gathered by the experiment. For example, an analysis may perform some data processing of the measured data and a fit to a function to extract a parameter. Analysis subclasses must implement the abstract method `_run_analysis`. This method should not have side-effects on the analysis class itself since it could potentially be called asynchronously in multiple threads. Any configurable option values should be specified in the `_default_options` class method. These values can be overriden by a user by calling the `set_options` method or for a single-run can be specified by passing kwarg options to the :meth:`run` method. """ def __init__(self): """Initialize the analysis object.""" # Analysis options self._options = self._default_options() # Store keys of non-default options self._set_options = set()
[docs] def config(self) -> AnalysisConfig: """Return the config dataclass for this analysis""" args = tuple(getattr(self, "__init_args__", OrderedDict()).values()) kwargs = dict(getattr(self, "__init_kwargs__", OrderedDict())) # Only store non-default valued options options = dict((key, getattr(self._options, key)) for key in self._set_options) return AnalysisConfig( cls=type(self), args=args, kwargs=kwargs, options=options, )
[docs] @classmethod def from_config(cls, config: Union[AnalysisConfig, Dict]) -> "BaseAnalysis": """Initialize an analysis class from analysis config""" if isinstance(config, dict): config = AnalysisConfig(**config) ret = cls(*config.args, **config.kwargs) if config.options: ret.set_options(**config.options) return ret
[docs] def copy(self) -> "BaseAnalysis": """Return a copy of the analysis""" # 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) ret._options = copy.copy(self._options) ret._set_options = copy.copy(self._set_options) return ret
@classmethod def _default_options(cls) -> Options: """Default analysis options common to all analyses. Analysis Options: figure_names (str or List[str]): Identifier of figures that appear in the experiment data to sort figures by name. """ options = Options() # figure names can be set for each analysis by calling # experiment_obj.analysis.set_options(figure_names=FIGURE_NAMES) options.figure_names = None return options @property def options(self) -> Options: """Return the analysis options for :meth:`run` method.""" return self._options
[docs] def set_options(self, **fields): """Set the analysis options for :meth:`run` method. Args: fields: The fields to update the options """ self._options.update_options(**fields) self._set_options = self._set_options.union(fields)
[docs] def run( self, experiment_data: ExperimentData, replace_results: bool = False, **options, ) -> ExperimentData: """Run analysis and update ExperimentData with analysis result. Args: experiment_data: the experiment data to analyze. replace_results: If True clear any existing analysis results, figures, and artifacts in the experiment data and replace with new results. See note for additional information. options: additional analysis options. See class documentation for supported options. Returns: An experiment data object containing analysis results, figures, and artifacts. Raises: QiskitError: If experiment_data container is not valid for analysis. .. note:: **Updating Results** If analysis is run with ``replace_results=True`` then any analysis results, figures, and artifacts in the experiment data will be cleared and replaced with the new analysis results. Saving this experiment data will replace any previously saved data in a database service using the same experiment ID. If analysis is run with ``replace_results=False`` and the experiment data being analyzed has already been saved to a database service, or already contains analysis results or figures, a copy with a unique experiment ID will be returned containing only the new analysis results and figures. This data can then be saved as its own experiment to a database service. """ # Make a new copy of experiment data if not updating results if not replace_results and _requires_copy(experiment_data): experiment_data = experiment_data.copy() # Set Analysis options if not options: analysis = self else: analysis = self.copy() analysis.set_options(**options) def run_analysis(expdata: ExperimentData): # Clearing previous analysis data experiment_data._clear_results() if not warnings.warn("ExperimentData object data is empty.\n") # Making new analysis results, figures = analysis._run_analysis(expdata) if results: for result in results: if isinstance(result, AnalysisResultData): # Populate missing data fields if not result.experiment_id: result.experiment_id = expdata.experiment_id if not result.experiment: result.experiment = expdata.experiment_type if not result.device_components: result.device_components = analysis._get_experiment_components(expdata) if not result.backend: result.backend = expdata.backend_name if not result.created_time: result.created_time = if not result.run_time: result.run_time = expdata.running_time # To canonical kwargs to add to the analysis table. table_format = as_table_element(result) # Remove result_id to make sure the id is unique in the scope of the container. # This will let the container generate a unique id. del table_format["result_id"] expdata.add_analysis_results(**table_format) elif isinstance(result, ArtifactData): if not result.experiment_id: result.experiment_id = expdata.experiment_id if not result.device_components: result.device_components = analysis._get_experiment_components(expdata) if not result.experiment: result.experiment = expdata.experiment_type expdata.add_artifacts(result) else: raise TypeError( f"Invalid object type {result.__class__.__name__} for analysis results. " "This data cannot be stored in the experiment data." ) if figures: figure_to_add = [] for figure in figures: if not isinstance(figure, FigureData): qubits_repr = "_".join( map(str, expdata.metadata.get("device_components", [])[:5]) ) short_id = expdata.experiment_id[:8] figure = FigureData( figure=figure, name=f"{expdata.experiment_type}_{qubits_repr}_{short_id}.svg", ) figure_to_add.append(figure) expdata.add_figures(figure_to_add, figure_names=analysis.options.figure_names) experiment_data.add_analysis_callback(run_analysis) return experiment_data
def _get_experiment_components(self, experiment_data: ExperimentData): """Subclasses may override this method to specify the experiment components.""" if "physical_qubits" in experiment_data.metadata: experiment_components = [ Qubit(qubit) for qubit in experiment_data.metadata["physical_qubits"] ] else: experiment_components = [] return experiment_components @abstractmethod def _run_analysis( self, experiment_data: ExperimentData, ) -> Tuple[List[Union[AnalysisResultData, ArtifactData]], List[FigureType]]: """Run analysis on circuit data. Args: experiment_data: the experiment data to analyze. Returns: A pair ``(analysis_results, figures)`` where ``analysis_results`` is a list of :class:`.AnalysisResultData` objects, and ``figures`` is a list of any figures for the experiment. Raises: AnalysisError: If the analysis fails. """ # NOTE: passing kwarg options to _run_analysis should be removed once pass def __json_encode__(self): return self.config() @classmethod def __json_decode__(cls, value): return cls.from_config(value)
def _requires_copy(experiment_data) -> bool: """Return True if a copy of the experiment data should be made.""" # If data is from DB or contains analysis results it should be copied if ( experiment_data._created_in_db or experiment_data._analysis_results or experiment_data._figures ): return True # Check child data: if hasattr(experiment_data, "_child_data"): for subdata in experiment_data._child_data.values(): if _requires_copy(subdata): return True # No Copy required return False