geoprior.models.subsidence.utils#

GeoPrior subsidence model utilities.

Functions

`affine_from_cfg`(scaling_kwargs, *, ...[, ...])	Return (a,b) for y_si = y_model*a + b.
`assert_dynamic_names_match_tensor`(Xh, ...)	Check dynamic_feature_names length matches Xh.
`canonicalize_scaling_kwargs`(scaling_kwargs, *)	Return a canonicalized scaling dict.
`coord_ranges`(scaling_kwargs)	Return (tR,xR,yR) if coords_normalized.
`coord_ranges_si`(sk)	Return coordinate spans in SI (t in seconds; x/y in meters).
`deg_to_m`(axis, scaling_kwargs)	Meters per degree factor for lon/lat coords.
`enforce_scaling_alias_consistency`(...[, where])	Enforce that canonical keys and aliases agree.
`finalize_scaling_kwargs`(sk)	Add derived SI conversion constants to `scaling_kwargs`.
`from_si_subsidence`(s_si, scaling_kwargs)	Inverse of to_si_subsidence: s_model = (s_si - b) / a.
`get_gwl_dyn_index_cached`(model)	Cache gwl_dyn_index on model after first resolve.
`get_h_hist_si`(model, inputs, *[, want_head])	Return head (or depth) history in SI meters.
`get_h_ref_si`(model, inputs, like)	Return h_ref in SI meters, broadcast to like.
`get_s_init_si`(model, inputs, like)	Return initial settlement (cumulative subsidence) in SI meters.
`get_sk`(scaling_kwargs, key, *aliases[, ...])	Fetch a key from scaling_kwargs with aliases + default.
`get_subs_dyn_index_cached`(model)	Cache subs_dyn_index on model after first resolve.
`gwl_to_head_m`(v_m, scaling_kwargs, *[, inputs])	Convert depth-bgs to head if possible.
`infer_dt_units_from_t`(t_BH1, scaling_kwargs, *)	Infer per-step dt in time_units from time tensor t(B,H,1).
`load_scaling_kwargs`(scaling_kwargs, *[, copy])	Load scaling kwargs from a dict-like object or JSON.
`policy_gate`(step, policy, *[, warmup_steps, ...])	Return a scalar gate in `[0,1]` based on a policy + step.
`resolve_gwl_dyn_index`(scaling_kwargs)	Resolve GWL channel index for dynamic_features.
`resolve_subs_dyn_index`(scaling_kwargs)	Resolve subsidence channel index for dynamic_features.
`slice_dynamic_channel`(Xh, idx)	Slice (B,T,F) -> (B,T,1) at idx.
`to_si_head`(h_model, scaling_kwargs)	Convert head/depth to SI meters.
`to_si_subsidence`(s_model, scaling_kwargs)	Convert subsidence to SI meters.
`to_si_thickness`(H_model, scaling_kwargs)	Convert thickness to SI.
`validate_scaling_kwargs`(scaling_kwargs)	Basic scaling sanity checks.

geoprior.models.subsidence.utils.enforce_scaling_alias_consistency(scaling_kwargs, *, where='validate')[source]#

Enforce that canonical keys and aliases agree.

If both canonical and an alias exist and their values differ, apply the scaling error policy.

Parameters:

scaling_kwargs (dict[str, Any] | None)
where (str)

Return type:

None

geoprior.models.subsidence.utils.canonicalize_scaling_kwargs(scaling_kwargs, *, copy=True)[source]#

Return a canonicalized scaling dict.

If a canonical key is missing, but one of its aliases exists, copy alias -> canonical.
Keeps existing canonical values unchanged.

Parameters:

scaling_kwargs (dict[str, Any] | None)
copy (bool)

Return type:

dict[str, Any]

geoprior.models.subsidence.utils.load_scaling_kwargs(scaling_kwargs, *, copy=True)[source]#

Load scaling kwargs from a dict-like object or JSON.

Parameters:

scaling_kwargs (Any | None)
copy (bool)

Return type:

dict[str, Any]

geoprior.models.subsidence.utils.get_sk(scaling_kwargs, key, *aliases, default=None, required=False, cast=None)[source]#

Fetch a key from scaling_kwargs with aliases + default.

Tries: key -> built-in aliases -> explicit aliases
Treats None and blank strings as “missing” and keeps searching.

Parameters:

key (str)
aliases (str)
required (bool)

geoprior.models.subsidence.utils.validate_scaling_kwargs(scaling_kwargs)[source]#

Basic scaling sanity checks.

This includes policy-controlled heuristic checks for common “silent fallback” cases.

Parameters:: scaling_kwargs (dict[str, Any] | None)
Return type:: None

geoprior.models.subsidence.utils.affine_from_cfg(scaling_kwargs, *, scale_key, bias_key, meta_keys=(), unit_key=None)[source]#

Return (a,b) for y_si = y_model*a + b.

Parameters:

scaling_kwargs (dict[str, Any] | None)
scale_key (str)
bias_key (str)
meta_keys (tuple[str, ...])
unit_key (str | None)

Return type:

tuple[Tensor, Tensor]

geoprior.models.subsidence.utils.to_si_thickness(H_model, scaling_kwargs)[source]#

Convert thickness to SI.

Parameters:

H_model (Tensor)
scaling_kwargs (dict[str, Any] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.to_si_head(h_model, scaling_kwargs)[source]#

Convert head/depth to SI meters.

Parameters:

h_model (Tensor)
scaling_kwargs (dict[str, Any] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.to_si_subsidence(s_model, scaling_kwargs)[source]#

Convert subsidence to SI meters.

Parameters:

s_model (Tensor)
scaling_kwargs (dict[str, Any] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.from_si_subsidence(s_si, scaling_kwargs)[source]#

Inverse of to_si_subsidence: s_model = (s_si - b) / a.

Parameters:

s_si (Tensor)
scaling_kwargs (dict[str, Any] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.deg_to_m(axis, scaling_kwargs)[source]#

Meters per degree factor for lon/lat coords.

If coords_in_degrees=True and deg_to_m_lon/lat are missing, we try to compute them from lat0_deg (recommended).

Parameters:

axis (str)
scaling_kwargs (dict[str, Any] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.coord_ranges(scaling_kwargs)[source]#

Return (tR,xR,yR) if coords_normalized.

Parameters:: scaling_kwargs (dict[str, Any] | None)
Return type:: tuple[float | None, float | None, float | None]

geoprior.models.subsidence.utils.resolve_gwl_dyn_index(scaling_kwargs)[source]#

Resolve GWL channel index for dynamic_features.

Parameters:: scaling_kwargs (dict[str, Any] | None)
Return type:: int

geoprior.models.subsidence.utils.get_gwl_dyn_index_cached(model)[source]#

Cache gwl_dyn_index on model after first resolve.

Return type:: int

geoprior.models.subsidence.utils.resolve_subs_dyn_index(scaling_kwargs)[source]#

Resolve subsidence channel index for dynamic_features.

This is optional: v3.2 can use historical subsidence as a dynamic driver to provide a physics-friendly initial condition for the mean settlement path.

geoprior.models.subsidence.utils.get_subs_dyn_index_cached(model)[source]#

Cache subs_dyn_index on model after first resolve.

Return type:: int

geoprior.models.subsidence.utils.slice_dynamic_channel(Xh, idx)[source]#

Slice (B,T,F) -> (B,T,1) at idx.

Parameters:

Xh (Tensor)
idx (int)

Return type:

Tensor

geoprior.models.subsidence.utils.assert_dynamic_names_match_tensor(Xh, scaling_kwargs)[source]#

Check dynamic_feature_names length matches Xh.

Parameters:

Xh (Tensor)
scaling_kwargs (dict[str, Any] | None)

Return type:

None

geoprior.models.subsidence.utils.gwl_to_head_m(v_m, scaling_kwargs, *, inputs=None)[source]#

Convert depth-bgs to head if possible.

Parameters:

v_m (Tensor)
scaling_kwargs (dict[str, Any] | None)
inputs (dict[str, Tensor] | None)

Return type:

Tensor

geoprior.models.subsidence.utils.get_h_hist_si(model, inputs, *, want_head=True)[source]#

Return head (or depth) history in SI meters.

Parameters:

model (object) – The model instance (provides scaling_kwargs and cached indices).
inputs (dict) – Batch inputs; expects dynamic_features unless an explicit head history key is provided.
want_head (bool, default True) – If True, convert depth-bgs to hydraulic head when possible.

Returns:

(B,T,1) tensor in SI meters.

Return type:

Tensor

geoprior.models.subsidence.utils.get_s_init_si(model, inputs, like)[source]#

Return initial settlement (cumulative subsidence) in SI meters.

Priority: 1) explicit keys in inputs (s_init_si/subs_hist_last_si/…) 2) last historical value from dynamic_features if subs_dyn_index exists 3) zeros (broadcast)

Parameters:

inputs (dict[str, Tensor] | None)
like (Tensor)

Return type:

Tensor

geoprior.models.subsidence.utils.get_h_ref_si(model, inputs, like)[source]#

Return h_ref in SI meters, broadcast to like.

Parameters:

inputs (dict[str, Tensor] | None)
like (Tensor)

Return type:

Tensor

geoprior.models.subsidence.utils.infer_dt_units_from_t(t_BH1, scaling_kwargs, *, eps=1e-12)[source]#

Infer per-step dt in time_units from time tensor t(B,H,1).

Parameters:

t_BH1 (Tensor)
scaling_kwargs (dict[str, Any] | None)
eps (float)

Return type:

Tensor

geoprior.models.subsidence.utils.policy_gate(step, policy, *, warmup_steps=0, ramp_steps=0, dtype=tf.float32)[source]#

Return a scalar gate in [0,1] based on a policy + step.

Parameters:

step (Tensor) – Global step counter (typically optimizer.iterations).
policy ({"always_on","always_off","warmup_off"}) – Gating behavior. always_on returns 1, always_off returns 0, and warmup_off returns 0 for step < warmup_steps before ramping to 1 over ramp_steps when ramp_steps > 0 or switching immediately at warmup_steps otherwise.
warmup_steps (int, default 0) – Number of steps to keep the gate at 0 (only for warmup_off).
ramp_steps (int, default 0) – Number of steps for a linear ramp from 0->1 after warmup. If 0, the gate is a hard step.
dtype (dtype, default tf_float32) – Output dtype.

Return type:

Tensor

geoprior.models.subsidence.utils.finalize_scaling_kwargs(sk)[source]#

Add derived SI conversion constants to scaling_kwargs.

Adds (when possible): - seconds_per_time_unit: float - coord_ranges_si: dict with keys t (seconds), x/y (meters) - coord_inv_ranges_si: inverse of the above (safe floor).

Notes

This helper is designed to be called once when assembling scaling_kwargs (e.g., in your stage2 script) so the model can reuse those constants without recomputing unit conversions in the hot training loop.

Parameters:: sk (dict[str, Any])
Return type:: dict[str, Any]

geoprior.models.subsidence.utils.coord_ranges_si(sk)[source]#

Return coordinate spans in SI (t in seconds; x/y in meters).

If coord_ranges_si is present in sk, it is used directly. Otherwise, this is computed from coord_ranges and time_units (and degree-to-meter factors when applicable).

Parameters:: sk (dict[str, Any])
Return type:: tuple[float | None, float | None, float | None]