pywellsfm.simulator

The pywellsfm.simulator package contains the forward modelling engines (simulators) and orchestration helpers used to run stratigraphic forward modelling scenarios.

Files documented

This page is built from the following Python modules:

__init__.py (pywellsfm.simulator)
AccommodationSimulator.py (pywellsfm.simulator.AccommodationSimulator)
AccumulationSimulator.py (pywellsfm.simulator.AccumulationSimulator)
FSSimulator.py (pywellsfm.simulator.FSSimulator)
DepositionalEnvironmentSimulator.py (pywellsfm.simulator.DepositionalEnvironmentSimulator)
EnvironmentConditionSimulator.py (pywellsfm.simulator.EnvironmentConditionSimulator)
TimeStepController.py (pywellsfm.simulator.TimeStepController)

API reference

Public API.

This package contains the simulators.

The symbols re-exported here form the supported, stable entry points. Callers should prefer importing from pywellsfm.model instead of submodules.

class pywellsfm.simulator.AccommodationSimulator

Bases: object

Simulate the accommodation space in wells based on a scenario.

The simulator uses the eustatism at the start of the simulation as the datum. All elevations are computed relatively to this datum.

Warning

Subsidence curve can be the cumulative subsidence or the subsidence rate.

At the start of the simulation, we have:

initial sea level elevation is 0

initial topography (basement) is -initial bathymetry

initial accommodation is initial bathymetry

At a given time t, we have:

sea level(t) = eustatism(t) - eustatism(0)

basement(t) = initial topography - subsidence(t)

accommodation(t) = sea level + subsidence(t)

getAccommodationAt(age: float) → float

Get the cumulative accommodation at a given age from the start.

Parameters:: age (float) – age to compute the accommodation for.
Returns:: accommodation at the given age.

getBasementElevationAt(age: float) → float

Get the basement elevation at a given age.

Parameters:: age (float) – age to get the basement elevation for.
Return float:: basement elevation at the given age.

getEustatismAt(age: float) → float

Get the eustatism value at a given age.

Parameters:: age (float) – age to get the eustatism for.
Return float:: eustatism value at the given age.

getSeaLevelAt(age: float) → float

Get the sea level value at a given age.

Parameters:: age (float) – age to get the sea level for.
Return float:: sea level value at the given age.

getSubsidenceAt(age: float) → float

Get the subsidence value at a given age.

Parameters:: age (float) – age to get the subsidence for.
Return float:: subsidence value at the given age.

getSubsidenceType() → SubsidenceType

Get the subsidence type used by the simulator.

Return SubsidenceType:: subsidence type used by the simulator.

initialEustacy(age: float) → None

Set the initial eustacy used by the simulator.

Parameters:: age (float) – age to set the initial eustacy for.

prepare() → None: Prepare the simulator for computations.

setEustaticCurve(eustaticCurve: Curve) → None

Set the eustatic curve used by the simulator.

Parameters:: eustaticCurve (Curve) – eustatic curve to set.

setInitialBathymetry(bathymetry: float) → None

Set the initial bathymetry used by the simulator.

Parameters:: bathymetry (float) – initial bathymetry to set.

setSubsidenceCurve(subsidenceCurve: Curve | None, type: SubsidenceType) → None

Set the subsidence curve used by the simulator.

Parameters:

subsidenceCurve (Curve | None) – subsidence curve to set.
type (SubsidenceType) – type of subsidence (cumulative or rate).

eustaticCurve: Curve | None: eustatic curve used by the simulator

eustatismStart: float: Eustatism at the start of the simulation

subsidenceCurve: Curve | None: subsidence curve (cumulative or rate) used by the simulator

subsidenceType: SubsidenceType: subsidence type (cumulative or rate) used by the simulator

topographyStart: float: initial topography value at the start of the simulation

class pywellsfm.simulator.AccommodationStorage(basement: float, seaLevel: float, accommodation: float)

Bases: object

Storage class for accommodation computation results.

Elevation datum is given by sea level at the start of the simulation.

Parameters:

basement (float) – basement elevation value
seaLevel (float) – sea level value
accommodation (float) – accommodation value

accommodation: float

basement: float

seaLevel: float

class pywellsfm.simulator.AccumulationSimulator

Bases: object

Simulate accumulation in wells based on an accumulation model.

computeAccumulationRatesForAllElements(environmentConditions: dict[str, float], age: float | None = None) → dict[str, float]

Compute the accumulation rate of each element from the model.

Parameters:

environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return dict[str, float]:

dictionary where keys are element names and values are the corresponding accumulation rates.

computeElementAccumulationRate(elementName: str, environmentConditions: dict[str, float], age: float | None = None) → float

Compute the accumulation rate of a given element from the model.

Parameters:

elementName (str) – name of the element to compute the accumulation rate for.
environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return float:

accumulation rate of the given element, or 0.0 if the element is not in the model.

computeTotalAccumulationRate(environmentConditions: dict[str, float], age: float | None = None) → float

Compute the total accumulation rate in the accumulation model.

Parameters:

environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return float:

total accumulation rate.

prepare() → None: Prepare the simulator for computations.

setAccumulationModel(accumulationModel: AccumulationModel) → None

Set the accumulation model used by the simulator.

Parameters:: accumulationModel (AccumulationModel) – accumulation model to set.

class pywellsfm.simulator.DESimulatorParameters(waterDepth_sigma: float = 2.0, waterDepth_weight: float = 1.0, transition_sigma: float = 1.0, transition_weight: float = 1.0, trend_sigma: float = 2.0, trend_window: int = 5, trend_weight: float = 1.0, interval_distance_method: IntervalDistanceMethod = IntervalDistanceMethod.GAP_OVERLAPPING_WIDTH)

Bases: object

Configuration parameters for DepositionalEnvironmentSimulator.

Parameters:

waterDepth_sigma (float) – standard-deviation (metres) for the Gaussian waterDepth likelihood kernel. Controls tolerance to waterDepth mismatches.
waterDepth_weight (float) – weight for the waterDepth likelihood in the posterior computation.
transition_sigma (float) – standard-deviation (metres) for the Gaussian transition (adjacency) likelihood kernel.
transition_weight (float) – weight for the transition likelihood in the posterior computation.
trend_sigma (float) – standard-deviation for the Gaussian distality trend likelihood kernel. Controls tolerance to distality trend mismatches.
trend_window (int) – number of previous environments to consider for distality trend likelihood.
trend_weight (float) – weight for the distality trend likelihood in the posterior computation.
interval_distance_method (IntervalDistanceMethod) – method to compute the distance between waterDepth intervals for likelihood computation.

interval_distance_method: IntervalDistanceMethod = 'gap_overlapping_width'

transition_sigma: float = 1.0

transition_weight: float = 1.0

trend_sigma: float = 2.0

trend_weight: float = 1.0

trend_window: int = 5

waterDepth_sigma: float = 2.0

waterDepth_weight: float = 1.0

