diff --git a/tests/test_vqls.py b/tests/test_vqls.py index 7504159..a386c2a 100644 --- a/tests/test_vqls.py +++ b/tests/test_vqls.py @@ -19,9 +19,13 @@ from qiskit import QuantumCircuit from qiskit.circuit.library import RealAmplitudes +from qiskit.primitives import Estimator, Sampler from qiskit_algorithms.optimizers import ADAM -from qiskit.primitives import Estimator, Sampler + +from qiskit_aer.primitives import EstimatorV2 as aer_EstimatorV2 +from qiskit_aer.primitives import SamplerV2 as aer_SamplerV2 + from vqls_prototype import VQLS # 8-11-2023 @@ -43,11 +47,13 @@ def setUp(self): self.estimators = ( Estimator(), + aer_EstimatorV2(), # AerEstimator(), ) self.samplers = ( Sampler(), + aer_SamplerV2(), # AerSampler(), ) @@ -66,12 +72,8 @@ def test_numpy_input(self): rhs = np.array([0.1] * 4) ansatz = RealAmplitudes(num_qubits=2, reps=3, entanglement="full") - for iprim, (estimator, sampler) in enumerate( - zip(self.estimators, self.samplers) - ): - for iopt, opt in enumerate(self.options): - if iprim == 1 and iopt == 2: - continue + for _, (estimator, sampler) in enumerate(zip(self.estimators, self.samplers)): + for _, opt in enumerate(self.options): vqls = VQLS( estimator, ansatz, @@ -101,12 +103,8 @@ def test_circuit_input_statevector(self): qc2.x(1) qc2.cx(0, 1) - for iprim, (estimator, sampler) in enumerate( - zip(self.estimators, self.samplers) - ): - for iopt, opt in enumerate(self.options): - if iprim == 1 and iopt == 2: - continue + for _, (estimator, sampler) in enumerate(zip(self.estimators, self.samplers)): + for _, opt in enumerate(self.options): vqls = VQLS( estimator, ansatz, diff --git a/vqls_prototype/hadamard_test/direct_hadamard_test.py b/vqls_prototype/hadamard_test/direct_hadamard_test.py index 776009b..b67c00e 100644 --- a/vqls_prototype/hadamard_test/direct_hadamard_test.py +++ b/vqls_prototype/hadamard_test/direct_hadamard_test.py @@ -4,6 +4,10 @@ import numpy as np import numpy.typing as npt +from qiskit.primitives.sampler import SamplerResult +from qiskit.primitives.containers import PrimitiveResult +from vqls_prototype.primitives_run_builder import SamplerRunBuilder + class BatchDirectHadammardTest: r"""Class that execute batches of Hadammard Test""" @@ -31,14 +35,18 @@ def get_values(self, sampler, parameter_sets: List, zne_strategy=None) -> List: """ ncircuits = len(self.circuits) + all_parameter_sets = [parameter_sets] * ncircuits + + sampler_run_builder = SamplerRunBuilder( + sampler, + self.circuits, + all_parameter_sets, + options={"shots": self.shots}, + ) try: if zne_strategy is None: - job = sampler.run( - self.circuits, - [parameter_sets] * ncircuits, - shots=self.shots, - ) + job = sampler_run_builder.build_run() else: job = sampler.run( self.circuits, @@ -133,8 +141,24 @@ def post_processing(self, sampler_result) -> npt.NDArray[np.cdouble]: Returns: List: value of the overlap hadammard test """ + if isinstance(sampler_result, SamplerResult): + quasi_dist = sampler_result.quasi_dists + + elif isinstance(sampler_result, PrimitiveResult): + quasi_dist = [ + { + key: value / result.data.meas.num_shots + for key, value in result.data.meas.get_int_counts().items() + } + for result in sampler_result + ] + + else: + raise NotImplementedError( + f"Cannot post processing for {type(sampler_result)} type class." + f"Please, refer to {self.__class__.__name__}.post_processing()." + ) - quasi_dist = sampler_result.quasi_dists val = [] for qdist in quasi_dist: # add missing keys @@ -158,14 +182,16 @@ def get_value( Returns: List: value of the test """ + sampler_run_builder = SamplerRunBuilder( + sampler, + self.circuits, + parameter_sets, + options={"shots": self.shots}, + ) try: if zne_strategy is None: - job = sampler.run( - self.circuits, - parameter_sets, - shots=self.shots, - ) + job = sampler_run_builder.build_run() else: job = sampler.run( self.circuits, diff --git a/vqls_prototype/hadamard_test/hadamard_overlap_test.py b/vqls_prototype/hadamard_test/hadamard_overlap_test.py index 526f4ff..266ef8e 100644 --- a/vqls_prototype/hadamard_test/hadamard_overlap_test.py +++ b/vqls_prototype/hadamard_test/hadamard_overlap_test.py @@ -4,6 +4,10 @@ import numpy as np import numpy.typing as npt +from qiskit.primitives.sampler import SamplerResult +from qiskit.primitives.containers import PrimitiveResult +from vqls_prototype.primitives_run_builder import SamplerRunBuilder + class BatchHadammardOverlapTest: r"""Class that execute batches of Hadammard Test""" @@ -31,14 +35,18 @@ def get_values(self, sampler, parameter_sets: List, zne_strategy=None) -> List: """ ncircuits = len(self.circuits) + all_parameter_sets = [parameter_sets] * ncircuits + + sampler_run_builder = SamplerRunBuilder( + sampler, + self.circuits, + all_parameter_sets, + options={"shots": self.shots}, + ) try: if zne_strategy is None: - job = sampler.run( - self.circuits, - [parameter_sets] * ncircuits, - shots=self.shots, - ) + job = sampler_run_builder.build_run() else: job = sampler.run( self.circuits, @@ -240,8 +248,24 @@ def post_processing(self, sampler_result) -> npt.NDArray[np.cdouble]: Returns: List: value of the overlap hadammard test """ + if isinstance(sampler_result, SamplerResult): + quasi_dist = sampler_result.quasi_dists + + elif isinstance(sampler_result, PrimitiveResult): + quasi_dist = [ + { + key: value / result.data.meas.num_shots + for key, value in result.data.meas.get_int_counts().items() + } + for result in sampler_result + ] + + else: + raise NotImplementedError( + f"Cannot post processing for {type(sampler_result)} type class." + f"Please, refer to {self.__class__.__name__}.post_processing()." + ) - quasi_dist = sampler_result.quasi_dists output = [] for qdist in quasi_dist: @@ -269,7 +293,15 @@ def get_value(self, sampler, parameter_sets: List) -> float: float: value of the overlap hadammard test """ ncircuits = len(self.circuits) - job = sampler.run(self.circuits, [parameter_sets] * ncircuits, shots=self.shots) + all_parameter_sets = [parameter_sets] * ncircuits + + sampler_run_builder = SamplerRunBuilder( + sampler, + self.circuits, + all_parameter_sets, + options={"shots": self.shots}, + ) + job = sampler_run_builder.build_run() results = self.post_processing(job.result()) results *= np.array([1.0, 1.0j]) diff --git a/vqls_prototype/hadamard_test/hadamard_test.py b/vqls_prototype/hadamard_test/hadamard_test.py index 72e629f..34040c7 100644 --- a/vqls_prototype/hadamard_test/hadamard_test.py +++ b/vqls_prototype/hadamard_test/hadamard_test.py @@ -8,6 +8,10 @@ import numpy as np import numpy.typing as npt +from qiskit.primitives.estimator import EstimatorResult +from qiskit.primitives.containers import PrimitiveResult +from vqls_prototype.primitives_run_builder import EstimatorRunBuilder + class BatchHadammardTest: r"""Class that execute batches of Hadammard Test""" @@ -36,15 +40,19 @@ def get_values(self, primitive, parameter_sets: List, zne_strategy=None) -> List """ ncircuits = len(self.circuits) + all_parameter_sets = [parameter_sets] * ncircuits + + estimator_run_builder = EstimatorRunBuilder( + primitive, + self.circuits, + self.observable, + all_parameter_sets, + options={"shots": self.shots}, + ) try: if zne_strategy is None: - job = primitive.run( - self.circuits, - self.observable, - [parameter_sets] * ncircuits, - shots=self.shots, - ) + job = estimator_run_builder.build_run() else: job = primitive.run( self.circuits, @@ -236,8 +244,19 @@ def post_processing(self, estimator_result) -> npt.NDArray[np.cdouble]: Returns: npt.NDArray[np.cdouble]: value of the test """ - return np.array([1.0 - 2.0 * val for val in estimator_result.values]).astype( - "complex128" + if isinstance(estimator_result, EstimatorResult): + return np.array( + [1.0 - 2.0 * val for val in estimator_result.values] + ).astype("complex128") + + if isinstance(estimator_result, PrimitiveResult): + return np.array( + [1.0 - 2.0 * val.data.evs for val in estimator_result] + ).astype("complex128") + + raise NotImplementedError( + f"Cannot post processing for {type(estimator_result)} type class." + f"Please, refer to {self.__class__.__name__}.post_processing()." ) def get_value(self, estimator, parameter_sets: List, zne_strategy=None) -> List: @@ -252,15 +271,20 @@ def get_value(self, estimator, parameter_sets: List, zne_strategy=None) -> List: """ ncircuits = len(self.circuits) + all_parameter_sets = [parameter_sets] * ncircuits + all_observables = [self.observable] * ncircuits + + estimator_run_builder = EstimatorRunBuilder( + estimator, + self.circuits, + all_observables, + all_parameter_sets, + options={"shots": self.shots}, + ) try: if zne_strategy is None: - job = estimator.run( - self.circuits, - [self.observable] * ncircuits, - [parameter_sets] * ncircuits, - shots=self.shots, - ) + job = estimator_run_builder.build_run() else: job = estimator.run( self.circuits, diff --git a/vqls_prototype/primitives_run_builder/__init__.py b/vqls_prototype/primitives_run_builder/__init__.py new file mode 100644 index 0000000..42bdaae --- /dev/null +++ b/vqls_prototype/primitives_run_builder/__init__.py @@ -0,0 +1,10 @@ +"""Primitives builder package.""" + +from .estimator_run_builder import EstimatorRunBuilder +from .sampler_run_builder import SamplerRunBuilder + + +__all__ = [ + "EstimatorRunBuilder", + "SamplerRunBuilder", +] diff --git a/vqls_prototype/primitives_run_builder/base_run_builder.py b/vqls_prototype/primitives_run_builder/base_run_builder.py new file mode 100644 index 0000000..b296ea7 --- /dev/null +++ b/vqls_prototype/primitives_run_builder/base_run_builder.py @@ -0,0 +1,78 @@ +"""This module defines a base class for primitive run builders.""" + +from typing import Union, List, Tuple, Dict, Any +from qiskit import QuantumCircuit +from qiskit.primitives import PrimitiveJob +from qiskit_ibm_runtime import RuntimeJobV2 + + +class BasePrimitiveRunBuilder: + """ + Base class for building and configuring primitive runs based on their provenance and options. + """ + + def __init__( + self, + primitive, + circuits: List[QuantumCircuit], + parameter_sets: List[List[float]], + options: Dict[str, Any], + ): + """ + Initializes BasePrimitiveRunBuilder for given primitive, circuits, parameters, and options. + + Args: + primitive (Union[SamplerValidType, EstimatorValidType]): The primitive to use for runs. + circuits (List[QuantumCircuit]): The quantum circuits to run. + parameter_sets (List[List[float]]): The parameters to vary in the circuits. + options (Dict[str, Any]): Configuration options such as number of shots. + """ + self.primitive = primitive + self.circuits = circuits + self.parameter_sets = parameter_sets + self.shots = options.pop("shots", None) + self.seed = options.pop("seed", None) + self.provenance = self.find_provenance() + + def find_provenance(self) -> Tuple[str, str]: + """Determines the provenance of the primitive based on its class and module.""" + return ( + self.primitive.__class__.__module__.split(".")[0], + self.primitive.__class__.__name__, + ) + + def build_run(self) -> Union[PrimitiveJob, RuntimeJobV2]: + """ + Configures and returns primitive runs based on its provenance. + + Raises: + NotImplementedError: If the primitive's provenance is not supported. + + Returns: + Union[PrimitiveJob, RuntimeJobV2]: A primitive job. + """ + primitive_job = self._select_run_builder() + return primitive_job() + + def _select_run_builder(self) -> Union[PrimitiveJob, RuntimeJobV2]: + """Selects the appropriate builder function based on the primitive's provenance.""" + raise NotImplementedError("This method should be implemented by subclasses.") + + def _build_native_qiskit_run(self) -> PrimitiveJob: + """Builds a run function for a standard qiskit primitive.""" + raise NotImplementedError("This method should be implemented by subclasses.") + + def _build_v2_run(self) -> Union[PrimitiveJob, RuntimeJobV2]: + """Builds a run function for qiskit-aer and qiskit-ibm-runtime V2 primitives.""" + raise NotImplementedError("This method should be implemented by subclasses.") + + def _build_v1_run(self): + """ + Attempts to build a run function for primitives V1, which will be soon deprecated. + + Raises: + NotImplementedError: Indicates that V1 will be soon deprecated. + """ + raise NotImplementedError( + "Primitives V1 will be soon deprecated. Please, use V2 implementation." + ) diff --git a/vqls_prototype/primitives_run_builder/estimator_run_builder.py b/vqls_prototype/primitives_run_builder/estimator_run_builder.py new file mode 100644 index 0000000..2875ea4 --- /dev/null +++ b/vqls_prototype/primitives_run_builder/estimator_run_builder.py @@ -0,0 +1,92 @@ +"""This module defines the estimator run builder class.""" + +from typing import Union, List, Dict, Any +from qiskit import QuantumCircuit +from qiskit.quantum_info import SparsePauliOp +from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager +from qiskit.primitives import Estimator, PrimitiveJob +from qiskit_aer.primitives import Estimator as aer_Estimator +from qiskit_aer.primitives import EstimatorV2 as aer_EstimatorV2 +from qiskit_ibm_runtime import Estimator as ibm_runtime_Estimator +from qiskit_ibm_runtime import EstimatorV2 as ibm_runtime_EstimatorV2 +from qiskit_ibm_runtime import RuntimeJobV2 +from .base_run_builder import BasePrimitiveRunBuilder + +EstimatorValidType = Union[ + Estimator, + aer_Estimator, + aer_EstimatorV2, + ibm_runtime_Estimator, + ibm_runtime_EstimatorV2, +] + + +class EstimatorRunBuilder(BasePrimitiveRunBuilder): # pylint: disable=abstract-method + """ + A class to build and configure estimator runs based on their provenance and options. + + Attributes: + estimator (EstimatorValidType): The quantum estimator instance. + circuits (List[QuantumCircuit]): List of quantum circuits. + observables (List[SparsePauliOp]): List of observables. + parameter_sets (List[List[float]]): List of parameter sets. + """ + + def __init__( # pylint: disable=too-many-arguments + self, + estimator: EstimatorValidType, + circuits: List[QuantumCircuit], + observables: List[SparsePauliOp], + parameter_sets: List[List[float]], + options: Dict[str, Any], + ): + """ + Initializes the EstimatorRunBuilder with the given estimator, circuits, observables, + parameter sets, and options. + + Args: + estimator (EstimatorValidType): The estimator to use for runs. + circuits (List[QuantumCircuit]): The quantum circuits to run. + observables (List[SparsePauliOp]): The observables to measure. + parameter_sets (List[List[float]]): The parameters to vary in the circuits. + options (Dict[str, Any]): Configuration options such as number of shots. + """ + super().__init__(estimator, circuits, parameter_sets, options) + self.observables = observables + + def _select_run_builder(self) -> Union[PrimitiveJob, RuntimeJobV2]: + builders = { + ("qiskit", "Estimator"): self._build_native_qiskit_run, + ("qiskit_aer", "EstimatorV2"): self._build_v2_run, + ("qiskit_aer", "Estimator"): self._build_v1_run, + ("qiskit_ibm_runtime", "EstimatorV2"): self._build_v2_run, + ("qiskit_ibm_runtime", "EstimatorV1"): self._build_v1_run, + } + try: + return builders[self.provenance] + except KeyError as err: + raise NotImplementedError( + f"{self.__class__.__name__} not compatible with {self.provenance}." + ) from err + + def _build_native_qiskit_run(self) -> PrimitiveJob: + """Builds a run function for a standard qiskit Estimator.""" + return self.primitive.run( + self.circuits, + self.observables, + self.parameter_sets, + shots=self.shots, + seed=self.seed, + ) + + def _build_v2_run(self) -> Union[PrimitiveJob, RuntimeJobV2]: + """Builds a run function for qiskit-aer and qiskit-ibm-runtime EstimatorV2.""" + backend = self.primitive._backend # pylint: disable=protected-access + optimization_level = 1 + pm = generate_preset_pass_manager(optimization_level, backend) + pubs = [] + for qc, obs, param in zip(self.circuits, self.observables, self.parameter_sets): + isa_circuit = pm.run(qc) + isa_obs = obs.apply_layout(isa_circuit.layout) + pubs.append((isa_circuit, isa_obs, param)) + return self.primitive.run(pubs) diff --git a/vqls_prototype/primitives_run_builder/sampler_run_builder.py b/vqls_prototype/primitives_run_builder/sampler_run_builder.py new file mode 100644 index 0000000..b64e82b --- /dev/null +++ b/vqls_prototype/primitives_run_builder/sampler_run_builder.py @@ -0,0 +1,76 @@ +"""This module defines the sampler run builder class.""" + +from typing import Union, List, Dict, Any +from qiskit import QuantumCircuit +from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager +from qiskit.primitives import Sampler, PrimitiveJob +from qiskit_aer.primitives import Sampler as aer_Sampler +from qiskit_aer.primitives import SamplerV2 as aer_SamplerV2 +from qiskit_ibm_runtime import Sampler as ibm_runtime_Sampler +from qiskit_ibm_runtime import SamplerV2 as ibm_runtime_SamplerV2 +from qiskit_ibm_runtime import RuntimeJobV2 + +from .base_run_builder import BasePrimitiveRunBuilder + +SamplerValidType = Union[ + Sampler, + aer_Sampler, + aer_SamplerV2, + ibm_runtime_Sampler, + ibm_runtime_SamplerV2, +] + + +class SamplerRunBuilder(BasePrimitiveRunBuilder): # pylint: disable=abstract-method + """ + A class to build and configure sampler runs based on their provenance and options. + + Attributes: + sampler (SamplerValidType): The quantum sampler instance. + circuits (List[QuantumCircuit]): List of quantum circuits. + parameter_sets (List[List[float]]): List of parameter sets. + """ + + def __init__( + self, + sampler: SamplerValidType, + circuits: List[QuantumCircuit], + parameter_sets: List[List[float]], + options: Dict[str, Any], + ): + super().__init__(sampler, circuits, parameter_sets, options) + + def _select_run_builder(self) -> Union[PrimitiveJob, RuntimeJobV2]: + builders = { + ("qiskit", "Sampler"): self._build_native_qiskit_run, + ("qiskit_aer", "SamplerV2"): self._build_v2_run, + ("qiskit_aer", "Sampler"): self._build_v1_run, + ("qiskit_ibm_runtime", "SamplerV2"): self._build_v2_run, + ("qiskit_ibm_runtime", "SamplerV1"): self._build_v1_run, + } + try: + return builders[self.provenance] + except KeyError as err: + raise NotImplementedError( + f"{self.__class__.__name__} not compatible with {self.provenance}." + ) from err + + def _build_native_qiskit_run(self) -> PrimitiveJob: + """Builds a run function for a standard qiskit Sampler.""" + return self.primitive.run( + self.circuits, + self.parameter_sets, + shots=self.shots, + seed=self.seed, + ) + + def _build_v2_run(self) -> Union[PrimitiveJob, RuntimeJobV2]: + """Builds a run function for qiskit-aer and qiskit-ibm-runtime SamplerV2.""" + backend = self.primitive._backend # pylint: disable=protected-access + optimization_level = 1 + pm = generate_preset_pass_manager(optimization_level, backend) + pubs = [] + for qc, param in zip(self.circuits, self.parameter_sets): + isa_circuit = pm.run(qc) + pubs.append((isa_circuit, param)) + return self.primitive.run(pubs)