Update BaseEstimator import to support qiskit 2.0

qiskit_aer/noise/noise_model.py

@@ -444,6 +444,7 @@ def from_backend(
                     dt=dt,
                     op_types=Delay,
                     excited_state_populations=excited_state_populations,
+                    target=target,
                 )
                 noise_model._custom_noise_passes.append(delay_pass)
             except ValueError:

qiskit_aer/noise/passes/relaxation_noise_pass.py

@@ -18,6 +18,7 @@
 import numpy as np
 
 from qiskit.circuit import Instruction, QuantumCircuit
+from qiskit.transpiler import Target
 from qiskit.utils.units import apply_prefix
 from .local_noise_pass import LocalNoisePass
 from ..errors.standard_errors import thermal_relaxation_error
@@ -34,6 +35,7 @@ def __init__(
         dt: Optional[float] = None,
         op_types: Optional[Union[type, Sequence[type]]] = None,
         excited_state_populations: Optional[List[float]] = None,
+        target: Target = None,
     ):
         """Initialize RelaxationNoisePass.
 
@@ -47,6 +49,7 @@ def __init__(
             excited_state_populations: Optional, list of excited state populations
                 for each qubit at thermal equilibrium. If not supplied or obtained
                 from the backend this will be set to 0 for each qubit.
+            target: Optional, target instance with instruction duration information.
         """
         self._t1s = np.asarray(t1s)
         self._t2s = np.asarray(t2s)