class pywellsfm.simulator.DepositionalEnvironmentSimulator(depositionalEnvironmentModel: DepositionalEnvironmentModel, weights: dict[str, float] | None = None, params: DESimulatorParameters | None = None)

Bases: object

Bayesian simulator that selects a depositional environment.

At a single time-slice and location the simulator:

Computes prior probabilities from environment weights.
Computes waterDepth likelihood from a measured (or uncertain) waterDepth value / range.
Computes transition likelihood from an optional previous-step environment (spatial continuity / Walther’s law).
Multiplies prior × likelihoods and normalises to obtain the posterior.
Samples one environment from the posterior.

Parameters:

environments (list[DepositionalEnvironment]) – non-empty list of environment definitions with unique names. Consider ordering environments from proximal to distal if DepositionalEnvironment.distality is not provided.
weights (dict[str, float] | None) – optional explicit mapping from environment name to weights. If None, weights are equal to 1.0 for all environments.
params (DESimulatorParameters | None) – simulator tuning knobs; when None the defaults are used.

Raises:

ValueError – if environments is empty or contains duplicate names.

Initialise the simulator with environments and parameters.

static _estimate_distality_slope(distality_series: list[float]) → float: Estimate the trend slope from a distality history series.

static _gaussian_kernel(delta: float, sigma: float) → float: Unnormalised Gaussian kernel.

\[k = \exp\!\left(-\frac{\delta^2}{2\,\sigma^2}\right)\]

static _interval_distance(min1: float, max1: float, min2: float, max2: float, method: IntervalDistanceMethod = IntervalDistanceMethod.WASSERSTEIN2) → float

Compute the interval distance between two intervals.

Returns 0 when the two intervals are identical.

static _normalize(values: dict[str, float]) → tuple[dict[str, float], float]

Normalise a dict of non-negative values to sum to 1.

Returns (normalised_dict, total_before_normalisation).

static _normalize_weights(waterDepth_weight: float, transition_weight: float, trend_weight: float) → tuple[float, float, float]: Normalize a dict of non-negative weights to sum to 1.

static _weightedLikelihood(likelihood: dict[str, float] | None, weight: float, names: list[str] | None = None) → dict[str, float]

static _weightedLikelihoods(L_bathy: dict[str, float] | None, L_trans: dict[str, float] | None, L_trend: dict[str, float] | None, waterDepth_weight: float, transition_weight: float, trend_weight: float, names: list[str] | None = None) → tuple[dict[str, float], dict[str, float], dict[str, float]]

_compute_posterior(prior: dict[str, float], L_bathy: dict[str, float] | None = None, L_trans: dict[str, float] | None = None, L_trend: dict[str, float] | None = None, waterDepth_weight: float = 1.0, transition_weight: float = 1.0, trend_weight: float = 1.0) → tuple[dict[str, float], float]

compute_distality_trend_likelihood(previous_environments: list[str] | None = None, *, _sigma_override: float | None = None) → dict[str, float]

Compute distality trend likelihood from previous environments.

Compute the slope of the derivative of the distality series, and compare to the implied increment from each candidate environment. The more the mismatch, the lower the likelihood.

Parameters:

previous_environments (list[str] | None) – list of previous environment names, or None.
_sigma_override (float | None) – optional override for the trend sigma (for testing and fallback purposes).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if any name in previous_environments is not a known environment.

compute_posterior(*, waterDepth_value: float | None = None, waterDepth_range: tuple[float, float] | None = None, previous_environments: list[str] | None = None) → dict[str, float]

Compute posterior probabilities via Bayesian conditioning.

\[P(e \mid D, S, R) \propto P(e)\,P(D \mid e)\,P(S \mid e)\,P(R \mid e)\]

where:

\(P(e)\) is the prior probability of environment \(e\).
\(P(D \mid e)\) is the water depth likelihood of environment \(e\) given waters depth evidence \(D\). \(P(D \mid e) = w_{wd} P(D \mid e)_0 + (1 - w_{wd})\), where \(P(D \mid e)_0\) is the initial likelihood computed from the water depth ranges, and \(w_{wd}\) is the weight for the waterDepth likelihood.
\(P(S \mid e)\) is the transition likelihood of environment \(e\) given previous environment evidence \(S\). \(P(S \mid e) = w_{trans} P(S \mid e)_0 + (1-w_{trans})\), where \(P(S \mid e)_0\) is the initial likelihood computed from the previous environment, and \(w_{trans}\) is the weight for the transition likelihood.
\(P(T \mid e)\) is the distality trend likelihood of environment \(e\) given previous environment history evidence \(T\). \(P(T \mid e) = w_{trend} P(T \mid e)_0 + (1-w_{trend})\), where \(P(T \mid e)_0\) is the initial likelihood computed from the previous environment history, and \(w_{trend}\) is the weight for the distality trend likelihood.

If the posterior is numerically zero everywhere, the method applies a five-level fallback strategy:

Relax the transition constraint (σ_transition × 10) while keeping waterDepth and trend constraints.

Drop the transition constraint entirely; keep waterDepth and trend constraints.

Relax trend constraint (σ_trend × 10); while keeping waterDepth constraint and dropping transition constraint.

Drop trend constraint entirely (and trend constraint); while keeping only waterDepth constraint.

Return the prior (no likelihoods at all).

Parameters:

waterDepth_value (float | None) – exact waterDepth measurement (mutually exclusive with waterDepth_range).
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth interval (min, max).
previous_environments (list[str] | None) – full ordered history of previous environments.

Returns:

normalised posterior probabilities.

compute_prior() → dict[str, float]

Compute prior probabilities from environment weights.

\[P(e) = \frac{w_e}{\sum_{e'} w_{e'}}\]

Returns:: normalised prior probability for each environment.

compute_transition_likelihood(previous_environment: str | None = None, *, _sigma_override: float | None = None) → dict[str, float]

Compute transition (adjacency) likelihood.

When previous_environment is None all likelihoods are 1 (unconstrained).

The likelihood is a Gaussian kernel on the combined distance of waterDepth range interval distance and the distality difference between the previous environment’s and each candidate’s range.

Parameters:

previous_environment (str | None) – name of the environment selected at the previous step, or None.
_sigma_override (float | None) – optional override for the transition sigma (for testing and fallback purposes).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if previous_environment is not a known name.

compute_waterDepth_likelihood(*, waterDepth_value: float | None = None, waterDepth_range: tuple[float, float] | None = None) → dict[str, float]

Compute likelihood of each environment given waterDepth evidence.

The likelihood uses a Gaussian kernel on the interval distance between the observation and each environment’s waterDepth range:

\[P(D \mid e) = \exp\!\left( -\frac{\delta(D,\, \text{range}(e))^2} {2\,\sigma_{\text{waterDepth}}^2} \right)\]

If neither waterDepth_value nor waterDepth_range is given, all likelihoods are 1 (unconstrained).

Parameters:

waterDepth_value (float | None) – exact waterDepth measurement.
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth interval (min, max).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if both parameters are specified simultaneously.

prepare() → None

Precompute and cache invariant probabilities.

Caches:

prior probabilities (environment weights),
transition likelihood dictionaries for None and each environment name at default transition sigma.
distality trend likelihood dictionaries for all combinations of previous environment histories up to trend_window length at default trend sigma.

The cache is safe because environments and parameters are fixed after initialisation.

Compute posterior and sample one environment.

If water depth is lower than the minimum of all environments, the simulator will select the more proximal environment if water depth range is inside, None otherwise.

Parameters:

waterDepth_value (float | None) – exact waterDepth value.
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth range.
previous_environments (list[str] | None) – ordered history of previous-step environments.
distality_by_environment (dict[str, float] | None) – optional explicit distality mapping.
seed (int | None) – optional seed for deterministic sampling.

Returns:

(posterior, sampled_environment).

sample_environment(posterior: dict[str, float], *, rng: Generator | None = None, seed: int | None = None) → str

Sample one environment from posterior probabilities.

Parameters:

posterior (dict[str, float]) – mapping {name: probability}.
rng (numpy.random.Generator | None) – random number generator. If None, a new generator is created from seed.
seed (int | None) – seed for a new generator when rng is None.

Returns:

name of the sampled environment.

Raises:

ValueError – if all posterior probabilities are zero.

_cached_prior: dict[str, float] | None

_cached_transition_likelihood: dict[str | None, dict[str, float]]

_distality_by_environment: dict[str, float]: mapping of environment name to distality value.

_environments: dict[str, DepositionalEnvironment]: mapping from environment name to definition

_params: DESimulatorParameters: simulator parameters

_trend_threshold: float

_weights: dict[str, float]

depositionalEnvironmentModel: depositional environment model with environment definitions

property environment_names: list[str]: Return a copy of the environment name list.

property environments: list[DepositionalEnvironment]: Return a copy of the environments.

property params: DESimulatorParameters: Return the simulator parameters.

class pywellsfm.simulator.EnvironmentConditionSimulator

Bases: object

Simulate environmental conditions based on a conditions model.

The model is valid for a given environment.

computeEnvironmentalConditions(environmentName: str, waterDepth: float, age: float) → dict[str, float]

Compute environmental conditions for the given environment.

Parameters:

environmentName (str) – name of the depositional environment for the location.
waterDepth (float) – waterDepth value for the location.
age (float) – age at which to compute the conditions.

Return dict[str, float]:

dictionary containing environmental conditions for the given environment.

prepare() → None: Prepare the simulator for computations.

setEnvironmentModel(envModel: DepositionalEnvironmentModel) → None

Set the environment model used by the simulator.

Parameters:

envModel (DepositionalEnvironmentModel) – environment model to set.
envConditionsModel (EnvironmentConditionsModel) – environment conditions model to set.

class pywellsfm.simulator.FSSimulator(scenario: Scenario, realizationDataList: list[RealizationData], use_depositional_environment_simulator: bool = False, deSimulator_weights: dict[str, float] | None = None, deSimulator_params: DESimulatorParameters | None = None, fsSimulator_params: FSSimulatorParameters = FSSimulatorParameters(max_waterDepth_change_per_step=0.5, dt_min=0.001, dt_max=0.1, safety=0.9, max_steps=1000000000))

Bases: object

Defines a Forward Stratigraphic Simulator runner.

The FS simulator runner is used to run one or multiple realizations of a scenario over time using adaptive time steps. Time step is adjusted such as accumulated thickness and waterDepth change do not exceed the given threshold (by default 0.5m).

The FS simulator runner also manages uncertainties over multiple realizations.

Parameters:

scenario (Scenario) – scenario to simulate.
realizationDataList (list[RealizationData]) – list of realization data.
use_depositional_environment_simulator (bool) – whether to use the depositional environment simulator. Default is False.
deSimulator_weights (dict[str, float] | None) – weights for the depositional environment simulator. Keys are depositional environment names, values are weights. If None, all environments have equal weight. Default is None.
deSimulator_params (DESimulatorParameters | None) – parameters for the depositional environment simulator. If None, default parameters are used. Default is None.
fsSimulator_params (FSSimulatorParameters) – parameters for the FS simulator.

static _getAccumulatedThickness(accumulationRate: ndarray[tuple[Any, ...], dtype[float64]], dt: float) → ndarray[tuple[Any, ...], dtype[float64]]

Get the deposited thickness at a step given the accumulation rate.

Parameters:

accumulationRate (npt.NDArray[np.float64]) – total accumulation rate for each realization (e.g., in m/Myr).
dt (float) – time step duration (e.g., in Myr).

Return npt.NDArray[np.float64]:

deposited thickness over the time step.

static _getWaterDepthVariation(delta_seaLevel: float, delta_subs: ndarray[tuple[Any, ...], dtype[float64]], thickness_step: ndarray[tuple[Any, ...], dtype[float64]]) → float | ndarray[tuple[Any, ...], dtype[float64]]

Compute water depth variation over a time step.

Parameters:

delta_seaLevel (float) – change in sea level over the time step.
delta_subs (npt.NDArray[np.float64]) – change in subsidence of each realization over the time step.
thickness_step (npt.NDArray[np.float64]) – deposited thickness of each realization over the time step.

Return float | npt.NDArray[np.float64]:

water depth variation over the time step.

_buildEnsembleDataset() → Dataset

Build the ensemble dataset after running all realizations.

Return xr.Dataset:: xarray.Dataset containing ensemble results.

_computeAccumulationRates(env_list: list[dict[str, float]], age: float) → ndarray[tuple[Any, ...], dtype[float64]]

Compute accumulation rates for all realizations from the model.

Parameters:

env_list (list[dict[str, float]]) – list of environment conditions for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return npt.NDArray[np.float64]:

accumulation rates for all realizations.

_computeDepositionalEnvironment(waterDepth: ndarray[tuple[Any, ...], dtype[float64]], prev_env: list[list[str] | None]) → list[DepositionalEnvironment | None]

Compute depositional environment for each realization.

Parameters:: waterDepth (npt.NDArray[np.float64]) – waterDepth value for each realization.
Return list[list[str] | None]:: list of previous depositional environment names for each realization.

_computeElementAccumulationRates(elementName: str, env_list: list[dict[str, float]], age: float) → ndarray[tuple[Any, ...], dtype[float64]]

Compute accumulation rates for all realizations from the model.

Parameters:

elementName (str) – name of the accumulation element to compute.
env_list (list[dict[str, float]]) – list of environment conditions for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return npt.NDArray[np.float64]:

accumulation rates for all realizations.

_computeEnvironmentalConditions(waterDepth: ndarray[tuple[Any, ...], dtype[float64]], envs: list[DepositionalEnvironment | None], age: float) → list[dict[str, float]]

Compute environmental conditions based on DepositionalEnvironment.

