Replace results with bare arrays

doc/source/getting-started/experiment.rst

@@ -242,6 +242,7 @@ We leave to the dedicated tutorial a full explanation of the experiment, but her
 
     from qibolab import create_platform
     from qibolab.pulses import PulseSequence
+    from qibolab.result import magnitude
     from qibolab.sweeper import Sweeper, SweeperType, Parameter
     from qibolab.execution_parameters import (
         ExecutionParameters,
@@ -276,7 +277,7 @@ We leave to the dedicated tutorial a full explanation of the experiment, but her
     probe_pulse = next(iter(sequence.probe_pulses))
 
     # plot the results
-    amplitudes = results[probe_pulse.id][0].magnitude
+    amplitudes = magnitude(results[probe_pulse.id][0])
     frequencies = np.arange(-2e8, +2e8, 1e6) + platform.config(qubit.probe.name).frequency
 
     plt.title("Resonator Spectroscopy")

doc/source/tutorials/calibration.rst

@@ -31,6 +31,7 @@ around the pre-defined frequency.
     import numpy as np
     from qibolab import create_platform
     from qibolab.pulses import PulseSequence
+    from qibolab.result import magnitude
     from qibolab.sweeper import Sweeper, SweeperType, Parameter
     from qibolab.execution_parameters import (
         ExecutionParameters,
@@ -72,7 +73,7 @@ In few seconds, the experiment will be finished and we can proceed to plot it.
     import matplotlib.pyplot as plt
 
     probe_pulse = next(iter(sequence.probe_pulses))
-    amplitudes = results[probe_pulse.id][0].magnitude
+    amplitudes = magnitude(results[probe_pulse.id][0])
     frequencies = np.arange(-2e8, +2e8, 1e6) + platform.config(qubit.probe.name).frequency
 
     plt.title("Resonator Spectroscopy")
@@ -110,6 +111,7 @@ complex pulse sequence. Therefore with start with that:
     import matplotlib.pyplot as plt
     from qibolab import create_platform
     from qibolab.pulses import Pulse, PulseSequence, Delay, Gaussian
+    from qibolab.result import magnitude
     from qibolab.sweeper import Sweeper, SweeperType, Parameter
     from qibolab.execution_parameters import (
         ExecutionParameters,
@@ -156,7 +158,7 @@ We can now proceed to launch on hardware:
     results = platform.execute([sequence], options, [[sweeper]])
 
     probe_pulse = next(iter(sequence.probe_pulses))
-    amplitudes = results[probe_pulse.id][0].magnitude
+    amplitudes = magnitude(results[probe_pulse.id][0])
     frequencies = np.arange(-2e8, +2e8, 1e6) + platform.config(qubit.drive.name).frequency
 
     plt.title("Resonator Spectroscopy")
@@ -208,6 +210,7 @@ and its impact on qubit states in the IQ plane.
     import matplotlib.pyplot as plt
     from qibolab import create_platform
     from qibolab.pulses import PulseSequence, Delay
+    from qibolab.result import unpack
     from qibolab.sweeper import Sweeper, SweeperType, Parameter
     from qibolab.execution_parameters import (
         ExecutionParameters,
@@ -246,13 +249,12 @@ and its impact on qubit states in the IQ plane.
     plt.xlabel("I [a.u.]")
     plt.ylabel("Q [a.u.]")
     plt.scatter(
-        results_one[probe_pulse1.id][0].voltage_i,
-        results_one[probe_pulse1.id][0].voltage_q,
+        results_one[probe_pulse1.id][0],
+        results_one[probe_pulse1.id][0],
         label="One state",
     )
     plt.scatter(
-        results_zero[probe_pulse2.id][0].voltage_i,
-        results_zero[probe_pulse2.id][0].voltage_q,
+        *unpack(results_zero[probe_pulse2.id][0]),
         label="Zero state",
     )
     plt.show()

src/qibolab/backends.py

@@ -69,7 +69,7 @@ def assign_measurements(self, measurement_map, readout):
                 containing the readout measurement shots. This is created in ``execute_circuit``.
         """
         for gate, sequence in measurement_map.items():
-            _samples = (readout[pulse.id].samples for pulse in sequence.probe_pulses)
+            _samples = (readout[pulse.id] for pulse in sequence.probe_pulses)
             samples = list(filter(lambda x: x is not None, _samples))
             gate.result.backend = self
             gate.result.register_samples(np.array(samples).T)
@@ -160,8 +160,7 @@ def execute_circuits(self, circuits, initial_states=None, nshots=1000):
             )
             for gate, sequence in measurement_map.items():
                 samples = [
-                    readout[pulse.id].popleft().samples
-                    for pulse in sequence.probe_pulses
+                    readout[pulse.id].popleft() for pulse in sequence.probe_pulses
                 ]
                 gate.result.backend = self
                 gate.result.register_samples(np.array(samples).T)

src/qibolab/execution_parameters.py

@@ -1,15 +1,8 @@
 from enum import Enum, auto
 from typing import Any, Optional
 
-from qibolab.result import (
-    AveragedIntegratedResults,
-    AveragedRawWaveformResults,
-    AveragedSampleResults,
-    IntegratedResults,
-    RawWaveformResults,
-    SampleResults,
-)
 from qibolab.serialize_ import Model
+from qibolab.sweeper import ParallelSweepers
 
 
 class AcquisitionType(Enum):
@@ -36,19 +29,10 @@ class AveragingMode(Enum):
     SEQUENTIAL = auto()
     """SEQUENTIAL: Worse averaging for noise[Avoid]"""
 
-
-RESULTS_TYPE = {
-    AveragingMode.CYCLIC: {
-        AcquisitionType.INTEGRATION: AveragedIntegratedResults,
-        AcquisitionType.RAW: AveragedRawWaveformResults,
-        AcquisitionType.DISCRIMINATION: AveragedSampleResults,
-    },
-    AveragingMode.SINGLESHOT: {
-        AcquisitionType.INTEGRATION: IntegratedResults,
-        AcquisitionType.RAW: RawWaveformResults,
-        AcquisitionType.DISCRIMINATION: SampleResults,
-    },
-}
+    @property
+    def average(self) -> bool:
+        """Whether an average is performed or not."""
+        return self is not AveragingMode.SINGLESHOT
 
 
 ConfigUpdate = dict[str, dict[str, Any]]
@@ -87,7 +71,20 @@ class ExecutionParameters(Model):
     top of platform defaults.
     """
 
-    @property
-    def results_type(self):
-        """Returns corresponding results class."""
-        return RESULTS_TYPE[self.averaging_mode][self.acquisition_type]
+    def results_shape(
+        self, sweepers: list[ParallelSweepers], samples: Optional[int] = None
+    ) -> tuple[int, ...]:
+        """Compute the expected shape for collected data."""
+
+        shots = (
+            (self.nshots,) if self.averaging_mode is AveragingMode.SINGLESHOT else ()
+        )
+        sweeps = tuple(
+            min(len(sweep.values) for sweep in parsweeps) for parsweeps in sweepers
+        )
+        inner = {
+            AcquisitionType.DISCRIMINATION: (),
+            AcquisitionType.INTEGRATION: (2,),
+            AcquisitionType.RAW: (samples, 2),
+        }[self.acquisition_type]
+        return shots + sweeps + inner

src/qibolab/instruments/abstract.py

@@ -68,7 +68,7 @@ def dump(self):
 
     @property
     @abstractmethod
-    def sampling_rate(self):
+    def sampling_rate(self) -> int:
         """Sampling rate of control electronics in giga samples per second
         (GSps)."""
 

src/qibolab/instruments/dummy.py

@@ -3,6 +3,7 @@
 
 from qibolab import AcquisitionType, AveragingMode, ExecutionParameters
 from qibolab.pulses import PulseSequence
+from qibolab.pulses.pulse import Pulse
 from qibolab.sweeper import ParallelSweepers
 from qibolab.unrolling import Bounds
 
@@ -62,7 +63,7 @@ class DummyInstrument(Controller):
     BOUNDS = Bounds(1, 1, 1)
 
     @property
-    def sampling_rate(self):
+    def sampling_rate(self) -> int:
         return SAMPLING_RATE
 
     def connect(self):
@@ -74,22 +75,12 @@ def disconnect(self):
     def setup(self, *args, **kwargs):
         log.info(f"Setting up {self.name} instrument.")
 
-    def get_values(self, options, ro_pulse, shape):
+    def values(self, options: ExecutionParameters, shape: tuple[int, ...]):
         if options.acquisition_type is AcquisitionType.DISCRIMINATION:
             if options.averaging_mode is AveragingMode.SINGLESHOT:
-                values = np.random.randint(2, size=shape)
-            elif options.averaging_mode is AveragingMode.CYCLIC:
-                values = np.random.rand(*shape)
-        elif options.acquisition_type is AcquisitionType.RAW:
-            samples = int(ro_pulse.duration * SAMPLING_RATE)
-            waveform_shape = tuple(samples * dim for dim in shape)
-            values = (
-                np.random.rand(*waveform_shape) * 100
-                + 1j * np.random.rand(*waveform_shape) * 100
-            )
-        elif options.acquisition_type is AcquisitionType.INTEGRATION:
-            values = np.random.rand(*shape) * 100 + 1j * np.random.rand(*shape) * 100
-        return values
+                return np.random.randint(2, size=shape)
+            return np.random.rand(*shape)
+        return np.random.rand(*shape) * 100
 
     def play(
         self,
@@ -99,19 +90,10 @@ def play(
         integration_setup: dict[str, tuple[np.ndarray, float]],
         sweepers: list[ParallelSweepers],
     ):
-        if options.averaging_mode is not AveragingMode.CYCLIC:
-            shape = (options.nshots,) + tuple(
-                min(len(sweep.values) for sweep in parsweeps) for parsweeps in sweepers
-            )
-        else:
-            shape = tuple(
-                min(len(sweep.values) for sweep in parsweeps) for parsweeps in sweepers
+        def values(pulse: Pulse):
+            samples = int(pulse.duration * self.sampling_rate)
+            return np.array(
+                self.values(options, options.results_shape(sweepers, samples))
             )
 
-        results = {}
-        for seq in sequences:
-            for ro_pulse in seq.probe_pulses:
-                values = self.get_values(options, ro_pulse, shape)
-                results[ro_pulse.id] = options.results_type(values)
-
-        return results
+        return {ro.id: values(ro) for seq in sequences for ro in seq.probe_pulses}

src/qibolab/instruments/emulator/pulse_simulator.py

@@ -6,6 +6,7 @@
 from typing import Dict, List, Union
 
 import numpy as np
+import numpy.typing as npt
 
 from qibolab import AcquisitionType, AveragingMode, ExecutionParameters
 from qibolab.couplers import Coupler
@@ -14,7 +15,7 @@
 from qibolab.instruments.emulator.models import general_no_coupler_model
 from qibolab.pulses import PulseSequence, PulseType
 from qibolab.qubits import Qubit, QubitId
-from qibolab.result import IntegratedResults, SampleResults
+from qibolab.result import average, collect
 from qibolab.sweeper import Parameter, Sweeper, SweeperType
 
 AVAILABLE_SWEEP_PARAMETERS = {
@@ -135,7 +136,7 @@ def play(
         couplers: Dict[QubitId, Coupler],
         sequence: PulseSequence,
         execution_parameters: ExecutionParameters,
-    ) -> dict[str, Union[IntegratedResults, SampleResults]]:
+    ) -> dict[str, npt.NDArray]:
         """Executes the sequence of instructions and generates readout results,
         as well as simulation-related time and states data.
 
@@ -189,7 +190,7 @@ def sweep(
         sequence: PulseSequence,
         execution_parameters: ExecutionParameters,
         *sweeper: List[Sweeper],
-    ) -> dict[str, Union[IntegratedResults, SampleResults, dict]]:
+    ) -> dict[str, Union[npt.NDArray, dict]]:
         """Executes the sweep and generates readout results, as well as
         simulation-related time and states data.
 
@@ -379,9 +380,9 @@ def _sweep_play(
 
     @staticmethod
     def merge_sweep_results(
-        dict_a: """dict[str, Union[IntegratedResults, SampleResults, list]]""",
-        dict_b: """dict[str, Union[IntegratedResults, SampleResults, list]]""",
-    ) -> """dict[str, Union[IntegratedResults, SampleResults, list]]""":
+        dict_a: """dict[str, Union[npt.NDArray, list]]""",
+        dict_b: """dict[str, Union[npt.NDArray, list]]""",
+    ) -> """dict[str, Union[npt.NDArray, list]]""":
         """Merges two dictionary mapping pulse serial to Qibolab results
         object.
 
@@ -646,7 +647,7 @@ def get_results_from_samples(
     samples: dict[Union[str, int], list],
     execution_parameters: ExecutionParameters,
     prepend_to_shape: list = [],
-) -> dict[str, Union[IntegratedResults, SampleResults]]:
+) -> dict[str, npt.NDArray]:
     """Converts samples into Qibolab results format.
 
     Args:
@@ -673,20 +674,19 @@ def get_results_from_samples(
         values = np.array(samples[ro_pulse.qubit]).reshape(shape).transpose(tshape)
 
         if execution_parameters.acquisition_type is AcquisitionType.DISCRIMINATION:
-            processed_values = SampleResults(values)
+            processed_values = values
 
         elif execution_parameters.acquisition_type is AcquisitionType.INTEGRATION:
-            processed_values = IntegratedResults(values.astype(np.complex128))
+            vals = values.astype(np.complex128)
+            processed_values = collect(vals.real, vals.imag)
 
         else:
             raise ValueError(
                 f"Current emulator does not support requested AcquisitionType {execution_parameters.acquisition_type}"
             )
 
         if execution_parameters.averaging_mode is AveragingMode.CYCLIC:
-            processed_values = (
-                processed_values.average
-            )  # generates AveragedSampleResults
+            processed_values = average(processed_values)
 
         results[ro_pulse.qubit] = results[ro_pulse.serial] = processed_values
     return results

src/qibolab/instruments/icarusqfpga.py

@@ -15,7 +15,7 @@
 from qibolab.instruments.abstract import Controller
 from qibolab.pulses import Pulse, PulseSequence, PulseType
 from qibolab.qubits import Qubit, QubitId
-from qibolab.result import IntegratedResults, SampleResults
+from qibolab.result import average, average_iq, collect
 from qibolab.sweeper import Parameter, Sweeper, SweeperType
 
 DAC_SAMPLNG_RATE_MHZ = 5898.24
@@ -260,12 +260,12 @@ def play(
 
             if options.averaging_mode is not AveragingMode.SINGLESHOT:
                 res = {
-                    qunit_mapping[qunit]: IntegratedResults(i + 1j * q).average
+                    qunit_mapping[qunit]: average_iq(i, q)
                     for qunit, (i, q) in raw.items()
                 }
             else:
                 res = {
-                    qunit_mapping[qunit]: IntegratedResults(i + 1j * q)
+                    qunit_mapping[qunit]: average_iq(i, q)
                     for qunit, (i, q) in raw.items()
                 }
             # Temp fix for readout pulse sweepers, to be removed with IcarusQ v2
@@ -276,8 +276,7 @@ def play(
         elif options.acquisition_type is AcquisitionType.DISCRIMINATION:
             self.device.set_adc_trigger_mode(1)
             self.device.set_qunit_mode(1)
-            raw = self.device.start_qunit_acquisition(options.nshots, readout_qubits)
-            res = {qubit: SampleResults(states) for qubit, states in raw.items()}
+            res = self.device.start_qunit_acquisition(options.nshots, readout_qubits)
             # Temp fix for readout pulse sweepers, to be removed with IcarusQ v2
             for ro_pulse in readout_pulses:
                 res[ro_pulse.qubit] = res[ro_pulse.serial]
@@ -306,9 +305,9 @@ def process_readout_signal(
 
             i = np.dot(raw_signal, cos)
             q = np.dot(raw_signal, sin)
-            singleshot = IntegratedResults(i + 1j * q)
+            singleshot = collect(i, q)
             results[readout_pulse.serial] = (
-                singleshot.average
+                average(singleshot)
                 if options.averaging_mode is not AveragingMode.SINGLESHOT
                 else singleshot
             )

src/qibolab/instruments/qblox/controller.py

@@ -12,7 +12,6 @@
 from qibolab.instruments.qblox.cluster_qrm_rf import QrmRf
 from qibolab.instruments.qblox.sequencer import SAMPLING_RATE
 from qibolab.pulses import PulseSequence, PulseType
-from qibolab.result import SampleResults
 from qibolab.sweeper import Parameter, Sweeper, SweeperType
 from qibolab.unrolling import Bounds
 
@@ -523,12 +522,9 @@ def _sweep_recursion(
     def _combine_result_chunks(chunks):
         some_chunk = next(iter(chunks))
         some_result = next(iter(some_chunk.values()))
-        attribute = "samples" if isinstance(some_result, SampleResults) else "voltage"
         return {
             key: some_result.__class__(
-                np.concatenate(
-                    [getattr(chunk[key], attribute) for chunk in chunks], axis=0
-                )
+                np.concatenate([chunk[key] for chunk in chunks], axis=0)
             )
             for key in some_chunk.keys()
         }