pytams.trajectory

Exceptions

WallTimeLimitError

Exception for running into wall time limit.

Classes

Snapshot	A dataclass defining a snapshot.
Trajectory	A class defining a stochastic trajectory.

Functions

form_trajectory_id(→ str)	Helper to assemble a trajectory ID string.
get_index_from_id(→ tuple[int, int])	Helper to get trajectory index from ID string.

Module Contents

exception WallTimeLimitError

Bases: Exception

Exception for running into wall time limit.

form_trajectory_id(n: int, nb: int = 0) → str

Helper to assemble a trajectory ID string.

Parameters:

• n – trajectory index

• nb – number of branching

Returns:

trajectory ID

get_index_from_id(identity: str) → tuple[int, int]

Helper to get trajectory index from ID string.

Parameters:

identity – trajectory ID

Returns:

trajectory index and number of branching

class Snapshot

A dataclass defining a snapshot.

Gathering what defines a snapshot into an object. The time and score are of float type, but the actual type of the noise and state are completely determined by the forward model. A snapshot is allowed to have a state or not to accomodate memory savings.

Variables:

• time – snapshot time

• score – score function value

• noise – noise used to reach this snapshot

• state – model state

time: float

score: float

noise: Any

state: Any | None = None

has_state() → bool

Check if snapshot has state.

Returns:

True if state is not None

Return type:

bool

class Trajectory(traj_id: int, fmodel_t: type[pytams.fmodel.ForwardModelBaseClass] | None, parameters: dict[Any, Any], workdir: pathlib.Path | None = None, frozen: bool = False)

A class defining a stochastic trajectory.

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 TAMS reside 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, …).

Variables:

• _parameters_full – the full parameters dictionary

• _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

• _t_cur – the current time

• _t_end – the end time

• _dt – the stochastic time step size

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 Id.

Returns:

the trajectory id

idstr() → str

Return trajectory Id as a padded string.

Returns:

the trajectory id as a string

advance(t_end: float = 1000000000000.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 TAMS workers.

Parameters:

• t_end – the end time of the advance

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

Returns:

None

Raises:

WallTimeLimitError – if the walltime limit is reached

classmethod restore_from_checkfile(checkfile: pathlib.Path, fmodel_t: type[pytams.fmodel.ForwardModelBaseClass], parameters: dict[Any, Any], workdir: pathlib.Path | None = None, frozen: bool = False) → Trajectory

Return a trajectory restored from an XML chkfile.

classmethod branch_from_trajectory(from_traj: Trajectory, rst_traj: Trajectory, score: 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

store(traj_file: pathlib.Path | None = None) → None

Store the trajectory to an XML chkfile.

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_ended.

current_time() → float

Return the current trajectory time.

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.

has_ended() → 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_length() → int

Return the trajectory length.

get_nbranching() → int

Return the number of branching events.

get_last_state() → Any | None

Return the last state in the trajectory.