Parameters:

waterDepth (npt.NDArray[np.float64]) – waterDepth value for each realization.
envs (list[DepositionalEnvironment|None]) – list of depositional environments for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return list[dict[str, float]]:

list of dictionaries containing environmental conditions for each realization.

_computeMaxWaterDepthChange(t1: float, dt: float, rates: ndarray[tuple[Any, ...], dtype[float64]]) → float

Compute maximum water depth change across realizations for a dt.

Parameters:

t1 (float) – start time of the step.
rates (npt.NDArray[np.float64]) – deposition rates at t1 for each realization.
dt (float) – candidate time step.

Return float:

maximum absolute water depth change.

_getDeltaSeaLevel(t1: float, t2: float) → float

Get the change in sea level between t1 and t2.

Parameters:

t1 (float) – start time.
t2 (float) – end time.

Return float:

change in sea level (sea level at t2 - sea level at t1).

_getDeltaSubsidence(t1: float, t2: float) → ndarray[tuple[Any, ...], dtype[float64]]

Get the change in subsidence between t1 and t2 for all realizations.

Parameters:

t1 (float) – start time.
t2 (float) – end time.

Return npt.NDArray[np.float64]:

change in subsidence (subsidence at t2 - subsidence at t1) for all realizations.

_initializeDepositionalEnvironments(waterDepth: ndarray[tuple[Any, ...], dtype[float64]]) → list[DepositionalEnvironment | None]

finalize() → None: Finalize the FS simulator after running.

getAgeEnd(markerEnd: Marker | None = None) → float

Get the end age of the simulation.

Return float:: end age.

getStartAge(markerStart: Marker | None = None) → float

Get the start age of the simulation.

Return float:: start age.

prepare() → None: Prepare the FS simulator for running.

run(markerEnd: Marker | None = None, exactAges: set[float] | None = None) → None

Run the FS simulator until a given marker or to the top.

Time decreases from start to stop (e.g., from 100 Myr to 0 Myr).

Parameters:

markerEnd (Optional[Marker]) – marker until which to run the simulation. If None, the simulation runs to the top of wells.
exactAges (Optional[set[float]]) – set of exact ages to include in the simulation. These ages will be included in the set of ages at which the simulator state is recorded. If None, only marker ages are included. Default is None.

class pywellsfm.simulator.FSSimulatorParameters(max_waterDepth_change_per_step: float = 0.5, dt_min: float = 0.001, dt_max: float = 0.1, safety: float = 0.9, max_steps: int = 1000000000)

Bases: object

Parameters for the Forward Stratigraphic Simulator (FSSimulator).

dt_max: float = 0.1: maximum time step (in Myr). Default is 0.1 Myr.

dt_min: float = 0.001: minimum time step (in Myr). Default is 1e-3 Myr.

max_steps: int = 1000000000: maximum number of steps. Default is 1e9.

max_waterDepth_change_per_step: float = 0.5: maximum waterDepth change and accumulated thickness per step (in meters). Default is 0.5 m.

safety: float = 0.9: safety factor for time step adjustment. Must be in (0, 1]. Default is 0.9.

class pywellsfm.simulator.IntervalDistanceMethod(*values)

Bases: StrEnum

static _generate_next_value_(name, start, count, last_values): Return the lower-cased version of the member name.

CENTER = 'center'

GAP = 'gap'

GAP_OVERLAPPING_WIDTH = 'gap_overlapping_width'

GAP_TIMES_CENTER = 'gap_times_center'

HAUSDORFF = 'hausdorff'

WASSERSTEIN2 = 'wasserstein2'

class pywellsfm.simulator.TimeStepController(params: FSSimulatorParameters, compute_max_change: Callable[[float, float, ndarray[tuple[Any, ...], dtype[float64]]], float])

Bases: object

Domain-agnostic adaptive time-step controller.

Chooses an optimal time step duration such that a user-defined constraint (maximum change per step) is not exceeded. The actual evaluation of the constraint is delegated to a callback so that the controller remains independent of the geological domain.

Initialize the TimeStepController.

Parameters:

params (FSSimulatorParameters) – stepping parameters.
compute_max_change – callback (t, dt, rates) that returns the maximum state change for a candidate time step dt at time t.

_binary_search(t: float, curWaterDepths: ndarray[tuple[Any, ...], dtype[float64]], rates: ndarray[tuple[Any, ...], dtype[float64]], dt_lo: float, dt_hi: float) → float

Binary search for optimal time step.

Parameters:

t (float) – current time.
curWaterDepths (npt.NDArray[np.float64]) – current water depth for each realization.
rates (npt.NDArray[np.float64]) – deposition rates at time t.
dt_lo (float) – lower bound for time step.
dt_hi (float) – upper bound for time step.

Return float:

optimal time step.

adapt(t: float, curWaterDepths: ndarray[tuple[Any, ...], dtype[float64]], rates: ndarray[tuple[Any, ...], dtype[float64]], remaining: float) → float

Choose an adaptive time step based on the change constraint.

Parameters:

t (float) – current time.
curWaterDepths (npt.NDArray[np.float64]) – current water depth for each realization.
rates (npt.NDArray[np.float64]) – deposition rates for each realization.
remaining (float) – remaining time to stop.

Return float:

chosen time step.

set_exact_ages(exactAges: set[float]) → None

Pre-sort exact ages in descending order for efficient lookup.

Must be called before the first call to adapt().

Parameters:: exactAges (set[float]) – exact ages to include.

class pywellsfm.simulator.AccommodationSimulator.AccommodationSimulator

Bases: object

Simulate the accommodation space in wells based on a scenario.

The simulator uses the eustatism at the start of the simulation as the datum. All elevations are computed relatively to this datum.

Warning

Subsidence curve can be the cumulative subsidence or the subsidence rate.

At the start of the simulation, we have:

initial sea level elevation is 0

initial topography (basement) is -initial bathymetry

initial accommodation is initial bathymetry

At a given time t, we have:

sea level(t) = eustatism(t) - eustatism(0)

basement(t) = initial topography - subsidence(t)

accommodation(t) = sea level + subsidence(t)

getAccommodationAt(age: float) → float

Get the cumulative accommodation at a given age from the start.

Parameters:: age (float) – age to compute the accommodation for.
Returns:: accommodation at the given age.

getBasementElevationAt(age: float) → float

Get the basement elevation at a given age.

Parameters:: age (float) – age to get the basement elevation for.
Return float:: basement elevation at the given age.

getEustatismAt(age: float) → float

Get the eustatism value at a given age.

Parameters:: age (float) – age to get the eustatism for.
Return float:: eustatism value at the given age.

getSeaLevelAt(age: float) → float

Get the sea level value at a given age.

Parameters:: age (float) – age to get the sea level for.
Return float:: sea level value at the given age.

getSubsidenceAt(age: float) → float

Get the subsidence value at a given age.

