Rewrite Amplitude Estimators with Primitives

qiskit/algorithms/amplitude_estimators/ae.py

@@ -1,6 +1,6 @@
 # This code is part of Qiskit.
 #
-# (C) Copyright IBM 2018, 2020.
+# (C) Copyright IBM 2018, 2022.
 #
 # 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
@@ -12,18 +12,23 @@
 
 """The Quantum Phase Estimation-based Amplitude Estimation algorithm."""
 
-from typing import Optional, Union, List, Tuple, Dict
+from __future__ import annotations
+from typing import Union, List, Tuple, Dict
 from collections import OrderedDict
+import warnings
 import numpy as np
 from scipy.stats import chi2, norm
 from scipy.optimize import bisect
 
 from qiskit import QuantumCircuit, ClassicalRegister
-from qiskit.providers import Backend
+from qiskit.providers import Backend, JobStatus
+from qiskit.primitives import BaseSampler
 from qiskit.utils import QuantumInstance
+from qiskit.utils.deprecation import deprecate_function
 from .amplitude_estimator import AmplitudeEstimator, AmplitudeEstimatorResult
 from .ae_utils import pdf_a, derivative_log_pdf_a, bisect_max
 from .estimation_problem import EstimationProblem
+from ..exceptions import AlgorithmError
 
 
 class AmplitudeEstimation(AmplitudeEstimator):
@@ -56,9 +61,10 @@ class AmplitudeEstimation(AmplitudeEstimator):
     def __init__(
         self,
         num_eval_qubits: int,
-        phase_estimation_circuit: Optional[QuantumCircuit] = None,
-        iqft: Optional[QuantumCircuit] = None,
-        quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
+        phase_estimation_circuit: None | QuantumCircuit = None,
+        iqft: None | QuantumCircuit = None,
+        quantum_instance: None | Union[QuantumInstance, Backend] = None,
+        sampler: None | BaseSampler = None,
     ) -> None:
         r"""
         Args:
@@ -68,7 +74,9 @@ def __init__(
                 `qiskit.circuit.library.PhaseEstimation` when None.
             iqft: The inverse quantum Fourier transform component, defaults to using a standard
                 implementation from `qiskit.circuit.library.QFT` when None.
-            quantum_instance: The backend (or `QuantumInstance`) to execute the circuits on.
+            quantum_instance: Pending deprecation\: The backend (or `QuantumInstance`) to execute
+                the circuits on.
+            sampler: base sampler
 
         Raises:
             ValueError: If the number of evaluation qubits is smaller than 1.
@@ -79,27 +87,49 @@ def __init__(
         super().__init__()
 
         # set quantum instance
-        self.quantum_instance = quantum_instance
+        if quantum_instance is not None:
+            warnings.warn(
+                "The quantum_instance argument has been superseded by the sampler argument. "
+                "This argument will be deprecated in a future release and subsequently "
+                "removed after that.",
+                category=PendingDeprecationWarning,
+            )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            self.quantum_instance = quantum_instance
 
         # get parameters
         self._m = num_eval_qubits  # pylint: disable=invalid-name
         self._M = 2**num_eval_qubits  # pylint: disable=invalid-name
 
         self._iqft = iqft
         self._pec = phase_estimation_circuit
+        self._sampler = sampler
 
     @property
-    def quantum_instance(self) -> Optional[QuantumInstance]:
-        """Get the quantum instance.
+    @deprecate_function(
+        "The AmplitudeEstimation.quantum_instance getter is pending deprecation. "
+        "This property will be deprecated in a future release and subsequently "
+        "removed after that.",
+        category=PendingDeprecationWarning,
+    )
+    def quantum_instance(self) -> None | QuantumInstance:
+        """Pending deprecation: Get the quantum instance.
 
         Returns:
             The quantum instance used to run this algorithm.
         """
         return self._quantum_instance
 
     @quantum_instance.setter
+    @deprecate_function(
+        "The AmplitudeEstimation.quantum_instance setter is pending deprecation. "
+        "This property will be deprecated in a future release and subsequently "
+        "removed after that.",
+        category=PendingDeprecationWarning,
+    )
     def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) -> None:
-        """Set quantum instance.
+        """Pending deprecation: Set quantum instance.
 
         Args:
             quantum_instance: The quantum instance used to run this algorithm.
@@ -201,7 +231,9 @@ def _evaluate_count_results(self, counts):
         # construct probabilities
         measurements = OrderedDict()
         samples = OrderedDict()
-        shots = self._quantum_instance._run_config.shots
+        shots = (
+            self._quantum_instance._run_config.shots if self._quantum_instance is not None else 1
+        )
 
         for state, count in counts.items():
             y = int(state.replace(" ", "")[: self._m][::-1], 2)
@@ -283,12 +315,16 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio
         Raises:
             ValueError: If `state_preparation` or `objective_qubits` are not set in the
                 `estimation_problem`.
