Release v0.19.0

.github/workflows/publish.yml

@@ -66,7 +66,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3,12"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
     steps:
       - name: Check out Pulser
         uses: actions/checkout@v4

VERSION.txt

@@ -1 +1 @@
-0.18.1
+0.19.0

pulser-core/pulser/backend/abc.py

@@ -17,6 +17,7 @@
 import typing
 from abc import ABC, abstractmethod
 
+from pulser.devices import Device
 from pulser.result import Result
 from pulser.sequence import Sequence
 
@@ -26,20 +27,46 @@
 class Backend(ABC):
     """The backend abstract base class."""
 
-    def __init__(self, sequence: Sequence) -> None:
+    def __init__(self, sequence: Sequence, mimic_qpu: bool = False) -> None:
         """Starts a new backend instance."""
-        self.validate_sequence(sequence)
+        self.validate_sequence(sequence, mimic_qpu=mimic_qpu)
         self._sequence = sequence
+        self._mimic_qpu = bool(mimic_qpu)
 
     @abstractmethod
     def run(self) -> Results | typing.Sequence[Results]:
         """Executes the sequence on the backend."""
         pass
 
-    def validate_sequence(self, sequence: Sequence) -> None:
+    @staticmethod
+    def validate_sequence(sequence: Sequence, mimic_qpu: bool = False) -> None:
         """Validates a sequence prior to submission."""
         if not isinstance(sequence, Sequence):
             raise TypeError(
                 "'sequence' should be a `Sequence` instance"
                 f", not {type(sequence)}."
             )
+        if not mimic_qpu:
+            return
+
+        if not isinstance(device := sequence.device, Device):
+            raise TypeError(
+                "To be sent to a QPU, the device of the sequence "
+                "must be a real device, instance of 'Device'."
+            )
+        reg = sequence.get_register(include_mappable=True)
+        if device.requires_layout and (layout := reg.layout) is None:
+            raise ValueError(
+                f"'{device.name}' requires the sequence's register to be"
+                " defined from a `RegisterLayout`."
+            )
+        if (
+            not device.accepts_new_layouts
+            and layout is not None
+            and layout not in device.pre_calibrated_layouts
+        ):
+            raise ValueError(
+                f"'{device.name}' does not accept new register layouts so "
+                "the register's layout must be one of the layouts available "
+                f"in '{device.name}.calibrated_register_layouts'."
+            )

pulser-core/pulser/backend/qpu.py

@@ -17,12 +17,29 @@
 from typing import cast
 
 from pulser import Sequence
-from pulser.backend.remote import JobParams, RemoteBackend, RemoteResults
-from pulser.devices import Device
+from pulser.backend.remote import (
+    JobParams,
+    RemoteBackend,
+    RemoteConnection,
+    RemoteResults,
+)
 
 
 class QPUBackend(RemoteBackend):
