Restless measurements data processing nodes

qiskit_experiments/data_processing/nodes.py

@@ -13,13 +13,16 @@
 """Different data analysis steps."""
 
 from abc import abstractmethod
+from abc import ABC
 from enum import Enum
 from numbers import Number
 from typing import Union, Sequence
+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.framework import Options
@@ -596,3 +599,172 @@ class ProjectorType(Enum):
     ABS = ToAbs
     REAL = ToReal
     IMAG = ToImag
+
+
+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, circuits_first: bool = True):
+        """Initialize a restless node.
+
+        Args:
+            validate: If set to True the node will validate its input.
+            circuits_first: If set to True 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.
+        """
+        super().__init__(validate)
+        self._n_shots = None
+        self._n_circuits = None
+        self._circuits_first = circuits_first
+
+    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 != (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._circuits_first:
+            return unordered_data.T.flatten()
+        else:
+            return unordered_data.flatten()
+
+
+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)

test/data_processing/test_nodes.py

@@ -24,6 +24,7 @@
     AverageData,
     MinMaxNormalize,
     Probability,
+    RestlessToCounts,
 )
 from qiskit_experiments.framework.json import ExperimentDecoder, ExperimentEncoder
 from . import BaseDataProcessorTest
@@ -377,3 +378,59 @@ def test_json(self):
         """Check if the node is serializable."""
         node = Probability(outcome="00", alpha_prior=0.2)
         self.assertRoundTripSerializable(node, check_func=self.json_equiv)
+
+
+class TestRestless(QiskitExperimentsTestCase):
+    """Test the restless measurements node."""
+
+    def test_restless_classify_1(self):
+        """Test the classification of restless shots for two single-qubit shots.
+        This example corresponds to running two single-qubit circuits without qubit reset where
+        the first and second circuit would be, e.g. an X gate and an identity gate, respectively.
+        We measure the qubit in the 1 state for the first circuit and measure 1 again for the
+        second circuit. The second shot is reclassified as a 0 since there was no state change."""
+        previous_shot = "1"
+        shot = "1"
+
+        restless_classified_shot = RestlessToCounts._restless_classify(shot, previous_shot)
+        self.assertEqual(restless_classified_shot, "0")
+
+    def test_restless_classify_2(self):
+        """Test the classification of restless shots for two eight-qubit shots.
+        In this example we run two eight qubit circuits. The first circuit applies an
+        X, X, Id, Id, Id, X, X and Id gate, the second an Id, Id, X, Id, Id, X, Id and Id gate
+        to qubits one to eight, respectively."""
+        previous_shot = "11000110"
+        shot = "11100010"
+
+        restless_classified_shot = RestlessToCounts._restless_classify(shot, previous_shot)
+        self.assertEqual(restless_classified_shot, "00100100")
+
+    def test_restless_process_1(self):
+        """Test that a single-qubit restless memory is correctly post-processed.
+        This example corresponds to running an X gate and a SX gate with four shots
+        in an ideal restless setting."""
+        n_qubits = 1
+        node = RestlessToCounts(n_qubits)
+
+        data = [["0x1", "0x1", "0x0", "0x0"], ["0x0", "0x1", "0x1", "0x0"]]
+        processed_data = node(data=np.array(data))
+        # time-ordered data: ["1", "0", "1", "1", "0", "1", "0", "0"]
+        # classification: ["1", "1", "1", "0", "1", "1", "1", "0"]
+        expected_data = np.array([{"1": 4}, {"1": 2, "0": 2}])
+        self.assertTrue(processed_data.all() == expected_data.all())
+
+    def test_restless_process_2(self):
+        """Test if a two-qubit restless memory is correctly post-processed.
+        This example corresponds to running two two-qubit circuits in an ideal restless setting.
+        The first circuit applies an X gate to the first and a SX gate to the second qubit. The
+        second circuit applies two identity gates."""
+        n_qubits = 2
+        node = RestlessToCounts(n_qubits)
+
+        data = [["0x3", "0x1", "0x2", "0x0"], ["0x3", "0x1", "0x2", "0x0"]]
+        processed_data = node(data=np.array(data))
+        # time-ordered data: ["11", "11", "01", "01", "10", "10", "00", "00"]
+        # classification: ["11", "00", "10", "00", "11", "00", "10", "00"]
+        expected_data = np.array([{"10": 2, "11": 2}, {"00": 4}])
+        self.assertTrue(processed_data.all() == expected_data.all())