qoolqit.execution

`execution`

Execute quantum programs on QPUs or local/remote emulators.

Modules:

BackendType –

A module gathering all available backends.
backends –

Backends to run quantum programs on local/remote emulators and QPUs.
compilation_functions –
job –

Job submission and results retrieval on local/remote emulators and QPUs.
sequence_compiler –

Classes:

BitStrings –

Stores bitstrings sampled from the state at the evaluation times.
CorrelationMatrix –

Stores the correlation matrix for the current state.
EmulationConfig –

Configures an emulation on a backend.
Energy –

Stores the energy of the system at the evaluation times.
EnergySecondMoment –

Stores the expectation value of H(t)^2 at the evaluation times.
EnergyVariance –

Stores the variance of the Hamiltonian at the evaluation times.
Expectation –

Stores the expectation of the given operator on the current state.
Fidelity –

Stores the fidelity with a pure state at the evaluation times.
Job –

Abstract representation of an asynchronous job or task.
LocalEmulator –

Run QoolQit QuantumPrograms on a Pasqal local emulator backends.
Occupation –

Stores the occupation number of an eigenstate on each qudit.
QPU –

Execute QoolQit QuantumPrograms on Pasqal quantum processing units.
RemoteEmulator –

Run QoolQit QuantumPrograms on a Pasqal remote emulator backends.
Results –

A collection of results.
StateResult –

Stores the quantum state at the evaluation times.

Functions:

get_batch_id –

Return the batch ID associated with a remote job.
retrieve_remote_job –

Retrieve a previously submitted remote job by its identifiers.

Attributes:

JobStatus (TypeAlias) –

Alias for :class:~pulser.backend.remote.JobStatus.

`JobStatus: TypeAlias = remote.JobStatus` `module-attribute`

Alias for :class:~pulser.backend.remote.JobStatus.

Terminal states (where :meth:Job.has_ended returns True):

:attr:~remote.JobStatus.DONE
:attr:~remote.JobStatus.ERROR
:attr:~remote.JobStatus.CANCELED

Returned by:

API reference qoolqit.execution execution Job get_status

`BitStrings(*, evaluation_times: Sequence[float] | None = None, num_shots: int | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)`

Stores bitstrings sampled from the state at the evaluation times.

Error rates are taken from the NoiseModel passed to the backend via the EmulationConfig. The bitstrings are stored as a Counter[str].

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to sample bitstrings. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
num_shots
(int | None, default: None ) –

How many bitstrings to sample each time this observable is computed. If left as None, will use default_num_shots of the backend's EmulationConfig.
one_state
(Eigenstate | None, default: None ) –

The eigenstate that measures to 1. Can be left undefined if the state's eigenstates form a known eigenbasis with a defined "one state".
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

num_shots (int | None) –

How many bitstrings to sample each time this observable is computed.
tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/default_observables.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    num_shots: int | None = None,
    one_state: Eigenstate | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__(
        evaluation_times=evaluation_times, tag_suffix=tag_suffix
    )
    # Relies on the validation in the setter
    self.num_shots = num_shots
    self.one_state = one_state

`num_shots: int | None` `property` `writable`

How many bitstrings to sample each time this observable is computed.

Warning

The default value for this parameter was changed from 1000 to None in Pulser v1.7. When left as None, it now relies on EmulationConfig.default_num_shots to decide how many shots to take.

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, config: EmulationConfig, state: State, **kwargs: Any) -> Counter[str]`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self,
    *,
    config: EmulationConfig,
    state: State,
    **kwargs: Any,
) -> Counter[str]:
    """Calculates the observable to store in the Results."""
    return state.sample(
        num_shots=(
            self._num_shots
            if self._num_shots is not None
            else config.default_num_shots
        ),
        one_state=self.one_state,
        p_false_pos=config.noise_model.p_false_pos,
        p_false_neg=config.noise_model.p_false_neg,
    )

`CorrelationMatrix(*, evaluation_times: Sequence[float] | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)`

Stores the correlation matrix for the current state.

The correlation matrix is calculated as [[<φ(t)|n_i n_j|φ(t)> for j in qubits] for i in qubits] where n_k = |one_state><one_state|.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the correlation matrix. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
one_state
(Eigenstate | None, default: None ) –

The eigenstate to measure the population of in the correlation matrix. Can be left undefined if the state's eigenstates form a known eigenbasis with a defined "one state".
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/default_observables.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    one_state: Eigenstate | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__(
        evaluation_times=evaluation_times, tag_suffix=tag_suffix
    )
    self.one_state = one_state

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> list[list]`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self, *, state: State, hamiltonian: Operator, **kwargs: Any
) -> list[list]:
    """Calculates the observable to store in the Results."""

    @functools.cache
    def calc_expectation(qudit_ids: frozenset[int]) -> Any:
        return self._get_number_operator(
            qudit_ids,
            state.n_qudits,
            state.eigenstates,
            self.one_state or state.infer_one_state(),
            type(hamiltonian),
        ).expect(state)

    return [
        [
            calc_expectation(frozenset((i, j)))
            for j in range(state.n_qudits)
        ]
        for i in range(state.n_qudits)
    ]

`EmulationConfig(*, callbacks: Sequence[Callback] = (), observables: Sequence[Observable] = (), default_evaluation_times: Sequence[SupportsFloat] | Literal['Full'] = (1.0,), initial_state: StateType | None = None, with_modulation: bool = False, interaction_matrix: ArrayLike | None = None, prefer_device_noise_model: bool = False, noise_model: NoiseModel | None = None, n_trajectories: int | None = None, default_num_shots: int = 1000, **backend_options: Any)`

Configures an emulation on a backend.

Note

Additional parameters may be provided. It is up to the emulation backend that receives a configuration with extra parameters to assess whether it recognizes them and how it will use them. To know all parameters expected by an EmulatorBackend, consult its associated EmulationConfig subclass found under EmulatorBackend.config_type.

Parameters:

observables
(Sequence[Observable], default: () ) –

A sequence of observables to compute at specific evaluation times. The observables without specified evaluation times will use this configuration's 'default_evaluation_times'.
callbacks
(Sequence[Callback], default: () ) –

A general callback that is not an observable. Observables must be fed into the observables arg, since they all interact with the Results, and are subject to additional validation. Unlike observables, these are called at every emulation step.
default_evaluation_times
(Sequence[SupportsFloat] | Literal['Full'], default: (1.0,) ) –

The default times at which observables are computed. Can be a sequence of unique relative times between 0 (the start of the sequence) and 1 (the end of the sequence), in ascending order. Can also be specified as "Full", in which case every step in the emulation will also be an evaluation time.
initial_state
(StateType | None, default: None ) –

The initial state from which emulation starts. If specified, the state type needs to be compatible with the emulator backend (i.e. it must match EmulatorBackend.config_type.state_type). If left undefined, defaults to starting with all qudits in the ground state.
with_modulation
(bool, default: False ) –

Whether to emulate the sequence with the programmed input or the expected output.
interaction_matrix
(ArrayLike | None, default: None ) –

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
(bool, default: False ) –

If True, uses the noise model of the sequence's device (if the sequence's device has one), regardless of the noise model given with this configuration.
noise_model
(NoiseModel | None, default: None ) –

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.
n_trajectories
(int | None, default: None ) –

The number of trajectories to average over when the emulation includes stochastic noise or is using a Monte Carlo solver. When left undefined:

(1) If 'prefer_device_noise_model=False', tries to use the (now deprecated) value of NoiseModel.runs. If that is also not defined, defaults to 1 trajectory.

(2) If 'prefer_device_noise_model=True', defaults to 40 trajectories.
default_num_shots
(int, default: 1000 ) –