-    """Backend for sequence execution on a QPU."""
+    """Backend for sequence execution on a QPU.
+
+    Args:
+        sequence: A Sequence or a list of Sequences to execute on a
+            backend accessible via a remote connection.
+        connection: The remote connection through which the jobs
+            are executed.
+    """
+
+    def __init__(
+        self, sequence: Sequence, connection: RemoteConnection
+    ) -> None:
+        """Starts a new QPU backend instance."""
+        super().__init__(sequence, connection, mimic_qpu=True)
 
     def run(
         self, job_params: list[JobParams] | None = None, wait: bool = False
@@ -45,11 +62,23 @@ def run(
             The results, which can be accessed once all sequences have been
             successfully executed.
         """
+        self.validate_job_params(
+            job_params or [], self._sequence.device.max_runs
+        )
+        results = self._connection.submit(
+            self._sequence, job_params=job_params, wait=wait
+        )
+        return cast(RemoteResults, results)
+
+    @staticmethod
+    def validate_job_params(
+        job_params: list[JobParams], max_runs: int | None
+    ) -> None:
+        """Validates a list of job parameters prior to submission."""
         suffix = " when executing a sequence on a real QPU."
         if not job_params:
             raise ValueError("'job_params' must be specified" + suffix)
-
-        max_runs = self._sequence.device.max_runs
+        RemoteBackend._type_check_job_params(job_params)
         for j in job_params:
             if "runs" not in j:
                 raise ValueError(
@@ -60,20 +89,3 @@ def run(
                     "All 'runs' must be below the maximum allowed by the "
                     f"device ({max_runs})" + suffix
                 )
-        results = self._connection.submit(
-            self._sequence, job_params=job_params, wait=wait
-        )
-        return cast(RemoteResults, results)
-
-    def validate_sequence(self, sequence: Sequence) -> None:
-        """Validates a sequence prior to submission.
-
-        Args:
-            sequence: The sequence to validate.
-        """
-        super().validate_sequence(sequence)
-        if not isinstance(sequence.device, Device):
-            raise TypeError(
-                "To be sent to a QPU, the device of the sequence "
-                "must be a real device, instance of 'Device'."
-            )

pulser-core/pulser/backend/remote.py

@@ -131,17 +131,33 @@ class RemoteBackend(Backend):
             backend accessible via a remote connection.
         connection: The remote connection through which the jobs
             are executed.
+        mimic_qpu: Whether to mimic the validations necessary for
+            execution on a QPU.
     """
 
     def __init__(
         self,
         sequence: Sequence,
         connection: RemoteConnection,
+        mimic_qpu: bool = False,
     ) -> None:
         """Starts a new remote backend instance."""
-        super().__init__(sequence)
+        super().__init__(sequence, mimic_qpu=mimic_qpu)
         if not isinstance(connection, RemoteConnection):
             raise TypeError(
                 "'connection' must be a valid RemoteConnection instance."
             )
         self._connection = connection
+
+    @staticmethod
+    def _type_check_job_params(job_params: list[JobParams] | None) -> None:
+        if not isinstance(job_params, list):
+            raise TypeError(
+                f"'job_params' must be a list; got {type(job_params)} instead."
+            )
+        for d in job_params:
+            if not isinstance(d, dict):
+                raise TypeError(
+                    "All elements of 'job_params' must be dictionaries; "
+                    f"got {type(d)} instead."
+                )

pulser-core/pulser/channels/eom.py

@@ -17,7 +17,7 @@
 from dataclasses import dataclass, fields
 from enum import Flag
 from itertools import chain
-from typing import Any, cast
+from typing import Any, Literal, cast, overload
 
 import numpy as np
 
@@ -26,7 +26,12 @@
 # Conversion factor from modulation bandwith to rise time
 # For more info, see https://tinyurl.com/bdeumc8k
 MODBW_TO_TR = 0.48
-OPTIONAL_ABSTR_EOM_FIELDS = ("multiple_beam_control", "custom_buffer_time")
+OPTIONAL_ABSTR_EOM_FIELDS = (
+    "multiple_beam_control",
+    "custom_buffer_time",
+    "blue_shift_coeff",
+    "red_shift_coeff",
+)
 
 
 class RydbergBeam(Flag):
@@ -132,6 +137,8 @@ class _RydbergEOM:
 @dataclass(frozen=True)
 class _RydbergEOMDefaults:
     multiple_beam_control: bool = True
+    blue_shift_coeff: float = 1.0
+    red_shift_coeff: float = 1.0
 
 
 @dataclass(frozen=True)
@@ -149,11 +156,20 @@ class RydbergEOM(_RydbergEOMDefaults, BaseEOM, _RydbergEOM):
         custom_buffer_time: A custom wait time to enforce during EOM buffers.
         multiple_beam_control: Whether both EOMs can be used simultaneously.
             Ignored when only one beam can be controlled.
+        blue_shift_coeff: The weight coefficient of the blue beam's
+            contribution to the lightshift.
+        red_shift_coeff: The weight coefficient of the red beam's contribution
+            to the lightshift.
     """
 
     def __post_init__(self) -> None:
         super().__post_init__()
-        for param in ["max_limiting_amp", "intermediate_detuning"]:
+        for param in [
+            "max_limiting_amp",
+            "intermediate_detuning",
+            "blue_shift_coeff",
+            "red_shift_coeff",
+        ]:
             value = getattr(self, param)
             if value <= 0.0:
                 raise ValueError(
@@ -182,9 +198,42 @@ def __post_init__(self) -> None:
                     f" enumeration, not {self.limiting_beam}."
                 )
 
+    @property
+    def _switching_beams_combos(self) -> list[tuple[RydbergBeam, ...]]:
+        switching_beams: list[tuple[RydbergBeam, ...]] = [
+            (beam,) for beam in self.controlled_beams
+        ]
+        if len(self.controlled_beams) > 1 and self.multiple_beam_control:
+            switching_beams.append(tuple(RydbergBeam))
+        return switching_beams
+
+    @overload
     def calculate_detuning_off(
-        self, amp_on: float, detuning_on: float, optimal_detuning_off: float
+        self,
+        amp_on: float,
+        detuning_on: float,
+        optimal_detuning_off: float,
+        return_switching_beams: Literal[False],
     ) -> float:
+        pass
+
+    @overload
+    def calculate_detuning_off(
+        self,
+        amp_on: float,
+        detuning_on: float,
+        optimal_detuning_off: float,
+        return_switching_beams: Literal[True],
+    ) -> tuple[float, tuple[RydbergBeam, ...]]:
+        pass
+
+    def calculate_detuning_off(
+        self,
+        amp_on: float,
+        detuning_on: float,
+        optimal_detuning_off: float,
+        return_switching_beams: bool = False,
+    ) -> float | tuple[float, tuple[RydbergBeam, ...]]:
         """Calculates the detuning when the amplitude is off in EOM mode.
 
         Args:
@@ -193,13 +242,20 @@ def calculate_detuning_off(
             optimal_detuning_off: The optimal value of detuning (in rad/µs)
                 when there is no pulse being played. It will choose the closest
                 value among the existing options.
+            return_switching_beams: Whether to return the beams that switch on
+                on and off.
         """
         off_options = self.detuning_off_options(amp_on, detuning_on)
         closest_option = np.abs(off_options - optimal_detuning_off).argmin()
-        return cast(float, off_options[closest_option])
+        best_det_off = cast(float, off_options[closest_option])
+        if not return_switching_beams:
+            return best_det_off
+        return best_det_off, self._switching_beams_combos[closest_option]
 
     def detuning_off_options(
-        self, rabi_frequency: float, detuning_on: float
+        self,
+        rabi_frequency: float,
+        detuning_on: float,
     ) -> np.ndarray:
         """Calculates the possible detuning values when the amplitude is off.
 
@@ -216,24 +272,12 @@ def detuning_off_options(
         # offset takes into account the lightshift when both beams are on
         # which is not zero when the Rabi freq of both beams is not equal
         offset = detuning_on - self._lightshift(rabi_frequency, *RydbergBeam)
-        if len(self.controlled_beams) == 1:
-            # When only one beam is controlled, the lighshift during delays
-            # corresponds to having only the other beam (which can't be
-            # switched off) on.
-            lightshifts = [
-                self._lightshift(rabi_frequency, ~self.controlled_beams[0])
-            ]
-
-        else:
-            # When both beams are controlled, we have three options for the
-            # lightshift: (ON, OFF), (OFF, ON) and (OFF, OFF)
-            lightshifts = [
-                self._lightshift(rabi_frequency, beam)
-                for beam in self.controlled_beams
-            ]
-            if self.multiple_beam_control:
-                # Case where both beams are off ie (OFF, OFF) -> no lightshift
-                lightshifts.append(0.0)
+        all_beams: set[RydbergBeam] = set(RydbergBeam)
+        lightshifts = []
+        for beams_off in self._switching_beams_combos:
+            # The beams that don't switch off contribute to the lightshift
+            beams_on: set[RydbergBeam] = all_beams - set(beams_off)
+            lightshifts.append(self._lightshift(rabi_frequency, *beams_on))
 
         # We sum the offset to all lightshifts to get the effective detuning
         return np.array(lightshifts) + offset
@@ -243,7 +287,10 @@ def _lightshift(
     ) -> float:
         # lightshift = (rabi_blue**2 - rabi_red**2) / 4 * int_detuning
         rabi_freqs = self._rabi_freq_per_beam(rabi_frequency)
-        bias = {RydbergBeam.RED: -1, RydbergBeam.BLUE: 1}
+        bias = {
+            RydbergBeam.RED: -self.red_shift_coeff,
+            RydbergBeam.BLUE: self.blue_shift_coeff,
+        }
         # beam off -> beam_rabi_freq = 0
         return sum(bias[beam] * rabi_freqs[beam] ** 2 for beam in beams_on) / (
             4 * self.intermediate_detuning
@@ -252,16 +299,25 @@ def _lightshift(
     def _rabi_freq_per_beam(
         self, rabi_frequency: float
     ) -> dict[RydbergBeam, float]:
+        shift_factor = np.sqrt(
+            self.red_shift_coeff / self.blue_shift_coeff
+            if self.limiting_beam == RydbergBeam.RED
+            else self.blue_shift_coeff / self.red_shift_coeff
+        )
         # rabi_freq = (rabi_red * rabi_blue) / (2 * int_detuning)
-        limit_rabi_freq = self.max_limiting_amp**2 / (
-            2 * self.intermediate_detuning
+        limit_rabi_freq = (
+            shift_factor
+            * self.max_limiting_amp**2
+            / (2 * self.intermediate_detuning)
         )
         # limit_rabi_freq is the maximum effective rabi frequency value
-        # below which the rabi frequency of both beams can be matched
+        # below which the lightshift can be zero
         if rabi_frequency <= limit_rabi_freq:
-            # Both beams the same rabi_freq
-            beam_amp = np.sqrt(2 * rabi_frequency * self.intermediate_detuning)
-            return {beam: beam_amp for beam in RydbergBeam}
+            base_amp_squared = 2 * rabi_frequency * self.intermediate_detuning
+            return {
+                self.limiting_beam: np.sqrt(base_amp_squared / shift_factor),
+                ~self.limiting_beam: np.sqrt(base_amp_squared * shift_factor),
+            }
 
         # The limiting beam is at its maximum amplitude while the other
         # has the necessary amplitude to reach the desired effective rabi freq

pulser-core/pulser/devices/_devices.py

@@ -80,7 +80,9 @@
     max_radial_distance=35,
     min_atom_distance=5,
     max_sequence_duration=4000,
-    # TODO: Define max_runs
+    max_runs=2000,
+    requires_layout=True,
+    accepts_new_layouts=False,
     channel_objects=(
         Rydberg.Global(
             max_abs_detuning=2 * np.pi * 20,