Long pulses

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

@@ -2,6 +2,7 @@
     Acquisition,
     Align,
     Delay,
+    LongPulse,
     Pulse,
     Readout,
     VirtualZ,
@@ -17,6 +18,7 @@
     Move,
     Play,
     Register,
+    SetAwgOffs,
     SetPhDelta,
     UpdParam,
     Wait,
@@ -59,6 +61,81 @@ def _play_duration_swept(registers: dict[ParamRole, Register]) -> list[Instructi
     ]
 
 
+def _process_longpulse(pulse: LongPulse, params: set[Param], merged_vzs: bool):
+    """Emit Q1ASM for a LongPulse.
+
+    Uses ``set_awg_offs`` to produce a continuous CW tone without storing
+    any per-duration waveforms.  The signal chain on the RF module is:
+
+        output = (waveform * gain + offset)
+
+    Timing:  ``upd_param(4)`` starts the tone (4 ns), then ``wait(dur - 4)``
+    holds it.  The DURATION sweep register already holds ``total_duration - 4``
+    (set by ``_longpulse_duration``).  After the wait, ``set_awg_offs(0, 0)``
+    is latched and applied by the next real-time instruction.
+    """
+    uid = pulse.id
+    duration_sweep = {
+        p.role: p.reg for p in params if p.role.value[1] is Parameter.duration
+    }
+    amplitude_sweep = {
+        p.role: p.reg for p in params if p.role.value[1] is Parameter.amplitude
+    }
+
+    if merged_vzs:
+        assert pulse.relative_phase == 0.0
+        phase_pre: list[Instruction] = []
+        phase_post: list[Instruction] = []
+    else:
+        phase = int(convert(pulse.relative_phase, Parameter.relative_phase))
+        minus_phase = int(convert(-pulse.relative_phase, Parameter.relative_phase))
+        phase_pre = (
+            [
+                Add(
+                    a=Registers.phase_delta.value,
+                    b=phase,
+                    destination=Registers.phase_delta.value,
+                )
+            ]
+            if phase != 0
+            else []
+        ) + [SetPhDelta(value=Registers.phase_delta.value)]
+        phase_post = [Move(source=minus_phase, destination=Registers.phase_delta.value)]
+
+    if duration_sweep:
+        hold: list[Instruction] = [Wait(duration=duration_sweep[ParamRole.DURATION])]
+    else:
+        hold = [Wait(duration=int(pulse.duration) - 4)] if pulse.duration > 4 else []
+
+    if amplitude_sweep:
+        pseudo_pulse = [
+            SetAwgOffs(value_0=amplitude_sweep[ParamRole.AMPLITUDE], value_1=0),
+            Line(
+                instruction=UpdParam(duration=4),
+                comment=f"longpulse id: 0x{uid.hex[:5]}",
+            ),
+        ]
+    else:
+        pseudo_pulse = [
+            SetAwgOffs(
+                value_0=int(convert(pulse.amplitude, Parameter.amplitude)), value_1=0
+            ),
+            Line(
+                instruction=UpdParam(duration=4),
+                comment=f"longpulse id: 0x{uid.hex[:5]}",
+            ),
+        ]
+
+    return (
+        phase_pre
+        + pseudo_pulse
+        + hold
+        + phase_post
+        + [SetAwgOffs(value_0=0, value_1=0)]
+        # Line(instruction=UpdParam(duration=4))] # This may be neeed if this is the last pulse in the sequence, otherwise it will never turn off the tone.
+    )
+
+
 def _process_pulse(
     pulse: Pulse, params: set[Param], waveforms: WaveformIndices, merged_vzs: bool
 ):