@@ -55,29 +58,61 @@ def __init__(
         else:
             self._p1s = np.zeros(len(t1s))
         self._dt = dt
+        self._target = target
         super().__init__(self._thermal_relaxation_error, op_types=op_types, method="append")
 
     def _thermal_relaxation_error(self, op: Instruction, qubits: Sequence[int]):
         """Return thermal relaxation error on each operand qubit"""
-        if not op.duration:
-            if op.duration is None:
-                warnings.warn(
-                    "RelaxationNoisePass ignores instructions without duration,"
-                    " you may need to schedule circuit in advance.",
-                    UserWarning,
-                )
-            return None
 
-        # Convert op duration to seconds
-        if op.unit == "dt":
-            if self._dt is None:
-                raise NoiseError(
-                    "RelaxationNoisePass cannot apply noise to a 'dt' unit duration"
-                    " without a dt time set."
-                )
-            duration = op.duration * self._dt
+        duration = None
+
+        if self._target is not None:
+            if op.name == "delay":
+                duration = op.duration
+                if duration is not None:
+                    # Convert op duration to seconds
+                    if op.unit == "dt":
+                        if self._dt is None:
+                            raise NoiseError(
+                                "RelaxationNoisePass cannot apply noise to a 'dt' unit duration"
+                                " without a dt time set."
+                            )
+                        duration = op.duration * self._dt
+                    else:
+                        duration = apply_prefix(op.duration, op.unit)
+            else:
+                op_props = self._target.get(op.name)
+                if op_props is not None:
+                    inst_props = op_props.get(tuple(qubits))
+                    if inst_props is not None:
+                        # get duration in seconds
+                        duration = getattr(inst_props, "duration")
         else:
-            duration = apply_prefix(op.duration, op.unit)
+            try:
+                duration = op.duration
+                if duration is not None:
+                    # Convert op duration to seconds
+                    if op.unit == "dt":
+                        if self._dt is None:
+                            raise NoiseError(
+                                "RelaxationNoisePass cannot apply noise to a 'dt' unit duration"
+                                " without a dt time set."
+                            )
+                        duration = op.duration * self._dt
+                    else:
+                        duration = apply_prefix(op.duration, op.unit)
+
+            except AttributeError:
+                duration = None
+
+        if duration is None:
+            warnings.warn(
+                r"Instruction duration not found for {op.name}. RelaxationNoisePass ignores "
+                "instructions without duration. "
+                "To avoid this warning, provide the corresponding duration information via `target`.",
+                UserWarning,
+            )
+            return None
 
         t1s = self._t1s[qubits]
         t2s = self._t2s[qubits]

qiskit_aer/primitives/estimator.py

@@ -23,12 +23,12 @@
 import numpy as np
 from qiskit.circuit import ParameterExpression, QuantumCircuit
 from qiskit.compiler import transpile
-from qiskit.primitives import BaseEstimator, EstimatorResult
+from qiskit.primitives import BaseEstimatorV1, EstimatorResult
 from qiskit.primitives.primitive_job import PrimitiveJob
-from qiskit.primitives.utils import _circuit_key, _observable_key, init_observable
 from qiskit.providers import Options
-from qiskit.quantum_info import Pauli, PauliList
+from qiskit.quantum_info import Pauli, PauliList, SparsePauliOp
 from qiskit.quantum_info.operators.base_operator import BaseOperator
+from qiskit.quantum_info.operators.symplectic.base_pauli import BasePauli
 from qiskit.result.models import ExperimentResult
 from qiskit.transpiler import CouplingMap, PassManager
 from qiskit.transpiler.passes import (
@@ -41,9 +41,49 @@
 from qiskit.utils import deprecate_func
 
 from .. import AerError, AerSimulator
+from .sampler import _circuit_key
 
 
-class Estimator(BaseEstimator):
+def init_observable(observable: BaseOperator | str) -> SparsePauliOp:
+    """Initialize observable by converting the input to a :class:`~qiskit.quantum_info.SparsePauliOp`.
+
+    Args:
+        observable: The observable.
+
+    Returns:
+        The observable as :class:`~qiskit.quantum_info.SparsePauliOp`.
+
+    Raises:
+        AerError: when observable type cannot be converted to SparsePauliOp.
+    """
+
+    if isinstance(observable, SparsePauliOp):
+        return observable
+    elif isinstance(observable, BaseOperator) and not isinstance(observable, BasePauli):
+        raise AerError(f"observable type not supported: {type(observable)}")
+    else:
+        if isinstance(observable, PauliList):
+            raise AerError(f"observable type not supported: {type(observable)}")
+        return SparsePauliOp(observable)
+
+
+def _observable_key(observable: SparsePauliOp) -> tuple:
+    """Private key function for SparsePauliOp.
+    Args:
+        observable: Input operator.
+
+    Returns:
+        Key for observables.
+    """
+    return (
+        observable.paulis.z.tobytes(),
+        observable.paulis.x.tobytes(),
+        observable.paulis.phase.tobytes(),
+        observable.coeffs.tobytes(),
+    )
+
+
+class Estimator(BaseEstimatorV1):
     """
     Aer implmentation of Estimator.
 
@@ -592,7 +632,12 @@ def run(self, results: list[ExperimentResult]) -> tuple[float, dict]:
                 result = results[c_i]
                 count = result.data.counts
                 shots = sum(count.values())
-                basis = result.header.metadata["basis"]
+                # added for compatibility with qiskit 1.4 (metadata as attribute)
+                # and qiskit 2.0 (header as dict)
+                try:
+                    basis = result.header.metadata["basis"]
+                except AttributeError:
+                    basis = result.header["metadata"]["basis"]
                 indices = np.where(basis.z | basis.x)[0]
                 measured_paulis = PauliList.from_symplectic(
                     paulis.z[:, indices], paulis.x[:, indices], 0

qiskit_aer/primitives/sampler.py

@@ -16,20 +16,134 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Sequence, Iterable
 
 from warnings import warn
 import numpy as np
-from qiskit.circuit import ParameterExpression, QuantumCircuit
+from qiskit.circuit import ParameterExpression, QuantumCircuit, Qubit
+from qiskit.circuit.library.data_preparation import Initialize
 from qiskit.compiler import transpile
 from qiskit.exceptions import QiskitError
 from qiskit.primitives import BaseSamplerV1, SamplerResult
-from qiskit.primitives.utils import final_measurement_mapping, init_circuit
+from qiskit.quantum_info import Statevector
 from qiskit.result import QuasiDistribution
 
 from .. import AerSimulator
 
 
+def init_circuit(state: QuantumCircuit | Statevector) -> QuantumCircuit:
+    """Initialize state by converting the input to a quantum circuit.
+
+    Args:
+        state: The state as quantum circuit or statevector.
+
+    Returns:
+        The state as quantum circuit.
+    """
+    if isinstance(state, QuantumCircuit):
+        return state
+    if not isinstance(state, Statevector):
+        state = Statevector(state)
+    qc = QuantumCircuit(state.num_qubits)
+    qc.append(Initialize(state), qargs=range(state.num_qubits))
+    return qc
+
+
+def final_measurement_mapping(circuit: QuantumCircuit) -> dict[int, int]:
+    """Return the final measurement mapping for the circuit.
+
+    Dict keys label measured qubits, whereas the values indicate the
+    classical bit onto which that qubits measurement result is stored.
+
+    Parameters:
+        circuit: Input quantum circuit.
+
+    Returns:
+        Mapping of qubits to classical bits for final measurements.
+    """
+    active_qubits = list(range(circuit.num_qubits))
+    active_cbits = list(range(circuit.num_clbits))
+
+    # Find final measurements starting in back
+    mapping = {}
+    for item in circuit._data[::-1]:
+        if item.operation.name == "measure":
+            cbit = circuit.find_bit(item.clbits[0]).index
+            qbit = circuit.find_bit(item.qubits[0]).index
+            if cbit in active_cbits and qbit in active_qubits:
+                mapping[qbit] = cbit
+                active_cbits.remove(cbit)
+                active_qubits.remove(qbit)
+        elif item.operation.name not in ["barrier", "delay"]:
+            for qq in item.qubits:
+                _temp_qubit = circuit.find_bit(qq).index
+                if _temp_qubit in active_qubits:
+                    active_qubits.remove(_temp_qubit)
+
+        if not active_cbits or not active_qubits:
+            break
+
+    # Sort so that classical bits are in numeric order low->high.
+    mapping = dict(sorted(mapping.items(), key=lambda item: item[1]))
+    return mapping
+
+
+def _bits_key(bits: tuple[Qubit, ...], circuit: QuantumCircuit) -> tuple:
+    return tuple(
+        (
+            circuit.find_bit(bit).index,
+            tuple((reg[0].size, reg[0].name, reg[1]) for reg in circuit.find_bit(bit).registers),
+        )
+        for bit in bits
+    )
+
+
+def _format_params(param):
+    if isinstance(param, np.ndarray):
+        return param.data.tobytes()
+    elif isinstance(param, QuantumCircuit):
+        return _circuit_key(param)
+    elif isinstance(param, Iterable):
+        return tuple(param)
+    return param
+
+
+def _circuit_key(circuit: QuantumCircuit, functional: bool = True) -> tuple:
+    """Private key function for QuantumCircuit.
+
+    This is the workaround until :meth:`QuantumCircuit.__hash__` will be introduced.
+    If key collision is found, please add elements to avoid it.
+
+    Args:
+        circuit: Input quantum circuit.
+        functional: If True, the returned key only includes functional data (i.e. execution related).
+
+    Returns:
+        Composite key for circuit.
+    """
+    functional_key: tuple = (
+        circuit.num_qubits,
+        circuit.num_clbits,
+        circuit.num_parameters,
+        tuple(  # circuit.data
+            (
+                _bits_key(data.qubits, circuit),  # qubits
+                _bits_key(data.clbits, circuit),  # clbits
+                data.operation.name,  # operation.name
+                tuple(_format_params(param) for param in data.operation.params),  # operation.params
+            )
+            for data in circuit.data
+        ),
+        None if circuit._op_start_times is None else tuple(circuit._op_start_times),
+    )
+    if functional:
+        return functional_key
+    return (
+        circuit.name,
+        *functional_key,
+    )
+
+
 class Sampler(BaseSamplerV1):
     """
     Aer implementation of Sampler class.
@@ -155,7 +269,6 @@ def _run(
         from typing import List
 
         from qiskit.primitives.primitive_job import PrimitiveJob
-        from qiskit.primitives.utils import _circuit_key
 
         circuit_indices: List[int] = []
         for circuit in circuits:

test/benchmark/simulator_benchmark.py

@@ -15,7 +15,7 @@
 import sys
 import numpy as np
 from time import time
-from qiskit.compiler import transpile, assemble
+from qiskit.compiler import transpile
 from qiskit_aer import AerSimulator, UnitarySimulator
 from qiskit_aer.noise import NoiseModel, amplitude_damping_error, depolarizing_error
 
@@ -186,31 +186,23 @@ def add_expval(base, num_terms):
             if runtime in self.TRANSPLIERS:
                 runtime_circuit = eval(self.TRANSPLIERS[runtime])(circuit)
                 if (runtime, app, measure, measure_count, qubit) not in QOBJS:
-                    QOBJS[(runtime, app, measure, measure_count, qubit)] = assemble(
-                        runtime_circuit, simulator, shots=measure_count
-                    )
+                    QOBJS[(runtime, app, measure, measure_count, qubit)] = runtime_circuit
                 return QOBJS[(runtime, app, measure, measure_count, qubit)]
             else:
                 runtime_circuit = circuit
                 if (simulator, app, measure, measure_count, qubit) not in QOBJS:
-                    QOBJS[(simulator, app, measure, measure_count, qubit)] = assemble(
-                        runtime_circuit, simulator, shots=measure_count
-                    )
+                    QOBJS[(simulator, app, measure, measure_count, qubit)] = runtime_circuit
                 return QOBJS[(simulator, app, measure, measure_count, qubit)]
         elif measure == self.MEASUREMENT_EXPVAL:
             if runtime in self.TRANSPLIERS:
                 runtime_circuit = eval(self.TRANSPLIERS[runtime])(circuit)
                 if (runtime, app, measure, measure_count, qubit) not in QOBJS:
-                    QOBJS[(runtime, app, measure, measure_count, qubit)] = assemble(
-                        runtime_circuit, simulator, shots=1
-                    )
+                    QOBJS[(runtime, app, measure, measure_count, qubit)] = runtime_circuit
                 return QOBJS[(runtime, app, measure, measure_count, qubit)]
             else:
                 runtime_circuit = circuit
                 if (simulator, app, measure, measure_count, qubit) not in QOBJS:
-                    QOBJS[(simulator, app, measure, measure_count, qubit)] = assemble(
-                        runtime_circuit, simulator, shots=1
-                    )
+                    QOBJS[(simulator, app, measure, measure_count, qubit)] = runtime_circuit
                 return QOBJS[(simulator, app, measure, measure_count, qubit)]
 
     def _transpile(self, circuit, basis_gates):