# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# 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
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Different data analysis steps."""
from abc import abstractmethod
from abc import ABC
from enum import Enum
from numbers import Number
from typing import List, Union, Sequence, Set
from collections import defaultdict
import numpy as np
from uncertainties import unumpy as unp, ufloat
from qiskit.result.postprocess import format_counts_memory
from qiskit_experiments.data_processing.data_action import DataAction, TrainableDataAction
from qiskit_experiments.data_processing.exceptions import DataProcessorError
from qiskit_experiments.data_processing.discriminator import BaseDiscriminator
from qiskit_experiments.framework import Options
[docs]
class AverageData(DataAction):
"""A node to average data representable as numpy arrays."""
def __init__(self, axis: int, validate: bool = True):
r"""Initialize a data averaging node.
Args:
axis: The axis along which to average.
validate: If set to False the DataAction will not validate its input.
Notes:
If the input array has no standard error, then this node will compute the
standard error of the mean, i.e. the standard deviation of the datum divided by
:math:`\sqrt{N}` where :math:`N` is the number of data points.
Otherwise the standard error is given by the square root of :math:`N^{-1}` times
the sum of the squared errors.
"""
super().__init__(validate)
self._axis = axis
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Format the data into numpy arrays.
Args:
data: A data array to format. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The data that has been validated and formatted.
Raises:
DataProcessorError: When the specified axis does not exist in given array.
"""
if self._validate:
if len(data.shape) <= self._axis:
raise DataProcessorError(
f"Cannot average the {len(data.shape)} dimensional "
f"array along axis {self._axis}."
)
return data
def _process(self, data: np.ndarray) -> np.ndarray:
"""Average the data.
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
Arrays with one less dimension than the given data.
"""
ax = self._axis
reduced_array = np.mean(data, axis=ax)
nominals = unp.nominal_values(reduced_array)
with np.errstate(invalid="ignore"):
# Setting std_devs to NaN will trigger floating point exceptions
# which we can ignore. See https://stackoverflow.com/q/75656026
errors = unp.std_devs(reduced_array)
if np.any(np.isnan(errors)):
# replace empty elements with SEM
sem = np.std(unp.nominal_values(data), axis=ax) / np.sqrt(data.shape[ax])
errors = np.where(np.isnan(errors), sem, errors)
return unp.uarray(nominals, errors)
def __repr__(self):
"""String representation of the node."""
return f"{self.__class__.__name__}(validate={self._validate}, axis={self._axis})"
[docs]
class MinMaxNormalize(DataAction):
"""Normalizes the data."""
def _process(self, data: np.ndarray) -> np.ndarray:
"""Normalize the data to the interval [0, 1].
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The normalized data.
Notes:
This doesn't consider the uncertainties of the minimum or maximum values.
Input data array is just scaled by the data range.
"""
nominals = unp.nominal_values(data)
min_y, max_y = np.min(nominals), np.max(nominals)
return (data - min_y) / (max_y - min_y)
[docs]
class SVD(TrainableDataAction):
"""Singular Value Decomposition of averaged IQ data."""
def __init__(self, validate: bool = True):
"""Create new action.
Args:
validate: If set to False the DataAction will not validate its input.
"""
super().__init__(validate=validate)
self._n_circs = 0
self._n_shots = 0
self._n_slots = 0
self._n_iq = 0
@classmethod
def _default_parameters(cls) -> Options:
"""Default parameters.
Parameters are defined for each qubit in the data and thus
represented as an array-like.
Trainable parameters:
main_axes: A unit vector representing the main axis in IQ plane.
i_means: Mean value of training data along I quadrature.
q_means: Mean value of training data along Q quadrature.
scales: Scaling of IQ signal.
"""
params = super()._default_parameters()
params.main_axes = None
params.i_means = None
params.q_means = None
params.scales = None
return params
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Check that the IQ data is 2D and convert it to a numpy array.
Args:
data: A data array to format. This is a single numpy array containing
all circuit results input to the data processor.
This data has different dimensions depending on whether
single-shot or averaged data is being processed.
Single-shot data is four dimensional, i.e., ``[#circuits, #shots, #slots, 2]``,
while averaged IQ data is three dimensional, i.e., ``[#circuits, #slots, 2]``.
Here, ``#slots`` is the number of classical registers used in the circuit.
Returns:
data and any error estimate as a numpy array.
Raises:
DataProcessorError: If the datum does not have the correct format.
"""
self._n_circs = 0
self._n_shots = 0
self._n_slots = 0
self._n_iq = 0
# identify shape
try:
# level1 single-shot data
self._n_circs, self._n_shots, self._n_slots, self._n_iq = data.shape
except ValueError:
try:
# level1 data averaged over shots
self._n_circs, self._n_slots, self._n_iq = data.shape
except ValueError as ex:
raise DataProcessorError(
f"Data given to {self.__class__.__name__} is not likely level1 data."
) from ex
if self._validate:
if self._n_iq != 2:
raise DataProcessorError(
f"IQ data given to {self.__class__.__name__} does not have two-dimensions "
f"(I and Q). Instead, {self._n_iq} dimensions were found."
)
return data
def _process(self, data: np.ndarray) -> np.ndarray:
"""Project the IQ data onto the axis defined by an SVD and scale it.
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
A Tuple of 1D arrays of the result of the SVD and the associated error. Each entry
is the real part of the averaged IQ data of a qubit. The data has the shape
n_circuits x n_slots for averaged data and n_circuits x n_shots x n_slots for
single-shot data.
Raises:
DataProcessorError: If the SVD has not been previously trained on data.
"""
if not self.is_trained:
raise DataProcessorError("SVD must be trained on data before it can be used.")
# IQ axis is reduced by projection
if self._n_shots == 0:
# level1 average mode
dims = self._n_circs, self._n_slots
else:
# level1 single mode
dims = self._n_circs, self._n_shots, self._n_slots
projected_data = np.zeros(dims, dtype=object)
for idx in range(self._n_slots):
scale = self.parameters.scales[idx]
axis = self.parameters.main_axes[idx]
mean_i = self.parameters.i_means[idx]
mean_q = self.parameters.q_means[idx]
if self._n_shots != 0:
# Single shot
for circ_idx in range(self._n_circs):
centered = [
data[circ_idx, :, idx, 0] - mean_i,
data[circ_idx, :, idx, 1] - mean_q,
]
projected_data[circ_idx, :, idx] = axis @ np.array(centered) / scale
else:
# Averaged
centered = [data[:, idx, 0] - mean_i, data[:, idx, 1] - mean_q]
projected_data[:, idx] = axis @ np.array(centered) / scale
return projected_data
[docs]
def train(self, data: np.ndarray):
"""Train the SVD on the given data.
Each element of the given data will be converted to a 2D array of dimension
n_qubits x 2. The number of qubits is inferred from the shape of the data.
For each qubit the data is collected into an array of shape 2 x n_data_points.
The mean of the in-phase a quadratures is subtracted before passing the data
to numpy's svd function. The dominant axis and the scale is saved for each
qubit so that future data points can be projected onto the axis.
.. note::
Before training the SVD the mean of the training data is subtracted from the
training data to avoid large offsets in the data.
These means can be retrieved with :attr:`.parameters.i_means` or
:attr:`parameters.q_means` for I and Q quadrature, respectively.
Args:
data: A data array to be trained. This is a single numpy array containing
all circuit results input to the data processor.
"""
if data is None:
return
# TODO do not remove standard error. Currently svd is not supported.
data = unp.nominal_values(self._format_data(data))
main_axes = []
scales = []
i_means = []
q_means = []
for idx in range(self._n_slots):
datums = np.vstack([datum[idx] for datum in data]).T
# Calculate the mean of the data to recenter it in the IQ plane.
mean_i = np.average(datums[0, :])
mean_q = np.average(datums[1, :])
i_means.append(mean_i)
q_means.append(mean_q)
datums[0, :] = datums[0, :] - mean_i
datums[1, :] = datums[1, :] - mean_q
mat_u, mat_s, _ = np.linalg.svd(datums)
# There is an arbitrary sign in the direction of the matrix which we fix to
# positive to make the SVD node more reliable in tests and real settings.
main_axes.append(np.sign(mat_u[0, 0]) * mat_u[:, 0])
scales.append(mat_s[0])
self.set_parameters(
main_axes=main_axes,
scales=scales,
i_means=i_means,
q_means=q_means,
)
class IQPart(DataAction):
"""Abstract class for IQ data post-processing."""
def __init__(self, scale: float = 1.0, validate: bool = True):
"""
Args:
scale: Float with which to multiply the IQ data. Defaults to 1.0.
validate: If set to False the DataAction will not validate its input.
"""
self.scale = scale
super().__init__(validate)
self._n_circs = 0
self._n_shots = 0
self._n_slots = 0
self._n_iq = 0
@abstractmethod
def _process(self, data: np.ndarray) -> np.ndarray:
"""Defines how the IQ point is processed.
The last dimension of the array should correspond to [real, imaginary] part of data.
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The data that has been processed.
"""
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Format and validate the input.
Args:
data: A data array to format. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The data that has been validated and formatted.
Raises:
DataProcessorError: When input data is not likely IQ data.
"""
self._n_shots = 0
# identify shape
try:
# level1 single-shot data
self._n_circs, self._n_shots, self._n_slots, self._n_iq = data.shape
except ValueError:
try:
# level1 data averaged over shots
self._n_circs, self._n_slots, self._n_iq = data.shape
except ValueError as ex:
raise DataProcessorError(
f"Data given to {self.__class__.__name__} is not likely level1 data."
) from ex
if self._validate:
if data.shape[-1] != 2:
raise DataProcessorError(
f"IQ data given to {self.__class__.__name__} must be a multi-dimensional array"
"of dimension [d0, d1, ..., 2] in which the last dimension "
"corresponds to IQ elements."
f"Input data contains element with length {data.shape[-1]} != 2."
)
return data
def __repr__(self):
"""String representation of the node."""
return f"{self.__class__.__name__}(validate={self._validate}, scale={self.scale})"
[docs]
class ToReal(IQPart):
"""IQ data post-processing. Isolate the real part of single-shot IQ data."""
def _process(self, data: np.ndarray) -> np.ndarray:
"""Take the real part of the IQ data.
Args:
data: An N-dimensional array of complex IQ point as [real, imaginary].
Returns:
A N-1 dimensional array, each entry is the real part of the given IQ data.
"""
return data[..., 0] * self.scale
[docs]
class ToImag(IQPart):
"""IQ data post-processing. Isolate the imaginary part of single-shot IQ data."""
def _process(self, data: np.ndarray) -> np.ndarray:
"""Take the imaginary part of the IQ data.
Args:
data: An N-dimensional array of complex IQ point as [real, imaginary].
Returns:
A N-1 dimensional array, each entry is the imaginary part of the given IQ data.
"""
return data[..., 1] * self.scale
class ToAbs(IQPart):
"""IQ data post-processing. Take the absolute value of the IQ point."""
def _process(self, data: np.array) -> np.array:
"""Take the absolute value of the IQ data.
Args:
data: An N-dimensional array of complex IQ point as [real, imaginary].
Returns:
A N-1 dimensional array, each entry is the absolute value of the given IQ data.
"""
# pylint: disable=no-member
return unp.sqrt(data[..., 0] ** 2 + data[..., 1] ** 2) * self.scale
[docs]
class DiscriminatorNode(DataAction):
"""A class to discriminate kerneled data, e.g., IQ data, to produce counts.
This node integrates into the data processing chain a serializable
:class:`.BaseDiscriminator` subclass instance which must have a
:meth:`~.BaseDiscriminator.predict` method that takes as input a list of lists and
returns a list of labels. Crucially, this node can be initialized with a single
discriminator which applies to each memory slot or it can be initialized with a list
of discriminators, i.e., one for each slot.
.. note::
Future versions may see this class become a sub-class of :class:`.TrainableDataAction`.
.. note::
This node will drop uncertainty from unclassified nodes.
Returned labels don't have uncertainty.
"""
def __init__(
self,
discriminators: Union[BaseDiscriminator, List[BaseDiscriminator]],
validate: bool = True,
):
"""Initialize the node with an object that can discriminate.
Args:
discriminators: The entity that will perform the discrimination. This needs to
be a :class:`.BaseDiscriminator` or a list thereof that takes
as input a list of lists and returns a list of labels. If a list of
discriminators is given then there should be as many discriminators as there
will be slots in the memory. The discriminator at the i-th index will be applied
to the i-th memory slot.
validate: If set to False the DataAction will not validate its input.
"""
super().__init__(validate)
self._discriminator = discriminators
self._n_circs = 0
self._n_shots = 0
self._n_slots = 0
self._n_iq = 0
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Check that there are as many discriminators as there are slots."""
self._n_shots = 0
# identify shape
try:
# level1 single-shot data
self._n_circs, self._n_shots, self._n_slots, self._n_iq = data.shape
except ValueError as ex:
raise DataProcessorError(
f"The data given to {self.__class__.__name__} does not have the shape of "
"single-shot IQ data; expecting a 4D array."
) from ex
if self._validate:
if data.shape[-1] != 2:
raise DataProcessorError(
f"IQ data given to {self.__class__.__name__} must be a multi-dimensional array"
"of dimension [d0, d1, ..., 2] in which the last dimension "
"corresponds to IQ elements."
f"Input data contains element with length {data.shape[-1]} != 2."
)
if self._validate:
if isinstance(self._discriminator, list):
if self._n_slots != len(self._discriminator):
raise DataProcessorError(
f"The Discriminator node has {len(self._discriminator)} which does "
f"not match the {self._n_slots} slots in the data."
)
return unp.nominal_values(data)
def _process(self, data: np.ndarray) -> np.ndarray:
"""Discriminate the data.
Args:
data: The IQ data as a list of points to discriminate. This data should have
the shape dim_1 x dim_2 x ... x dim_k x 2.
Returns:
The discriminated data as a list of labels with shape dim_1 x ... x dim_k.
"""
# Case where one discriminator is applied to all the data.
if not isinstance(self._discriminator, list):
# Reshape the IQ data to an array of size n x 2
shape, data_length = data.shape, 1
for dim in shape[:-1]:
data_length *= dim
data = data.reshape((data_length, 2)) # the last dim is guaranteed by _process
# Classify the data using the discriminator and reshape it to dim_1 x ... x dim_k
classified = np.array(self._discriminator.predict(data)).reshape(shape[0:-1])
# case where a discriminator is applied to each slot.
else:
classified = np.empty((self._n_circs, self._n_shots, self._n_slots), dtype=str)
for idx, discriminator in enumerate(self._discriminator):
sub_data = data[:, :, idx, :].reshape((self._n_circs * self._n_shots, 2))
sub_classified = np.array(discriminator.predict(sub_data))
sub_classified = sub_classified.reshape((self._n_circs, self._n_shots))
classified[:, :, idx] = sub_classified
# Concatenate the bit-strings together.
labeled_data = []
for idx in range(self._n_circs):
labeled_data.append(
["".join(classified[idx, jdx, :][::-1]) for jdx in range(self._n_shots)]
)
return np.array(labeled_data).reshape((self._n_circs, self._n_shots))
[docs]
class MemoryToCounts(DataAction):
"""A data action that takes discriminated data and transforms it into a counts dict.
This node is intended to be used after the :class:`.DiscriminatorNode` node. It will convert
the classified memory into a list of count dictionaries wrapped in a numpy array.
"""
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Validate the input data."""
if self._validate:
if len(data.shape) <= 1:
raise DataProcessorError(
"The data should be an array with at least two dimensions."
)
return data
def _process(self, data: np.ndarray) -> np.ndarray:
"""
Args:
data: The classified data to format into a counts dictionary. The first dimension
is assumed to correspond to the different circuit executions.
Returns:
A list of dictionaries where each dict is a count dictionary over the labels of
the input data.
"""
all_counts = []
for datum in data:
counts = {}
for bit_string in set(datum):
counts[bit_string] = sum(datum == bit_string)
all_counts.append(counts)
return np.array(all_counts)
class CountsAction(DataAction):
"""An abstract DataAction that acts on count dictionaries."""
@abstractmethod
def _process(self, data: np.ndarray) -> np.ndarray:
"""Defines how the counts are processed."""
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""
Checks that the given data has a counts format.
Args:
data: A data array to format. This is a single numpy array containing
all circuit results input to the data processor.
This is usually an object data type containing Python dictionaries of
count data keyed on the measured bitstring.
A count value is a discrete quantity representing the frequency of an event.
Therefore, count values do not have an uncertainty.
Returns:
The ``data`` as given.
Raises:
DataProcessorError: If the data is not a counts dict or a list of counts dicts.
"""
valid_count_type = int, np.integer
if self._validate:
for datum in data:
if not isinstance(datum, dict):
raise DataProcessorError(
f"Data entry must be dictionary of counts, received {type(datum)}."
)
for bit_str, count in datum.items():
if not isinstance(bit_str, str):
raise DataProcessorError(
f"Key {bit_str} is not a valid count key in {self.__class__.__name__}."
)
if not isinstance(count, valid_count_type):
raise DataProcessorError(
f"Count {bit_str} is not a valid count for {self.__class__.__name__}. "
"The uncertainty of probability is computed based on sampling error, "
"thus the count should be an error-free discrete quantity "
"representing the frequency of event."
)
return data
[docs]
class MarginalizeCounts(CountsAction):
r"""A data action to marginalize count dictionaries.
This data action takes a count dictionary and returns a new count dictionary that
is marginalized over a set of specified qubits. For example, given the count
dictionary :code:`{"010": 1, "110": 10, "100": 100}` this node will return the
count dictionary :code:`{"10": 11, "00": 100}` when marginalized over qubit 2.
.. note::
This data action can be used to discard one or more qubits in the counts
dictionary. This is, for example, useful when processing two-qubit restless
experiments but can be used in a more general context. In composite
experiments the counts marginalization is already done in the data container.
"""
def __init__(self, qubits_to_keep: Set[int], validate: bool = True):
"""Initialize a counts marginalization node.
Args:
qubits_to_keep: A set of qubits to retain during the marginalization process.
validate: If set to False the DataAction will not validate its input.
"""
super().__init__(validate)
self._qubits_to_keep = sorted(qubits_to_keep, reverse=True)
def _process(self, data: np.ndarray) -> np.ndarray:
"""Perform counts marginalization."""
marginalized_counts = []
for datum in data:
new_counts = defaultdict(int)
for bit_str, count in datum.items():
new_counts["".join([bit_str[::-1][idx] for idx in self._qubits_to_keep])] += count
marginalized_counts.append(new_counts)
return np.array(marginalized_counts)
def __repr__(self):
"""String representation of the node."""
options_str = ", ".join(
[
f"qubits_to_keep={self._qubits_to_keep}",
f"validate={self._validate}",
]
)
return f"{self.__class__.__name__}({options_str})"
[docs]
class Probability(CountsAction):
r"""Compute the mean probability of a single measurement outcome from counts.
This node returns the mean and standard deviation of a single measurement
outcome probability :math:`p` estimated from the observed counts. The mean and
variance are computed from the posterior Beta distribution
:math:`B(\alpha_0^\prime,\alpha_1^\prime)` estimated from a Bayesian update
of a prior Beta distribution :math:`B(\alpha_0, \alpha_1)` given the observed
counts.
The mean and variance of the Beta distribution :math:`B(\alpha_0, \alpha_1)` are:
.. math::
\text{E}[p] = \frac{\alpha_0}{\alpha_0 + \alpha_1}, \quad
\text{Var}[p] = \frac{\text{E}[p] (1 - \text{E}[p])}{\alpha_0 + \alpha_1 + 1}
Given a prior Beta distribution :math:`B(\alpha_0, \alpha_1)`, the posterior
distribution for the observation of :math:`F` counts of a given
outcome out of :math:`N` total shots is a
:math:`B(\alpha_0^\prime,\alpha_1^\prime):math:` with
.. math::
\alpha_0^\prime = \alpha_0 + F, \quad
\alpha_1^\prime = \alpha_1 + N - F.
.. note::
The default value for the prior distribution is *Jeffery's Prior*
:math:`\alpha_0 = \alpha_1 = 0.5` which represents ignorance about the true
probability value. Note that for this prior the mean probability estimate
from a finite number of counts can never be exactly 0 or 1. The estimated
mean and variance are given by
.. math::
\text{E}[p] = \frac{F + 0.5}{N + 1}, \quad
\text{Var}[p] = \frac{\text{E}[p] (1 - \text{E}[p])}{N + 2}
This node will deprecate standard error provided by the previous node.
"""
def __init__(
self,
outcome: str,
alpha_prior: Union[float, Sequence[float]] = 0.5,
validate: bool = True,
):
"""Initialize a counts to probability data conversion.
Args:
outcome: The bitstring for which to return the probability and variance.
alpha_prior: A prior Beta distribution parameter ``[alpha0, alpha1]``.
If specified as float this will use the same value for
``alpha0`` and ``alpha1`` (Default: 0.5).
validate: If set to False the DataAction will not validate its input.
Raises:
DataProcessorError: When the dimension of the prior and expected parameter vector
do not match.
"""
self._outcome = outcome
if isinstance(alpha_prior, Number):
self._alpha_prior = [alpha_prior, alpha_prior]
else:
if validate and len(alpha_prior) != 2:
raise DataProcessorError(
"Prior for probability node must be a float or pair of floats."
)
self._alpha_prior = list(alpha_prior)
super().__init__(validate)
def _process(self, data: np.ndarray) -> np.ndarray:
"""Compute mean and standard error from the beta distribution.
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
This is usually an object data type containing Python dictionaries of
count data keyed on the measured bitstring.
Returns:
The data that has been processed.
"""
probabilities = np.empty(data.size, dtype=object)
for idx, counts_dict in enumerate(data):
shots = sum(counts_dict.values())
freq = counts_dict.get(self._outcome, 0)
alpha_posterior = [freq + self._alpha_prior[0], shots - freq + self._alpha_prior[1]]
alpha_sum = sum(alpha_posterior)
p_mean = alpha_posterior[0] / alpha_sum
p_var = p_mean * (1 - p_mean) / (alpha_sum + 1)
with np.errstate(invalid="ignore"):
# Setting std_devs to NaN will trigger floating point exceptions
# which we can ignore. See https://stackoverflow.com/q/75656026
probabilities[idx] = ufloat(nominal_value=p_mean, std_dev=np.sqrt(p_var))
return probabilities
def __repr__(self):
"""String representation of the node."""
options_str = ", ".join(
[
f"validate={self._validate}",
f"outcome={self._outcome}",
f"alpha_prior={self._alpha_prior}",
]
)
return f"{self.__class__.__name__}({options_str})"
[docs]
class BasisExpectationValue(DataAction):
"""Compute expectation value of measured basis from probability.
Note:
The sign becomes P(0) -> 1, P(1) -> -1.
"""
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Format and validate the input.
Args:
data: A data array to format. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The data that has been validated and formatted.
Raises:
DataProcessorError: When input value is not in [0, 1]
"""
if self._validate:
if not all(0.0 <= p <= 1.0 for p in data):
raise DataProcessorError(
f"Input data for node {self.__class__.__name__} is not likely probability."
)
return data
def _process(self, data: np.ndarray) -> np.ndarray:
"""Compute basis eigenvalue.
Args:
data: A data array to process. This is a single numpy array containing
all circuit results input to the data processor.
Returns:
The data that has been processed.
"""
return 2 * (0.5 - data)
class ProjectorType(Enum):
"""Types of projectors for data dimensionality reduction."""
SVD = SVD
ABS = ToAbs
REAL = ToReal
IMAG = ToImag
[docs]
class ShotOrder(Enum):
"""Shot order allowed values.
Generally, there are two possible modes in which a backend measures m
circuits with n shots:
- In the "circuit_first" mode, the backend subsequently first measures
all m circuits and then repeats this n times.
- In the "shot_first" mode, the backend first measures the 1st circuit
n times, then the 2nd circuit n times, and it proceeds with the remaining
circuits in the same way until it measures the m-th circuit n times.
The current default mode of IBM Quantum devices is "circuit_first".
"""
# pylint: disable=invalid-name
circuit_first = "c"
shot_first = "s"
[docs]
class RestlessNode(DataAction, ABC):
"""An abstract node for restless data processing nodes.
In restless measurements, the qubit is not reset after each measurement. Instead, the
outcome of the previous quantum non-demolition measurement is the initial state for the
current circuit. Restless measurements therefore require special data processing nodes
that are implemented as sub-classes of ``RestlessNode``. Restless experiments provide a
fast alternative for several calibration and characterization tasks, for details
see https://arxiv.org/pdf/2202.06981.pdf.
This node takes as input an array of arrays (2d array) where the sub-arrays are
the memories of each measured circuit. The sub-arrays therefore have a length
given by the number of shots. This data is reordered into a one dimensional array where
the element at index j was the jth measured shot. This node assumes by default that
a list of circuits :code:`[circ_1, cric_2, ..., circ_m]` is measured :code:`n_shots`
times according to the following order:
.. parsed-literal::
[
circuit 1 - shot 1,
circuit 2 - shot 1,
...
circuit m - shot 1,
circuit 1 - shot 2,
circuit 2 - shot 2,
...
circuit m - shot 2,
circuit 1 - shot 3,
...
circuit m - shot n,
]
Once the shots have been ordered in this fashion the data can be post-processed.
"""
def __init__(
self, validate: bool = True, memory_allocation: ShotOrder = ShotOrder.circuit_first
):
"""Initialize a restless node.
Args:
validate: If set to True the node will validate its input.
memory_allocation: If set to "c" the node assumes that the backend
subsequently first measures all circuits and then repeats this
n times, where n is the total number of shots. The default value
is "c". If set to "s" it is assumed that the backend subsequently
measures each circuit n times.
"""
super().__init__(validate)
self._n_shots = None
self._n_circuits = None
self._memory_allocation = memory_allocation
def _format_data(self, data: np.ndarray) -> np.ndarray:
"""Convert the data to an array.
This node will also set all the attributes needed to process the data such as
the number of shots and the number of circuits.
Args:
data: An array representing the memory.
Returns:
The data that has been processed.
Raises:
DataProcessorError: If the datum has the wrong shape.
"""
self._n_shots = len(data[0])
self._n_circuits = len(data)
if self._validate:
if data.shape[:2] != (self._n_circuits, self._n_shots):
raise DataProcessorError(
f"The datum given to {self.__class__.__name__} does not convert "
"of an array with dimension (number of circuit, number of shots)."
)
return data
def _reorder(self, unordered_data: np.ndarray) -> np.ndarray:
"""Reorder the measured data according to the measurement sequence.
Here, by default, it is assumed that the inner loop of the measurement
is done over the circuits and the outer loop is done over the shots.
The returned data is a one-dimensional array of time-ordered shots.
"""
if unordered_data is None:
return unordered_data
if self._memory_allocation == ShotOrder.circuit_first:
return unordered_data.T.flatten()
else:
return unordered_data.flatten()
[docs]
class RestlessToCounts(RestlessNode):
"""Post-process restless data and convert restless memory to counts.
This node first orders the measured restless data according to the measurement
sequence and then compares each bit in a shot with its value in the previous shot.
If they are the same then the bit corresponds to a 0, i.e. no state change, and if
they are different then the bit corresponds to a 1, i.e. there was a state change.
"""
def __init__(self, num_qubits: int, validate: bool = True):
"""
Args:
num_qubits: The number of qubits which is needed to construct the header needed
by :code:`qiskit.result.postprocess.format_counts_memory` to convert the memory
into a bit-string of counts.
validate: If set to False the DataAction will not validate its input.
"""
super().__init__(validate)
self._num_qubits = num_qubits
def _process(self, data: np.ndarray) -> np.ndarray:
"""Reorder the shots and assign values to them based on the previous outcome.
Args:
data: An array representing the memory.
Returns:
A counts dictionary processed according to the restless methodology.
"""
# Step 1. Reorder the data.
memory = self._reorder(data)
# Step 2. Do the restless classification into counts.
counts = [defaultdict(int) for _ in range(self._n_circuits)]
prev_shot = "0" * self._num_qubits
header = {"memory_slots": self._num_qubits}
for idx, shot in enumerate(memory):
shot = format_counts_memory(shot, header)
restless_adjusted_shot = RestlessToCounts._restless_classify(shot, prev_shot)
circuit_idx = idx % self._n_circuits
counts[circuit_idx][restless_adjusted_shot] += 1
prev_shot = shot
return np.array([dict(counts_dict) for counts_dict in counts])
@staticmethod
def _restless_classify(shot: str, prev_shot: str) -> str:
"""Adjust the measured shot based on the previous shot.
Each bit in shot is compared to its value in the previous shot. If both are equal
the restless adjusted bit is 0 (no state change) otherwise it is 1 (the
qubit changed state). This corresponds to taking the exclusive OR operation
between each bit and its previous outcome.
Args:
shot: A measured shot as a binary string, e.g. "0110100".
prev_shot: The shot that was measured in the previous circuit.
Returns:
The restless adjusted string computed by comparing the shot with the previous shot.
"""
restless_adjusted_bits = []
for idx, bit in enumerate(shot):
restless_adjusted_bits.append("0" if bit == prev_shot[idx] else "1")
return "".join(restless_adjusted_bits)
[docs]
class RestlessToIQ(RestlessNode):
"""Post-process restless data and convert restless memory to IQ data.
This node first orders the measured restless IQ point (measurement level 1) data
according to the measurement sequence and then subtracts an IQ point from the previous
one, i.e. :math:`(I_2 - I_1) + i(Q_2 - Q_1)` for consecutively measured IQ points
:math:`I_1 + iQ_1` and :math:`I_2 + iQ_2`. Following this, it takes the absolute
value of the in-phase and quadrature component and returns a sequence of circuit-
ordered IQ values, e.g. containing :math:`|I_2 - I_1| + i|Q_2 - Q_1|`.
This procedure is based on M. Werninghaus, et al., PRX Quantum 2, 020324 (2021).
"""
def __init__(self, validate: bool = True):
"""
Args:
validate: If set to False the DataAction will not validate its input.
"""
super().__init__(validate)
def _process(self, data: np.ndarray) -> np.ndarray:
"""Reorder the IQ shots and assign values to them based on the previous outcome.
Args:
data: An array representing the memory.
Returns:
An array of arrays of IQ shots processed according to the restless methodology.
"""
# Step 1. Reorder the data.
memory = self._reorder_iq(data)
# Step 2. Subtract and take absolute value of consecutive IQ points in
# the reordered memory.
post_processed_memory = np.abs(np.diff(memory, axis=0))
# The first element of the post-processed data is the first element
# of the reordered memory from step 1.
post_processed_memory = np.insert(post_processed_memory, 0, memory[0], axis=0)
# Step 3. Order post-processed IQ points by circuit.
iq_memory = [[] for _ in range(self._n_circuits)]
for idx, iq_point in enumerate(post_processed_memory):
iq_memory[idx % self._n_circuits].append(iq_point)
return np.array(iq_memory)
def _reorder_iq(self, unordered_data: np.ndarray) -> np.ndarray:
"""Reorder IQ data according to the measurement sequence."""
if unordered_data is None:
return unordered_data
ordered_data = [None] * self._n_shots * self._n_circuits
count = 0
if self._memory_allocation == ShotOrder.circuit_first:
for shot_idx in range(self._n_shots):
for circ_idx in range(self._n_circuits):
ordered_data[count] = unordered_data[circ_idx][shot_idx]
count += 1
else:
for circ_idx in range(self._n_circuits):
for shot_idx in range(self._n_shots):
ordered_data[count] = unordered_data[circ_idx][shot_idx]
count += 1
return np.array(ordered_data)