@@ -175,6 +252,8 @@ def play(
     """Process the individual pulse in experiment."""
     pulse = parpulse[0]
     params = parpulse[1]
+    if isinstance(pulse, LongPulse):
+        return _process_longpulse(pulse, params, merged_vzs)
     if isinstance(pulse, Pulse):
         return _process_pulse(pulse, params, waveforms, merged_vzs)
     if isinstance(pulse, Delay):

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

@@ -14,6 +14,7 @@
 )
 from qibolab._core.instruments.qblox.sequence.asm import Registers
 from qibolab._core.pulses.pulse import (
+    LongPulse,
     Pulse,
     PulseId,
     PulseLike,
@@ -59,7 +60,7 @@ def from_sweeper(cls, sweep: Sweeper) -> "ParamRole":
     def unique(cls, sweep: Sweeper) -> bool:
         return sweep.parameter is not Parameter.duration or (
             sweep.pulses is not None
-            and not any(isinstance(p, Pulse) for p in sweep.pulses)
+            and not any(isinstance(p, (Pulse, LongPulse)) for p in sweep.pulses)
         )
 
     @property

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

@@ -180,12 +180,18 @@ def waveforms(
 
     pulses_not_swept: list[Pulse] = []
     pulses_swept: list[tuple[Pulse, Sweeper]] = []
+    _seen_swept: set = set()
     for p in sequence:
         if isinstance(p, (Pulse, Readout)):
             if p.id in duration_swept:
-                pulses_swept.append(
-                    (_pulse(p, p.id in amplitude_swept), duration_swept[p.id])
-                )
+                # Deduplicate by UUID: the same pulse object may appear N times
+                # in the sequence (pulse-train approach) but should only
+                # contribute one set of waveforms to avoid N-fold memory usage.
+                if p.id not in _seen_swept:
+                    _seen_swept.add(p.id)
+                    pulses_swept.append(
+                        (_pulse(p, p.id in amplitude_swept), duration_swept[p.id])
+                    )
             else:
                 pulses_not_swept.append(_pulse(p, p.id in amplitude_swept))
 

src/qibolab/_core/pulses/pulse.py

@@ -13,6 +13,7 @@
     "Acquisition",
     "Align",
     "Delay",
+    "LongPulse",
     "Pulse",
     "PulseId",
     "PulseLike",
@@ -98,6 +99,26 @@ def envelopes(self, sampling_rate: float) -> IqWaveform:
         return np.array([self.i(sampling_rate), self.q(sampling_rate)])
 
 
+class LongPulse(_PulseLike):
+    """Long rectangular pulse for hardware with limited waveform memory.
+
+    On Qblox the backend stores a minimal 4-sample waveform and extends the
+    output with a Q1ASM ``wait``, avoiding the per-duration waveform storage
+    that exhausts AWG memory for swept long pulses.
+    """
+
+    kind: Literal["longpulse"] = "longpulse"
+
+    duration: float
+    """Total pulse duration [ns]."""
+
+    amplitude: float
+    """Pulse digital amplitude (unitless), normalised between -1 and 1."""
+
+    relative_phase: float = 0.0
+    """Relative phase of the pulse, in radians."""
+
+
 class Delay(_PulseLike):
     """Wait instruction.
 
@@ -198,6 +219,6 @@ class Align(_PulseLike):
 
 
 PulseLike = Annotated[
-    Union[Align, Pulse, Delay, VirtualZ, Acquisition, Readout],
+    Union[Align, LongPulse, Pulse, Delay, VirtualZ, Acquisition, Readout],
     Field(discriminator="kind"),
 ]

src/qibolab/_core/sequence.py

@@ -12,7 +12,7 @@
 from qibolab._core.pulses.pulse import PulseId, VirtualZ
 
 from .identifier import ChannelId
-from .pulses import Acquisition, Align, Delay, Pulse, PulseLike, Readout
+from .pulses import Acquisition, Align, Delay, LongPulse, Pulse, PulseLike, Readout
 
 __all__ = ["PulseSequence"]
 
@@ -284,7 +284,8 @@ def to_vzs(self) -> "PulseSequence":
                 el
                 for els in (
                     [(ch, ev)]
-                    if not isinstance(ev, Pulse) or np.isclose(ev.relative_phase, 0)
+                    if not isinstance(ev, (Pulse, LongPulse))
+                    or np.isclose(ev.relative_phase, 0)
                     else [
                         (ch, VirtualZ(phase=ev.relative_phase)),
                         (ch, ev.model_copy(update={"relative_phase": 0})),