Skip to content

Commit

Permalink
lint: fix some mypy stuff
Browse files Browse the repository at this point in the history
More to come from Elena soon... 🙂
  • Loading branch information
mrossinek committed Sep 8, 2023
1 parent c1b068f commit 7dd27f4
Show file tree
Hide file tree
Showing 29 changed files with 126 additions and 98 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from __future__ import annotations

from collections.abc import Callable
from typing import Any
from typing import Any, cast

from qiskit.circuit import QuantumCircuit
from qiskit.circuit.library import GroverOperator
Expand Down Expand Up @@ -167,13 +167,13 @@ def is_good_state(self) -> Callable[[str], bool]:
return self._is_good_state # returns None if no is_good_state arg has been set
elif isinstance(self._is_good_state, list):
if all(isinstance(good_bitstr, str) for good_bitstr in self._is_good_state):
return lambda bitstr: bitstr in self._is_good_state
return lambda bitstr: bitstr in cast(list[str], self._is_good_state)
else:
return lambda bitstr: all(
bitstr[good_index] == "1" for good_index in self._is_good_state
bitstr[good_index] == "1" for good_index in cast(list[int], self._is_good_state)
)

return lambda bitstr: bitstr in self._is_good_state.probabilities_dict()
return lambda bitstr: bitstr in cast(Statevector, self._is_good_state).probabilities_dict()