The default number of shots for BitStrings, used whenever the observable doesn't define its own. Defaults to 1000.

Returned by:

API reference qoolqit.execution execution EmulationConfig from_abstract_repr

Used by:

API reference qoolqit.execution execution

Methods:

from_abstract_repr –

Deserialize an EmulationConfig from an abstract JSON object.
is_evaluation_time –

Assesses whether a relative time is an evaluation time.
is_time_in_evaluation_times –

Checks if a time is within a collection of evaluation times.
operator_type –

The preferred operator type to use with this config class.
state_type –

The preferred state type to use with this config class.
to_abstract_repr –

Serialize EmulationConfig to a JSON formatted str.
with_changes –

Returns a copy of the config with the given changes.

Attributes:

callbacks (Sequence[Callback]) –

A sequence of callbacks that are not observables.
default_evaluation_times (NDArray[floating[Any]] | Literal['Full']) –

The default times at which observables are computed.
default_num_shots (int) –

The default number of shots for BitStrings.
initial_state (StateType | None) –

A custom initial state from which emulation starts.
interaction_matrix (AbstractArray | None) –

An interaction matrix to replace the Hamiltonian interaction terms.
n_trajectories (int) –

The number of trajectories to average over (when applicable).
noise_model (NoiseModel) –

A custom noise model to emulate the sequence with.
observables (Sequence[Observable]) –

A sequence of observables to compute at specific evaluation times.
prefer_device_noise_model (bool) –

Whether to use the noise model of the sequence's Device.
with_modulation (bool) –

Whether to emulate the sequence with output modulation effects.

Source code in pulser/backend/config.py

def __init__(
    self,
    *,
    callbacks: Sequence[Callback] = (),
    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 | None = None,
    n_trajectories: int | None = None,
    default_num_shots: int = 1000,
    **backend_options: Any,
) -> None:
    """Initializes the EmulationConfig."""
    obs_tags = []
    if not observables and not callbacks:
        warnings.warn(
            f"{self.__class__.__name__!r} was initialized without any "
            "observables. The corresponding emulation results will be"
            " empty.",
            stacklevel=2,
        )
    for cb in callbacks:
        if isinstance(cb, Observable):
            raise TypeError(
                "All entries in 'callbacks' must not be instances of "
                "Observable, since those go in 'observables'."
            )
        if not isinstance(cb, Callback):
            raise TypeError(
                "All entries in 'callbacks' must be instances of "
                f"Callback. Instead, got instance of type {type(cb)}."
            )
    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 not (
        isinstance(default_evaluation_times, str)
        and default_evaluation_times == "Full"
    ):
        eval_times_arr = Observable._validate_eval_times(
            list(map(float, default_evaluation_times))
        )
        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."
            )
        matrix_arr = interaction_matrix.as_array(detach=True)
        if not np.allclose(matrix_arr, matrix_arr.transpose()):
            raise ValueError(
                "The received interaction matrix is not symmetric."
            )
        if np.any(np.diag(matrix_arr) != 0):
            warnings.warn(
                "The received interaction matrix has non-zero values in "
                "its diagonal; keep in mind that these values are "
                "ignored.",
                stacklevel=2,
            )

    if noise_model is None:
        noise_model = NoiseModel()
    elif not isinstance(noise_model, NoiseModel):
        raise TypeError(
            "When defined, 'noise_model' must be a NoiseModel instance,"
            f" not {type(noise_model)}."
        )

    if (
        n_trajectories is not None
        and noise_model.runs is not None
        and n_trajectories != noise_model.runs
    ):
        raise ValueError(
            "`EmulationConfig.n_trajectories` and `NoiseModel.runs` "
            "can't be simultaneously defined. Please favour using only"
            " `EmulationConfig.n_trajectories`."
        )

    if n_trajectories is None:
        if prefer_device_noise_model:
            n_trajectories = DEFAULT_N_TRAJECTORIES
        else:
            n_trajectories = (
                noise_model.runs if noise_model.runs is not None else 1
            )

    if n_trajectories < 1 or n_trajectories != int(n_trajectories):
        raise ValueError(
            "`n_trajectories` must be a strictly positive integer, "
            f"not {n_trajectories}."
        )

    super().__init__(
        callbacks=tuple(callbacks),
        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,
        n_trajectories=int(n_trajectories),
        default_num_shots=int(default_num_shots),
        **backend_options,
    )

`callbacks: Sequence[Callback]` `instance-attribute`

A sequence of callbacks that are not observables.

`default_evaluation_times: NDArray[np.floating[Any]] | Literal['Full']` `instance-attribute`

The default times at which observables are computed.

`default_num_shots: int` `instance-attribute`

The default number of shots for BitStrings.

`initial_state: StateType | None` `instance-attribute`

A custom initial state from which emulation starts.

`interaction_matrix: pm.AbstractArray | None` `instance-attribute`

An interaction matrix to replace the Hamiltonian interaction terms.

`n_trajectories: int` `instance-attribute`

The number of trajectories to average over (when applicable).

`noise_model: NoiseModel` `instance-attribute`

A custom noise model to emulate the sequence with.

`observables: Sequence[Observable]` `instance-attribute`

A sequence of observables to compute at specific evaluation times.

`prefer_device_noise_model: bool` `instance-attribute`

Whether to use the noise model of the sequence's Device.

`with_modulation: bool` `instance-attribute`

Whether to emulate the sequence with output modulation effects.

`from_abstract_repr(obj_str: str) -> EmulationConfig` `classmethod`

Deserialize an EmulationConfig from an abstract JSON object.

Parameters:

obj_str
(str) –

The JSON string representing the config encoded in the abstract JSON format.

Returns:

EmulationConfig ( EmulationConfig ) –

The EmulationConfig instance.

Source code in pulser/backend/config.py

@classmethod
def from_abstract_repr(cls, obj_str: str) -> EmulationConfig:
    """Deserialize an EmulationConfig from an abstract JSON object.

    Args:
        obj_str (str): The JSON string representing the config encoded
            in the abstract JSON format.

    Returns:
        EmulationConfig: The EmulationConfig instance.
    """
    if not isinstance(obj_str, str):
        raise TypeError(
            "The serialized EmulationConfig must be given as a string. "
            f"Instead, got object of type {type(obj_str)}."
        )
    validate_abstract_repr(obj_str, "config")
    return _deserialize_emulation_config(
        json.loads(obj_str),
        cls,
        cls.state_type,
        cls.operator_type,
    )

`is_evaluation_time(t: float, tol: float = 1e-06) -> bool`

Assesses whether a relative time is an evaluation time.

Source code in pulser/backend/config.py

def is_evaluation_time(self, t: float, tol: float = 1e-6) -> bool:
    """Assesses whether a relative time is an evaluation time."""
    return (
        self.default_evaluation_times == "Full" and 0.0 <= t <= 1.0
    ) or (
        self.is_time_in_evaluation_times(
            t, self.default_evaluation_times, tol=tol
        )
    )

`is_time_in_evaluation_times(t: float, evaluation_times: ArrayLike, tol: float = 1e-06) -> bool` `staticmethod`

Checks if a time is within a collection of evaluation times.

Source code in pulser/backend/config.py

@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 0.0 <= t <= 1.0 and bool(
        np.any(np.abs(np.array(evaluation_times, dtype=float) - t) <= tol)
    )

`operator_type() -> Type[Operator]`

The preferred operator type to use with this config class.

Source code in pulser/backend/config.py

@classproperty
def operator_type(cls) -> Type[Operator]:
    """The preferred operator type to use with this config class."""
    return cls._operator_type

`state_type() -> Type[State]`

The preferred state type to use with this config class.

