# Copyright 2026 Qilimanjaro Quantum Tech
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
import heapq
import operator
from dataclasses import dataclass
from pprint import pformat
from typing import TYPE_CHECKING, Any, Generic, Never, Protocol, Self, TypeGuard, TypeVar, overload
import numpy as np
from loguru import logger
from qilisdk.core import QTensor, expect_val
from qilisdk.core.result import Result
from qilisdk.settings import get_settings
from qilisdk.yaml import yaml
from .readout import ExpectationReadout, ReadoutMethod, SamplingReadout, StateTomographyReadout
if TYPE_CHECKING:
from qilisdk.core.types import Number
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _real_if_close(number: Number) -> Number:
if isinstance(number, complex) and abs(number.imag) < get_settings().atol:
return number.real
return number
def _assert_real(number: Number) -> float:
if isinstance(number, complex):
if abs(number.imag) < get_settings().atol:
return number.real
raise ValueError("Complex Number encountered when expecting only real values to be present.")
return number
# ---------------------------------------------------------------------------
# ReadoutResult base class
# ---------------------------------------------------------------------------
@yaml.register_class
[documents]
class ReadoutResult(Result, Generic[C]):
"""Abstract base class for a single readout result.
Every concrete subclass must expose a :attr:`readout` property that returns the :class:`~qilisdk.readout.ReadoutMethod`
configuration used to produce the result.
"""
# ---------------------------------------------------------------------------
# Concrete result classes
# ---------------------------------------------------------------------------
@yaml.register_class
[documents]
class SamplingReadoutResult(ReadoutResult[SamplingReadout]):
"""Result produced by a :class:`~qilisdk.readout.SamplingReadout`.
Holds bitstring measurement counts and the corresponding probability
distribution. The object can be constructed in two ways:
1. From explicit ``samples`` (and optionally ``probabilities``).
2. From a quantum ``state``, in which case samples are drawn stochastically according to the state amplitudes.
Args:
readout (SamplingReadout): The readout configuration that produced this result.
samples (dict[str, int] | None): Mapping of bitstring to measurement count. Mutually exclusive with ``state``.
probabilities (dict[str, float] | None): Pre-computed probability distribution. If omitted when ``samples``
is given, it is derived from the counts.
state (QTensor | None): Quantum state from which to derive samples and probabilities. Mutually exclusive
with ``samples``.
Raises:
ValueError: If neither ``samples``/``probabilities`` nor ``state``
is provided.
"""
@staticmethod
def _filter_samples(
samples_to_filter: dict[str, int], qubits_to_measure: list[int], expand_samples: bool = True
) -> dict[str, int]:
"""
Filter the input samples to include only the requested qubits.
Args:
samples_to_filter (dict[str, int]): The original samples to be filtered.
qubits_to_measure (list[int]): The list of qubit indices that were measured.
expand_samples (bool): Whether to include placeholders for unmeasured qubits in the filtered results.
Returns:
dict[str, int]: The filtered samples containing only the requested qubits.
"""
filtered_samples: dict[str, int] = {}
for bitstring, count in samples_to_filter.items():
if expand_samples:
filtered_bitstring = "".join(
"_" if i not in qubits_to_measure else bit for i, bit in enumerate(bitstring)
)
else:
filtered_bitstring = "".join(bit for i, bit in enumerate(bitstring) if i in qubits_to_measure)
filtered_samples[filtered_bitstring] = filtered_samples.get(filtered_bitstring, 0) + count
return filtered_samples
@staticmethod
def _expand_samples(
samples_to_expand: dict[str, int], qubits_to_measure: list[int], nqubits_total: int
) -> dict[str, int]:
"""
Expand the input samples to include placeholders for unmeasured qubits.
Args:
samples_to_expand (dict[str, int]): The original samples to be expanded.
qubits_to_measure (list[int]): The list of qubit indices that were measured.
nqubits_total (int): The total number of qubits in the system.
Returns:
dict[str, int]: The expanded samples containing placeholders for unmeasured qubits.
"""
expanded_samples: dict[str, int] = {}
for bitstring, count in samples_to_expand.items():
bitstring_remaining = bitstring
expanded_bitstring = ""
for i in range(nqubits_total):
if i in qubits_to_measure:
expanded_bitstring += bitstring_remaining[0]
bitstring_remaining = bitstring_remaining[1:]
else:
expanded_bitstring += "_"
expanded_samples[expanded_bitstring] = expanded_samples.get(expanded_bitstring, 0) + count
return expanded_samples
@classmethod
def _adjust_samples(
cls,
samples_to_filter: dict[str, int],
qubits_to_measure: list[int],
nqubits: int | None = None,
expand_samples: bool = True,
) -> dict[str, int]:
"""
Adjust the input samples to match the requested qubits to measure and total number of qubits.
This method handles both filtering and expansion of samples based on the relationship between
the number of qubits in the input samples and the requested total number of qubits.
- If the input samples have more qubits than requested, it filters out the unmeasured qubits.
- If the input samples have fewer qubits than requested, it expands them by adding placeholders for the unmeasured qubits.
Args:
samples_to_filter (dict[str, int]): The original samples to be adjusted.
qubits_to_measure (list[int]): The list of qubit indices that were measured.
nqubits (int | None): The total number of qubits in the system.
expand_samples (bool): Whether to expand samples with placeholders for unmeasured qubits.
Returns:
dict[str, int]: The adjusted samples matching the requested qubits and total number of qubits.
"""
if len(samples_to_filter) >= 1:
nqubits_samples = len(next(iter(samples_to_filter.keys())))
nqubits_total = nqubits if nqubits is not None else nqubits_samples
# Assuming 3 qubits, if asked for 0 and 1 and we have xxx, return xx_
if nqubits_samples >= nqubits_total:
return cls._filter_samples(samples_to_filter, qubits_to_measure, expand_samples=expand_samples)
# Assuming 3 qubits, if asked for 0 and 1 and we have xx, return xx_
if nqubits_samples < nqubits_total and expand_samples:
return cls._expand_samples(samples_to_filter, qubits_to_measure, nqubits_total)
return samples_to_filter
@classmethod
[documents]
def from_samples(
cls,
samples: dict[str, int],
qubits_to_measure: list[int] | None = None,
nqubits: int | None = None,
expand_samples: bool | None = None,
) -> Self:
"""
Construct a SamplingReadoutResult from raw samples.
Args:
samples (dict[str, int]): Mapping of bitstring to measurement count.
qubits_to_measure (list[int] | None): Optional list of qubit indices that were measured.
nqubits (int | None): Total number of qubits in the system.
expand_samples (bool | None): Whether to expand samples with placeholders for unmeasured qubits.
Returns:
SamplingReadoutResult: The constructed result object.
Raises:
ValueError: If samples are not provided.
ValueError: If not all bitstring keys have the same length.
ValueError: If qubits_to_measure is provided and has more qubits than are present in the bitstrings.
"""
if not samples:
raise ValueError("can't initialize Sampling Results if samples are not provided.")
nshots = sum(samples.values())
bitstrings = list(samples.keys())
nqubits_samples = len(bitstrings[0])
nqubits_total = nqubits if nqubits is not None else nqubits_samples
if not all(len(bitstring) == nqubits_samples for bitstring in bitstrings):
raise ValueError("Not all bitstring keys have the same length.")
if qubits_to_measure and len(qubits_to_measure) > nqubits_samples:
raise ValueError("Can't filter samples for more qubits than are present in the bitstrings.")
if nqubits_total != nqubits_samples and qubits_to_measure is None:
raise ValueError(
"Must provide qubits_to_measure if nqubits is different from the number of qubits in the samples."
)
expand_samples = expand_samples if expand_samples is not None else True
# Calculate probabilities
probabilities: dict[str, float] = {
bitstring: (counts / nshots if nshots and nshots > 0 else 0.0) for bitstring, counts in samples.items()
}
if qubits_to_measure is not None:
samples = cls._adjust_samples(samples, qubits_to_measure, nqubits_total, expand_samples=expand_samples)
return cls(samples=samples, probabilities=probabilities)
@classmethod
[documents]
def from_state(
cls,
sampling_readout: SamplingReadout,
state: QTensor,
qubits_to_measure: list[int] | None = None,
expand_samples: bool = True,
) -> Self:
f_string = "{:0" + str(state.nqubits) + "b}"
probabilities: dict[str, float] = {(f_string).format(i): p for i, p in enumerate(state.probabilities())}
samples: dict[str, int] = _samples_from_probabilities(probabilities, nshots=sampling_readout.nshots)
if qubits_to_measure is not None:
samples = cls._adjust_samples(samples, qubits_to_measure, state.nqubits, expand_samples=expand_samples)
return cls(samples=samples, probabilities=probabilities)
def __init__(self, samples: dict[str, int], probabilities: dict[str, float] | None = None) -> None:
if samples is None:
raise ValueError("Can't construct the Sampling results if samples are not provided.")
self._samples: dict[str, int] = samples or {}
self._probabilities: dict[str, float] = probabilities or {}
@property
[documents]
def samples(self) -> dict[str, int]:
"""dict[str, int]: Mapping of measured bitstring to count."""
return self._samples
@property
[documents]
def probabilities(self) -> dict[str, float]:
"""Estimated probability distribution over bitstrings.
Returns:
dict[str, float]: Mapping of bitstring to probability.
"""
return self._probabilities
[documents]
def get_probability(self, bitstring: str) -> float:
"""Return the probability for a single bitstring.
Args:
bitstring (str): The bitstring to look up (e.g. ``"010"``).
Returns:
float: The probability associated with ``bitstring``, or ``0.0``
if it was not observed.
"""
return self._probabilities.get(bitstring, 0.0)
[documents]
def get_probabilities(self, n: int | None = None) -> list[tuple[str, float]]:
"""Return the most probable bitstrings in descending order.
Args:
n (int | None): Maximum number of entries to return. ``None``
(the default) returns all outcomes.
Returns:
list[tuple[str, float]]: Up to ``n`` ``(bitstring, probability)``
pairs sorted by probability in descending order.
"""
if n is None:
n = len(self._probabilities)
return heapq.nlargest(n, self._probabilities.items(), key=operator.itemgetter(1))
def __repr__(self) -> str:
return f"Sampling Results: (\n\tnshots={sum(self.samples.values())},\n\tsamples={pformat(self.samples)}\n)\n\n"
__str__ = __repr__
@yaml.register_class
[documents]
class ExpectationReadoutResult(ReadoutResult[ExpectationReadout]):
"""Result produced by an :class:`~qilisdk.readout.ExpectationReadout`.
Contains the computed expectation values for each observable specified in
the readout configuration. The object can be constructed in two ways:
1. From pre-computed ``expectation_values``.
2. From a quantum ``state``, in which case the expectation values are derived via :func:`~qilisdk.core.expect_val`.
Args:
readout (ExpectationReadout): The readout configuration that produced this result.
expectation_values (list[Number] | None): Pre-computed expectation values, one per observable.
Mutually exclusive with ``state``.
state (QTensor | None): Quantum state used to compute expectation values on-the-fly. Mutually exclusive with
``expectation_values``.
Raises:
ValueError: If neither ``expectation_values`` nor ``state`` is provided.
"""
@classmethod
[documents]
def from_expectations(cls, expectation_values: list[float], nshots: int | None = None) -> Self:
return cls(expectation_values=expectation_values, nshots=nshots)
@classmethod
[documents]
def from_state(cls, expectation_readout: ExpectationReadout, state: QTensor) -> Self:
try:
expectation_values: list[int | float] = [
_assert_real((expect_val(o, state)))
for o in expectation_readout.expanded_observables(nqubits=state.nqubits)
]
except ValueError:
raise ValueError(
"Encountered an imaginary expectation value while computing the expectation values, try reducing the total tolerance or improving simulation precision."
)
return cls(expectation_values=expectation_values, nshots=expectation_readout.nshots)
def __init__(self, expectation_values: list[float], nshots: int | None = None) -> None:
if expectation_values is None:
raise ValueError("Can't initialize Expectation Readout if the expectation values are not provided.")
self._expectation_values: list[int | float] = expectation_values
self._nshots: int | None = nshots
@property
[documents]
def expectation_values(self) -> list[float]:
"""list[Number]: Expectation values, one per observable, in the same order as specified in the readout."""
return self._expectation_values
def __repr__(self) -> str:
return (
"Expectation Value Results: (\n"
+ (f"\tnshots = {self._nshots},\n" if self._nshots and self._nshots > 0 else "")
+ f"\texpectation_values={pformat(self._expectation_values)},\n"
+ ")\n\n"
)
__str__ = __repr__
@yaml.register_class
[documents]
class StateTomographyReadoutResult(ReadoutResult[StateTomographyReadout]):
"""Result produced by a :class:`~qilisdk.readout.StateTomographyReadout`.
Contains the full quantum state after execution and, optionally, the computational-basis probability distribution derived from it.
Args:
readout (StateTomographyReadout): The readout configuration that produced this result.
state (QTensor): The reconstructed quantum state (ket or density matrix).
"""
@classmethod
[documents]
def from_state(cls, state: QTensor) -> Self:
return cls(state=state)
def __init__(
self,
state: QTensor,
) -> None:
self._state: QTensor = state
@property
[documents]
def state(self) -> QTensor:
"""QTensor: The reconstructed quantum state (ket or density matrix)."""
return self._state
@property
[documents]
def probabilities(self) -> dict[str, float]:
"""Computational-basis probability distribution derived from the state.
Returns:
dict[str, float]: Mapping of bitstring to probability.
"""
f_string = "{:0" + str(self.state.nqubits) + "b}"
return {(f_string).format(i): p for i, p in enumerate(self.state.probabilities())}
[documents]
def get_probability(self, bitstring: str) -> float:
"""Return the probability for a single bitstring.
Args:
bitstring (str): The bitstring to look up (e.g. ``"010"``).
Returns:
float: The probability associated with ``bitstring``, or ``0.0``
if it was not observed.
"""
return self.probabilities.get(bitstring, 0.0)
[documents]
def get_probabilities(self, n: int | None = None) -> list[tuple[str, float]]:
"""Return the most probable bitstrings in descending order.
Args:
n (int | None): Maximum number of entries to return. ``None`` (the default) returns all outcomes.
Returns:
list[tuple[str, float]]: Up to ``n`` ``(bitstring, probability)`` pairs sorted by probability in descending order.
"""
probs = self.probabilities
if n is None:
n = len(probs)
return heapq.nlargest(n, probs.items(), key=operator.itemgetter(1))
def __repr__(self) -> str:
return "State Tomography Results: (\n" + (f"\tfinal_state={pformat(self.state)}\n") + ")\n\n"
__str__ = __repr__
# ---------------------------------------------------------------------------
# Type variables for generic readout result containers
#
# Defined AFTER the concrete result classes so the constraints reference
# real types, not string literals. TypeVar arguments are evaluated at
# runtime (they are function parameters, not annotations), so forward
# references via strings would silently pass as literal strings.
# ---------------------------------------------------------------------------
#: Type variable tracking whether sampling results are present.
[documents]
S = TypeVar("S", SamplingReadoutResult, None)
#: Type variable tracking whether expectation-value results are present.
[documents]
E = TypeVar("E", ExpectationReadoutResult, None)
#: Type variable tracking whether state-tomography results are present.
[documents]
T = TypeVar("T", StateTomographyReadoutResult, None)
# ---------------------------------------------------------------------------
# ReadoutCompositeResults — generic aggregate container
# ---------------------------------------------------------------------------
class _HasSampling(Protocol):
sampling: SamplingReadoutResult
[documents]
def has_sampling(obj: ReadoutCompositeResults) -> TypeGuard[_HasSampling]:
"""Return ``True`` if the composite contains a sampling result."""
return obj.sampling is not None
class _HasExpectation(Protocol):
expectation_values: ExpectationReadoutResult
[documents]
def has_expectation_values(obj: ReadoutCompositeResults) -> TypeGuard[_HasExpectation]:
"""Return ``True`` if the composite contains an expectation-value result."""
return obj.expectation_values is not None
class _HasStateTomography(Protocol):
state_tomography: StateTomographyReadoutResult
[documents]
def has_state_tomography(obj: ReadoutCompositeResults) -> TypeGuard[_HasStateTomography]:
"""Return ``True`` if the composite contains a state-tomography result."""
return obj.state_tomography is not None
@yaml.register_class
@dataclass(frozen=True)
[documents]
class ReadoutCompositeResults(Result, Generic[S, E, T]):
"""Aggregated container for readout results from a single execution step.
This class is returned internally by the backend and is accessible via
:attr:`FunctionalResult.readout_results <qilisdk.functionals.FunctionalResult.readout_results>`.
Most users should access results through the convenience properties on
:class:`~qilisdk.functionals.FunctionalResult` instead.
The three type parameters ``S``, ``E``, ``T`` encode at the type level which readout results are present
and correspond to the readout types declared in the :class:`~qilisdk.readout.Readout`:
* ``S`` is :class:`SamplingReadoutResult` or ``None``
* ``E`` is :class:`ExpectationReadoutResult` or ``None``
* ``T`` is :class:`StateTomographyReadoutResult` or ``None``
When the concrete type parameter is the result class (not ``None``), the corresponding field is
guaranteed to be populated and the type checker can verify access without runtime guards.
"""
[documents]
sampling: S = None # type: ignore[assignment] # ty:ignore[invalid-assignment]
[documents]
expectation_values: E = None # type: ignore[assignment] # ty:ignore[invalid-assignment]
[documents]
state_tomography: T = None # type: ignore[assignment] # ty:ignore[invalid-assignment]
@classmethod
[documents]
def from_list(cls, results: list) -> ReadoutCompositeResults:
"""Construct a :class:`ReadoutCompositeResults` from a flat list of readout results.
Each element is matched by type to the appropriate field.
Args:
results (list): A list containing any combination of
:class:`SamplingReadoutResult`, :class:`ExpectationReadoutResult`,
and :class:`StateTomographyReadoutResult` instances.
Returns:
ReadoutCompositeResults: The constructed result container.
"""
sampling = None
expectation_values = None
state_tomography = None
for result in results:
if isinstance(result, SamplingReadoutResult):
sampling = result
elif isinstance(result, ExpectationReadoutResult):
expectation_values = result
elif isinstance(result, StateTomographyReadoutResult):
state_tomography = result
return cls(sampling=sampling, expectation_values=expectation_values, state_tomography=state_tomography)
@classmethod
[documents]
def from_dict(cls, data: dict) -> ReadoutCompositeResults:
"""Construct a :class:`ReadoutCompositeResults` from a plain dictionary.
Args:
data (dict): Mapping with optional keys ``"sampling"``,
``"expectation_values"``, and ``"state_tomography"``.
Returns:
ReadoutCompositeResults: The constructed result container.
Raises:
TypeError: If any provided value has the wrong type.
"""
sampling = data.get("sampling")
expectation_values = data.get("expectation_values")
state_tomography = data.get("state_tomography")
if sampling is not None and not isinstance(sampling, SamplingReadoutResult):
raise TypeError("sampling must be SamplingReadoutResult or None")
if expectation_values is not None and not isinstance(expectation_values, ExpectationReadoutResult):
raise TypeError("expectation_values must be ExpectationReadoutResult or None")
if state_tomography is not None and not isinstance(state_tomography, StateTomographyReadoutResult):
raise TypeError("state_tomography must be StateTomographyReadoutResult or None")
return cls(sampling=sampling, expectation_values=expectation_values, state_tomography=state_tomography)
@overload
[documents]
def get_samples(self: ReadoutCompositeResults[SamplingReadoutResult, Any, Any]) -> dict[str, int]: ...
@overload
def get_samples(self: ReadoutCompositeResults[None, Any, Any]) -> Never: ...
def get_samples(self) -> dict[str, int]:
"""Return the sampling results as a bitstring-to-count mapping.
Returns:
dict[str, int]: Mapping of measured bitstring to count.
Raises:
ValueError: If no sampling results are present.
"""
if self.sampling is not None:
return self.sampling.samples
raise ValueError("Can't find samples because no SamplingReadoutResult is present in this composite.")
@overload
[documents]
def get_probabilities(
self: ReadoutCompositeResults[SamplingReadoutResult | None, Any, StateTomographyReadoutResult | None],
) -> dict[str, float]: ...
@overload
def get_probabilities(self: ReadoutCompositeResults[None, Any, None]) -> Never: ...
def get_probabilities(self) -> dict[str, float]:
"""Return the probability distribution as a bitstring-to-probability mapping.
Returns:
dict[str, float]: Mapping of bitstring to probability.
Raises:
ValueError: If no sampling or state-tomography results are present.
"""
if self.sampling is not None:
return self.sampling.probabilities
if self.state_tomography is not None:
return self.state_tomography.probabilities
raise ValueError(
"Can't find probabilities because neither SamplingReadoutResult nor StateTomographyReadoutResult is present in this composite."
)
@overload
[documents]
def get_expectation_values(self: ReadoutCompositeResults[Any, ExpectationReadoutResult, Any]) -> list[float]: ...
@overload
def get_expectation_values(self: ReadoutCompositeResults[Any, None, Any]) -> Never: ...
def get_expectation_values(self) -> list[float]:
"""Return the expectation values as a list of numbers.
Returns:
list[float]: Expectation values, one per observable, in the same order as specified in the readout.
Raises:
ValueError: If no expectation-value results are present.
"""
if self.expectation_values is not None:
return self.expectation_values.expectation_values
raise ValueError(
"Can't find expectation values because no ExpectationReadoutResult is present in this composite."
)
@overload
[documents]
def get_state(self: ReadoutCompositeResults[Any, Any, StateTomographyReadoutResult]) -> QTensor: ...
@overload
def get_state(self: ReadoutCompositeResults[Any, Any, None]) -> Never: ...
def get_state(self) -> QTensor:
"""Return the reconstructed quantum state.
Returns:
QTensor: The reconstructed quantum state (ket or density matrix).
Raises:
ValueError: If no state-tomography results are present.
"""
if self.state_tomography is not None:
return self.state_tomography.state
raise ValueError("Can't find the state because no StateTomographyReadoutResult is present in this composite.")
def __repr__(self) -> str:
out = ""
if self.sampling:
out += str(self.sampling)
if self.expectation_values:
out += str(self.expectation_values)
if self.state_tomography:
out += str(self.state_tomography)
return out or f"{type(self).__name__}(empty)"
# ---------------------------------------------------------------------------
# Utility functions
# ---------------------------------------------------------------------------
def _samples_from_state(state: QTensor, nshots: int = 100, seed: int | None = None) -> dict[str, int]:
f_string = "{:0" + str(state.nqubits) + "b}"
probabilities: dict[str, int | float] = {(f_string).format(i): p for i, p in enumerate(state.probabilities())}
return _samples_from_probabilities(probabilities=probabilities, nshots=nshots, seed=seed)
def _samples_from_probabilities(
probabilities: dict[str, float], nshots: int = 100, seed: int | None = None
) -> dict[str, int]:
states = np.array(list(probabilities.keys()))
probs = np.array(list(probabilities.values()), dtype=np.float64)
if not np.isclose(probs.sum(), 1):
logger.warning("Renormalizing probabilities obtained as they don't sum up to 1.")
probs /= probs.sum()
rng = np.random.default_rng(seed)
draws = rng.choice(len(states), size=nshots, p=probs)
counts = np.bincount(draws, minlength=len(states))
return {str(states[i]): int(counts[i]) for i in range(len(states)) if counts[i] > 0}