Parameters:: age (float) – age to get the subsidence for.
Return float:: subsidence value at the given age.

getSubsidenceType() → SubsidenceType

Get the subsidence type used by the simulator.

Return SubsidenceType:: subsidence type used by the simulator.

initialEustacy(age: float) → None

Set the initial eustacy used by the simulator.

Parameters:: age (float) – age to set the initial eustacy for.

prepare() → None: Prepare the simulator for computations.

setEustaticCurve(eustaticCurve: Curve) → None

Set the eustatic curve used by the simulator.

Parameters:: eustaticCurve (Curve) – eustatic curve to set.

setInitialBathymetry(bathymetry: float) → None

Set the initial bathymetry used by the simulator.

Parameters:: bathymetry (float) – initial bathymetry to set.

setSubsidenceCurve(subsidenceCurve: Curve | None, type: SubsidenceType) → None

Set the subsidence curve used by the simulator.

Parameters:

subsidenceCurve (Curve | None) – subsidence curve to set.
type (SubsidenceType) – type of subsidence (cumulative or rate).

eustaticCurve: Curve | None: eustatic curve used by the simulator

eustatismStart: float: Eustatism at the start of the simulation

subsidenceCurve: Curve | None: subsidence curve (cumulative or rate) used by the simulator

subsidenceType: SubsidenceType: subsidence type (cumulative or rate) used by the simulator

topographyStart: float: initial topography value at the start of the simulation

class pywellsfm.simulator.AccommodationSimulator.AccommodationStorage(basement: float, seaLevel: float, accommodation: float)

Bases: object

Storage class for accommodation computation results.

Elevation datum is given by sea level at the start of the simulation.

Parameters:

basement (float) – basement elevation value
seaLevel (float) – sea level value
accommodation (float) – accommodation value

accommodation: float

basement: float

seaLevel: float

class pywellsfm.simulator.AccumulationSimulator.AccumulationSimulator

Bases: object

Simulate accumulation in wells based on an accumulation model.

computeAccumulationRatesForAllElements(environmentConditions: dict[str, float], age: float | None = None) → dict[str, float]

Compute the accumulation rate of each element from the model.

Parameters:

environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return dict[str, float]:

dictionary where keys are element names and values are the corresponding accumulation rates.

computeElementAccumulationRate(elementName: str, environmentConditions: dict[str, float], age: float | None = None) → float

Compute the accumulation rate of a given element from the model.

Parameters:

elementName (str) – name of the element to compute the accumulation rate for.
environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return float:

accumulation rate of the given element, or 0.0 if the element is not in the model.

computeTotalAccumulationRate(environmentConditions: dict[str, float], age: float | None = None) → float

Compute the total accumulation rate in the accumulation model.

Parameters:

environmentConditions (dict[str, float]) – dictionary of environmental conditions where keys are the names of the environmental factors and values are the corresponding values.
age (float | None) – age of the accumulation (in My). If None, age is not used in the computation of the accumulation rate.

Return float:

total accumulation rate.

prepare() → None: Prepare the simulator for computations.

setAccumulationModel(accumulationModel: AccumulationModel) → None

Set the accumulation model used by the simulator.

Parameters:: accumulationModel (AccumulationModel) – accumulation model to set.

This module contains the DepositionalEnvironmentSimulator class, which simulates the depositional environment at a given point based on its location, waterDepth, and energy conditions.

Let’s assume a list of possible depositional environments. Each environment is defined by its location on a 2D map, and by waterDepth and energy ranges. For instance, for a protected carbonate platform, the list includes from proximal to distal:

shore: between the subaerial beach and the lagoon, waterDepth <5m, low energy zone.
lagoon: between shore and reef environments, waterDepth <15m, very low energy zone.
back reef: located between laggon and reef crest, waterDepth <5m, intermediate energy.
reef crest: very high energy zone (wave swash zone), waterDepth <2m
fore reef: in the front of the reef, high energy zone (wave surf zone), waterDepth 2-20m
outer platform: more distal than fore reef, before shelf, low energy zone (intermediate if storms), waterDepth 20-50m
shelf: more distal than outer platform, waterDepth 50-100s m, very low energy
basin: deep than shelf, waterDepth >100s m, very low energy.

Exact waterDepth ranges depends on user inputs including lagoon maximum depth, fairweather wave breaking depth, fairweather wave base depth, storm wave base depth, and shelf break depth.

The Depositional environment simulator stochastically simulate in which zone a given point is. The unconstrained case uses waterDepth ranges to compute probabilities to be in a given zone, then pick one of the zones according to these probabilities.

A constrained simulation by the waterDepth at the given point, recompute the probabilities such as the given waterDepth at the point preferentially fall within the range of the selected zone.

A forward constrained simulation computes the probabilities by assuming a progressive transition from one zone to another one. So at a given point, probabilities are computed according to:

the selected environment at the previous step. The probability to be in given environment increases if this environment is the same (1.0) or adjacent to the previous one. Adjacency is computed by combining water depth range and distality (if defined) differences.
the distality trend from previously selected environments. The probability to be in a given environment increases if the trend (transgressive or regressive) is respected. In pratice, the probability increases as the variance of the derivative of the distality increases.

All constraints can be combined together.

The conditioning algorithm is based on bayesian approach, where the prior probabilities are computed based on the waterDepth ranges defined for each environment, and the likelihood is computed based on the waterDepth constraint and the previous environment constraints. The posterior probabilities are then computed by multiplying the prior and the likelihood, and normalizing the result. Finally, the environment is selected by sampling from the posterior probabilities.

Likelihoods can be weighted to control the influence of each constraint (i.e., water depth, transition, distality trend). When a constraint weight differs from 1, the initial likelihood values for the given constraint are weighted against 1 (unconstrained), so that the more the weight, the more the likelihood values are close to the initial ones, and the less the weight, the more the likelihood values are close to 1 for all environments (unconstrained). By default, all weights are equal (1.0), meaning that all constraints are equally important.

This module contains the DepositionalEnvironmentSimulator class, which selects a depositional environment (a discrete state) at a given time and given location/order using Bayesian conditioning.

The algorithm combines:

Prior probabilities computed from environment weights.
waterDepth likelihood from a measured or uncertain waterDepth value / range.
Transition likelihood from an optional previous-step environment (Walther’s law / spatial continuity).
Distality trend likelihood from an optional list of previous-step environments (progressive transitions).
Weights for each likelihood to control their influence.

Posterior:

\[P(e \mid D, S) \propto P(e)\,P(D \mid e)\,P(S \mid e)\,P(T \mid e)\]

where:

