pyrevs.trajectory

Submodules

• pyrevs.trajectory.config
• pyrevs.trajectory.trajectory

Classes

TrajectoryConfig	Trajectory configuration.
Trajectory	A class defining a stochastic trajectory.

Package Contents

class TrajectoryConfig

Trajectory configuration.

step_size: float = -1.0

end_time: float = -1.0

targetscore: float = 0.95

sparse_freq: int = 1

sparse_start: int = 0

chkfile_dump_all: bool = False

validate() → None

Validate trajectory configuration.

class Trajectory(traj_id: int, weight: float, fmodel_t: type[pyrevs.core.ForwardModelBaseClass[T_Noise, T_State]] | None, traj_cfg: pyrevs.trajectory.config.TrajectoryConfig, diag_configs: dict[str, pyrevs.core.Config] | None = None, model_params: dict[str, Any] | None = None, workdir: pathlib.Path | None = None, deterministic: bool = False, frozen: bool = False)

Bases: Generic[T_Noise, T_State]

A class defining a stochastic trajectory.

Each trajectory is a MCMC simulation of the forward model.

The trajectory class is a container for time-ordered snapshots. It contains an instance of the forward model, current and end times, and a list of the model snapshots. Note that the class uses a plain list of snapshots and not a more computationally efficient data structure such as a numpy array for convenience. It is assumed that the computational cost of running the sampling algorithm resides in the forward model and the overhead of the trajectory class is negligible.

It also provide the forward model with the necessary context to advance in time, method to move forward in time, methods to save/load the trajectory to/from disk as well as accessor to the trajectory history (time, state, score, …).

The _computed_steps variable store the number of steps actually taken by the trajectory. It differs from the _step variable when a trajectory is branched from an ancestor.

Variables:

• _tid – the trajectory index

• _checkFile – the trajectory checkpoint file

• _workdir – the model working directory

• _score_max – the maximum score

• _snaps – a list of snapshots

• _step – the current step counter

• _computed_steps – the number of steps explicitly advanced by the trajectory

• _t_cur – the current time

• _t_end – the end time

• _dt – the stochastic time step size

max_ntraj = 10000

max_nbranch = 9999

noise_backlog: list[T_Noise] = []

set_checkfile(path: pathlib.Path) → None

Setter of the trajectory checkFile.

Parameters:

path – the new checkFile

set_workdir(path: pathlib.Path) → None

Setter of the trajectory working directory.

And propagate the workdir to the forward model.

Parameters:

path – the new working directory

get_workdir() → pathlib.Path

Get the trajectory working directory.

Returns:

the working directory

id() → int

Return trajectory index.

This is the index of the trajectory in the ensemble.

Returns:

the trajectory id

unique_id() → int

Return trajectory unique Id.

Combining the index and the number of branching events. This makes a unique ID.

Returns:

the trajectory unique id

idstr() → str

Return trajectory Id as a padded string.

Returns:

the trajectory id as a string

advance(termination_criteria: list[pyrevs.strategies.base.TerminationCriterion] | None = None, nstep_end: int = -1, t_end: float = -1.0, walltime: float = 1000000000000.0) → None

Advance the trajectory to a prescribed end time.

This is the main time loop of the trajectory object. Unless specified otherwise, the trajectory will advance until the end time is reached or the model has converged.

If the walltime limit is reached, a WallTimeLimitError exception is raised. Note that this exception is treated as a warning not an error by the pyREVS workers.

Parameters:

• nstep_end – the number of steps to advance

• t_end – the end time of the advance

• walltime – a walltime limit to advance the model to t_end

• termination_criteria – a list of termination criterion

Returns:

None

Raises:

• WallTimeLimitError – if the walltime limit is reached

• RuntimeError – if the model advance run into a problem

setup_noise() → None

Prepare the noise for the next step.