+            ValueError: A quantum instance or Sampler must be provided.
+            AlgorithmError: Sampler run error
         """
         # check if A factory or state_preparation has been set
         if estimation_problem.state_preparation is None:
             raise ValueError(
                 "The state_preparation property of the estimation problem must be set."
             )
+        if self._quantum_instance is None and self._sampler is None:
+            raise ValueError("A quantum instance or sampler must be provided.")
 
         if estimation_problem.objective_qubits is None:
             raise ValueError("The objective_qubits property of the estimation problem must be set.")
@@ -297,15 +333,37 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio
         result.num_evaluation_qubits = self._m
         result.post_processing = estimation_problem.post_processing
 
-        if self._quantum_instance.is_statevector:
+        measurements = OrderedDict()
+        samples = OrderedDict()
+        if self._sampler is not None:
+            circuit = self.construct_circuit(estimation_problem, measurement=True)
+            job = self._sampler.run([circuit])
+            if job.status() is not JobStatus.DONE:
+                raise AlgorithmError("The job was not completed successfully. ")
+            ret = job.result()
+            result.shots = 1
+            circuit_results = {
+                np.binary_repr(k, circuit.num_qubits): v for k, v in ret.quasi_dists[0].items()
+            }
+            for state, probability in circuit_results.items():
+                state = state[::-1]
+                y = int(state[: self._m], 2)
+                measurements[y] = probability
+                a = np.round(np.power(np.sin(y * np.pi / 2**self._m), 2), decimals=7)
+                samples[a] = samples.get(a, 0.0) + probability
+            # cutoff probabilities below the threshold
+            threshold: float = 1e-6
+            samples = {a: p for a, p in samples.items() if p > threshold}
+            measurements = {y: p for y, p in measurements.items() if p > threshold}
+        elif self._quantum_instance.is_statevector:
             circuit = self.construct_circuit(estimation_problem, measurement=False)
             # run circuit on statevector simulator
             statevector = self._quantum_instance.execute(circuit).get_statevector()
             result.circuit_results = statevector
-
             # store number of shots: convention is 1 shot for statevector,
             # needed so that MLE works!
             result.shots = 1
+            samples, measurements = self.evaluate_measurements(result.circuit_results)
         else:
             # run circuit on QASM simulator
             circuit = self.construct_circuit(estimation_problem, measurement=True)
@@ -314,8 +372,7 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio
 
             # store shots
             result.shots = sum(counts.values())
-
-        samples, measurements = self.evaluate_measurements(result.circuit_results)
+            samples, measurements = self.evaluate_measurements(result.circuit_results)
 
         result.samples = samples
         result.samples_processed = {

qiskit/algorithms/amplitude_estimators/fae.py

@@ -1,6 +1,6 @@
 # This code is part of Qiskit.
 #
-# (C) Copyright IBM 2017, 2020.
+# (C) Copyright IBM 2017, 2022.
 #
 # 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
@@ -12,12 +12,16 @@
 
 """Faster Amplitude Estimation."""
 
-from typing import Optional, Union, List, Tuple
+from __future__ import annotations
+from typing import Union, List, Tuple
+import warnings
 import numpy as np
 
 from qiskit.circuit import QuantumCircuit, ClassicalRegister
-from qiskit.providers import Backend
+from qiskit.providers import Backend, JobStatus
+from qiskit.primitives import BaseSampler
 from qiskit.utils import QuantumInstance
+from qiskit.utils.deprecation import deprecate_function
 from qiskit.algorithms.exceptions import AlgorithmError
 
 from .amplitude_estimator import AmplitudeEstimator, AmplitudeEstimatorResult
@@ -50,14 +54,17 @@ def __init__(
         delta: float,
         maxiter: int,
         rescale: bool = True,
-        quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
+        quantum_instance: None | Union[QuantumInstance, Backend] = None,
+        sampler: None | BaseSampler = None,
     ) -> None:
         r"""
         Args:
             delta: The probability that the true value is outside of the final confidence interval.
             maxiter: The number of iterations, the maximal power of Q is `2 ** (maxiter - 1)`.
             rescale: Whether to rescale the problem passed to `estimate`.
-            quantum_instance: The quantum instance or backend to run the circuits.
+            quantum_instance: Pending deprecation\: The quantum instance or backend
+                to run the circuits.
+            sampler: base sampler
 
         .. note::
 
