refactor: Move non-sequence-related acquisition code

src/qibolab/_core/instruments/qblox/cluster.py

@@ -19,8 +19,8 @@
 from .config import PortAddress, SlotId
 from .identifiers import SequencerMap
 from .log import Logger
-from .sequence import Sequence, acquisition, compile
-from .utils import integration_lenghts
+from .results import AcquiredData, extract, integration_lenghts
+from .sequence import Sequence, compile
 
 __all__ = ["Cluster"]
 
@@ -93,7 +93,7 @@ def play(
         options: ExecutionParameters,
         sweepers: list[ParallelSweepers],
     ) -> dict[int, Result]:
-        results = {}
+        results_ = {}
         log = Logger(configs)
 
         for ps in sequences:
@@ -104,8 +104,8 @@ def play(
             data = self._execute(sequencers, options.estimate_duration([ps], sweepers))
             log.data(data)
             lenghts = integration_lenghts(sequences_, sequencers, self._modules)
-            results |= acquisition.extract(data, lenghts)
-        return results
+            results_ |= extract(data, lenghts)
+        return results_
 
     def _configure(
         self,
@@ -136,7 +136,7 @@ def _configure(
 
     def _execute(
         self, sequencers: SequencerMap, duration: float
-    ) -> dict[ChannelId, acquisition.AcquiredData]:
+    ) -> dict[ChannelId, AcquiredData]:
         for mod, seqs in sequencers.items():
             module = self._modules[mod]
             for seq in seqs.values():

src/qibolab/_core/instruments/qblox/results.py

@@ -0,0 +1,89 @@
+from functools import reduce
+from operator import or_
+from typing import Optional, TypedDict
+
+from qblox_instruments.qcodes_drivers.module import Module
+
+from qibolab._core.identifier import ChannelId, Result
+from qibolab._core.pulses.pulse import PulseId
+
+from .identifiers import SequencerId, SequencerMap, SlotId
+from .sequence import Sequence, acquisition
+
+__all__ = []
+
+
+def _fill_empty_lenghts(
+    weighted: dict[acquisition.MeasureId, Optional[int]],
+    defaults: dict[tuple[SlotId, SequencerId], int],
+    locations: dict[acquisition.MeasureId, tuple[SlotId, SequencerId]],
+) -> dict[acquisition.MeasureId, int]:
+    return {
+        k: v if v is not None else defaults[locations[k]] for k, v in weighted.items()
+    }
+
+
+def integration_lenghts(
+    sequences: dict[ChannelId, Sequence],
+    sequencers: SequencerMap,
+    modules: dict[SlotId, Module],
+) -> dict[acquisition.MeasureId, int]:
+    channels_to_sequencer = {
+        ch: (mod, seq) for mod, chs in sequencers.items() for ch, seq in chs.items()
+    }
+    return _fill_empty_lenghts(
+        reduce(or_, (seq.integration_lengths for seq in sequences.values())),
+        {
+            (mod_id, seq): seq.integration_length_acq()
+            for mod_id, mod in modules.items()
+            for seq in mod.sequencers
+        },
+        {
+            acq: channels_to_sequencer[ch]
+            for ch, seq in sequences.items()
+            for acq in seq.acquisitions
+        },
+    )
+
+
+IndividualScope = TypedDict(
+    "IndividualScope",
+    {"data": list[float], "out-of-range": list[bool], "avg_cnt": list[bool]},
+)
+
+
+class ScopeData(TypedDict):
+    path0: IndividualScope
+    path1: IndividualScope
+
+
+class Integration(TypedDict):
+    path0: list[float]
+    path1: list[float]
+
+
+class BinData(TypedDict):
+    integration: Integration
+    threshold: list[int]
+    valid: list
+    avg_cnt: list[int]
+
+
+class Data(TypedDict):
+    scope: ScopeData
+    bins: BinData
+
+
+class IndexedData(TypedDict):
+    index: int
+    acquisition: Data
+
+
+AcquiredData = dict[acquisition.MeasureId, IndexedData]
+
+
+def extract(
+    acquisitions: dict[ChannelId, AcquiredData],
+    lenghts: dict[acquisition.MeasureId, int],
+) -> dict[PulseId, Result]:
+    return {}

src/qibolab/_core/instruments/qblox/sequence/acquisition.py

@@ -1,12 +1,7 @@
-from typing import Optional, TypedDict
-
 from qibolab._core import pulses
-from qibolab._core.identifier import ChannelId, Result
-from qibolab._core.pulses.pulse import PulseId
 from qibolab._core.sequence import PulseSequence
 from qibolab._core.serialize import Model
 
-from ..identifiers import SequencerId, SlotId
 from .waveforms import Waveform
 
 __all__ = []
@@ -31,55 +26,3 @@ def acquisitions(sequence: PulseSequence, num_bins: int) -> Acquisitions:
             p for _, p in sequence if isinstance(p, pulses.Acquisition)
         )
     }
-
-
-IndividualScope = TypedDict(
-    "IndividualScope",
-    {"data": list[float], "out-of-range": list[bool], "avg_cnt": list[bool]},
-)
-
-
-class ScopeData(TypedDict):
-    path0: IndividualScope
-    path1: IndividualScope
-
-
-class Integration(TypedDict):
-    path0: list[float]
-    path1: list[float]
-
-
-class BinData(TypedDict):
-    integration: Integration
-    threshold: list[int]
-    valid: list
-    avg_cnt: list[int]
-
-
-class Data(TypedDict):
-    scope: ScopeData
-    bins: BinData
-
-
-class IndexedData(TypedDict):
-    index: int
-    acquisition: Data
-
-
-AcquiredData = dict[MeasureId, IndexedData]
-
-
-def integration_lengths(
-    weighted: dict[MeasureId, Optional[int]],
-    defaults: dict[tuple[SlotId, SequencerId], int],
-    locations: dict[MeasureId, tuple[SlotId, SequencerId]],
-) -> dict[MeasureId, int]:
-    return {
-        k: v if v is not None else defaults[locations[k]] for k, v in weighted.items()
-    }
-
-
-def extract(
-    acquisitions: dict[ChannelId, AcquiredData], lenghts: dict[MeasureId, int]
-) -> dict[PulseId, Result]:
-    return {}