feat: Lay out execution workflow

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

@@ -1,3 +1,5 @@
+from collections import defaultdict
+from functools import cached_property
 from typing import Optional
 
 import qblox_instruments as qblox
@@ -6,7 +8,7 @@
 
 from qibolab._core.components.configs import Config
 from qibolab._core.execution_parameters import ExecutionParameters
-from qibolab._core.identifier import Result
+from qibolab._core.identifier import ChannelId, Result
 from qibolab._core.instruments.abstract import Controller
 from qibolab._core.sequence import PulseSequence
 from qibolab._core.serialize import Model
@@ -19,17 +21,53 @@
 SAMPLING_RATE = 1
 
 
+SlotId = int
+SequencerId = int
+
+
 class PortAddress(Model):
-    module: int
-    port: int
+    slot: SlotId
+    ports: tuple[int, Optional[int]]
     input: bool = False
 
     @classmethod
     def from_path(cls, path: str):
         """Load address from :attr:`qibolab.Channel.path`."""
         els = path.split("/")
         assert len(els) == 2
-        return cls(module=int(els[0]), port=int(els[1][1:]), input=els[1][0] == "i")
+        ports = els[1][1:].split("_")
+        assert 1 <= len(ports) <= 2
+        return cls(
+            slot=int(els[0]),
+            ports=(int(ports[0]), int(ports[1]) if len(ports) == 2 else None),
+            input=els[1][0] == "i",
+        )
+
+    @property
+    def local_address(self):
+        """Physical address within the module.
+
+        It will generate a string in the format ``<direction><channel>`` or
+        ``<direction><I-channel>_<Q-channel>``.
+        ``<direction>`` is ``in`` for a connection between an input and the acquisition
+        path, ``out`` for a connection from the waveform generator to an output, or
+        ``io`` to do both.
+        The channels must be integer channel indices.
+        Only one channel is present for a real mode operating sequencer; two channels
+        are used for complex mode.
+
+        .. note::
+
+            Description adapted from
+            https://docs.qblox.com/en/main/api_reference/cluster.html#qblox_instruments.Cluster.connect_sequencer
+        """
+        direction = "in" if self.input else "out"
+        channels = (
+            str(self.ports[0])
+            if self.ports[1] is None
+            else f"{self.ports[0]}_{self.ports[1]}"
+        )
+        return f"{direction}{channels}"
 
 
 class Cluster(Controller):
@@ -42,6 +80,11 @@ class Cluster(Controller):
     bounds: str = "qblox/bounds"
     _cluster: Optional[qblox.Cluster] = None
 
+    @cached_property
+    def _modules(self) -> dict[SlotId, Module]:
+        assert self._cluster is not None
+        return {mod.slot_idx: mod for mod in self._cluster.modules if mod.present()}
+
     @property
     def sampling_rate(self) -> int:
         return SAMPLING_RATE
@@ -50,6 +93,7 @@ def connect(self):
         self._cluster = find_or_create_instrument(
             qblox.Cluster, recreate=True, name=self.name, identifier=self.address
         )
+        self._cluster.reset()
 
     @property
     def is_connected(self) -> bool:
@@ -58,16 +102,11 @@ def is_connected(self) -> bool:
     def disconnect(self):
         assert self._cluster is not None
 
-        for module in self.modules.values():
+        for module in self._modules.values():
             module.stop_sequencer()
         self._cluster.reset()
         self._cluster = None
 
-    @property
-    def modules(self) -> dict[int, Module]:
-        assert self._cluster is not None
-        return {mod.slot_idx: mod for mod in self._cluster.modules if mod.present()}
-
     def play(
         self,
         configs: dict[str, Config],
@@ -77,9 +116,57 @@ def play(
     ) -> dict[int, Result]:
         results = {}
         for ps in sequences:
-            seq = Sequence.from_pulses(ps, sweepers, options)
-            results |= self._execute([seq])
+            sequences_ = _prepare(ps, sweepers, options)
+            sequencers = self._upload(sequences_)
+            results |= self._execute(sequencers)
         return results
 
-    def _execute(self, sequences: list[Sequence]) -> dict:
+    def _upload(
+        self, sequences: dict[ChannelId, Sequence]
+    ) -> dict[SlotId, dict[ChannelId, SequencerId]]:
+        channels_by_module = _channels_by_module()
+        sequencers = defaultdict(dict)
+        for mod, chs in channels_by_module.items():
+            module = self._modules[mod]
+            assert len(module.sequencers) > len(chs)
+            # Map sequencers to specific outputs (but first disable all sequencer connections)
+            module.disconnect_outputs()
+            for idx, (ch, sequencer) in enumerate(zip(chs, module.sequencers)):
+                sequencers[mod][ch] = idx
+                sequencer.sequence(sequences[ch].model_dump())
+                # Configure the sequencers to synchronize
+                sequencer.sync_en(True)
+                sequencer.connect_sequencer(
+                    PortAddress.from_path(self.channels[ch].path).local_address
+                )
+
+        return sequencers
+
+    def _execute(self, sequencers: dict[SlotId, dict[ChannelId, SequencerId]]) -> dict:
+        # TODO: implement
+        for mod, seqs in sequencers.items():
+            module = self._modules[mod]
+            for seq in seqs.values():
+                module.arm_sequencer(seq)
+            module.start_sequencer()
+
         return {}
+
+
+def _channels_by_module() -> dict[SlotId, list[ChannelId]]:
+    # TODO: implement
+    return {}
+
+
+def _prepare(
+    sequence: PulseSequence,
+    sweepers: list[ParallelSweepers],
+    options: ExecutionParameters,
+) -> dict[ChannelId, Sequence]:
+    sequences = {}
+    for channel in sequence.channels:
+        filtered = PulseSequence((ch, pulse) for ch, pulse in sequence if ch == channel)
+        seq = Sequence.from_pulses(filtered, sweepers, options)
+        sequences[channel] = seq
+
+    return sequences