Source code in pulser/backend/config.py

@classproperty
def state_type(cls) -> Type[State]:
    """The preferred state type to use with this config class."""
    return cls._state_type

`to_abstract_repr(skip_validation: bool = False) -> str`

Serialize EmulationConfig to a JSON formatted str.

Source code in pulser/backend/config.py

def to_abstract_repr(self, skip_validation: bool = False) -> str:
    """Serialize `EmulationConfig` to a JSON formatted str."""
    obj_str = json.dumps(self, cls=AbstractReprEncoder)
    if not skip_validation:
        validate_abstract_repr(obj_str, "config")
    return obj_str

`with_changes(**changes: Any) -> Self`

Returns a copy of the config with the given changes.

Source code in pulser/backend/config.py

def with_changes(self: Self, **changes: Any) -> Self:
    """Returns a copy of the config with the given changes."""
    return type(self)(**(self._backend_options | changes))

`Energy(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the energy of the system at the evaluation times.

The energy is calculated as the expectation value of the Hamiltonian, i.e. <φ(t)|H(t)|φ(t)>.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the energy. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/observable.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__()
    self.evaluation_times = (
        self._validate_eval_times(evaluation_times)
        if evaluation_times is not None
        else None
    )
    self._tag_suffix = tag_suffix

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> Any`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self, *, state: State, hamiltonian: Operator, **kwargs: Any
) -> Any:
    """Calculates the observable to store in the Results."""
    return hamiltonian.expect(state)

`EnergySecondMoment(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the expectation value of H(t)^2 at the evaluation times.

Useful for computing the variance when averaging over many executions of the program.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the variance. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/observable.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__()
    self.evaluation_times = (
        self._validate_eval_times(evaluation_times)
        if evaluation_times is not None
        else None
    )
    self._tag_suffix = tag_suffix

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> Any`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self, *, state: State, hamiltonian: Operator, **kwargs: Any
) -> Any:
    """Calculates the observable to store in the Results."""
    # This works for state vectors and density matrices and avoids
    # squaring the hamiltonian
    h_state = hamiltonian.apply_to(state)
    identity = hamiltonian.from_operator_repr(
        eigenstates=state.eigenstates,
        n_qudits=state.n_qudits,
        operations=[(1.0, [])],
    )
    return identity.expect(h_state)

`EnergyVariance(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the variance of the Hamiltonian at the evaluation times.

The variance of the Hamiltonian at time t is calculated by <φ(t)|H(t)^2|φ(t)> - <φ(t)|H(t)|φ(t)>^2

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the variance. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/observable.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__()
    self.evaluation_times = (
        self._validate_eval_times(evaluation_times)
        if evaluation_times is not None
        else None
    )
    self._tag_suffix = tag_suffix

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> Any`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self, *, state: State, hamiltonian: Operator, **kwargs: Any
) -> Any:
    """Calculates the observable to store in the Results."""
    # This works for state vectors and density matrices and avoids
    # squaring the hamiltonian
    h_state = hamiltonian.apply_to(state)
    identity = hamiltonian.from_operator_repr(
        eigenstates=state.eigenstates,
        n_qudits=state.n_qudits,
        operations=[(1.0, [])],
    )
    return identity.expect(h_state) - hamiltonian.expect(state) ** 2

`Expectation(operator: Operator, *, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the expectation of the given operator on the current state.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the expectation value. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
operator
(Operator) –

The operator to measure. Must be of the appropriate type for the backend.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/default_observables.py

def __init__(
    self,
    operator: Operator,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__(
        evaluation_times=evaluation_times, tag_suffix=tag_suffix
    )
    if not isinstance(operator, Operator):
        raise TypeError(
            "'operator' must be an Operator instance;"
            f" got {type(operator)} instead."
        )
    self.operator = operator

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, **kwargs: Any) -> Any`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(self, *, state: State, **kwargs: Any) -> Any:
    """Calculates the observable to store in the Results."""
    return self.operator.expect(state)

`Fidelity(state: State, *, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the fidelity with a pure state at the evaluation times.

Parameters:

state
(State) –

The state |ψ>. Note that this must be of an appropriate type for the backend.
evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the fidelity. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/default_observables.py

def __init__(
    self,
    state: State,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__(
        evaluation_times=evaluation_times, tag_suffix=tag_suffix
    )
    if not isinstance(state, State):
        raise TypeError(
            f"'state' must be a State instance; got {type(state)} instead."
        )
    self.state = state

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, **kwargs: Any) -> Any`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(self, *, state: State, **kwargs: Any) -> Any:
    """Calculates the observable to store in the Results."""
    return self.state.overlap(state)

`Job`

Abstract representation of an asynchronous job or task.

A :class:Job tracks the lifecycle of a unit of work submitted for execution. It provides methods to inspect its state, wait for its result, and cancel it.

Lifecycle::

PENDING -> RUNNING -> DONE
                   -> ERROR
                   -> CANCELED

Concrete backend implementations (:class:_LocalJob, :class:_RemoteJob) must override all abstract methods. The leading underscore in their names indicates that they are internal to the library and must not be instantiated or referenced directly by library consumers. Instead, obtain a :class:Job from a backend submission function such as backend.run(program).

Note

A submitted job is not necessarily asynchronous. Whether execution is synchronous or asynchronous depends on the function that creates and returns the :class:Job instance (e.g. a local backend may execute synchronously and return an already-completed job, while a remote backend submits the work and returns a job that must be polled).

Returned by:

API reference qoolqit.execution execution retrieve_remote_job

Used by:

API reference qoolqit.execution execution get_batch_id

Methods:

cancel –

Request cancellation of the job.
get_status –

Return the current lifecycle state of the job.
has_ended –

Return whether the job has reached a terminal state.
job_id –

Return the unique identifier of the job.
results –

Block until the job completes and return its result.

`cancel() -> None` `abstractmethod`

Request cancellation of the job.

Has no effect if the job has already reached a terminal state. After a successful cancellation, :meth:get_status returns :attr:~remote.JobStatus.CANCELED and :meth:has_ended returns True.

Note

Some implementations do not support cancellation. In that case this method is a no-op and the job status will never be :attr:~remote.JobStatus.CANCELED.

Source code in qoolqit/execution/job.py

@abstractmethod
def cancel(self) -> None:
    """Request cancellation of the job.

    Has no effect if the job has already reached a terminal state.
    After a successful cancellation, :meth:`get_status` returns
    :attr:`~remote.JobStatus.CANCELED` and :meth:`has_ended` returns
    ``True``.

    Note:
        Some implementations do not support cancellation. In that
        case this method is a no-op and the job status will never
        be :attr:`~remote.JobStatus.CANCELED`.
    """
    ...

`get_status() -> JobStatus` `abstractmethod`

Return the current lifecycle state of the job.

Returns:

JobStatus ( JobStatus ) –

The current state of the job.

Source code in qoolqit/execution/job.py

@abstractmethod
def get_status(self) -> JobStatus:
    """Return the current lifecycle state of the job.

    Returns:
        JobStatus: The current state of the job.
    """
    ...

`has_ended() -> bool`

Return whether the job has reached a terminal state.

A job has ended when its status is one of :attr:~remote.JobStatus.DONE, :attr:~remote.JobStatus.ERROR, or :attr:~remote.JobStatus.CANCELED.

Returns:

bool ( bool ) –

True if the job is in a terminal state,
bool –

False otherwise.

Source code in qoolqit/execution/job.py