@@ -66,25 +73,48 @@ def __init__(
 
         """
         super().__init__()
-        self.quantum_instance = quantum_instance
+        # set quantum instance
+        if quantum_instance is not None:
+            warnings.warn(
+                "The quantum_instance argument has been superseded by the sampler argument. "
+                "This argument will be deprecated in a future release and subsequently "
+                "removed after that.",
+                category=PendingDeprecationWarning,
+            )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            self.quantum_instance = quantum_instance
         self._shots = (int(1944 * np.log(2 / delta)), int(972 * np.log(2 / delta)))
         self._rescale = rescale
         self._delta = delta
         self._maxiter = maxiter
         self._num_oracle_calls = 0
+        self._sampler = sampler
 
     @property
-    def quantum_instance(self) -> Optional[QuantumInstance]:
-        """Get the quantum instance.
+    @deprecate_function(
+        "The FasterAmplitudeEstimation.quantum_instance getter is pending deprecation. "
+        "This property will be deprecated in a future release and subsequently "
+        "removed after that.",
+        category=PendingDeprecationWarning,
+    )
+    def quantum_instance(self) -> None | QuantumInstance:
+        """Pending deprecation: Get the quantum instance.
 
         Returns:
             The quantum instance used to run this algorithm.
         """
         return self._quantum_instance
 
     @quantum_instance.setter
+    @deprecate_function(
+        "The FasterAmplitudeEstimation.quantum_instance setter is pending deprecation. "
+        "This property will be deprecated in a future release and subsequently "
+        "removed after that.",
+        category=PendingDeprecationWarning,
+    )
     def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) -> None:
-        """Set quantum instance.
+        """Pending deprecation: Set quantum instance.
 
         Args:
             quantum_instance: The quantum instance used to run this algorithm.
@@ -94,10 +124,30 @@ def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) ->
         self._quantum_instance = quantum_instance
 
     def _cos_estimate(self, estimation_problem, k, shots):
-        if self._quantum_instance is None:
-            raise AlgorithmError("Quantum instance must be set.")
+        if self._quantum_instance is None and self._sampler is None:
+            raise ValueError("A quantum instance or sampler must be provided.")
 
-        if self._quantum_instance.is_statevector:
+        if self._sampler is not None:
+            circuit = self.construct_circuit(estimation_problem, k, measurement=True)
+            job = self._sampler.run([circuit])
+            if job.status() is not JobStatus.DONE:
+                raise AlgorithmError("The job was not completed successfully. ")
+            result = job.result()
+
+            # sum over all probabilities where the objective qubits are 1
+            prob = 0
+            for bit, probabilities in result.quasi_dists[0].items():
+                i = int(bit)
+                # get bitstring of objective qubits
+                full_state = bin(i)[2:].zfill(circuit.num_qubits)[::-1]
+                state = "".join([full_state[i] for i in estimation_problem.objective_qubits])
+
+                # check if it is a good state
+                if estimation_problem.is_good_state(state[::-1]):
+                    prob = prob + probabilities
+
+            cos_estimate = 1 - 2 * prob
+        elif self._quantum_instance.is_statevector:
             circuit = self.construct_circuit(estimation_problem, k, measurement=False)
             statevector = self._quantum_instance.execute(circuit).get_statevector()
 
@@ -179,21 +229,37 @@ def construct_circuit(
         return circuit
 
     def estimate(self, estimation_problem: EstimationProblem) -> "FasterAmplitudeEstimationResult":
+        """Run the amplitude estimation algorithm on provided estimation problem.
+
+        Args:
+            estimation_problem: The estimation problem.
+
+        Returns:
+            An amplitude estimation results object.
+
+        Raises:
+            ValueError: A quantum instance or Sampler must be provided.
+            AlgorithmError: Sampler run error
+        """
+        if self._quantum_instance is None and self._sampler is None:
+            raise ValueError("A quantum instance or sampler must be provided.")
+
         self._num_oracle_calls = 0
-        user_defined_shots = self.quantum_instance._run_config.shots
+        user_defined_shots = (
+            self._quantum_instance._run_config.shots if self._quantum_instance is not None else 0
+        )
 
         if self._rescale:
             problem = estimation_problem.rescale(0.25)
         else:
             problem = estimation_problem
 
-        if self._quantum_instance.is_statevector:
+        if self._sampler is not None or self._quantum_instance.is_statevector:
             cos = self._cos_estimate(problem, k=0, shots=1)
             theta = np.arccos(cos) / 2
             theta_ci = [theta, theta]
             theta_cis = [theta_ci]
             num_steps = num_first_stage_steps = 1
-
         else:
             theta_ci = [0, np.arcsin(0.25)]
             first_stage = True
@@ -240,7 +306,7 @@ def cos_estimate(power, shots):
         result.num_oracle_queries = self._num_oracle_calls
         result.num_steps = num_steps
         result.num_first_state_steps = num_first_stage_steps
-        if self._quantum_instance.is_statevector:
+        if self._sampler is not None or self._quantum_instance.is_statevector:
             result.success_probability = 1
         else:
             result.success_probability = 1 - (2 * self._maxiter - j_0) * self._delta
@@ -252,7 +318,8 @@ def cos_estimate(power, shots):
         result.theta_intervals = theta_cis
 
         # reset shots to what the user had defined
-        self.quantum_instance._run_config.shots = user_defined_shots
+        if self._quantum_instance is not None:
+            self._quantum_instance._run_config.shots = user_defined_shots
         return result