Defining the new Backend API classes

pulser-core/pulser/backend/__init__.py

@@ -17,5 +17,6 @@
 from pulser.backend.config import EmulatorConfig
 from pulser.noise_model import NoiseModel  # For backwards compat
 from pulser.backend.qpu import QPUBackend
+from pulser.backend.results import Results
 
-__all__ = ["EmulatorConfig", "NoiseModel", "QPUBackend"]
+__all__ = ["EmulatorConfig", "NoiseModel", "QPUBackend", "Results"]

pulser-core/pulser/backend/abc.py

@@ -14,34 +14,38 @@
 """Base class for the backend interface."""
 from __future__ import annotations
 
-import typing
 from abc import ABC, abstractmethod
+from collections.abc import Sequence
+from typing import ClassVar
 
+import pulser
+from pulser.backend.config import EmulationConfig
+from pulser.backend.results import Results
 from pulser.devices import Device
-from pulser.result import Result
-from pulser.sequence import Sequence
-
-Results = typing.Sequence[Result]
 
 
 class Backend(ABC):
     """The backend abstract base class."""
 
-    def __init__(self, sequence: Sequence, mimic_qpu: bool = False) -> None:
+    def __init__(
+        self, sequence: pulser.Sequence, mimic_qpu: bool = False
+    ) -> None:
         """Starts a new backend instance."""
         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]:
+    def run(self) -> Results | Sequence[Results]:
         """Executes the sequence on the backend."""
         pass
 
     @staticmethod
-    def validate_sequence(sequence: Sequence, mimic_qpu: bool = False) -> None:
+    def validate_sequence(
+        sequence: pulser.Sequence, mimic_qpu: bool = False
+    ) -> None:
         """Validates a sequence prior to submission."""
-        if not isinstance(sequence, Sequence):
+        if not isinstance(sequence, pulser.Sequence):
             raise TypeError(
                 "'sequence' should be a `Sequence` instance"
                 f", not {type(sequence)}."
@@ -70,3 +74,27 @@ def validate_sequence(sequence: Sequence, mimic_qpu: bool = False) -> None:
                 "the register's layout must be one of the layouts available "
                 f"in '{device.name}.calibrated_register_layouts'."
             )
+
+
+class EmulatorBackend(Backend):
+    """The emulator backend parent class."""
+
+    default_config: ClassVar[EmulationConfig]
+
+    def __init__(
+        self,
+        sequence: pulser.Sequence,
+        config: EmulationConfig | None = None,
+        mimic_qpu: bool = False,
+    ) -> None:
+        """Initializes the backend."""
+        super().__init__(sequence, mimic_qpu=mimic_qpu)
+        config = config or self.default_config
+        if not isinstance(config, EmulationConfig):
+            raise TypeError(
+                "'config' must be an instance of 'EmulationConfig', "
+                f"not {type(config)}."
+            )
+        self._config = config
+        # See the BackendConfig definition to see how this would work
+        self._config = type(self.default_config)(**config._backend_options)

pulser-core/pulser/backend/config.py

@@ -14,28 +14,234 @@
 """Defines the backend configuration classes."""
 from __future__ import annotations
 
+import copy
+import warnings
+from collections import Counter
 from dataclasses import dataclass, field
-from typing import Any, Literal, Sequence, get_args
+from typing import (
+    Any,
+    Generic,
+    Literal,
+    Sequence,
+    SupportsFloat,
+    TypeVar,
+    cast,
+    get_args,
+)
 
 import numpy as np
+from numpy.typing import ArrayLike
 
+import pulser.math as pm
+from pulser.backend.observable import Observable
+from pulser.backend.state import State
 from pulser.noise_model import NoiseModel
 
 EVAL_TIMES_LITERAL = Literal["Full", "Minimal", "Final"]
 
+StateType = TypeVar("StateType", bound=State)
+
 
-@dataclass(frozen=True)
 class BackendConfig:
-    """The base backend configuration.
+    """The base backend configuration."""
 
-    Attributes:
-        backend_options: A dictionary of backend specific options.
+    _backend_options: dict[str, Any]
+
+    def __init__(self, **backend_options: Any) -> None:
+        """Initializes the backend config."""
+        cls_name = self.__class__.__name__
+        if invalid_kwargs := (
+            set(backend_options)
+            - (self._expected_kwargs() | {"backend_options"})
+        ):
+            warnings.warn(
+                f"{cls_name!r} received unexpected keyword arguments: "
+                f"{invalid_kwargs}; only the following keyword "
+                f"arguments are expected: {self._expected_kwargs()}.",
+                stacklevel=2,
+            )
+        # Prevents potential issues with mutable arguments
+        self._backend_options = copy.deepcopy(backend_options)
+        if "backend_options" in backend_options:
+            with warnings.catch_warnings():
+                warnings.filterwarnings("always")
+                warnings.warn(
+                    f"The 'backend_options' argument of {cls_name!r} "
+                    "has been deprecated. Please provide the options "
+                    f"as keyword arguments directly to '{cls_name}()'.",
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
+            self._backend_options.update(backend_options["backend_options"])
+
+    def _expected_kwargs(self) -> set[str]:
+        return set()
+
+    def __getattr__(self, name: str) -> Any:
+        if (
+            # Needed to avoid recursion error
+            "_backend_options" in self.__dict__
+            and name in self._backend_options
+        ):
+            return self._backend_options[name]
+        raise AttributeError(f"{name!r} has not been passed to {self!r}.")
+
+
+class EmulationConfig(BackendConfig, Generic[StateType]):
+    """Configures an emulation on a backend.
+
+    Args:
+        observables: A sequence of observables to compute at specific
+            evaluation times. The observables without specified evaluation
+            times will use this configuration's 'default_evaluation_times'.
+        default_evaluation_times: The default times at which observables
+            are computed. Can be a sequence of relative times between 0
+            (the start of the sequence) and 1 (the end of the sequence).
+            Can also be specified as "Full", in which case every step in the
+            emulation will also be an evaluation times.
+        initial_state: The initial state from which emulation starts. If
+            specified, the state type needs to be compatible with the emulator
+            backend. If left undefined, defaults to starting with all qudits
+            in the ground state.
+        with_modulation: Whether to emulate the sequence with the programmed
+            input or the expected output.
+        interaction_matrix: An optional interaction matrix to replace the
+            interaction terms in the Hamiltonian. For an N-qudit system,
+            must be an NxN symmetric matrix where entry (i, j) dictates
+            the interaction coefficient between qudits i and j, ie it replaces
+            the C_n/r_{ij}^n term.
+        prefer_device_noise_model: If the sequence's device has a default noise
+            model, this option signals the backend to prefer it over the noise
+            model given with this configuration.
+        noise_model: An optional noise model to emulate the sequence with.
+            Ignored if the sequence's device has default noise model and
+            `prefer_device_noise_model=True`.
     """
 
-    backend_options: dict[str, Any] = field(default_factory=dict)
+    observables: Sequence[Observable]
+    default_evaluation_times: np.ndarray | Literal["Full"]
+    initial_state: StateType | None
+    with_modulation: bool
+    interaction_matrix: pm.AbstractArray | None
+    prefer_device_noise_model: bool
+    noise_model: NoiseModel
+
+    def __init__(
+        self,
+        *,
+        observables: Sequence[Observable] = (),
+        # Default evaluation times for observables that don't specify one
+        default_evaluation_times: Sequence[SupportsFloat] | Literal["Full"] = (
+            1.0,
+        ),
+        initial_state: StateType | None = None,  # Default is ggg...
+        with_modulation: bool = False,
+        interaction_matrix: ArrayLike | None = None,
+        prefer_device_noise_model: bool = False,
+        noise_model: NoiseModel = NoiseModel(),
+        **backend_options: Any,
+    ) -> None:
+        """Initializes the EmulationConfig."""
+        obs_tags = []
+        for obs in observables:
+            if not isinstance(obs, Observable):
+                raise TypeError(
+                    "All entries in 'observables' must be instances of "
+                    f"Observable. Instead, got instance of type {type(obs)}."
+                )
+            obs_tags.append(obs.tag)
+        repeated_tags = [k for k, v in Counter(obs_tags).items() if v > 1]
+        if repeated_tags:
+            raise ValueError(
+                "Some of the provided 'observables' share identical tags. Use "
+                "'tag_suffix' when instantiating multiple instances of the "
+                "same observable so they can be distinguished. "
+                f"Repeated tags found: {repeated_tags}"
+            )
 
+        if default_evaluation_times != "Full":
+            eval_times_arr = np.array(default_evaluation_times, dtype=float)
+            if np.any((eval_times_arr < 0.0) | (eval_times_arr > 1.0)):
+                raise ValueError(
+                    "All evaluation times must be between 0. and 1."
+                )
+            default_evaluation_times = cast(Sequence[float], eval_times_arr)
+
+        if initial_state is not None and not isinstance(initial_state, State):
+            raise TypeError(
+                "When defined, 'initial_state' must be an instance of State;"
+                f" got object of type {type(initial_state)} instead."
+            )
+
+        if interaction_matrix is not None:
+            interaction_matrix = pm.AbstractArray(interaction_matrix)
+            _shape = interaction_matrix.shape
+            if len(_shape) != 2 or _shape[0] != _shape[1]:
+                raise ValueError(
+                    "'interaction_matrix' must be a square matrix. Instead, "
+                    f"an array of shape {_shape} was given."
+                )
+            if (
+                initial_state is not None
+                and _shape[0] != initial_state.n_qudits
+            ):
+                raise ValueError(
+                    f"The received interaction matrix of shape {_shape} is "
+                    "incompatible with the received initial state of "
+                    f"{initial_state.n_qudits} qudits."
+                )
 
-@dataclass(frozen=True)
+        if not isinstance(noise_model, NoiseModel):
+            raise TypeError(
+                "'noise_model' must be a NoiseModel instance,"
+                f" not {type(noise_model)}."
+            )
+
+        super().__init__(
+            observables=tuple(observables),
+            default_evaluation_times=default_evaluation_times,
+            initial_state=initial_state,
+            with_modulation=bool(with_modulation),
+            interaction_matrix=interaction_matrix,
+            prefer_device_noise_model=bool(prefer_device_noise_model),
+            noise_model=noise_model,
+            **backend_options,
+        )
+
+    def _expected_kwargs(self) -> set[str]:
+        return super()._expected_kwargs() | {
+            "observables",
+            "default_evaluation_times",
+            "initial_state",
+            "with_modulation",
+            "interaction_matrix",
+            "prefer_device_noise_model",
+            "noise_model",
+        }
+
+    def is_evaluation_time(self, t: float, tol: float = 1e-6) -> bool:
+        """Assesses whether a relative time is an evaluation time."""
+        return 0.0 <= t <= 1.0 and (
+            self.default_evaluation_times == "Full"
+            or self.is_time_in_evaluation_times(
+                t, self.default_evaluation_times, tol=tol
+            )
+        )
+
+    @staticmethod
+    def is_time_in_evaluation_times(
+        t: float, evaluation_times: ArrayLike, tol: float = 1e-6
+    ) -> bool:
+        """Checks if a time is within a collection of evaluation times."""
+        return bool(
+            np.any(np.abs(np.array(evaluation_times, dtype=float) - t) <= tol)
+        )
+
+
+# Legacy class
+
+
+@dataclass
 class EmulatorConfig(BackendConfig):
     """The configuration for emulator backends.
 
@@ -74,6 +280,7 @@ class EmulatorConfig(BackendConfig):
             `prefer_device_noise_model=True`.
     """
 
+    backend_options: dict[str, Any] = field(default_factory=dict)
     sampling_rate: float = 1.0
     evaluation_times: float | Sequence[float] | EVAL_TIMES_LITERAL = "Full"
     initial_state: Literal["all-ground"] | Sequence[complex] = "all-ground"
@@ -82,6 +289,7 @@ class EmulatorConfig(BackendConfig):
     noise_model: NoiseModel = field(default_factory=NoiseModel)
 
     def __post_init__(self) -> None:
+        # TODO: Deprecate
         if not (0 < self.sampling_rate <= 1.0):
             raise ValueError(
                 "The sampling rate (`sampling_rate` = "