def has_ended(self) -> bool:
    """Return whether the job has reached a terminal state.

    A job has ended when its status is one of
    :attr:`~remote.JobStatus.DONE`, :attr:`~remote.JobStatus.ERROR`, or
    :attr:`~remote.JobStatus.CANCELED`.

    Returns:
        bool: ``True`` if the job is in a terminal state,
        ``False`` otherwise.
    """
    return self.get_status() in {
        JobStatus.DONE,
        JobStatus.ERROR,
        JobStatus.CANCELED,
    }

`job_id() -> str` `abstractmethod`

Return the unique identifier of the job.

For remote jobs this is the UUID used to track the job in the PASQAL Cloud Portal. For local jobs this is currently an empty string.

Returns:

str ( str ) –

A string uniquely identifying this job.

Source code in qoolqit/execution/job.py

@abstractmethod
def job_id(self) -> str:
    """Return the unique identifier of the job.

    For remote jobs this is the UUID used to track the job in
    the PASQAL Cloud Portal. For local jobs this is currently
    an empty string.

    Returns:
        str: A string uniquely identifying this job.
    """
    ...

`results(timeout: float = math.inf) -> ResultType` `abstractmethod`

Block until the job completes and return its result.

Waits until the job reaches a terminal state, then returns the result. Polling strategy (interval, backoff) is delegated to the underlying connection.

Parameters:

timeout
(float, default: inf ) –

Maximum seconds to wait. Defaults to math.inf (wait indefinitely).

Returns:

ResultType ( ResultType ) –

The result produced by the job.

Raises:

TimeoutError –

If the job does not complete within timeout seconds.
JobFailedError –

If the job reached a failed or error state.
JobCancelledError –

If the job was cancelled.

Source code in qoolqit/execution/job.py

@abstractmethod
def results(self, timeout: float = math.inf) -> ResultType:
    """Block until the job completes and return its result.

    Waits until the job reaches a terminal state, then returns
    the result. Polling strategy (interval, backoff) is delegated
    to the underlying connection.

    Args:
        timeout (float): Maximum seconds to wait.
            Defaults to ``math.inf`` (wait indefinitely).

    Returns:
        ResultType: The result produced by the job.

    Raises:
        TimeoutError: If the job does not complete within
            *timeout* seconds.
        JobFailedError: If the job reached a failed or error state.
        JobCancelledError: If the job was cancelled.
    """
    ...

`LocalEmulator(*, backend_type: type[EmulatorBackend] = QutipBackendV2, emulation_config: EmulationConfig | None = None, num_shots: int | None = None)`

Run QoolQit QuantumPrograms on a Pasqal local emulator backends.

This class serves as a primary interface between tools written using QoolQit (including solvers) and local emulator backends.

Parameters:

backend_type
(type, default: QutipBackendV2 ) –

backend type. Must be a subtype of pulser.backend.EmulatorBackend.
emulation_config
(EmulationConfig, default: None ) –

optional configuration object emulators.
num_shots
(int, default: None ) –

number of bitstring samples to collect from the final quantum state.

Examples:

from qoolqit.execution import LocalEmulator, BackendType
backend = LocalEmulator(backend_type=BackendType.QutipBackendV2)
result = backend.run(program)

Methods:

default_emulation_config –

Return a unique emulation config for all emulators.
run –

Run a compiled QuantumProgram and return the results.
validate_emulation_config –

Returns a valid config for emulator backends, if needed.

Source code in qoolqit/execution/backends.py

def __init__(
    self,
    *,
    backend_type: type[EmulatorBackend] = QutipBackendV2,
    emulation_config: EmulationConfig | None = None,
    num_shots: int | None = None,
) -> None:
    """Instantiates a LocalEmulator."""
    super().__init__(num_shots=num_shots)
    if not issubclass(backend_type, EmulatorBackend):
        raise TypeError(
            "Error in `LocalEmulator`: `backend_type` must be a EmulatorBackend type."
        )
    self._backend_type = backend_type
    self._emulation_config = self.validate_emulation_config(emulation_config)

`default_emulation_config() -> EmulationConfig`

Return a unique emulation config for all emulators.

Defaults to a configuration that asks for the final bitstring, sampled num_shots times.

Source code in qoolqit/execution/backends.py

def default_emulation_config(self) -> EmulationConfig:
    """Return a unique emulation config for all emulators.

    Defaults to a configuration that asks for the final bitstring, sampled `num_shots` times.
    """
    return EmulationConfig(observables=(BitStrings(num_shots=self._num_shots),))

`run(program: QuantumProgram) -> job.Job[Results]`

Run a compiled QuantumProgram and return the results.

Source code in qoolqit/execution/backends.py

def run(self, program: QuantumProgram) -> job.Job[Results]:
    """Run a compiled QuantumProgram and return the results."""
    self._backend = self._backend_type(program.compiled_sequence, config=self._emulation_config)
    results = self._backend.run()
    assert isinstance(results, Results)
    return job._LocalJob(results)

`validate_emulation_config(emulation_config: EmulationConfig | None) -> EmulationConfig`

Returns a valid config for emulator backends, if needed.

Parameters:

emulation_config
(EmulationConfig | None) –

optional base configuration class for all emulators backends. If no config is provided to an emulator backend, the backend default will used.

Source code in qoolqit/execution/backends.py

def validate_emulation_config(
    self, emulation_config: EmulationConfig | None
) -> EmulationConfig:
    """Returns a valid config for emulator backends, if needed.

    Args:
        emulation_config: optional base configuration class for all emulators backends.
            If no config is provided to an emulator backend, the backend default will used.
    """
    if emulation_config is None:
        emulation_config = self.default_emulation_config()
    else:
        has_bitstrings = any(
            isinstance(obs, BitStrings) for obs in emulation_config.observables
        )
        if not has_bitstrings:
            updated_obs = (*emulation_config.observables, BitStrings(num_shots=self._num_shots))
            emulation_config = emulation_config.with_changes(observables=updated_obs)

    if self._num_shots is not None:
        emulation_config = emulation_config.with_changes(default_num_shots=self._num_shots)

    return emulation_config

`Occupation(*, evaluation_times: Sequence[float] | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)`

Stores the occupation number of an eigenstate on each qudit.

For every qudit i, calculates <φ(t)|n_i|φ(t)>, where n_i = |one_state><one_state|.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to compute the correlation matrix. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
one_state
(Eigenstate | None, default: None ) –