classmethod init_from_metadata(metadata: dict[str, Any], fmodel_t: type[pyrevs.core.ForwardModelBaseClass[T_Noise, T_State]], traj_cfg: pyrevs.trajectory.config.TrajectoryConfig, diag_configs: dict[str, pyrevs.core.Config] | None = None, model_params: dict[str, Any] | None = None, workdir: pathlib.Path | None = None, frozen: bool = False) → Trajectory[T_Noise, T_State]

Initialize a trajectory from serialized metadata.

classmethod restore_from_checkfile(checkfile: pathlib.Path, metadata: dict[str, Any], fmodel_t: type[pyrevs.core.ForwardModelBaseClass[T_Noise, T_State]], traj_cfg: pyrevs.trajectory.config.TrajectoryConfig, diag_configs: dict[str, pyrevs.core.Config] | None = None, model_params: dict[str, Any] | None = None, workdir: pathlib.Path | None = None, frozen: bool = False) → Trajectory[T_Noise, T_State]

Return a trajectory restored from an XML chkfile.

classmethod branch_from_trajectory(from_traj: Trajectory[T_Noise, T_State], rst_traj: Trajectory[T_Noise, T_State], score: float, new_weight: float) → Trajectory

Create a new trajectory.

Loading the beginning of a provided trajectory for all entries with score below a given score. This effectively branches the trajectory.

Although the rst_traj is provided as an argument, it is only used to set metadata of the branched trajectory.

Parameters:

• from_traj – an already existing trajectory to restart from

• rst_traj – the trajectory being restarted

• score – a threshold score

• new_weight – the weight of the child trajectory

store(traj_file: pathlib.Path | None = None, write_metadata_json: bool = False) → None

Store the trajectory data to an XML chkfile.

The default behavior is to store the trajectory into the file specified by the attribute self._checkFile unless a different path is provided. The metadata are not writen to file by default, as the pyREVS database handle metadata in an SQL file. It can be triggered when using trajectories in stand-alone.

Parameters:

• traj_file – an optional path file to store the trajectory to

• write_metadata_json – an optional boolean to also write the metadata json dict

set_weight(weight: float) → None

Set the trajectory weight.

Parameters:

weight – the (new) trajectory weight in the ensemble

update_metadata() → None

Update trajectory score/ending metadata.

Update the maximum of the score function over the trajectory as well as the bool values for has_converged and has_terminated

set_current_time_and_step(time: float, step: int) → None

Set the current time and step.

current_time() → float

Return the current trajectory time.

current_step() → int

Return the current trajectory step.

step_size() → float

Return the time step size.

score_max() → float

Return the maximum of the score function.

is_converged() → bool

Return True for converged trajectory.

is_terminated() → bool

Return True for terminated trajectory.

has_started() → bool

Return True if computation has started.

get_checkfile() → pathlib.Path

Return the trajectory check file name.

get_time_array() → numpy.typing.NDArray[numpy.number]

Return the trajectory time instants.

get_score_array() → numpy.typing.NDArray[numpy.number]

Return the trajectory scores.

get_noise_array() → numpy.typing.NDArray[Any]

Return the trajectory noises.

get_state_list() → list[tuple[int, T_State | None]]

Return a list of states and associated indices.

Returns:

A list of tuples with index and states

get_length() → int

Return the trajectory length.

get_nbranching() → int

Return the number of branching events.

get_computed_steps_count() → int

Return the number of compute steps taken.

get_last_state() → T_State | None

Return the last state in the trajectory.

get_last_state_id() → int | None

Return the id of the last state in the trajectory.

get_metadata() → dict

Return a dict with traj metadata.

Returns:

The trajectory metadata in a dict.

serialize_metadata_json() → str

Return a json string with metadata.

Returns:

A json string with the trajectory metadata

classmethod deserialize_metadata(json_str: str) → dict[str, Any]

Load a json string into a properly typed metadata dict.

Returns:

A dictionary with the metadata

delete() → None

Clear the trajectory on-disk data.