\(P(e)\) is the prior probability of environment \(e\).
\(P(D \mid e)\) is the water depth likelihood of environment \(e\) given waters depth evidence \(D\). \(P(D \mid e) = w_{wd} P(D \mid e)_0 + (1 - w_{wd})\), where \(P(D \mid e)_0\) is the initial likelihood computed from the water depth ranges, and \(w_{wd}\) is the weight for the waterDepth likelihood.
\(P(S \mid e)\) is the transition likelihood of environment \(e\) given previous environment evidence \(S\). \(P(S \mid e) = w_{trans} P(S \mid e)_0 + (1-w_{trans})\), where \(P(S \mid e)_0\) is the initial likelihood computed from the previous environment, and \(w_{trans}\) is the weight for the transition likelihood.
\(P(T \mid e)\) is the distality trend likelihood of environment \(e\) given previous environment history evidence \(T\). \(P(T \mid e) = w_{trend} P(T \mid e)_0 + (1-w_{trend})\), where \(P(T \mid e)_0\) is the initial likelihood computed from the previous environment history, and \(w_{trend}\) is the weight for the distality trend likelihood.

class pywellsfm.simulator.DepositionalEnvironmentSimulator.DESimulatorParameters(waterDepth_sigma: float = 2.0, waterDepth_weight: float = 1.0, transition_sigma: float = 1.0, transition_weight: float = 1.0, trend_sigma: float = 2.0, trend_window: int = 5, trend_weight: float = 1.0, interval_distance_method: IntervalDistanceMethod = IntervalDistanceMethod.GAP_OVERLAPPING_WIDTH)

Bases: object

Configuration parameters for DepositionalEnvironmentSimulator.

Parameters:

waterDepth_sigma (float) – standard-deviation (metres) for the Gaussian waterDepth likelihood kernel. Controls tolerance to waterDepth mismatches.
waterDepth_weight (float) – weight for the waterDepth likelihood in the posterior computation.
transition_sigma (float) – standard-deviation (metres) for the Gaussian transition (adjacency) likelihood kernel.
transition_weight (float) – weight for the transition likelihood in the posterior computation.
trend_sigma (float) – standard-deviation for the Gaussian distality trend likelihood kernel. Controls tolerance to distality trend mismatches.
trend_window (int) – number of previous environments to consider for distality trend likelihood.
trend_weight (float) – weight for the distality trend likelihood in the posterior computation.
interval_distance_method (IntervalDistanceMethod) – method to compute the distance between waterDepth intervals for likelihood computation.

interval_distance_method: IntervalDistanceMethod = 'gap_overlapping_width'

transition_sigma: float = 1.0

transition_weight: float = 1.0

trend_sigma: float = 2.0

trend_weight: float = 1.0

trend_window: int = 5

waterDepth_sigma: float = 2.0

waterDepth_weight: float = 1.0

class pywellsfm.simulator.DepositionalEnvironmentSimulator.DepositionalEnvironmentSimulator(depositionalEnvironmentModel: DepositionalEnvironmentModel, weights: dict[str, float] | None = None, params: DESimulatorParameters | None = None)

Bases: object

Bayesian simulator that selects a depositional environment.

At a single time-slice and location the simulator:

Computes prior probabilities from environment weights.
Computes waterDepth likelihood from a measured (or uncertain) waterDepth value / range.
Computes transition likelihood from an optional previous-step environment (spatial continuity / Walther’s law).
Multiplies prior × likelihoods and normalises to obtain the posterior.
Samples one environment from the posterior.

Parameters:

environments (list[DepositionalEnvironment]) – non-empty list of environment definitions with unique names. Consider ordering environments from proximal to distal if DepositionalEnvironment.distality is not provided.
weights (dict[str, float] | None) – optional explicit mapping from environment name to weights. If None, weights are equal to 1.0 for all environments.
params (DESimulatorParameters | None) – simulator tuning knobs; when None the defaults are used.

Raises:

ValueError – if environments is empty or contains duplicate names.

Initialise the simulator with environments and parameters.

static _estimate_distality_slope(distality_series: list[float]) → float: Estimate the trend slope from a distality history series.

static _gaussian_kernel(delta: float, sigma: float) → float: Unnormalised Gaussian kernel.

\[k = \exp\!\left(-\frac{\delta^2}{2\,\sigma^2}\right)\]

static _interval_distance(min1: float, max1: float, min2: float, max2: float, method: IntervalDistanceMethod = IntervalDistanceMethod.WASSERSTEIN2) → float

Compute the interval distance between two intervals.

Returns 0 when the two intervals are identical.

static _normalize(values: dict[str, float]) → tuple[dict[str, float], float]

Normalise a dict of non-negative values to sum to 1.

Returns (normalised_dict, total_before_normalisation).

static _normalize_weights(waterDepth_weight: float, transition_weight: float, trend_weight: float) → tuple[float, float, float]: Normalize a dict of non-negative weights to sum to 1.

static _weightedLikelihood(likelihood: dict[str, float] | None, weight: float, names: list[str] | None = None) → dict[str, float]

static _weightedLikelihoods(L_bathy: dict[str, float] | None, L_trans: dict[str, float] | None, L_trend: dict[str, float] | None, waterDepth_weight: float, transition_weight: float, trend_weight: float, names: list[str] | None = None) → tuple[dict[str, float], dict[str, float], dict[str, float]]

_compute_posterior(prior: dict[str, float], L_bathy: dict[str, float] | None = None, L_trans: dict[str, float] | None = None, L_trend: dict[str, float] | None = None, waterDepth_weight: float = 1.0, transition_weight: float = 1.0, trend_weight: float = 1.0) → tuple[dict[str, float], float]

compute_distality_trend_likelihood(previous_environments: list[str] | None = None, *, _sigma_override: float | None = None) → dict[str, float]

Compute distality trend likelihood from previous environments.

Compute the slope of the derivative of the distality series, and compare to the implied increment from each candidate environment. The more the mismatch, the lower the likelihood.

Parameters:

previous_environments (list[str] | None) – list of previous environment names, or None.
_sigma_override (float | None) – optional override for the trend sigma (for testing and fallback purposes).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if any name in previous_environments is not a known environment.

compute_posterior(*, waterDepth_value: float | None = None, waterDepth_range: tuple[float, float] | None = None, previous_environments: list[str] | None = None) → dict[str, float]

Compute posterior probabilities via Bayesian conditioning.

\[P(e \mid D, S, R) \propto P(e)\,P(D \mid e)\,P(S \mid e)\,P(R \mid e)\]

where:

\(P(e)\) is the prior probability of environment \(e\).
\(P(D \mid e)\) is the water depth likelihood of environment \(e\) given waters depth evidence \(D\). \(P(D \mid e) = w_{wd} P(D \mid e)_0 + (1 - w_{wd})\), where \(P(D \mid e)_0\) is the initial likelihood computed from the water depth ranges, and \(w_{wd}\) is the weight for the waterDepth likelihood.
\(P(S \mid e)\) is the transition likelihood of environment \(e\) given previous environment evidence \(S\). \(P(S \mid e) = w_{trans} P(S \mid e)_0 + (1-w_{trans})\), where \(P(S \mid e)_0\) is the initial likelihood computed from the previous environment, and \(w_{trans}\) is the weight for the transition likelihood.
\(P(T \mid e)\) is the distality trend likelihood of environment \(e\) given previous environment history evidence \(T\). \(P(T \mid e) = w_{trend} P(T \mid e)_0 + (1-w_{trend})\), where \(P(T \mid e)_0\) is the initial likelihood computed from the previous environment history, and \(w_{trend}\) is the weight for the distality trend likelihood.