@is_good_state.setter
def is_good_state(
Expand Down
8 changes: 3 additions & 5 deletions qiskit_algorithms/eigensolvers/eigensolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,7 @@ class EigensolverResult(AlgorithmResult):
def __init__(self) -> None:
super().__init__()
self._eigenvalues: np.ndarray | None = None
self._aux_operators_evaluated: list[
ListOrDict[tuple[complex, dict[str, Any]]]
] | None = None
self._aux_operators_evaluated: list[ListOrDict[tuple[float, dict[str, Any]]]] | None = None

@property
def eigenvalues(self) -> np.ndarray | None:
Expand All @@ -90,7 +88,7 @@ def eigenvalues(self, value: np.ndarray) -> None:
@property
def aux_operators_evaluated(
self,
) -> list[ListOrDict[tuple[complex, dict[str, Any]]]] | None:
) -> list[ListOrDict[tuple[float, dict[str, Any]]]] | None:
"""Return the aux operator expectation values.
These values are in fact tuples formatted as (mean, metadata).
Expand All @@ -99,7 +97,7 @@ def aux_operators_evaluated(

@aux_operators_evaluated.setter
def aux_operators_evaluated(
self, value: list[ListOrDict[tuple[complex, dict[str, Any]]]]
self, value: list[ListOrDict[tuple[float, dict[str, Any]]]]
) -> None:
"""Set the aux operator eigenvalues."""
self._aux_operators_evaluated = value
18 changes: 11 additions & 7 deletions qiskit_algorithms/eigensolvers/numpy_eigensolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@

from __future__ import annotations

from typing import Callable, Union, List, Optional
from collections.abc import Iterable
from typing import Callable, Union, List, Optional, cast
import logging
import numpy as np
from scipy import sparse as scisparse
Expand All @@ -29,7 +30,9 @@

logger = logging.getLogger(__name__)

FilterType = Callable[[Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool]
FilterType = Callable[
[Union[List, np.ndarray], float, Optional[ListOrDict[tuple[float, dict[str, float]]]]], bool
]


class NumPyEigensolver(Eigensolver):
Expand Down Expand Up @@ -159,21 +162,22 @@ def _solve_sparse(op_matrix: scisparse.csr_matrix, k: int) -> tuple[np.ndarray,
def _solve_dense(op_matrix: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
if op_matrix.all() == op_matrix.conj().T.all():
# Operator is Hermitian
return np.linalg.eigh(op_matrix)
return cast(tuple[np.ndarray, np.ndarray], np.linalg.eigh(op_matrix))
else:
return np.linalg.eig(op_matrix)
return cast(tuple[np.ndarray, np.ndarray], np.linalg.eig(op_matrix))

@staticmethod
def _eval_aux_operators(
aux_operators: ListOrDict[BaseOperator],
wavefn: np.ndarray,
threshold: float = 1e-12,
) -> ListOrDict[tuple[complex, complex]]:
) -> ListOrDict[tuple[float, dict[str, float]]]:

values: ListOrDict[tuple[complex, complex]]
values: ListOrDict[tuple[float, dict[str, float]]]

# As a list, aux_operators can contain None operators for which None values are returned.
# As a dict, the None operators in aux_operators have been dropped in compute_eigenvalues.
key_op_iterator: Iterable[tuple[str | int, BaseOperator]]
if isinstance(aux_operators, list):
values = [None] * len(aux_operators)
key_op_iterator = enumerate(aux_operators)
Expand Down Expand Up @@ -217,7 +221,7 @@ def _eval_aux_operators(
# The metadata includes variance (and, for other eigensolvers, shots).
# Since this is an exact computation, there are no shots
# and the variance is known to be zero.
values[key] = (value, {"variance": 0.0})
values[key] = (value, {"variance": 0.0}) # type: ignore[index]
return values

def compute_eigenvalues(
Expand Down
8 changes: 6 additions & 2 deletions qiskit_algorithms/gradients/base/base_estimator_gradient.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,9 @@ def _preprocess(
parameter_values and parameters are updated to match the gradient circuit.
"""
translator = TranslateParameterizedGates(supported_gates)
g_circuits, g_parameter_values, g_parameters = [], [], []
g_circuits: list[QuantumCircuit] = []
g_parameter_values: list[Sequence[float]] = []
g_parameters: list[Sequence[Parameter]] = []
for circuit, parameter_value_, parameters_ in zip(circuits, parameter_values, parameters):
circuit_key = _circuit_key(circuit)
if circuit_key not in self._gradient_circuit_cache:
Expand All @@ -196,7 +198,9 @@ def _preprocess(
gradient_circuit = self._gradient_circuit_cache[circuit_key]
g_circuits.append(gradient_circuit.gradient_circuit)
g_parameter_values.append(
_make_gradient_parameter_values(circuit, gradient_circuit, parameter_value_)
_make_gradient_parameter_values( # type: ignore[arg-type]
circuit, gradient_circuit, parameter_value_
)
)
g_parameters.append(_make_gradient_parameters(gradient_circuit, parameters_))
return g_circuits, g_parameter_values, g_parameters
Expand Down
13 changes: 9 additions & 4 deletions qiskit_algorithms/gradients/base/base_qgt.py
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,9 @@ def _preprocess(
parameter_values and parameters are updated to match the gradient circuit.
"""
translator = TranslateParameterizedGates(supported_gates)
g_circuits, g_parameter_values, g_parameters = [], [], []
g_circuits: list[QuantumCircuit] = []
g_parameter_values: list[Sequence[float]] = []
g_parameters: list[Sequence[Parameter]] = []
for circuit, parameter_value_, parameters_ in zip(circuits, parameter_values, parameters):
circuit_key = _circuit_key(circuit)
if circuit_key not in self._gradient_circuit_cache:
Expand All @@ -207,7 +209,9 @@ def _preprocess(
gradient_circuit = self._gradient_circuit_cache[circuit_key]
g_circuits.append(gradient_circuit.gradient_circuit)
g_parameter_values.append(
_make_gradient_parameter_values(circuit, gradient_circuit, parameter_value_)
_make_gradient_parameter_values( # type: ignore[arg-type]
circuit, gradient_circuit, parameter_value_
)
)
g_parameters_ = [
g_param
Expand Down Expand Up @@ -253,7 +257,7 @@ def _postprocess(
for param in gradient_circuit.gradient_circuit.parameters
if param in g_parameters
]
g_parameter_indices = {param: i for i, param in enumerate(g_parameter_indices)}
g_parameter_indices_d = {param: i for i, param in enumerate(g_parameter_indices)}
rows, cols = np.triu_indices(len(parameters_))
for row, col in zip(rows, cols):
for g_parameter1, coeff1 in gradient_circuit.parameter_map[parameters_[row]]:
Expand All @@ -278,7 +282,8 @@ def _postprocess(
float(bound_coeff1)
* float(bound_coeff2)
* results.qgts[idx][
g_parameter_indices[g_parameter1], g_parameter_indices[g_parameter2]
g_parameter_indices_d[g_parameter1],
g_parameter_indices_d[g_parameter2],
]
)

Expand Down
10 changes: 7 additions & 3 deletions qiskit_algorithms/gradients/base/base_sampler_gradient.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ def _preprocess(
parameter_values: Sequence[Sequence[float]],
parameters: Sequence[Sequence[Parameter]],
supported_gates: Sequence[str],
) -> tuple[Sequence[QuantumCircuit], Sequence[Sequence[float]], Sequence[set[Parameter]]]:
) -> tuple[Sequence[QuantumCircuit], Sequence[Sequence[float]], Sequence[Sequence[Parameter]]]:
"""Preprocess the gradient. This makes a gradient circuit for each circuit. The gradient
circuit is a transpiled circuit by using the supported gates, and has unique parameters.
``parameter_values`` and ``parameters`` are also updated to match the gradient circuit.
Expand All @@ -144,7 +144,9 @@ def _preprocess(
parameter_values and parameters are updated to match the gradient circuit.
"""
translator = TranslateParameterizedGates(supported_gates)
g_circuits, g_parameter_values, g_parameters = [], [], []
g_circuits: list[QuantumCircuit] = []
g_parameter_values: list[Sequence[float]] = []
g_parameters: list[Sequence[Parameter]] = []
for circuit, parameter_value_, parameters_ in zip(circuits, parameter_values, parameters):
circuit_key = _circuit_key(circuit)
if circuit_key not in self._gradient_circuit_cache:
Expand All @@ -153,7 +155,9 @@ def _preprocess(
gradient_circuit = self._gradient_circuit_cache[circuit_key]
g_circuits.append(gradient_circuit.gradient_circuit)
g_parameter_values.append(
_make_gradient_parameter_values(circuit, gradient_circuit, parameter_value_)
_make_gradient_parameter_values( # type: ignore[arg-type]
circuit, gradient_circuit, parameter_value_
)
)
g_parameters.append(_make_gradient_parameters(gradient_circuit, parameters_))
return g_circuits, g_parameter_values, g_parameters
Expand Down
2 changes: 1 addition & 1 deletion qiskit_algorithms/gradients/base/qgt_result.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ class QGTResult:
"""The QGT."""
derivative_type: DerivativeType
"""The type of derivative."""
metadata: list[dict[str, Any]]
metadata: list[dict[str, Any]] | list[list[dict[str, Any]]]
"""Additional information about the job."""
options: Options
"""Primitive runtime options for the execution of the job."""
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ class SamplerGradientResult:

gradients: list[list[dict[int, float]]]
"""The gradients of the sample probabilities."""
metadata: list[dict[str, Any]]
metadata: list[dict[str, Any]] | list[list[dict[str, Any]]]
"""Additional information about the job."""
options: Options
"""Primitive runtime options for the execution of the job."""
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ def _run_unique(
all_n = []
for circuit, parameter_values_, parameters_ in zip(circuits, parameter_values, parameters):
# Prepare circuits for the gradient of the specified parameters.
# TODO: why is this not wrapped into another list level like it is done elsewhere?
metadata.append({"parameters": parameters_})
circuit_key = _circuit_key(circuit)
if circuit_key not in self._lin_comb_cache:
Expand Down
2 changes: 2 additions & 0 deletions qiskit_algorithms/gradients/reverse/bind.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,5 @@ def bind(

if not inplace:
return bound if return_list else bound[0]

return None
5 changes: 3 additions & 2 deletions qiskit_algorithms/gradients/reverse/reverse_gradient.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from __future__ import annotations
from collections.abc import Sequence
from typing import cast
import logging

import numpy as np
Expand Down Expand Up @@ -66,7 +67,7 @@ def __init__(self, derivative_type: DerivativeType = DerivativeType.REAL):
dummy_estimator = Estimator() # this is required by the base class, but not used
super().__init__(dummy_estimator, derivative_type=derivative_type)

@BaseEstimatorGradient.derivative_type.setter
@BaseEstimatorGradient.derivative_type.setter # type: ignore[attr-defined]
def derivative_type(self, derivative_type: DerivativeType) -> None:
"""Set the derivative type."""
self._derivative_type = derivative_type
Expand Down Expand Up @@ -148,7 +149,7 @@ def _run_unique(
bind(gate, parameter_binds, inplace=True)

# iterate the state variable
unitary_j_dagger = bind(unitary_j, parameter_binds).inverse()
unitary_j_dagger = cast(QuantumCircuit, bind(unitary_j, parameter_binds)).inverse()
phi = phi.evolve(unitary_j_dagger)

# compute current gradient
Expand Down
4 changes: 2 additions & 2 deletions qiskit_algorithms/gradients/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ class LinearCombGradientCircuit:
################################################################################
def _make_param_shift_parameter_values( # pylint: disable=invalid-name
circuit: QuantumCircuit,
parameter_values: np.ndarray | list[float],
parameter_values: np.ndarray | Sequence[float],
parameters: Sequence[Parameter],
) -> list[np.ndarray]:
"""Returns a list of parameter values with offsets for parameter shift rule.
Expand Down Expand Up @@ -329,7 +329,7 @@ def _assign_unique_parameters(
def _make_gradient_parameter_values(
circuit: QuantumCircuit,
gradient_circuit: GradientCircuit,
parameter_values: np.ndarray,
parameter_values: np.ndarray | Sequence[float],
) -> np.ndarray:
"""Makes parameter values for the gradient circuit.
Expand Down
22 changes: 11 additions & 11 deletions qiskit_algorithms/minimum_eigensolvers/diagonal_estimator.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

from __future__ import annotations

from collections.abc import Callable, Sequence, Mapping
from collections.abc import Callable, Sequence, Mapping, Iterable
from typing import Any

from dataclasses import dataclass
Expand Down Expand Up @@ -43,7 +43,7 @@ class _DiagonalEstimator(BaseEstimator):
def __init__(
self,
sampler: BaseSampler,
aggregation: float | Callable[[Sequence[tuple[float, float]]], float] | None = None,
aggregation: float | Callable[[Iterable[tuple[float, float]]], float] | None = None,
callback: Callable[[Sequence[Mapping[str, Any]]], None] | None = None,
**options,
) -> None:
Expand All @@ -65,8 +65,8 @@ def __init__(

self.aggregation = aggregation
self.callback = callback
self._circuit_ids = {}
self._observable_ids = {}
self._circuit_ids: dict[int, QuantumCircuit] = {}
self._observable_ids: dict[int, BaseOperator] = {}

def _run(
self,
Expand Down Expand Up @@ -119,7 +119,7 @@ def _call(
samples = sampler_result.quasi_dists

# a list of dictionaries containing: {state: (measurement probability, value)}
evaluations = [
evaluations: list[dict[int, tuple[float, float]]] = [
{
state: (probability, _evaluate_sparsepauli(state, self._observables[i]))
for state, probability in sampled.items()
Expand Down Expand Up @@ -151,7 +151,7 @@ def _call(
)


def _get_cvar_aggregation(alpha):
def _get_cvar_aggregation(alpha: float | None) -> Callable[[Iterable[tuple[float, float]]], float]:
"""Get the aggregation function for CVaR with confidence level ``alpha``."""
if alpha is None:
alpha = 1
Expand All @@ -161,17 +161,17 @@ def _get_cvar_aggregation(alpha):
# if alpha is close to 1 we can avoid the sorting
if np.isclose(alpha, 1):

def aggregate(measurements):
def aggregate(measurements: Iterable[tuple[float, float]]) -> float:
return sum(probability * value for probability, value in measurements)

else:

def aggregate(measurements):
def aggregate(measurements: Iterable[tuple[float, float]]) -> float:
# sort by values
sorted_measurements = sorted(measurements, key=lambda x: x[1])

accumulated_percent = 0 # once alpha is reached, stop
cvar = 0
accumulated_percent = 0.0 # once alpha is reached, stop
cvar = 0.0
for probability, value in sorted_measurements:
cvar += value * min(probability, alpha - accumulated_percent)
accumulated_percent += probability
Expand All @@ -186,7 +186,7 @@ def aggregate(measurements):
_PARITY = np.array([-1 if bin(i).count("1") % 2 else 1 for i in range(256)], dtype=np.complex128)


def _evaluate_sparsepauli(state: int, observable: SparsePauliOp) -> complex:
def _evaluate_sparsepauli(state: int, observable: SparsePauliOp) -> float:
packed_uint8 = np.packbits(observable.paulis.z, axis=1, bitorder="little")
state_bytes = np.frombuffer(state.to_bytes(packed_uint8.shape[1], "little"), dtype=np.uint8)
reduced = np.bitwise_xor.reduce(packed_uint8 & state_bytes, axis=1)
Expand Down
19 changes: 11 additions & 8 deletions qiskit_algorithms/observables_evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,10 +64,11 @@ def estimate_observables(
if len(observables_list) > 0:
observables_list = _handle_zero_ops(observables_list)
quantum_state = [quantum_state] * len(observables)
parameter_values_: Sequence[float] | Sequence[Sequence[float]] | None = parameter_values
if parameter_values is not None:
parameter_values = [parameter_values] * len(observables)
parameter_values_ = [parameter_values] * len(observables)
try:
estimator_job = estimator.run(quantum_state, observables_list, parameter_values)
estimator_job = estimator.run(quantum_state, observables_list, parameter_values_)
expectation_values = estimator_job.result().values
except Exception as exc:
raise AlgorithmError("The primitive job failed!") from exc
Expand Down Expand Up @@ -113,14 +114,16 @@ def _prepare_result(
A list or a dictionary of tuples (mean, metadata).
"""

observables_eigenvalues: ListOrDict[tuple[complex, dict]]

if isinstance(observables, list):
# by construction, all None values will be overwritten
observables_eigenvalues: ListOrDict[tuple[complex, complex]] = [None] * len(observables)
key_value_iterator = enumerate(observables_results)
observables_eigenvalues = []
for value in observables_results:
observables_eigenvalues.append(value)

else:
observables_eigenvalues = {}
key_value_iterator = zip(observables.keys(), observables_results)
for key, value in zip(observables.keys(), observables_results):
observables_eigenvalues[key] = value

for key, value in key_value_iterator:
observables_eigenvalues[key] = value
return observables_eigenvalues
Loading

0 comments on commit 7dd27f4

Please sign in to comment.