Convert all QM sweepers to absolute

src/qibolab/instruments/qm/controller.py

@@ -23,6 +23,7 @@
 from .components import QmChannel
 from .config import SAMPLING_RATE, QmConfig, operation
 from .program import ExecutionArguments, create_acquisition, program
+from .program.sweepers import check_frequency_bandwidth, sweeper_amplitude
 
 OCTAVE_ADDRESS_OFFSET = 11000
 """Offset to be added to Octave addresses, because they must be 11xxx, where
@@ -95,9 +96,15 @@ def fetch_results(result, acquisitions):
     }
 
 
-def find_duration_sweepers(sweepers: list[ParallelSweepers]) -> list[Sweeper]:
-    """Find duration sweepers in order to register multiple pulses."""
-    return [s for ps in sweepers for s in ps if s.parameter is Parameter.duration]
+def find_sweepers(
+    sweepers: list[ParallelSweepers], parameter: Parameter
+) -> list[Sweeper]:
+    """Find sweepers of given parameter in order to register specific pulses.
+
+    Duration and amplitude sweepers may require registering additional pulses
+    in the QM ``config``.
+    """
+    return [s for ps in sweepers for s in ps if s.parameter is parameter]
 
 
 @dataclass
@@ -309,19 +316,43 @@ def register_pulses(self, configs: dict[str, Config], sequence: PulseSequence):
     def register_duration_sweeper_pulses(
         self, args: ExecutionArguments, sweeper: Sweeper
     ):
-        """Register pulse with many different durations, in order to sweep
-        duration."""
+        """Register pulse with many different durations.
+
+        Needed when sweeping duration.
+        """
         for pulse in sweeper.pulses:
             if isinstance(pulse, (Align, Delay)):
                 continue
 
-            op = operation(pulse)
-            channel_name = str(args.sequence.pulse_channels(pulse.id)[0])
-            channel = self.channels[channel_name].logical_channel
+            params = args.parameters[operation(pulse)]
+            channel_ids = args.sequence.pulse_channels(pulse.id)
+            channel = self.channels[str(channel_ids[0])].logical_channel
+            original_pulse = (
+                pulse if params.amplitude_pulse is None else params.amplitude_pulse
+            )
             for value in sweeper.values:
-                sweep_pulse = pulse.model_copy(update={"duration": value})
+                sweep_pulse = original_pulse.model_copy(update={"duration": value})
                 sweep_op = self.register_pulse(channel, sweep_pulse)
-                args.parameters[op].pulses.append((value, sweep_op))
+                params.duration_ops.append((value, sweep_op))
+
+    def register_amplitude_sweeper_pulses(
+        self, args: ExecutionArguments, sweeper: Sweeper
+    ):
+        """Register pulse with different amplitude.
+
+        Needed when sweeping amplitude because the original amplitude
+        may not sufficient to reach all the sweeper values.
+        """
+        new_op = None
+        amplitude = sweeper_amplitude(sweeper.values)
+        for pulse in sweeper.pulses:
+            channel_ids = args.sequence.pulse_channels(pulse.id)
+            channel = self.channels[str(channel_ids[0])].logical_channel
+            sweep_pulse = pulse.model_copy(update={"amplitude": amplitude})
+
+            params = args.parameters[operation(pulse)]
+            params.amplitude_pulse = sweep_pulse
+            params.amplitude_op = self.register_pulse(channel, sweep_pulse)
 
     def register_acquisitions(
         self,
@@ -366,6 +397,23 @@ def register_acquisitions(
 
         return acquisitions
 
+    def preprocess_sweeps(
+        self,
+        sweepers: list[ParallelSweepers],
+        configs: dict[str, Config],
+        args: ExecutionArguments,
+    ):
+        """Preprocessing and checks needed before executing some sweeps.
+
+        Amplitude and duration sweeps require registering additional pulses in the QM ``config.
+        """
+        for sweeper in find_sweepers(sweepers, Parameter.frequency):
+            check_frequency_bandwidth(sweeper.channels, configs, sweeper.values)
+        for sweeper in find_sweepers(sweepers, Parameter.amplitude):
+            self.register_amplitude_sweeper_pulses(args, sweeper)
+        for sweeper in find_sweepers(sweepers, Parameter.duration):
+            self.register_duration_sweeper_pulses(args, sweeper)
+
     def execute_program(self, program):
         """Executes an arbitrary program written in QUA language."""
         machine = self.manager.open_qm(asdict(self.config))
@@ -409,10 +457,7 @@ def play(
         acquisitions = self.register_acquisitions(configs, sequence, options)
 
         args = ExecutionArguments(sequence, acquisitions, options.relaxation_time)
-
-        for sweeper in find_duration_sweepers(sweepers):
-            self.register_duration_sweeper_pulses(args, sweeper)
-
+        self.preprocess_sweeps(sweepers, configs, args)
         experiment = program(configs, args, options, sweepers)
 
         if self.script_file_name is not None:

src/qibolab/instruments/qm/program/arguments.py

@@ -4,6 +4,7 @@
 
 from qm.qua._dsl import _Variable  # for type declaration only
 
+from qibolab.pulses import Pulse
 from qibolab.sequence import PulseSequence
 
 from .acquisition import Acquisitions
@@ -13,10 +14,14 @@
 class Parameters:
     """Container of swept QUA variables."""
 
-    duration: Optional[_Variable] = None
     amplitude: Optional[_Variable] = None
+    amplitude_pulse: Optional[Pulse] = None
+    amplitude_op: Optional[str] = None
+
     phase: Optional[_Variable] = None
-    pulses: list[tuple[float, str]] = field(default_factory=list)
+
+    duration: Optional[_Variable] = None
+    duration_ops: list[tuple[float, str]] = field(default_factory=list)
     interpolated: bool = False
 
 

src/qibolab/instruments/qm/program/instructions.py

@@ -35,7 +35,7 @@ def _delay(pulse: Delay, element: str, parameters: Parameters):
 def _play_multiple_waveforms(element: str, parameters: Parameters):
     """Sweeping pulse duration using distinctly uploaded waveforms."""
     with qua.switch_(parameters.duration, unsafe=True):
-        for value, sweep_op in parameters.pulses:
+        for value, sweep_op in parameters.duration_ops:
             if parameters.amplitude is not None:
                 sweep_op = sweep_op * parameters.amplitude
             with qua.case_(value):
@@ -49,7 +49,7 @@ def _play_single_waveform(
     acquisition: Optional[Acquisition] = None,
 ):
     if parameters.amplitude is not None:
-        op = op * parameters.amplitude
+        op = parameters.amplitude_op * parameters.amplitude
     if acquisition is not None:
         acquisition.measure(op)
     else:
@@ -65,7 +65,7 @@ def _play(
     if parameters.phase is not None:
         qua.frame_rotation_2pi(parameters.phase, element)
 
-    if len(parameters.pulses) > 0:
+    if len(parameters.duration_ops) > 0:
         _play_multiple_waveforms(element, parameters)
     else:
         _play_single_waveform(op, element, parameters, acquisition)
@@ -148,9 +148,12 @@ def sweep(
             ):
                 method = SWEEPER_METHODS[sweeper.parameter]
                 if sweeper.pulses is not None:
-                    method(sweeper.pulses, values, variable, configs, args)
+                    for pulse in sweeper.pulses:
+                        params = args.parameters[operation(pulse)]
+                        method(variable, params)
                 else:
-                    method(sweeper.channels, values, variable, configs, args)
+                    for channel in sweeper.channels:
+                        method(variable, channel, configs)
 
             sweep(sweepers[1:], configs, args)