If the posterior is numerically zero everywhere, the method applies a five-level fallback strategy:

Relax the transition constraint (σ_transition × 10) while keeping waterDepth and trend constraints.

Drop the transition constraint entirely; keep waterDepth and trend constraints.

Relax trend constraint (σ_trend × 10); while keeping waterDepth constraint and dropping transition constraint.

Drop trend constraint entirely (and trend constraint); while keeping only waterDepth constraint.

Return the prior (no likelihoods at all).

Parameters:

waterDepth_value (float | None) – exact waterDepth measurement (mutually exclusive with waterDepth_range).
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth interval (min, max).
previous_environments (list[str] | None) – full ordered history of previous environments.

Returns:

normalised posterior probabilities.

compute_prior() → dict[str, float]

Compute prior probabilities from environment weights.

\[P(e) = \frac{w_e}{\sum_{e'} w_{e'}}\]

Returns:: normalised prior probability for each environment.

compute_transition_likelihood(previous_environment: str | None = None, *, _sigma_override: float | None = None) → dict[str, float]

Compute transition (adjacency) likelihood.

When previous_environment is None all likelihoods are 1 (unconstrained).

The likelihood is a Gaussian kernel on the combined distance of waterDepth range interval distance and the distality difference between the previous environment’s and each candidate’s range.

Parameters:

previous_environment (str | None) – name of the environment selected at the previous step, or None.
_sigma_override (float | None) – optional override for the transition sigma (for testing and fallback purposes).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if previous_environment is not a known name.

compute_waterDepth_likelihood(*, waterDepth_value: float | None = None, waterDepth_range: tuple[float, float] | None = None) → dict[str, float]

Compute likelihood of each environment given waterDepth evidence.

The likelihood uses a Gaussian kernel on the interval distance between the observation and each environment’s waterDepth range:

\[P(D \mid e) = \exp\!\left( -\frac{\delta(D,\, \text{range}(e))^2} {2\,\sigma_{\text{waterDepth}}^2} \right)\]

If neither waterDepth_value nor waterDepth_range is given, all likelihoods are 1 (unconstrained).

Parameters:

waterDepth_value (float | None) – exact waterDepth measurement.
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth interval (min, max).

Returns:

likelihood value for each environment (unnormalised).

Raises:

ValueError – if both parameters are specified simultaneously.

prepare() → None

Precompute and cache invariant probabilities.

Caches:

prior probabilities (environment weights),
transition likelihood dictionaries for None and each environment name at default transition sigma.
distality trend likelihood dictionaries for all combinations of previous environment histories up to trend_window length at default trend sigma.

The cache is safe because environments and parameters are fixed after initialisation.

Compute posterior and sample one environment.

If water depth is lower than the minimum of all environments, the simulator will select the more proximal environment if water depth range is inside, None otherwise.

Parameters:

waterDepth_value (float | None) – exact waterDepth value.
waterDepth_range (tuple[float, float] | None) – uncertain waterDepth range.
previous_environments (list[str] | None) – ordered history of previous-step environments.
distality_by_environment (dict[str, float] | None) – optional explicit distality mapping.
seed (int | None) – optional seed for deterministic sampling.

Returns:

(posterior, sampled_environment).

sample_environment(posterior: dict[str, float], *, rng: Generator | None = None, seed: int | None = None) → str

Sample one environment from posterior probabilities.

Parameters:

posterior (dict[str, float]) – mapping {name: probability}.
rng (numpy.random.Generator | None) – random number generator. If None, a new generator is created from seed.
seed (int | None) – seed for a new generator when rng is None.

Returns:

name of the sampled environment.

Raises:

ValueError – if all posterior probabilities are zero.

_cached_prior: dict[str, float] | None

_cached_transition_likelihood: dict[str | None, dict[str, float]]

_distality_by_environment: dict[str, float]: mapping of environment name to distality value.

_environments: dict[str, DepositionalEnvironment]: mapping from environment name to definition

_params: DESimulatorParameters: simulator parameters

_trend_threshold: float

_weights: dict[str, float]

depositionalEnvironmentModel: depositional environment model with environment definitions

property environment_names: list[str]: Return a copy of the environment name list.

property environments: list[DepositionalEnvironment]: Return a copy of the environments.

property params: DESimulatorParameters: Return the simulator parameters.

class pywellsfm.simulator.EnvironmentConditionSimulator.EnvironmentConditionSimulator

Bases: object

Simulate environmental conditions based on a conditions model.

The model is valid for a given environment.

computeEnvironmentalConditions(environmentName: str, waterDepth: float, age: float) → dict[str, float]

Compute environmental conditions for the given environment.

Parameters:

environmentName (str) – name of the depositional environment for the location.
waterDepth (float) – waterDepth value for the location.
age (float) – age at which to compute the conditions.

Return dict[str, float]:

dictionary containing environmental conditions for the given environment.

prepare() → None: Prepare the simulator for computations.

setEnvironmentModel(envModel: DepositionalEnvironmentModel) → None

Set the environment model used by the simulator.

Parameters:

envModel (DepositionalEnvironmentModel) – environment model to set.
envConditionsModel (EnvironmentConditionsModel) – environment conditions model to set.

class pywellsfm.simulator.FSSimulator.FSSimulator(scenario: Scenario, realizationDataList: list[RealizationData], use_depositional_environment_simulator: bool = False, deSimulator_weights: dict[str, float] | None = None, deSimulator_params: DESimulatorParameters | None = None, fsSimulator_params: FSSimulatorParameters = FSSimulatorParameters(max_waterDepth_change_per_step=0.5, dt_min=0.001, dt_max=0.1, safety=0.9, max_steps=1000000000))

Bases: object

Defines a Forward Stratigraphic Simulator runner.

The FS simulator runner is used to run one or multiple realizations of a scenario over time using adaptive time steps. Time step is adjusted such as accumulated thickness and waterDepth change do not exceed the given threshold (by default 0.5m).

The FS simulator runner also manages uncertainties over multiple realizations.

Parameters:

scenario (Scenario) – scenario to simulate.
realizationDataList (list[RealizationData]) – list of realization data.
use_depositional_environment_simulator (bool) – whether to use the depositional environment simulator. Default is False.
deSimulator_weights (dict[str, float] | None) – weights for the depositional environment simulator. Keys are depositional environment names, values are weights. If None, all environments have equal weight. Default is None.
deSimulator_params (DESimulatorParameters | None) – parameters for the depositional environment simulator. If None, default parameters are used. Default is None.
fsSimulator_params (FSSimulatorParameters) – parameters for the FS simulator.

