feat: Introduce waveforms extraction

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

@@ -121,7 +121,7 @@ def play(
     ) -> dict[int, Result]:
         results = {}
         for ps in sequences:
-            sequences_ = _prepare(ps, sweepers, options)
+            sequences_ = _prepare(ps, sweepers, options, self.sampling_rate)
             sequencers = self._upload(sequences_)
             results |= self._execute(sequencers)
         return results
@@ -166,11 +166,14 @@ def _prepare(
     sequence: PulseSequence,
     sweepers: list[ParallelSweepers],
     options: ExecutionParameters,
+    sampling_rate: float,
 ) -> dict[ChannelId, Sequence]:
     def ch_pulses(channel: ChannelId):
         return PulseSequence((ch, pulse) for ch, pulse in sequence if ch == channel)
 
     return {
-        channel: Sequence.from_pulses(ch_pulses(channel), sweepers, options)
+        channel: Sequence.from_pulses(
+            ch_pulses(channel), sweepers, options, sampling_rate
+        )
         for channel in sequence.channels
     }

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

@@ -0,0 +1,5 @@
+from .ast_ import Program
+
+
+def program():
+    return Program(elements=[])

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

@@ -3,12 +3,14 @@
 from pydantic import AfterValidator, PlainSerializer, PlainValidator
 
 from qibolab._core.execution_parameters import ExecutionParameters
+from qibolab._core.pulses.pulse import Pulse, PulseLike, Readout
 from qibolab._core.sequence import PulseSequence
 from qibolab._core.serialize import ArrayList, Model
 from qibolab._core.sweeper import ParallelSweepers
 
 from .ast_ import Program
 from .parse import parse
+from .program import program
 
 __all__ = []
 
@@ -19,15 +21,33 @@ class Waveform(Model):
 
 
 Weight = Waveform
+Waveforms = dict[str, Waveform]
 
 
 class Acquisition(Model):
     num_bins: int
     index: int
 
 
+def waveforms(sequence: PulseSequence, sampling_rate: float) -> Waveforms:
+    def id_(pulse):
+        return str(hash(pulse))
+
+    def waveform(pulse):
+        return Waveform(data=pulse.envelopes(sampling_rate), index=0)
+
+    def pulse_(event: PulseLike):
+        return event.probe if isinstance(event, Readout) else event
+
+    return {
+        id_(pulse_(event)): waveform(pulse_(event))
+        for _, event in sequence
+        if isinstance(event, (Pulse, Readout))
+    }
+
+
 class Sequence(Model):
-    waveforms: dict[str, Waveform]
+    waveforms: Waveforms
     weights: dict[str, Weight]
     acquisitions: dict[str, Acquisition]
     program: Annotated[
@@ -42,7 +62,11 @@ def from_pulses(
         sequence: PulseSequence,
         sweepers: list[ParallelSweepers],
         options: ExecutionParameters,
+        sampling_rate: float,
     ):
         return cls(
-            waveforms={}, weights={}, acquisitions={}, program=Program(elements=[])
+            waveforms=waveforms(sequence, sampling_rate),
+            weights={},
+            acquisitions={},
+            program=program(),
         )