The eigenstate to measure the population of. Can be left undefined if the state's eigenstates form a known eigenbasis with a defined "one state".
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/default_observables.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    one_state: Eigenstate | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__(
        evaluation_times=evaluation_times, tag_suffix=tag_suffix
    )
    self.one_state = one_state

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> list`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(
    self, *, state: State, hamiltonian: Operator, **kwargs: Any
) -> list:
    """Calculates the observable to store in the Results."""
    return [
        CorrelationMatrix._get_number_operator(
            frozenset((i,)),
            state.n_qudits,
            state.eigenstates,
            self.one_state or state.infer_one_state(),
            type(hamiltonian),
        ).expect(state)
        for i in range(state.n_qudits)
    ]

`QPU(*, connection: RemoteConnection, num_shots: int | None = None)`

Execute QoolQit QuantumPrograms on Pasqal quantum processing units.

This class provides the primary interface for running quantum programs on actual QPU hardware. It requires authenticated credentials through a connection object to submit and execute programs on remote quantum processors.

Parameters:

connection
(RemoteConnection) –

Authenticated connection to the remote QPU backend.
num_shots
(int | None, default: None ) –

Number of bitstring samples to collect from the final quantum state.

Examples:

Using Pasqal Cloud:

from pulser_pasqal import PasqalCloud
from qoolqit.execution import QPU

connection = PasqalCloud(
    username="your_username",
    password="your_password",
    project_id="your_project_id"
)
backend = QPU(connection=connection)
remote_results = backend.submit(program)

Using Atos MyQML:

from pulser_myqlm import PulserQLMConnection
from qoolqit.execution import QPU

connection = PulserQLMConnection()
backend = QPU(connection=connection)
results = backend.run(program)

Note

Contact your quantum computing provider for credentials and connection setup: - Pasqal Cloud Documentation - Atos MyQML Framework

Methods:

run –

Run a compiled QuantumProgram and return a job handler.
validate_connection –

Validate the required connection to instantiate a RemoteBackend.

Source code in qoolqit/execution/backends.py

def __init__(
    self,
    *,
    connection: RemoteConnection,
    num_shots: int | None = None,
) -> None:
    """Instantiates a QPU backend."""
    self._backend_type = QPUBackend
    self._connection = self.validate_connection(connection)
    if num_shots is None:
        raise ValueError(
            """Number of shots must be provided to use the QPU backend.
            Please specify `num_shots` when creating the QPU instance.
            For example: QPU(connection=..., num_shots=100)""",
        )
    self._config = BackendConfig(default_num_shots=num_shots)

`run(program: QuantumProgram) -> job.Job[Results]`

Run a compiled QuantumProgram and return a job handler.

The returned handler Job can be used to retrieve results with job.results()

Parameters:

program
(QuantumProgram) –

the compiled quantum program to run.

Source code in qoolqit/execution/backends.py

def run(self, program: QuantumProgram) -> job.Job[Results]:
    """Run a compiled QuantumProgram and return a job handler.

    The returned handler `Job` can be used to retrieve results with `job.results()`

    Args:
        program (QuantumProgram): the compiled quantum program to run.
    """
    self._backend = self._backend_type(
        program.compiled_sequence, connection=self._connection, config=self._config
    )
    remote_results = self._backend.run(wait=False)
    # If in open-batch mode, the job should be the last one in the batch ?
    return job._RemoteJob._from_remote_results(remote_results, -1)

`validate_connection(connection: RemoteConnection) -> RemoteConnection` `staticmethod`

Validate the required connection to instantiate a RemoteBackend.

Remote emulators and QPUs require a pulser.backend.remote.RemoteConnection or derived to send jobs. Validation also happens inside the backend. Early validation just makes the error easier to understand.

Source code in qoolqit/execution/backends.py

@staticmethod
def validate_connection(connection: RemoteConnection) -> RemoteConnection:
    """Validate the required connection to instantiate a RemoteBackend.

    Remote emulators and QPUs require a `pulser.backend.remote.RemoteConnection` or derived
    to send jobs. Validation also happens inside the backend. Early validation just makes the
    error easier to understand.
    """
    if not isinstance(connection, RemoteConnection):
        raise TypeError(f"""Error in `PulserRemoteBackend`:
            `connection` must be of type {RemoteConnection}.""")
    return connection

`RemoteEmulator(*, backend_type: type[RemoteEmulatorBackend] = EmuFreeBackendV2, connection: RemoteConnection, emulation_config: EmulationConfig | None = None, num_shots: int | None = None)`

Run QoolQit QuantumPrograms on a Pasqal remote emulator backends.

This class serves as a primary interface between tools written using QoolQit (including solvers) and remote emulator backends. The behavior is similar to LocalEmulator, but here, requires credentials through a connection to submit/run a program. To get your credentials and to create a connection object, please refer to the Pasqal Cloud interface documentation.

Parameters:

backend_type
(type, default: EmuFreeBackendV2 ) –

backend type. Must be a subtype of pulser_pasqal.backends.RemoteEmulatorBackend.
connection
(RemoteConnection) –

connection to execute the program on remote backends.
emulation_config
(EmulationConfig, default: None ) –

optional configuration object emulators.
num_shots
(int, default: None ) –

number of bitstring samples to collect from the final quantum state.

Examples:

from pulser_pasqal import PasqalCloud
from qoolqit.execution import RemoteEmulator, BackendType
connection = PasqalCloud(username=..., password=..., project_id=...)
backend = RemoteEmulator(backend_type=BackendType.EmuFreeBackendV2, connection=connection)

then

remote_results = backend.submit(program)

or

results = backend.run(program)

Methods:

default_emulation_config –

Return a unique emulation config for all emulators.
run –

Run a compiled QuantumProgram and return a job handler.
validate_connection –

Validate the required connection to instantiate a RemoteBackend.
validate_emulation_config –

Returns a valid config for emulator backends, if needed.

Source code in qoolqit/execution/backends.py

def __init__(
    self,
    *,
    backend_type: type[RemoteEmulatorBackend] = EmuFreeBackendV2,
    connection: RemoteConnection,
    emulation_config: EmulationConfig | None = None,
    num_shots: int | None = None,
) -> None:
    """Instantiates a RemoteEmulator."""
    super().__init__(num_shots=num_shots)
    if not issubclass(backend_type, RemoteEmulatorBackend):
        raise TypeError(
            "Error in `RemoteEmulator`: `backend_type` must be a RemoteEmulatorBackend type."
        )
    self._backend_type = backend_type
    self._connection = self.validate_connection(connection)
    self._emulation_config = self.validate_emulation_config(emulation_config)

`default_emulation_config() -> EmulationConfig`

Return a unique emulation config for all emulators.

Defaults to a configuration that asks for the final bitstring, sampled num_shots times.

Source code in qoolqit/execution/backends.py

def default_emulation_config(self) -> EmulationConfig:
    """Return a unique emulation config for all emulators.

    Defaults to a configuration that asks for the final bitstring, sampled `num_shots` times.
    """
    return EmulationConfig(observables=(BitStrings(num_shots=self._num_shots),))

`run(program: QuantumProgram) -> job.Job[Results]`

Run a compiled QuantumProgram and return a job handler.

The returned handler Job can be used to retrieve results with job.results()

Parameters:

program
(QuantumProgram) –

the compiled quantum program to run.

Source code in qoolqit/execution/backends.py

def run(self, program: QuantumProgram) -> job.Job[Results]:
    """Run a compiled QuantumProgram and return a job handler.

    The returned handler `Job` can be used to retrieve results with `job.results()`

    Args:
        program (QuantumProgram): the compiled quantum program to run.
    """
    # Instantiate backend
    self._backend = self._backend_type(
        program.compiled_sequence,
        connection=self._connection,
        config=self._emulation_config,
    )
    remote_results = self._backend.run(wait=False)
    # If in open-batch mode, the job should be the last one in the batch ?
    return job._RemoteJob._from_remote_results(remote_results, -1)

`validate_connection(connection: RemoteConnection) -> RemoteConnection` `staticmethod`

Validate the required connection to instantiate a RemoteBackend.

Remote emulators and QPUs require a pulser.backend.remote.RemoteConnection or derived to send jobs. Validation also happens inside the backend. Early validation just makes the error easier to understand.

Source code in qoolqit/execution/backends.py

@staticmethod
def validate_connection(connection: RemoteConnection) -> RemoteConnection:
    """Validate the required connection to instantiate a RemoteBackend.

    Remote emulators and QPUs require a `pulser.backend.remote.RemoteConnection` or derived
    to send jobs. Validation also happens inside the backend. Early validation just makes the
    error easier to understand.
    """
    if not isinstance(connection, RemoteConnection):
        raise TypeError(f"""Error in `PulserRemoteBackend`:
            `connection` must be of type {RemoteConnection}.""")
    return connection

`validate_emulation_config(emulation_config: EmulationConfig | None) -> EmulationConfig`

Returns a valid config for emulator backends, if needed.

Parameters:

emulation_config
(EmulationConfig | None) –

optional base configuration class for all emulators backends. If no config is provided to an emulator backend, the backend default will used.

Source code in qoolqit/execution/backends.py

def validate_emulation_config(
    self, emulation_config: EmulationConfig | None
) -> EmulationConfig:
    """Returns a valid config for emulator backends, if needed.

    Args:
        emulation_config: optional base configuration class for all emulators backends.
            If no config is provided to an emulator backend, the backend default will used.
    """
    if emulation_config is None:
        emulation_config = self.default_emulation_config()
    else:
        has_bitstrings = any(
            isinstance(obs, BitStrings) for obs in emulation_config.observables
        )
        if not has_bitstrings:
            updated_obs = (*emulation_config.observables, BitStrings(num_shots=self._num_shots))
            emulation_config = emulation_config.with_changes(observables=updated_obs)

    if self._num_shots is not None:
        emulation_config = emulation_config.with_changes(default_num_shots=self._num_shots)

    return emulation_config

`Results(atom_order: tuple[str, ...], total_duration: int)` `dataclass`

A collection of results.

Parameters:

atom_order
(tuple[str, ...]) –

The order of the atoms/qudits in the results.
total_duration
(int) –

The total duration of the sequence, in ns.

Returned by:

API reference qoolqit.execution execution
- Results
  - aggregate
  - from_abstract_repr
- retrieve_remote_job

Used by:

API reference qoolqit.execution execution

Methods:

aggregate –

Aggregate a Sequence of Results objects into a single Results.
from_abstract_repr –

Deserializes a Results object from json.
get_result –

Get the a specific result at a given time.
get_result_tags –

Get a list of results tags present in this object.
get_result_times –

Get a list of times for which the given result has been stored.
get_tagged_results –

Gets the results for every tag.
to_abstract_repr –

Serializes this object into a json string.

Attributes:

atom_order (tuple[str, ...]) –

The order of the atoms/qudits in the results.
final_bitstrings (dict[str, int]) –

The bitstrings at the end of the sequence, if available.
final_state (State) –

The state at the end of the sequence, if available.
total_duration (int) –

The total duration of the sequence, in ns.

`atom_order: tuple[str, ...]` `instance-attribute`

The order of the atoms/qudits in the results.

`final_bitstrings: dict[str, int]` `property`

The bitstrings at the end of the sequence, if available.

`final_state: State` `property`

The state at the end of the sequence, if available.

`total_duration: int` `instance-attribute`

The total duration of the sequence, in ns.

`aggregate(results_to_aggregate: typing.Sequence[Results], **aggregation_functions: Callable[[Any], Any]) -> Results` `classmethod`

Aggregate a Sequence of Results objects into a single Results.

This is meant to accumulate the results of several runs with different noise trajectories into a single averaged Results. By default, results are averaged, with the exception of BitStrings, where the counters are joined. StateResult and EnergyVariance are not supported by default.

Warning

The ability to access a result from an observable instance (e.g. via Results.get_result(obs)) is only preserved if all aggregated results originated from the same observable instance. When that is not the case, the aggregated result can still be accessed via the observable's tag.

Parameters:

results_to_aggregate
(Sequence[Results]) –

The list of Results to aggregate

Other Parameters:

observable_tag –

Overrides the default aggregator. The argument name should be the tag of the Observable. The value is a Callable taking a list of the type to aggregate. Note that this does not override the default aggregation behaviour of the aggregated results.

Returns:

Results –

The averaged Results object

Source code in pulser/backend/results.py

@classmethod
def aggregate(
    cls,
    results_to_aggregate: typing.Sequence[Results],
    **aggregation_functions: Callable[[Any], Any],
) -> Results:
    """Aggregate a Sequence of Results objects into a single Results.

    This is meant to accumulate the results of several runs with
    different noise trajectories into a single averaged Results.
    By default, results are averaged, with the exception of BitStrings,
    where the counters are joined.
    StateResult and EnergyVariance are not supported by default.

    Warning:
        The ability to access a result from an observable instance
        (e.g. via Results.get_result(obs)) is only preserved if all
        aggregated results originated from the same observable instance.
        When that is not the case, the aggregated result can still
        be accessed via the observable's tag.

    Args:
        results_to_aggregate: The list of Results to aggregate

    Keyword Args:
        observable_tag: Overrides the default aggregator.
            The argument name should be the tag of the Observable.
            The value is a Callable taking a list of the type to aggregate.
            Note that this does not override the default aggregation
            behaviour of the aggregated results.

    Returns:
        The averaged Results object
    """
    if len(results_to_aggregate) == 0:
        raise ValueError("No results to aggregate.")
    result_0 = results_to_aggregate[0]
    if len(results_to_aggregate) == 1:
        return result_0

    all_tags = set().union(
        *[set(x.get_result_tags()) for x in results_to_aggregate]
    )
    common_tags = all_tags.intersection(
        *[set(x.get_result_tags()) for x in results_to_aggregate]
    )

    for results in results_to_aggregate:
        if results._results and (not results._aggregation_methods):
            raise NotImplementedError(
                (
                    "You're trying to aggregate results from pulser<1.6,"
                    "aggregation is not supported in this case."
                )
            )
        for tag, uid in results._tagmap.items():
            if tag not in common_tags and not (
                results._aggregation_methods[uid].value
                in (AggregationMethod.SKIP, AggregationMethod.SKIP_WARN)
            ):
                raise ValueError(
                    "You're trying to aggregate incompatible results: "
                    f"result `{tag}` is not present in all results, "
                    "but it's not marked to be skipped."
                )
    if not all(
        {
            tag: results._aggregation_methods[results._find_uuid(tag)]
            for tag in common_tags
        }
        == {
            tag: result_0._aggregation_methods[result_0._find_uuid(tag)]
            for tag in common_tags
        }
        for results in results_to_aggregate
    ):
        raise ValueError(
            "You're trying to aggregate incompatible results: "
            "they do not all contain the same aggregation functions."
        )
    if not all(
        results.atom_order == result_0.atom_order
        for results in results_to_aggregate
    ):
        raise ValueError(
            "You're trying to aggregate incompatible results: "
            "they do not all have the same atom order."
        )
    if not all(
        results.total_duration == result_0.total_duration
        for results in results_to_aggregate
    ):
        raise ValueError(
            "You're trying to aggregate incompatible results: "
            "they do not all have the same sequence duration."
        )
    aggregated = Results(
        atom_order=result_0.atom_order,
        total_duration=result_0.total_duration,
    )
    for tag in common_tags:
        default_aggregation_method = result_0._aggregation_methods[
            result_0._tagmap[tag]
        ]
        aggregation_method = aggregation_functions.get(
            tag, default_aggregation_method
        )
        if (
            aggregation_method is AggregationMethod.SKIP
            or aggregation_method is AggregationMethod.SKIP_WARN
        ):
            if aggregation_method is AggregationMethod.SKIP_WARN:
                with warnings.catch_warnings():
                    warnings.simplefilter("once")
                    warnings.warn(f"Skipping aggregation of `{tag}`.")
            continue
        aggregation_function: Any = (
            AGGREGATOR_MAPPING[aggregation_method]
            if isinstance(aggregation_method, AggregationMethod)
            else aggregation_method
        )
        evaluation_times = results_to_aggregate[0].get_result_times(tag)
        if not all(
            results.get_result_times(tag) == evaluation_times
            for results in results_to_aggregate
        ):
            raise ValueError(
                "The Results come from "
                "incompatible simulations: "
                f"the times for `{tag}` are not all the same."
            )

        _uuids = set(res._tagmap[tag] for res in results_to_aggregate)
        if len(_uuids) == 1:
            # Preserve the UUID when all results share the same
            uid = list(_uuids)[0]
        else:
            # Otherwise, create a new one
            uid = uuid.uuid4()

        for t in result_0.get_result_times(tag):
            v = aggregation_function(
                [
                    result.get_result(tag, t)
                    for result in results_to_aggregate
                ]
            )

            aggregated._store_raw(
                uuid=uid,
                tag=tag,
                time=t,
                value=v,
                aggregation_method=default_aggregation_method,
            )

    return aggregated

`from_abstract_repr(repr: str) -> Results` `classmethod`

Deserializes a Results object from json.

Returns:

Results –

The deserialized Results object.

Source code in pulser/backend/results.py

@classmethod
def from_abstract_repr(cls, repr: str) -> Results:
    """Deserializes a Results object from json.

    Returns:
        The deserialized Results object.
    """
    validate_abstract_repr(repr, "results")
    d = json.loads(repr)
    return cls._from_abstract_repr(d)

`get_result(observable: Observable | str, time: float) -> Any`

Get the a specific result at a given time.

Parameters:

observable
(Observable | str) –

The observable instance used to calculate the result or its tag.
time
(float) –

Relative time at which to get the result.

Returns:

Any –

The result.

Source code in pulser/backend/results.py

def get_result(self, observable: Observable | str, time: float) -> Any:
    """Get the a specific result at a given time.

    Args:
        observable: The observable instance used to calculate the result
            or its tag.
        time: Relative time at which to get the result.

    Returns:
        The result.
    """
    obs_uuid = self._find_uuid(observable)
    try:
        ind = self._times[obs_uuid].index(time)
        return self._results[obs_uuid][ind]
    except (KeyError, ValueError):
        raise ValueError(
            f"{observable!r} is not available at time {time}."
        )

`get_result_tags() -> list[str]`

Get a list of results tags present in this object.

Source code in pulser/backend/results.py

def get_result_tags(self) -> list[str]:
    """Get a list of results tags present in this object."""
    return list(self._tagmap.keys())

`get_result_times(observable: Observable | str) -> list[float]`

Get a list of times for which the given result has been stored.

Parameters:

observable
(Observable | str) –

The observable instance used to calculate the result or its tag.

Returns:

list[float] –

List of relative times.

Source code in pulser/backend/results.py

def get_result_times(self, observable: Observable | str) -> list[float]:
    """Get a list of times for which the given result has been stored.

    Args:
        observable: The observable instance used to calculate the result
            or its tag.

    Returns:
        List of relative times.
    """
    return list(self._times[self._find_uuid(observable)])

`get_tagged_results() -> dict[str, list[Any]]`

Gets the results for every tag.

Returns:

dict[str, list[Any]] –

A mapping between a tag and the results associated to it,
dict[str, list[Any]] –

at every evaluation time.

Source code in pulser/backend/results.py

def get_tagged_results(self) -> dict[str, list[Any]]:
    """Gets the results for every tag.

    Returns:
        A mapping between a tag and the results associated to it,
        at every evaluation time.
    """
    return {
        tag: list(self._results[uuid_])
        for tag, uuid_ in self._tagmap.items()
    }

`to_abstract_repr(skip_validation: bool = False) -> str`

Serializes this object into a json string.

Numpy arrays and torch Tensors are converted into lists, and their original class is lost forever.

Parameters:

skip_validation
(bool, default: False ) –

Whether to skip validating the json against the schema used for deserialization.

Returns:

str –

The json string

Source code in pulser/backend/results.py

def to_abstract_repr(self, skip_validation: bool = False) -> str:
    """Serializes this object into a json string.

    Numpy arrays and torch Tensors are converted into lists,
    and their original class is lost forever.

    Args:
        skip_validation: Whether to skip validating the json against
            the schema used for deserialization.

    Returns:
        The json string
    """
    abstr_str = json.dumps(
        self._to_abstract_repr(), cls=AbstractReprEncoder
    )
    if not skip_validation:
        validate_abstract_repr(abstr_str, "results")
    return abstr_str

`StateResult(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

Stores the quantum state at the evaluation times.

Parameters:

evaluation_times
(Sequence[float] | None, default: None ) –

The relative times at which to store the state. If left as None, uses the default_evaluation_times of the backend's EmulationConfig.
tag_suffix
(str | None, default: None ) –

An optional suffix to append to the tag. Needed if multiple instances of the same observable are given to the same EmulationConfig.

Methods:

__call__ –

Specifies a call to the observable at a specific time.
apply –

Calculates the observable to store in the Results.

Attributes:

tag (str) –

Label for the observable, used to index the Results object.
uuid (UUID) –

A universal unique identifier for this instance.

Source code in pulser/backend/observable.py

def __init__(
    self,
    *,
    evaluation_times: Sequence[float] | None = None,
    tag_suffix: str | None = None,
):
    """Initializes the observable."""
    super().__init__()
    self.evaluation_times = (
        self._validate_eval_times(evaluation_times)
        if evaluation_times is not None
        else None
    )
    self._tag_suffix = tag_suffix

`tag: str` `property`

Label for the observable, used to index the Results object.

Within a Results instance, all computed observables must have different tags.

Returns:

str –

The tag of the observable.

`uuid: uuid.UUID` `property`

A universal unique identifier for this instance.

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

Specifies a call to the observable at a specific time.

This is called after each time step performed by the emulator. By default it calls apply() to compute a result and put it in Results if t in self.evaluation_times. It can be overloaded to define general callbacks that don't put results in the Results object.

Parameters:

config
(EmulationConfig) –

The config object passed to the backend.
t
(float) –

The relative time as a float between 0 and 1.
state
(State) –

The current state.
hamiltonian
(Operator) –

The Hamiltonian at this time.
result
(Results) –

The Results object to store the result in.

Source code in pulser/backend/observable.py

def __call__(
    self,
    config: EmulationConfig,
    t: float,
    state: State,
    hamiltonian: Operator,
    result: Results,
) -> None:
    """Specifies a call to the observable at a specific time.

    This is called after each time step performed by the emulator.
    By default it calls `apply()` to compute a result and put it in Results
    if t in self.evaluation_times.
    It can be overloaded to define general callbacks that don't put results
    in the Results object.

    Args:
        config: The config object passed to the backend.
        t: The relative time as a float between 0 and 1.
        state: The current state.
        hamiltonian: The Hamiltonian at this time.
        result: The Results object to store the result in.
    """
    time_tol = (
        (0.5 / result.total_duration) if result.total_duration else 1e-6
    )
    if (
        self.evaluation_times is not None
        and config.is_time_in_evaluation_times(
            t, self.evaluation_times, tol=time_tol
        )
    ) or (
        self.evaluation_times is None
        and config.is_evaluation_time(t, tol=time_tol)
    ):
        value_to_store = self.apply(
            config=config, state=state, hamiltonian=hamiltonian
        )
        result._store(observable=self, time=t, value=value_to_store)

`apply(*, state: StateType, **kwargs: Any) -> StateType`

Calculates the observable to store in the Results.

Source code in pulser/backend/default_observables.py

def apply(self, *, state: StateType, **kwargs: Any) -> StateType:
    """Calculates the observable to store in the Results."""
    return copy.deepcopy(state)

`get_batch_id(job: Job[Results]) -> str`

Return the batch ID associated with a remote job.

Temporary utility for cases where batch_id must be persisted and later passed to :func:retrieve_remote_job. Will be deprecated once RemoteConnection supports direct job lookup by job ID alone.

Parameters:

job
(Job[Results]) –

Any :class:Job instance.

Returns:

str ( str ) –

The batch ID if job is a :class:_RemoteJob,
str –

otherwise an empty string.

Source code in qoolqit/execution/job.py

def get_batch_id(job: Job[Results]) -> str:
    """Return the batch ID associated with a remote job.

    Temporary utility for cases where *batch_id* must be persisted
    and later passed to :func:`retrieve_remote_job`. Will be
    deprecated once ``RemoteConnection`` supports direct job lookup
    by job ID alone.

    Args:
        job (Job[Results]): Any :class:`Job` instance.

    Returns:
        str: The batch ID if *job* is a :class:`_RemoteJob`,
        otherwise an empty string.
    """
    if isinstance(job, _RemoteJob):
        return job._batch_id
    return ""

`retrieve_remote_job(connection: remote.RemoteConnection, job_id: str, *, batch_id: str = '') -> Job[Results]`

Retrieve a previously submitted remote job by its identifiers.

Use this to reconnect to a job in a new session, or to monitor a job submitted outside the current context.

Note

The batch_id argument is required by some RemoteConnection implementations (e.g. PasqalCloudConnection) to query job progress. Use :func:get_batch_id to recover the batch ID from an existing :class:Job if needed.

This dependency on batch_id is temporary and will be removed once RemoteConnection supports direct lookup by job ID alone.

Parameters:

connection
(RemoteConnection) –

The remote connection through which to query the job.
job_id
(str) –

The UUID of the job to retrieve.
batch_id
(str, default: '' ) –

The batch identifier the job belongs to. Defaults to an empty string.

Returns:

Job[Results] –

Job[Results]: A :class:_RemoteJob connected to the
Job[Results] –

specified job.

Source code in qoolqit/execution/job.py

def retrieve_remote_job(
    connection: remote.RemoteConnection,
    job_id: str,
    *,
    batch_id: str = "",
) -> Job[Results]:
    """Retrieve a previously submitted remote job by its identifiers.

    Use this to reconnect to a job in a new session, or to monitor a
    job submitted outside the current context.

    Note:
        The *batch_id* argument is required by some
        ``RemoteConnection`` implementations (e.g.
        ``PasqalCloudConnection``) to query job progress. Use
        :func:`get_batch_id` to recover the batch ID from an existing
        :class:`Job` if needed.

        This dependency on *batch_id* is temporary and will be removed
        once ``RemoteConnection`` supports direct lookup by job ID
        alone.

    Args:
        connection (remote.RemoteConnection): The remote connection
            through which to query the job.
        job_id (str): The UUID of the job to retrieve.
        batch_id (str): The batch identifier the job belongs to.
            Defaults to an empty string.

    Returns:
        Job[Results]: A :class:`_RemoteJob` connected to the
        specified job.
    """
    return _RemoteJob(connection, job_id, batch_id=batch_id)

qoolqit.execution

execution

JobStatus: TypeAlias = remote.JobStatus module-attribute

BitStrings(*, evaluation_times: Sequence[float] | None = None, num_shots: int | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)

evaluation_times

num_shots

one_state

tag_suffix

num_shots: int | None property writable

tag: str property

uuid: uuid.UUID property

__call__(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None

config

t

state

hamiltonian

result

apply(*, config: EmulationConfig, state: State, **kwargs: Any) -> Counter[str]

CorrelationMatrix(*, evaluation_times: Sequence[float] | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)

evaluation_times

one_state

tag_suffix

tag: str property

uuid: uuid.UUID property

__call__(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None

config

t

state

hamiltonian

result

apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> list[list]

observables

callbacks

default_evaluation_times

initial_state

with_modulation

interaction_matrix

prefer_device_noise_model

noise_model

n_trajectories

default_num_shots

callbacks: Sequence[Callback] instance-attribute

default_evaluation_times: NDArray[np.floating[Any]] | Literal['Full'] instance-attribute

default_num_shots: int instance-attribute

initial_state: StateType | None instance-attribute

interaction_matrix: pm.AbstractArray | None instance-attribute

n_trajectories: int instance-attribute

noise_model: NoiseModel instance-attribute

observables: Sequence[Observable] instance-attribute

prefer_device_noise_model: bool instance-attribute

with_modulation: bool instance-attribute

from_abstract_repr(obj_str: str) -> EmulationConfig classmethod

obj_str

is_evaluation_time(t: float, tol: float = 1e-06) -> bool

is_time_in_evaluation_times(t: float, evaluation_times: ArrayLike, tol: float = 1e-06) -> bool staticmethod

operator_type() -> Type[Operator]

state_type() -> Type[State]

to_abstract_repr(skip_validation: bool = False) -> str

with_changes(**changes: Any) -> Self

Energy(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)

evaluation_times

tag_suffix

tag: str property

uuid: uuid.UUID property

__call__(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None

config

t

state

hamiltonian

result

apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> Any

EnergySecondMoment(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)

evaluation_times

tag_suffix

tag: str property

uuid: uuid.UUID property

__call__(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None

config

t

state

`execution`

`JobStatus: TypeAlias = remote.JobStatus` `module-attribute`

`BitStrings(*, evaluation_times: Sequence[float] | None = None, num_shots: int | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)`

`evaluation_times`

`num_shots`

`one_state`

`tag_suffix`

`num_shots: int | None` `property` `writable`

`tag: str` `property`

`uuid: uuid.UUID` `property`

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

`config`

`t`

`state`

`hamiltonian`

`result`

`apply(*, config: EmulationConfig, state: State, **kwargs: Any) -> Counter[str]`

`CorrelationMatrix(*, evaluation_times: Sequence[float] | None = None, one_state: Eigenstate | None = None, tag_suffix: str | None = None)`

`evaluation_times`

`one_state`

`tag_suffix`

`tag: str` `property`

`uuid: uuid.UUID` `property`

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

`config`

`t`

`state`

`hamiltonian`

`result`

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> list[list]`

`observables`

`callbacks`

`default_evaluation_times`

`initial_state`

`with_modulation`

`interaction_matrix`

`prefer_device_noise_model`

`noise_model`

`n_trajectories`

`default_num_shots`

`callbacks: Sequence[Callback]` `instance-attribute`

`default_evaluation_times: NDArray[np.floating[Any]] | Literal['Full']` `instance-attribute`

`default_num_shots: int` `instance-attribute`

`initial_state: StateType | None` `instance-attribute`

`interaction_matrix: pm.AbstractArray | None` `instance-attribute`

`n_trajectories: int` `instance-attribute`

`noise_model: NoiseModel` `instance-attribute`

`observables: Sequence[Observable]` `instance-attribute`

`prefer_device_noise_model: bool` `instance-attribute`

`with_modulation: bool` `instance-attribute`

`from_abstract_repr(obj_str: str) -> EmulationConfig` `classmethod`

`obj_str`

`is_evaluation_time(t: float, tol: float = 1e-06) -> bool`

`is_time_in_evaluation_times(t: float, evaluation_times: ArrayLike, tol: float = 1e-06) -> bool` `staticmethod`

`operator_type() -> Type[Operator]`

`state_type() -> Type[State]`

`to_abstract_repr(skip_validation: bool = False) -> str`

`with_changes(**changes: Any) -> Self`

`Energy(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

`evaluation_times`

`tag_suffix`

`tag: str` `property`

`uuid: uuid.UUID` `property`

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

`config`

`t`

`state`

`hamiltonian`

`result`

`apply(*, state: State, hamiltonian: Operator, **kwargs: Any) -> Any`

`EnergySecondMoment(*, evaluation_times: Sequence[float] | None = None, tag_suffix: str | None = None)`

`evaluation_times`

`tag_suffix`

`tag: str` `property`

`uuid: uuid.UUID` `property`

`call(config: EmulationConfig, t: float, state: State, hamiltonian: Operator, result: Results) -> None`

`config`

`t`

`state`

`hamiltonian`