static _getAccumulatedThickness(accumulationRate: ndarray[tuple[Any, ...], dtype[float64]], dt: float) → ndarray[tuple[Any, ...], dtype[float64]]

Get the deposited thickness at a step given the accumulation rate.

Parameters:

accumulationRate (npt.NDArray[np.float64]) – total accumulation rate for each realization (e.g., in m/Myr).
dt (float) – time step duration (e.g., in Myr).

Return npt.NDArray[np.float64]:

deposited thickness over the time step.

static _getWaterDepthVariation(delta_seaLevel: float, delta_subs: ndarray[tuple[Any, ...], dtype[float64]], thickness_step: ndarray[tuple[Any, ...], dtype[float64]]) → float | ndarray[tuple[Any, ...], dtype[float64]]

Compute water depth variation over a time step.

Parameters:

delta_seaLevel (float) – change in sea level over the time step.
delta_subs (npt.NDArray[np.float64]) – change in subsidence of each realization over the time step.
thickness_step (npt.NDArray[np.float64]) – deposited thickness of each realization over the time step.

Return float | npt.NDArray[np.float64]:

water depth variation over the time step.

_buildEnsembleDataset() → Dataset

Build the ensemble dataset after running all realizations.

Return xr.Dataset:: xarray.Dataset containing ensemble results.

_computeAccumulationRates(env_list: list[dict[str, float]], age: float) → ndarray[tuple[Any, ...], dtype[float64]]

Compute accumulation rates for all realizations from the model.

Parameters:

env_list (list[dict[str, float]]) – list of environment conditions for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return npt.NDArray[np.float64]:

accumulation rates for all realizations.

_computeDepositionalEnvironment(waterDepth: ndarray[tuple[Any, ...], dtype[float64]], prev_env: list[list[str] | None]) → list[DepositionalEnvironment | None]

Compute depositional environment for each realization.

Parameters:: waterDepth (npt.NDArray[np.float64]) – waterDepth value for each realization.
Return list[list[str] | None]:: list of previous depositional environment names for each realization.

_computeElementAccumulationRates(elementName: str, env_list: list[dict[str, float]], age: float) → ndarray[tuple[Any, ...], dtype[float64]]

Compute accumulation rates for all realizations from the model.

Parameters:

elementName (str) – name of the accumulation element to compute.
env_list (list[dict[str, float]]) – list of environment conditions for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return npt.NDArray[np.float64]:

accumulation rates for all realizations.

_computeEnvironmentalConditions(waterDepth: ndarray[tuple[Any, ...], dtype[float64]], envs: list[DepositionalEnvironment | None], age: float) → list[dict[str, float]]

Compute environmental conditions based on DepositionalEnvironment.

Parameters:

waterDepth (npt.NDArray[np.float64]) – waterDepth value for each realization.
envs (list[DepositionalEnvironment|None]) – list of depositional environments for each realization.
age (float) – age at the location (only needed if some conditions depend on age).

Return list[dict[str, float]]:

list of dictionaries containing environmental conditions for each realization.

_computeMaxWaterDepthChange(t1: float, dt: float, rates: ndarray[tuple[Any, ...], dtype[float64]]) → float

Compute maximum water depth change across realizations for a dt.

Parameters:

t1 (float) – start time of the step.
rates (npt.NDArray[np.float64]) – deposition rates at t1 for each realization.
dt (float) – candidate time step.

Return float:

maximum absolute water depth change.

_getDeltaSeaLevel(t1: float, t2: float) → float

Get the change in sea level between t1 and t2.

Parameters:

t1 (float) – start time.
t2 (float) – end time.

Return float:

change in sea level (sea level at t2 - sea level at t1).

_getDeltaSubsidence(t1: float, t2: float) → ndarray[tuple[Any, ...], dtype[float64]]

Get the change in subsidence between t1 and t2 for all realizations.

Parameters:

t1 (float) – start time.
t2 (float) – end time.

Return npt.NDArray[np.float64]:

change in subsidence (subsidence at t2 - subsidence at t1) for all realizations.

_initializeDepositionalEnvironments(waterDepth: ndarray[tuple[Any, ...], dtype[float64]]) → list[DepositionalEnvironment | None]

finalize() → None: Finalize the FS simulator after running.

getAgeEnd(markerEnd: Marker | None = None) → float

Get the end age of the simulation.

Return float:: end age.

getStartAge(markerStart: Marker | None = None) → float

Get the start age of the simulation.

Return float:: start age.

prepare() → None: Prepare the FS simulator for running.

run(markerEnd: Marker | None = None, exactAges: set[float] | None = None) → None

Run the FS simulator until a given marker or to the top.

Time decreases from start to stop (e.g., from 100 Myr to 0 Myr).

Parameters:

markerEnd (Optional[Marker]) – marker until which to run the simulation. If None, the simulation runs to the top of wells.
exactAges (Optional[set[float]]) – set of exact ages to include in the simulation. These ages will be included in the set of ages at which the simulator state is recorded. If None, only marker ages are included. Default is None.

class pywellsfm.simulator.TimeStepController.TimeStepController(params: FSSimulatorParameters, compute_max_change: Callable[[float, float, ndarray[tuple[Any, ...], dtype[float64]]], float])

Bases: object

Domain-agnostic adaptive time-step controller.

Chooses an optimal time step duration such that a user-defined constraint (maximum change per step) is not exceeded. The actual evaluation of the constraint is delegated to a callback so that the controller remains independent of the geological domain.

Initialize the TimeStepController.

Parameters:

params (FSSimulatorParameters) – stepping parameters.
compute_max_change – callback (t, dt, rates) that returns the maximum state change for a candidate time step dt at time t.

_binary_search(t: float, curWaterDepths: ndarray[tuple[Any, ...], dtype[float64]], rates: ndarray[tuple[Any, ...], dtype[float64]], dt_lo: float, dt_hi: float) → float

Binary search for optimal time step.

Parameters:

t (float) – current time.
curWaterDepths (npt.NDArray[np.float64]) – current water depth for each realization.
rates (npt.NDArray[np.float64]) – deposition rates at time t.
dt_lo (float) – lower bound for time step.
dt_hi (float) – upper bound for time step.

Return float:

optimal time step.

adapt(t: float, curWaterDepths: ndarray[tuple[Any, ...], dtype[float64]], rates: ndarray[tuple[Any, ...], dtype[float64]], remaining: float) → float

Choose an adaptive time step based on the change constraint.

Parameters:

t (float) – current time.
curWaterDepths (npt.NDArray[np.float64]) – current water depth for each realization.
rates (npt.NDArray[np.float64]) – deposition rates for each realization.
remaining (float) – remaining time to stop.

Return float:

chosen time step.

set_exact_ages(exactAges: set[float]) → None

Pre-sort exact ages in descending order for efficient lookup.

Must be called before the first call to adapt().

Parameters:: exactAges (set[float]) – exact ages to include.