Calibration#

The calibration modules provide tools for observation well clustering, time interpolation of simulated heads to observation times, typical hydrograph computation, model file discovery, multi-layer observation well processing, and publication-quality calibration figures.

IWFM2OBS #

Interpolate simulated heads to observation timestamps and compute multi-layer transmissivity-weighted composite heads. Mirrors the Fortran IWFM2OBS utility. Includes the integrated iwfm2obs_from_model() workflow that auto-discovers .out files from the simulation main file.

IWFM2OBS — interpolate simulated heads to observation times.

Mirrors the Fortran IWFM2OBS utility with two core algorithms:

Time interpolation — linearly interpolate simulated time series to match observation timestamps.
Multi-layer T-weighted averaging — compute composite heads at wells that screen multiple aquifer layers, weighting by transmissivity.

The iwfm2obs_from_model() function combines both: it reads .out files directly from the IWFM simulation main file (like the old Fortran iwfm2obs_2015), performs time interpolation, and optionally applies multi-layer T-weighted averaging.

Example

>>> from pyiwfm.calibration.iwfm2obs import interpolate_to_obs_times
>>> result = interpolate_to_obs_times(observed_ts, simulated_ts)
>>> print(result.values)

class pyiwfm.calibration.iwfm2obs.InterpolationConfig(max_extrapolation_time=<factory>, sentinel_value=-999.0, interpolation_method='linear')[source]#

Bases: object

Configuration for time interpolation.

Variables:

max_extrapolation_time (timedelta) – Maximum allowed extrapolation beyond the simulated time range. Observation times beyond this are set to sentinel_value.
sentinel_value (float) – Value to use for observations outside the interpolation window.
interpolation_method (str) – Interpolation method: "linear" or "nearest".

max_extrapolation_time: timedelta#

sentinel_value: float = -999.0#

interpolation_method: Literal['linear', 'nearest'] = 'linear'#

__init__(max_extrapolation_time=<factory>, sentinel_value=-999.0, interpolation_method='linear')#

pyiwfm.calibration.iwfm2obs.interpolate_to_obs_times(observed, simulated, config=None)[source]#

Interpolate simulated values to observation timestamps.

Parameters:

observed (SMPTimeSeries) – Observed time series (provides target timestamps).
simulated (SMPTimeSeries) – Simulated time series to interpolate from.
config (InterpolationConfig | None) – Configuration options. Uses defaults if None.

Returns:

Interpolated simulated values at observation times.

Return type:

SMPTimeSeries

pyiwfm.calibration.iwfm2obs.interpolate_batch(observed, simulated, config=None)[source]#

Interpolate simulated values for all matching bores.

Parameters:

observed (dict[str, SMPTimeSeries]) – Observed time series by bore ID.
simulated (dict[str, SMPTimeSeries]) – Simulated time series by bore ID.
config (InterpolationConfig | None) – Configuration options.

Returns:

Interpolated results for bores present in both inputs.

Return type:

dict[str, SMPTimeSeries]

pyiwfm.calibration.iwfm2obs.expand_obs_to_layers(observed, n_layers, simulated_ids=None)[source]#

Expand base observation IDs to per-layer IDs for IWFM matching.

If observation bore IDs lack %N layer suffixes but the model expects per-layer IDs (e.g. WELL%1, WELL%2), this function duplicates each observation’s time series for layers 1..n_layers.

Detection logic:

If ALL obs IDs already contain % + digit suffix, return unchanged.
For each obs ID without a % suffix, expand to ID%1 .. ID%N.
IDs that already have % suffixes are kept as-is.

Parameters:

observed (dict[str, SMPTimeSeries]) – Observation time series keyed by bore ID.
n_layers (int) – Number of model layers to expand to.
simulated_ids (set[str] or None) – If provided, only expand IDs whose expanded form exists in simulated_ids. This avoids creating entries that would never match.

Returns:

Expanded observation dict (may be same object if no expansion needed).

Return type:

dict[str, SMPTimeSeries]

pyiwfm.calibration.iwfm2obs.deduplicate_smp(input_path, output_path)[source]#

Remove duplicate per-layer entries from an SMP file.

Strips %N suffixes and writes only unique base-ID entries. Verifies that all layer duplicates have identical timestamps and values before deduplication.

Parameters:

input_path (Path or str) – Path to the input SMP file with per-layer duplicates.
output_path (Path or str) – Path for the deduplicated output SMP file.

Returns:

(original_count, deduplicated_count) of unique bore IDs.

Return type:

tuple[int, int]

class pyiwfm.calibration.iwfm2obs.HeadDifferencePair(id1, id2)[source]#

Bases: object

A pair of well IDs for head difference computation.

Mirrors Fortran Class_HeadDifference.f90::HeadDiffPairType. Computes Head(id1) - Head(id2) at matching timesteps.

Variables:

id1 (str) – First well ID.
id2 (str) – Second well ID (subtracted from id1).

id1: str#

id2: str#

__init__(id1, id2)#

pyiwfm.calibration.iwfm2obs.read_head_difference_pairs(path)[source]#

Read head difference pairs from a text file.

Each line contains two whitespace-separated well IDs. IDs are uppercased for case-insensitive matching.

Parameters:: path (str or Path) – Path to the pairs file.
Return type:: list[HeadDifferencePair]
Raises:: ValueError – If a pair has identical IDs or a line has fewer than 2 tokens.

pyiwfm.calibration.iwfm2obs.compute_head_differences(interpolated, pairs)[source]#

Compute head differences for well pairs.

For each pair, computes interpolated[id1].values - interpolated[id2].values at matching timestamps. Mirrors Fortran Class_HeadDifference logic.

Parameters:

interpolated (dict[str, SMPTimeSeries]) – Interpolated time series keyed by bore ID (case-insensitive).
pairs (list[HeadDifferencePair]) – Well pairs to difference.

Returns:

Head differences keyed by "id1-id2".

Return type:

dict[str, SMPTimeSeries]

class pyiwfm.calibration.iwfm2obs.MultiLayerWellSpec(name, x, y, element_id, bottom_of_screen, top_of_screen)[source]#

Bases: object

Specification for a multi-layer observation well.

Variables:

name (str) – Well identifier.
x (float) – X coordinate of the well.
y (float) – Y coordinate of the well.
element_id (int) – Element containing the well (1-based).
bottom_of_screen (float) – Bottom elevation of the well screen.
top_of_screen (float) – Top elevation of the well screen.

name: str#

x: float#

y: float#

element_id: int#

bottom_of_screen: float#

top_of_screen: float#

__init__(name, x, y, element_id, bottom_of_screen, top_of_screen)#

pyiwfm.calibration.iwfm2obs.compute_multilayer_weights(well, grid, stratigraphy, hydraulic_conductivity, fe_node_ids=None, fe_weights=None)[source]#

Compute transmissivity-weighted layer weights for a well.

Parameters:

well (MultiLayerWellSpec) – Well specification with screen interval.
grid (AppGrid) – Model grid.
stratigraphy (Stratigraphy) – Model stratigraphy (layer elevations).
hydraulic_conductivity (NDArray[np.float64]) – Hydraulic conductivity array, shape (n_layers,) or (n_layers, n_nodes) for spatially varying HK.
fe_node_ids (tuple[int, ], optional) – Pre-computed FE interpolation node IDs (1-based). If provided along with fe_weights, skips the expensive FE search.
fe_weights (NDArray[np.float64], optional) – Pre-computed FE interpolation coefficients.

Returns:

Layer weights array, shape (n_layers,), summing to 1.

Return type:

NDArray[np.float64]

pyiwfm.calibration.iwfm2obs.compute_composite_subsidence(layer_subsidence)[source]#

Compute composite subsidence by summing per-layer values.

Unlike head which uses T-weighted averaging, subsidence is additive across layers (Fortran: Class_IWFM2OBS.f90:678-892).

Parameters:: layer_subsidence (NDArray[np.float64]) – Per-layer subsidence values, shape (n_layers,).
Returns:: Total subsidence (sum across all layers).
Return type:: float

pyiwfm.calibration.iwfm2obs.compute_composite_head(well, layer_heads, weights, grid)[source]#

Compute composite head at a multi-layer well.

Parameters:

well (MultiLayerWellSpec) – Well specification.
layer_heads (NDArray[np.float64]) – Head values by layer, shape (n_layers,) at the well location or (n_layers, n_nodes) for nodal heads.
weights (NDArray[np.float64]) – Layer weights from compute_multilayer_weights().
grid (AppGrid) – Model grid (used for FE interpolation if nodal heads provided).

Returns:

Composite head value.

Return type:

float

pyiwfm.calibration.iwfm2obs.iwfm2obs(obs_smp_path, sim_smp_path, output_path, well_specs=None, grid=None, stratigraphy=None, hydraulic_conductivity=None, config=None)[source]#

Run the full IWFM2OBS workflow.

Reads observed and simulated SMP files, performs time interpolation (and optionally multi-layer T-weighted averaging), and writes the result to an output SMP file.

Parameters:

obs_smp_path (Path) – Path to observed data SMP file.
sim_smp_path (Path) – Path to simulated data SMP file.
output_path (Path) – Path for output interpolated SMP file.
well_specs (list[MultiLayerWellSpec] | None) – Multi-layer well specifications (optional).
grid (AppGrid | None) – Model grid (required if well_specs provided).
stratigraphy (Stratigraphy | None) – Model stratigraphy (required if well_specs provided).
hydraulic_conductivity (NDArray[np.float64] | None) – HK array (required if well_specs provided).
config (InterpolationConfig | None) – Interpolation configuration.

Returns:

Interpolated time series by bore ID.

Return type:

dict[str, SMPTimeSeries]

class pyiwfm.calibration.iwfm2obs.IWFM2OBSConfig(interpolation=<factory>, date_format=2)[source]#

Bases: object

Configuration for the integrated IWFM2OBS workflow.

Variables:

interpolation (InterpolationConfig) – Time interpolation settings.
date_format (int) – Date format: 1 = dd/mm/yyyy, 2 = mm/dd/yyyy.

interpolation: InterpolationConfig#

date_format: int = 2#

__init__(interpolation=<factory>, date_format=2)#

class pyiwfm.calibration.iwfm2obs.IWFM2OBSHydBlock(model_smp='', obs_smp='', output_smp='', threshold=1.0, ins_file='', pcf_file='')[source]#

Bases: object

One hydrograph block from the IWFM2OBS input file.

Variables:

model_smp (str) – Model hydrograph SMP path (ignored in model-discovery mode).
obs_smp (str) – Observation SMP path (blank = skip this type).
output_smp (str) – Output SMP path.
threshold (float) – Extrapolation threshold in days.
ins_file (str) – PEST instruction file path (blank = skip).
pcf_file (str) – PEST PCF file path (blank = skip).

model_smp: str = ''#

obs_smp: str = ''#

output_smp: str = ''#

threshold: float = 1.0#

ins_file: str = ''#

pcf_file: str = ''#

__init__(model_smp='', obs_smp='', output_smp='', threshold=1.0, ins_file='', pcf_file='')#

class pyiwfm.calibration.iwfm2obs.IWFM2OBSInputFile(simulation_main_file='', date_format=2, gw=<factory>, stream=<factory>, tiledrain=<factory>, subsidence=<factory>, head_diff_enabled=False, head_diff_file='', multilayer_enabled=False, multilayer_obs_well_file='', multilayer_elements_file='', multilayer_nodes_file='', multilayer_stratigraphy_file='', multilayer_gw_main_file='')[source]#

Bases: object

Parsed IWFM2OBS input file (iwfm2obs_template.in format).

Mirrors the Fortran IWFM2OBS input file structure with 4 hydrograph blocks, head difference flag, and multi-layer target flag.

Variables:

simulation_main_file (str) – IWFM simulation main file (blank = explicit SMP mode).
date_format (int) – 1 = dd/mm/yyyy, 2 = mm/dd/yyyy.
gw (IWFM2OBSHydBlock) – Groundwater head hydrograph block.
stream (IWFM2OBSHydBlock) – Stream hydrograph block.
tiledrain (IWFM2OBSHydBlock) – Tile drain hydrograph block.
subsidence (IWFM2OBSHydBlock) – Subsidence hydrograph block.
head_diff_enabled (bool) – Whether to compute head differences.
head_diff_file (str) – Path to head difference pair file.
multilayer_enabled (bool) – Whether to enable multi-layer T-weighted averaging.
multilayer_obs_well_file (str) – Observation well locations + screen intervals.
multilayer_elements_file (str) – IWFM element connectivity file.
multilayer_nodes_file (str) – IWFM node coordinates file.
multilayer_stratigraphy_file (str) – IWFM stratigraphy (layer elevations) file.
multilayer_gw_main_file (str) – IWFM GW main file (for hydraulic conductivity).

simulation_main_file: str = ''#

date_format: int = 2#

gw: IWFM2OBSHydBlock#

stream: IWFM2OBSHydBlock#

tiledrain: IWFM2OBSHydBlock#

subsidence: IWFM2OBSHydBlock#

head_diff_enabled: bool = False#

head_diff_file: str = ''#

multilayer_enabled: bool = False#

multilayer_obs_well_file: str = ''#

multilayer_elements_file: str = ''#

multilayer_nodes_file: str = ''#

multilayer_stratigraphy_file: str = ''#

multilayer_gw_main_file: str = ''#

__init__(simulation_main_file='', date_format=2, gw=<factory>, stream=<factory>, tiledrain=<factory>, subsidence=<factory>, head_diff_enabled=False, head_diff_file='', multilayer_enabled=False, multilayer_obs_well_file='', multilayer_elements_file='', multilayer_nodes_file='', multilayer_stratigraphy_file='', multilayer_gw_main_file='')#

pyiwfm.calibration.iwfm2obs.read_iwfm2obs_config(path)[source]#

Parse an IWFM2OBS input file (iwfm2obs_template.in format).

Reads the structured Fortran-compatible config with 4 hydrograph blocks (GW, stream, tile drain, subsidence), head difference flag, and multi-layer target flag.

Parameters:: path (str or Path) – Path to the IWFM2OBS input file.
Returns:: Parsed configuration.
Return type:: IWFM2OBSInputFile

pyiwfm.calibration.iwfm2obs.iwfm2obs_from_model(simulation_main_file, obs_smp_paths, output_paths, config=None, obs_well_spec_path=None, multilayer_output_path=None, grid=None, stratigraphy=None, hydraulic_conductivity=None)[source]#

Full IWFM2OBS workflow reading .out files directly from the model.

Steps:

discover_hydrograph_files() — find .out paths and hydrograph metadata.
For each hydrograph type with observations: IWFMHydrographReader reads the .out file → convert to SMPTimeSeries dict → interpolate.
If multi-layer specified: compute T-weighted composite heads and write GW_MultiLayer.out and PEST .ins files.

Parameters:

simulation_main_file (Path or str) – IWFM simulation main file path.
obs_smp_paths (dict[str, Path]) – Observation SMP file paths keyed by type ("gw", "stream").
output_paths (dict[str, Path]) – Output SMP file paths keyed by type.
config (IWFM2OBSConfig or None) – Workflow configuration.
obs_well_spec_path (Path or None) – Multi-layer well specification file (enables T-weighted averaging).
multilayer_output_path (Path or None) – Path for GW_MultiLayer.out output.
grid (AppGrid or None) – Model grid (required for multi-layer).
stratigraphy (Stratigraphy or None) – Model stratigraphy (required for multi-layer).
hydraulic_conductivity (NDArray[np.float64] or None) – HK array (required for multi-layer).

Returns:

Interpolated results keyed by type then bore ID.

Return type:

dict[str, dict[str, SMPTimeSeries]]

pyiwfm.calibration.iwfm2obs.write_multilayer_output(results, well_specs, weights, output_path, n_layers)[source]#

Write GW_MultiLayer.out format output.

Format: Name Date Time Simulated T1 T2 T3 T4 NewTOS NewBOS

Parameters:

results (dict[str, list[tuple[datetime, float]]]) – Composite head results keyed by well name.
well_specs (list[ObsWellSpec]) – Well specifications.
weights (list[NDArray[np.float64]]) – Per-well layer weight arrays.
output_path (Path) – Output file path.
n_layers (int) – Number of model layers.

pyiwfm.calibration.iwfm2obs.compute_composite_continuous(per_layer_sim, well_specs, layer_weights, n_layers=4)[source]#

Compute T-weighted composite heads at ALL simulation timesteps.

Unlike the standard iwfm2obs_from_model workflow which first interpolates per-layer heads to observation dates and then averages, this function averages per-layer data first across all timesteps, producing a continuous composite time series per well.

This is the correct order of operations for CalcTypeHyd: the composite series preserves full temporal resolution (577 monthly timesteps), and downstream tools can sample or aggregate as needed.

Parameters:

per_layer_sim (dict[str, SMPTimeSeries]) – Per-layer simulation time series keyed by bore ID with %N layer suffix (e.g. "WELL_A%1", "WELL_A%2").
well_specs (list[MultiLayerWellSpec]) – Well specifications with screen intervals.
layer_weights (dict[str, NDArray[np.float64]]) – Pre-computed T-weights per well, keyed by base well name. Each array has shape (n_layers,).
n_layers (int) – Number of model layers (default 4).

Returns:

Composite time series keyed by base well name (no %N suffix).

Return type:

dict[str, SMPTimeSeries]

pyiwfm.calibration.iwfm2obs.average_to_seasonal(continuous, periods)[source]#

Aggregate continuous monthly time series to seasonal window averages.

For each well, groups timesteps by seasonal window (e.g. Jan-Apr for spring) and year, computes the mean, and assigns the result to the representative date. Both sim and obs data should be processed through this function with the same periods to ensure consistent comparison.

Parameters:

continuous (dict[str, SMPTimeSeries]) – Continuous monthly time series by bore ID.
periods (list[tuple[str, list[int], str]]) –
Seasonal period definitions as (name, months, representative_date) tuples. representative_date is "MM/DD" format. Example:
```
[("Spring", [1,2,3,4], "03/01"),
 ("Fall", [8,9,10,11], "10/01")]
```

Returns:

Seasonal-averaged time series. Each well has one record per period per year (e.g. 2 records/year for biannual periods).

Return type:

dict[str, SMPTimeSeries]

Model File Discovery #

Parse an IWFM simulation main file to auto-discover hydrograph .out file paths and observation metadata (bore IDs, layers, coordinates). Ports the model-file-discovery logic from the old Fortran IWFM2OBS program.

Discover hydrograph output files from an IWFM simulation main file.

Parses the simulation main file → component main files → .out file paths and hydrograph metadata (bore IDs, layer numbers). Ports the model-file- discovery logic from the old iwfm2obs_2015 Fortran program.

Reuses existing pyiwfm readers wherever possible:

pyiwfm.io.iwfm_reader.next_data_value() for line parsing
pyiwfm.io.iwfm_reader.resolve_path() for path resolution
pyiwfm.io.iwfm_reader.is_comment_line() for comment detection

class pyiwfm.calibration.model_file_discovery.HydrographFileInfo(gw_hydrograph_path=None, stream_hydrograph_path=None, subsidence_hydrograph_path=None, tiledrain_hydrograph_path=None, gw_locations=<factory>, stream_locations=<factory>, subsidence_locations=<factory>, tiledrain_locations=<factory>, n_model_layers=1, gw_main_path=None, stream_main_path=None, start_date_str='', time_unit='')[source]#

Bases: object

Paths and metadata for discovered hydrograph output files.

Variables:

gw_hydrograph_path (Path or None) – Path to the GW hydrograph .out file.
stream_hydrograph_path (Path or None) – Path to the stream hydrograph .out file.
subsidence_hydrograph_path (Path or None) – Path to the subsidence hydrograph .out file.
tiledrain_hydrograph_path (Path or None) – Path to the tile drain hydrograph .out file.
gw_locations (list[HydrographLocation]) – GW hydrograph locations (with layer info).
stream_locations (list[HydrographLocation]) – Stream hydrograph locations.
n_model_layers (int) – Number of model layers (from stratigraphy, if discoverable).
gw_main_path (Path or None) – Path to the GW main file.
stream_main_path (Path or None) – Path to the stream main file.
start_date_str (str) – Simulation start date string from the main file.
time_unit (str) – Simulation time step unit (e.g. "1MON").

gw_hydrograph_path: Path | None = None#

stream_hydrograph_path: Path | None = None#

subsidence_hydrograph_path: Path | None = None#

tiledrain_hydrograph_path: Path | None = None#

gw_locations: list[HydrographLocation]#

stream_locations: list[HydrographLocation]#

subsidence_locations: list[HydrographLocation]#

tiledrain_locations: list[HydrographLocation]#

n_model_layers: int = 1#

gw_main_path: Path | None = None#

stream_main_path: Path | None = None#

start_date_str: str = ''#

time_unit: str = ''#

__init__(gw_hydrograph_path=None, stream_hydrograph_path=None, subsidence_hydrograph_path=None, tiledrain_hydrograph_path=None, gw_locations=<factory>, stream_locations=<factory>, subsidence_locations=<factory>, tiledrain_locations=<factory>, n_model_layers=1, gw_main_path=None, stream_main_path=None, start_date_str='', time_unit='')#

pyiwfm.calibration.model_file_discovery.discover_hydrograph_files(simulation_main_file)[source]#

Parse an IWFM simulation main file to discover hydrograph output paths.

Parameters:: simulation_main_file (Path or str) – Path to the IWFM simulation main file (e.g. C2VSimFG.in).
Returns:: Discovered paths and metadata.
Return type:: HydrographFileInfo
Raises:: FileNotFoundError – If the simulation main file does not exist.

Observation Well Specification #

Read observation well specification files for multi-layer target processing with screen depth intervals and element locations.

Reader for observation well specification files (multi-layer target input).

The obs well spec file defines wells with screen intervals for transmissivity-weighted depth averaging of groundwater heads.

File format (whitespace-delimited, one header line):

Name                X           Y    Element   BOS    TOS  OverwriteLayer
S_380313N1219426W  6302184.5  2161430.2   1234  -175.44  -105.44  -1

OverwriteLayer = -1 means use the screen interval; a positive value forces all weight to that single layer.

class pyiwfm.calibration.obs_well_spec.ObsWellSpec(name, x, y, element_id, bottom_of_screen, top_of_screen, overwrite_layer=-1)[source]#

Bases: object

Specification for one observation well with screen interval.

Variables:

name (str) – Well identifier (up to 25 characters).
x (float) – X coordinate of the well.
y (float) – Y coordinate of the well.
element_id (int) – Element containing the well (1-based).
bottom_of_screen (float) – Bottom elevation of the well screen.
top_of_screen (float) – Top elevation of the well screen.
overwrite_layer (int) – Layer override (-1 = use screen interval, >0 = force single layer).

name: str#

x: float#

y: float#

element_id: int#

bottom_of_screen: float#

top_of_screen: float#

overwrite_layer: int = -1#

__init__(name, x, y, element_id, bottom_of_screen, top_of_screen, overwrite_layer=-1)#

pyiwfm.calibration.obs_well_spec.read_obs_well_spec(filepath)[source]#

Read an observation well specification file.

Parameters:

filepath (Path or str) – Path to the obs well spec file.

Returns:

Parsed well specifications.

Return type:

list[ObsWellSpec]

Raises:

FileNotFoundError – If the file does not exist.
ValueError – If a data line cannot be parsed.

Typical Hydrographs (CalcTypHyd)#

Compute typical hydrograph curves by cluster using seasonal averaging and membership-weighted combination. Includes Fortran .in config file parsing (read_calctyphyd_config, CalcTypHydFileConfig), Fortran-matching time-series computation (compute_typical_hydrographs_timeseries), and PEST .out/.ins output (write_pest_output). Date-range filtering and header-aware cluster weight reading are also supported.

CalcTypHyd — compute typical hydrographs from observation well data.

Mirrors the Fortran CalcTypHyd utility. The algorithm:

Group observations into seasonal periods (default: 4 seasons)
Compute seasonal averages per well
De-mean each well’s seasonal series
Compute cluster-weighted average of de-meaned series

Example

>>> from pyiwfm.calibration.calctyphyd import compute_typical_hydrographs
>>> result = compute_typical_hydrographs(water_levels, cluster_weights)
>>> for th in result.hydrographs:
...     print(f"Cluster {th.cluster_id}: {len(th.contributing_wells)} wells")

class pyiwfm.calibration.calctyphyd.SeasonalPeriod(name, months, representative_date)[source]#

Bases: object

Definition of a seasonal time period.

Variables:

name (str) – Season name (e.g., “Winter”).
months (list[int]) – Month numbers belonging to this season (1-12).
representative_date (str) – Representative date for this season in MM/DD format.

name: str#

months: list[int]#

representative_date: str#

__init__(name, months, representative_date)#

class pyiwfm.calibration.calctyphyd.CalcTypHydConfig(seasonal_periods=None, min_records_per_season=1, start_date=None, end_date=None)[source]#

Bases: object

Configuration for typical hydrograph computation.

Variables:

seasonal_periods (list[SeasonalPeriod] | None) – Seasonal period definitions. Uses 4 standard seasons if None.
min_records_per_season (int) – Minimum observations required per season to include a well.
start_date (datetime | None) – Optional start date for filtering water levels.
end_date (datetime | None) – Optional end date for filtering water levels.

seasonal_periods: list[SeasonalPeriod] | None = None#

min_records_per_season: int = 1#

start_date: datetime | None = None#

end_date: datetime | None = None#

__init__(seasonal_periods=None, min_records_per_season=1, start_date=None, end_date=None)#

class pyiwfm.calibration.calctyphyd.TypicalHydrograph(cluster_id, times, values, contributing_wells)[source]#

Bases: object

A typical hydrograph for one cluster.

Variables:

cluster_id (int) – Cluster identifier (0-based).
times (NDArray[np.datetime64]) – Representative timestamps for each season.
values (NDArray[np.float64]) – De-meaned typical water level values.
contributing_wells (list[str]) – Wells that contributed to this hydrograph.

cluster_id: int#

times: ndarray[tuple[Any, ...], dtype[datetime64]]#

values: ndarray[tuple[Any, ...], dtype[float64]]#

contributing_wells: list[str]#

__init__(cluster_id, times, values, contributing_wells)#

class pyiwfm.calibration.calctyphyd.CalcTypHydResult(hydrographs, well_means)[source]#

Bases: object

Result of typical hydrograph computation.

Variables:

hydrographs (list[TypicalHydrograph]) – One typical hydrograph per cluster.
well_means (dict[str, float]) – Mean water level per well (used for de-meaning).

hydrographs: list[TypicalHydrograph]#

well_means: dict[str, float]#

__init__(hydrographs, well_means)#

pyiwfm.calibration.calctyphyd.read_cluster_weights(filepath, n_clusters=None)[source]#

Read cluster membership weights from a text file.

The file format is whitespace-separated with columns: well_id w1 w2 ... wN [extra_cols...]

Header auto-detection is applied unconditionally: the second token of the first non-blank, non-comment line is tested with float(). If the conversion fails the line is treated as a column header and skipped. This allows the function to handle files both with and without a header row without requiring n_clusters to be supplied.

Parameters:

filepath (Path) – Path to the weights file.
n_clusters (int | None) – If provided, only read this many weight columns after the well ID. Extra columns (e.g. memb, cluster, subregion) are ignored. When None all numeric columns are read; header auto-detection is still applied so files with a text header row work correctly.

Returns:

Mapping of well ID to membership weight array.

Return type:

dict[str, NDArray[np.float64]]

pyiwfm.calibration.calctyphyd.compute_seasonal_averages(water_levels, config=None)[source]#

Compute seasonal average water levels per well.

Parameters:

water_levels (dict[str, SMPTimeSeries]) – Water level time series by well ID.
config (CalcTypHydConfig | None) – Configuration.

Returns:

Mapping of well ID to seasonal averages array with shape (n_seasons,). NaN where insufficient data.

Return type:

dict[str, NDArray[np.float64]]

pyiwfm.calibration.calctyphyd.compute_typical_hydrographs(water_levels, cluster_weights, config=None)[source]#

Compute typical hydrographs using cluster membership weights.

Parameters:

water_levels (dict[str, SMPTimeSeries]) – Water level time series by well ID.
cluster_weights (dict[str, NDArray[np.float64]]) – Cluster membership weights by well ID.
config (CalcTypHydConfig | None) – Configuration.

Returns:

Typical hydrographs and per-well means.

Return type:

CalcTypHydResult

class pyiwfm.calibration.calctyphyd.CalcTypHydFileConfig(water_level_path=<factory>, weights_path=<factory>, n_clusters=0, n_wells=0, desired_clusters=<factory>, start_date='', end_date='', pest_base_name='', periods=<factory>)[source]#

Bases: object

Configuration parsed from a Fortran-format CalcTypHyd .in file.

Variables:

water_level_path (Path) – SMP file with water level observations.
weights_path (Path) – Cluster membership weights file.
n_clusters (int) – Total number of clusters in the weights file.
n_wells (int) – Number of wells.
desired_clusters (list[int]) – 1-based cluster IDs to generate output for.
start_date (str) – Start date as "MM/DD/YYYY".
end_date (str) – End date as "MM/DD/YYYY".
pest_base_name (str) – Base name for PEST parameter naming.
periods (list[SeasonalPeriod]) – Averaging period definitions.

water_level_path: Path#

weights_path: Path#

n_clusters: int = 0#

n_wells: int = 0#

desired_clusters: list[int]#

start_date: str = ''#

end_date: str = ''#

pest_base_name: str = ''#

periods: list[SeasonalPeriod]#

__init__(water_level_path=<factory>, weights_path=<factory>, n_clusters=0, n_wells=0, desired_clusters=<factory>, start_date='', end_date='', pest_base_name='', periods=<factory>)#

pyiwfm.calibration.calctyphyd.read_calctyphyd_config(filepath)[source]#

Parse a Fortran-format CalcTypHyd .in config file.

Parameters:: filepath (Path) – Path to the .in file.
Returns:: Parsed configuration.
Return type:: CalcTypHydFileConfig

class pyiwfm.calibration.calctyphyd.TypicalHydrographTimeSeries(cluster_id, dates=<factory>, values=<factory>, pest_names=<factory>)[source]#

Bases: object

Per-period-year time series for one cluster.

Variables:

cluster_id (int) – Cluster identifier (1-based, matching desired_clusters).
dates (list[datetime]) – Representative dates for each period-year entry.
values (list[float]) – De-meaned cluster-weighted values.
pest_names (list[str]) – PEST parameter names (one per entry).

cluster_id: int#

dates: list[datetime]#

values: list[float]#

pest_names: list[str]#

__init__(cluster_id, dates=<factory>, values=<factory>, pest_names=<factory>)#

class pyiwfm.calibration.calctyphyd.CalcTypHydTimeSeriesResult(hydrographs, well_means)[source]#

Bases: object

Result of time-series typical hydrograph computation.

Variables:

hydrographs (list[TypicalHydrographTimeSeries]) – One time series per desired cluster.
well_means (dict[str, float]) – Mean water level per well.

hydrographs: list[TypicalHydrographTimeSeries]#

well_means: dict[str, float]#

__init__(hydrographs, well_means)#

pyiwfm.calibration.calctyphyd.compute_typical_hydrographs_timeseries(water_levels, cluster_weights, file_config, config=None)[source]#

Compute typical hydrographs as per-period-year time series.

Unlike compute_typical_hydrographs() which produces a single seasonal pattern, this function produces one value per period per year within the date range — matching the Fortran CalcTypHyd output.

The algorithm matches the Fortran Class_CalcTypeHyd:

Bin observations into period-year slots (like Fortran rMonAvg). Only observations in active months contribute.
Compute per-well mean as the mean of these slot averages (like Fortran ComputeMeans), NOT the mean of raw observations.
De-mean each slot average and compute cluster-weighted average.

Parameters:

water_levels (dict[str, SMPTimeSeries]) – Water level time series by well ID.
cluster_weights (dict[str, NDArray[np.float64]]) – Cluster membership weights by well ID.
file_config (CalcTypHydFileConfig) – Configuration from the .in file (date range, periods, clusters).
config (CalcTypHydConfig | None) – Additional configuration (min_records_per_season).

Returns:

Per-period-year time series for each desired cluster.

Return type:

CalcTypHydTimeSeriesResult

pyiwfm.calibration.calctyphyd.write_pest_output(result, file_config, output_dir)[source]#

Write PEST .out and .ins files matching Fortran CalcTypHyd format.

For each desired cluster, writes:

sim_{weights_stem}_cls{N}.out — data file
sim_{weights_stem}_cls{N}.ins — PEST instruction file

Parameters:

result (CalcTypHydTimeSeriesResult) – Time-series typical hydrograph results.
file_config (CalcTypHydFileConfig) – Config (for weights file stem and PEST base name).
output_dir (Path) – Directory for output files.

Returns:

Paths to all written files.

Return type:

list[Path]

pyiwfm.calibration.calctyphyd.compute_obs_type_hydrographs(water_levels, cluster_weights, file_config, output_dir, *, obs_prefix='obs_')[source]#

Compute observed type hydrographs and write PEST output files.

Convenience wrapper that computes period-year type hydrographs from observation data and writes obs_*.out / obs_*.ins files.

This function is intended to be called from the data processing pipeline (Step 8) to pre-compute the observed type hydrograph targets. The sim counterparts are produced by CalcTypeHyd reading the continuous sim SMP.

Both sim and obs type hydrographs use the same seasonal period aggregation (window-averaging) so they are directly comparable.

Parameters:

water_levels (dict[str, SMPTimeSeries]) – Observed water level time series by well ID (from GW_Obs.smp).
cluster_weights (dict[str, NDArray[np.float64]]) – Fuzzy cluster membership weights by well ID.
file_config (CalcTypHydFileConfig) – CalcTypHyd configuration (date range, periods, desired clusters, pest base name, weights path — for naming output files).
output_dir (Path) – Directory for output files.
obs_prefix (str) – Prefix for output file names (default "obs_").

Returns:

Paths to all written files.

Return type:

list[Path]

Fuzzy C-Means Clustering #

NumPy-only fuzzy c-means clustering of observation wells using combined spatial and temporal features.

Fuzzy c-means clustering for IWFM observation wells.

Pure NumPy implementation — no scikit-fuzzy dependency. Wells are clustered using a combination of spatial (x, y) and temporal features (cross-correlation, amplitude, trend, seasonality).

Example

>>> from pyiwfm.calibration.clustering import fuzzy_cmeans_cluster
>>> result = fuzzy_cmeans_cluster(well_locations, well_timeseries)
>>> print(f"FPC: {result.fpc:.3f}")
>>> print(result.get_cluster_wells(0))

class pyiwfm.calibration.clustering.ClusteringConfig(n_clusters=5, fuzziness=2.0, spatial_weight=0.3, temporal_weight=0.7, max_iterations=300, tolerance=1e-06, random_seed=None)[source]#

Bases: object

Configuration for fuzzy c-means clustering.

Variables:

n_clusters (int) – Number of clusters.
fuzziness (float) – Fuzziness parameter m (must be > 1).
spatial_weight (float) – Weight for spatial (x, y) features.
temporal_weight (float) – Weight for temporal features.
max_iterations (int) – Maximum number of FCM iterations.
tolerance (float) – Convergence tolerance on membership change.
random_seed (int | None) – Random seed for reproducibility.

n_clusters: int = 5#

fuzziness: float = 2.0#

spatial_weight: float = 0.3#

temporal_weight: float = 0.7#

max_iterations: int = 300#

tolerance: float = 1e-06#

random_seed: int | None = None#

__init__(n_clusters=5, fuzziness=2.0, spatial_weight=0.3, temporal_weight=0.7, max_iterations=300, tolerance=1e-06, random_seed=None)#

class pyiwfm.calibration.clustering.ClusteringResult(membership, cluster_centers, well_ids, n_clusters, fpc)[source]#

Bases: object

Result of fuzzy c-means clustering.

Variables:

membership (NDArray[np.float64]) – Membership matrix, shape (n_wells, n_clusters). Rows sum to 1.
cluster_centers (NDArray[np.float64]) – Cluster centers in feature space, shape (n_clusters, n_features).
well_ids (list[str]) – Well identifiers in row order.
n_clusters (int) – Number of clusters.
fpc (float) – Fuzzy Partition Coefficient (0 to 1, higher = crisper).

membership: ndarray[tuple[Any, ...], dtype[float64]]#

cluster_centers: ndarray[tuple[Any, ...], dtype[float64]]#

well_ids: list[str]#

n_clusters: int#

fpc: float#

get_dominant_cluster(well_id)[source]#

Return the dominant cluster ID for a well.

Parameters:: well_id (str) – Well identifier.
Returns:: Cluster index with highest membership.
Return type:: int

get_cluster_wells(cluster_id, threshold=0.5)[source]#

Return wells with membership above threshold for a cluster.

Parameters:

cluster_id (int) – Cluster index.
threshold (float) – Minimum membership value.

Returns:

Well IDs above the threshold.

Return type:

list[str]

to_weights_file(output_path)[source]#

Write cluster weights in CalcTypHyd format.

Parameters:: output_path (Path) – Output file path.

__init__(membership, cluster_centers, well_ids, n_clusters, fpc)#

pyiwfm.calibration.clustering.fuzzy_cmeans_cluster(well_locations, well_timeseries, config=None)[source]#

Cluster observation wells using fuzzy c-means.

Parameters:

well_locations (dict[str, tuple[float, float]]) – Mapping of well ID to (x, y) coordinates.
well_timeseries (dict[str, SMPTimeSeries]) – Mapping of well ID to water level time series.
config (ClusteringConfig | None) – Clustering configuration. Uses defaults if None.

Returns:

Clustering membership and cluster centers.

Return type:

ClusteringResult

Headall Extraction #

Extract and process HeadAll HDF5 output data for calibration workflows.

HeadAll extraction pipeline for IWFM models.

Extracts head time series at arbitrary well locations from HeadAll HDF5 output using finite element interpolation and transmissivity-weighted multi-layer averaging.

This module composes existing pyiwfm components (LazyHeadDataLoader, FEInterpolator, compute_multilayer_weights) into an end-to-end pipeline.

Example

>>> extractor = HeadAllExtractor(model, headall_path)
>>> extractor.prepare(wells)
>>> result = extractor.extract()
>>> extractor.write_cache(Path("continuous_sim.hdf5"))

class pyiwfm.calibration.headall_extraction.WellSpec(name, x, y, layer=-1, element=0, bos=nan, tos=nan)[source]#

Bases: object

Specification for a well extraction point.

Variables:

name (str) – Well identifier.
x (float) – X coordinate (in model units, typically feet).
y (float) – Y coordinate (in model units, typically feet).
layer (int) – Target model layer (1-based). Use -1 for multi-layer T-weighted.
element (int) – Model element containing the well (0 = auto-detect).
bos (float) – Bottom of screen elevation (feet).
tos (float) – Top of screen elevation (feet).

name: str#

x: float#

y: float#

layer: int = -1#

element: int = 0#

bos: float = nan#

tos: float = nan#

__init__(name, x, y, layer=-1, element=0, bos=nan, tos=nan)#

class pyiwfm.calibration.headall_extraction.HeadAllExtractor(model, headall_path)[source]#

Bases: object

Extract head time series at arbitrary well locations from HeadAll HDF5.

Parameters:

model (IWFMModel) – Loaded IWFM model (for mesh geometry and stratigraphy).
headall_path (Path) – Path to the HeadAll HDF5 output file.

__init__(model, headall_path)[source]#

prepare(wells)[source]#

Pre-compute FE interpolation weights and T-weights for all wells.

Parameters:: wells (list[WellSpec]) – Well specifications with coordinates in model units.

extract(timesteps=None)[source]#

Extract per-layer and T-weighted multi-layer heads at all wells.

Iterates timesteps in the outer loop so each HDF5 frame is read exactly once, regardless of the number of wells.

Parameters:: timesteps (list[int], optional) – Specific timestep indices to extract. If None, extract all.
Return type:: ExtractionResult

write_cache(output_path, result)[source]#

Write extraction results to HDF5 cache.

Layout: /times, /per_layer, /values, /names.

Parameters:

output_path (Path) – Output HDF5 file path.
result (ExtractionResult) – Results from extract().

static load_cache(cache_path)[source]#

Load previously cached extraction results.

Parameters:: cache_path (Path) – Path to the HDF5 cache file.
Return type:: ExtractionResult

Results Extraction #

Extract and process model results for calibration comparison and analysis.

Unified results extraction pipeline for IWFM models.

Extracts time series at arbitrary locations from any all-node HDF5 output (HeadAll or SubsidenceAll) using finite element interpolation.

For HEAD data, supports transmissivity-weighted multi-layer averaging. For SUBSIDENCE data, supports layer summation (additive compaction) and cumulative-to-incremental conversion.

This generalizes HeadAllExtractor to handle any all-node output type.

Example

>>> extractor = ResultsExtractor(model, results_path, data_type='SUBSIDENCE')
>>> extractor.prepare(specs)
>>> result = extractor.extract()
>>> extractor.write_smp(Path("SUB_OUT.smp"), result)

class pyiwfm.calibration.results_extraction.ExtractionSpec(name, x, y, layer=0, element=0, bos=nan, tos=nan)[source]#

Bases: object

Specification for an extraction point.

Variables:

name (str) – Location identifier.
x (float) – X coordinate (in model units, typically feet).
y (float) – Y coordinate (in model units, typically feet).
layer (int) – Target model layer (1-based). Use 0 for all-layer aggregation (average for HEAD, sum for SUBSIDENCE). Use -1 for T-weighted multi-layer average (HEAD only).
element (int) – Model element containing the point (0 = auto-detect).
bos (float) – Bottom of screen elevation (feet), for T-weighted averaging.
tos (float) – Top of screen elevation (feet), for T-weighted averaging.

name: str#

x: float#

y: float#

layer: int = 0#

element: int = 0#

bos: float = nan#

tos: float = nan#

__init__(name, x, y, layer=0, element=0, bos=nan, tos=nan)#

class pyiwfm.calibration.results_extraction.ExtractionResult(times, names=<factory>, values=<factory>, per_layer=<factory>, data_type='HEAD', incremental=False)[source]#

Bases: object

Results of extraction.

Variables:

times (NDArray) – Array of timestamps (datetime64).
names (list[str]) – Names of extraction points in order.
values (dict[str, NDArray[np.float64]]) – Extracted values: {name: array of shape (n_times,)}.
per_layer (dict[str, NDArray[np.float64]]) – Per-layer values: {name: array of shape (n_times, n_layers)}.
data_type (str) – ‘HEAD’ or ‘SUBSIDENCE’.
incremental (bool) – If True, values are incremental (diff between timesteps).

times: ndarray[tuple[Any, ...], dtype[datetime64]]#

names: list[str]#

values: dict[str, ndarray[tuple[Any, ...], dtype[float64]]]#

per_layer: dict[str, ndarray[tuple[Any, ...], dtype[float64]]]#

data_type: str = 'HEAD'#

incremental: bool = False#

__init__(times, names=<factory>, values=<factory>, per_layer=<factory>, data_type='HEAD', incremental=False)#

class pyiwfm.calibration.results_extraction.ResultsExtractor(model, results_path, data_type='HEAD', incremental=True)[source]#

Bases: object

Extract time series at arbitrary locations from all-node HDF5 output.

Parameters:

model (IWFMModel) – Loaded IWFM model (for mesh geometry and stratigraphy).
results_path (Path) – Path to the all-node HDF5 output file (HeadAll or SubsidenceAll).
data_type (str) – ‘HEAD’ or ‘SUBSIDENCE’.
incremental (bool) – If True and data_type is ‘SUBSIDENCE’, convert cumulative values to incremental (current - previous timestep). Ignored for HEAD.

__init__(model, results_path, data_type='HEAD', incremental=True)[source]#

prepare(specs)[source]#

Pre-compute FE interpolation weights for all extraction points.

Parameters:: specs (list[ExtractionSpec]) – Extraction specifications with coordinates in model units.

extract(timesteps=None)[source]#

Extract values at all locations.

Iterates timesteps in the outer loop so each HDF5 frame is read exactly once, regardless of the number of extraction points.

Parameters:: timesteps (list[int], optional) – Specific timestep indices to extract. If None, extract all.
Return type:: ExtractionResult

write_smp(output_path, result)[source]#

Write results to SMP format for PEST/IWFM2OBS.

Parameters:

output_path (Path) – Output SMP file path.
result (ExtractionResult) – Results from extract().

write_cache(output_path, result)[source]#

Write extraction results to HDF5 cache.

Parameters:

output_path (Path) – Output HDF5 file path.
result (ExtractionResult) – Results from extract().

static load_cache(cache_path)[source]#

Load previously cached extraction results.

Parameters:: cache_path (Path) – Path to the HDF5 cache file.
Return type:: ExtractionResult

class pyiwfm.calibration.results_extraction.FortranBackend(exe_path, sim_file, data_type='SUBSIDENCE', incremental=True)[source]#

Bases: object

Run ResultsExtract Fortran executable as a fast extraction backend.

Generates an input file, invokes ResultsExtract_x64.exe, and parses the IWFM columnar .out output back into an ExtractionResult.

This is ~300x faster than the pure-Python FE extraction for large datasets (e.g. 31K subsidence locations × 577 timesteps: 3s vs 17min).

Parameters:

exe_path (Path) – Path to ResultsExtract_x64.exe.
sim_file (Path) – Path to the IWFM simulation main file.
data_type (str) – 'HEAD' or 'SUBSIDENCE'.
incremental (bool) – If True and data_type is 'SUBSIDENCE', the Fortran code produces incremental output automatically.

__init__(exe_path, sim_file, data_type='SUBSIDENCE', incremental=True)[source]#

static find_exe(search_dirs=None)[source]#

Locate ResultsExtract_x64.exe in common locations.

generate_input(specs, output_file='RE_OUT.out', factxy=1.0)[source]#

Generate a ResultsExtract input file from ExtractionSpec objects.

Parameters:

specs (list[ExtractionSpec]) – Extraction specifications (coordinates already in model units).
output_file (str) – Name for the ResultsExtract output file.
factxy (float) – Coordinate conversion factor written to the input file.

Returns:

(input file content, ordered list of spec names)

Return type:

tuple[str, list[str]]

run(specs, work_dir=None, factxy=1.0)[source]#

Run ResultsExtract and parse the output.

Parameters:

specs (list[ExtractionSpec]) – Extraction specifications (coordinates in model units).
work_dir (Path, optional) – Working directory for the subprocess.
factxy (float) – Coordinate conversion factor.

Return type:

ExtractionResult

Raises:

FileNotFoundError – If the executable is not found.
RuntimeError – If ResultsExtract reports a FATAL error.

Calibration Plots #

Publication-quality composite figures for calibration reports: 1:1 plots, residual histograms, hydrograph panels, metrics tables, water budget summaries, cluster maps, and typical hydrograph curves.

Publication-quality composite figures for IWFM model calibration.

This module builds on the individual plotting functions in pyiwfm.visualization.plotting to create multi-panel figures suitable for reports and journal articles.

Functions #

plot_calibration_summary() — Multi-panel summary (1:1, spatial bias, histograms)
plot_hydrograph_panel() — Grid of observed vs simulated hydrographs
plot_metrics_table() — Metrics table rendered as a figure
plot_residual_histogram() — Residual distribution histogram
plot_water_budget_summary() — Stacked budget components over time
plot_zbudget_summary() — Zone budget summary
plot_cluster_map() — Spatial map of cluster memberships
plot_typical_hydrographs() — Typical hydrograph curves by cluster

pyiwfm.visualization.calibration_plots.plot_calibration_summary(well_comparisons, grid=None, output_dir=None, dpi=300)[source]#

Create multi-panel calibration summary figures.

Parameters:

well_comparisons (dict[str, tuple[NDArray, NDArray]]) – Mapping of well ID to (observed, simulated) value arrays.
grid (AppGrid | None) – Model grid for spatial plots.
output_dir (Path | None) – If provided, save figures to this directory.
dpi (int) – Output resolution.

Returns:

List of generated figures.

Return type:

list[Figure]

pyiwfm.visualization.calibration_plots.plot_hydrograph_panel(comparisons, n_cols=3, max_panels=12, output_path=None, dpi=300)[source]#

Plot a grid of observed vs simulated hydrographs.

Parameters:

comparisons (dict[str, tuple[NDArray, NDArray, NDArray]]) – Mapping of well ID to (times, observed, simulated).
n_cols (int) – Number of columns in the grid.
max_panels (int) – Maximum number of panels to show.
output_path (Path | None) – If provided, save the figure.
dpi (int) – Output resolution.

Returns:

The generated figure.

Return type:

Figure

pyiwfm.visualization.calibration_plots.plot_metrics_table(metrics_by_well, output_path=None, dpi=300)[source]#

Render calibration metrics as a table figure.

Parameters:

metrics_by_well (dict[str, ComparisonMetrics]) – Mapping of well ID to computed metrics.
output_path (Path | None) – If provided, save the figure.
dpi (int) – Output resolution.

Returns:

The generated figure.

Return type:

Figure

pyiwfm.visualization.calibration_plots.plot_residual_histogram(residuals, ax=None, show_normal_fit=True, figsize=(8, 6))[source]#

Plot a histogram of residuals (simulated - observed).

Parameters:

residuals (NDArray[np.float64]) – Residual values.
ax (Axes | None) – Existing axes to plot on.
show_normal_fit (bool) – Overlay a normal distribution fit.
figsize (tuple[float, float]) – Figure size.

Returns:

The figure and axes.

Return type:

tuple[Figure, Axes]

pyiwfm.visualization.calibration_plots.plot_water_budget_summary(budget_data, times, output_path=None, dpi=300)[source]#

Plot stacked water budget summary over time.

Parameters:

budget_data (dict[str, NDArray[np.float64]]) – Mapping of component name to time series of values.
times (NDArray[np.datetime64]) – Time stamps.
output_path (Path | None) – If provided, save the figure.
dpi (int) – Output resolution.

Returns:

The generated figure.

Return type:

Figure

pyiwfm.visualization.calibration_plots.plot_zbudget_summary(zone_budgets, times, output_path=None, dpi=300)[source]#

Plot zone budget summary.

Parameters:

zone_budgets (dict[str, dict[str, NDArray[np.float64]]]) – Mapping of zone name to component budget data.
times (NDArray[np.datetime64]) – Time stamps.
output_path (Path | None) – If provided, save the figure.
dpi (int) – Output resolution.

Returns:

The generated figure.

Return type:

Figure

pyiwfm.visualization.calibration_plots.plot_cluster_map(well_locations, clustering_result, grid=None, figsize=(10, 8))[source]#

Plot a spatial map of cluster memberships.

Parameters:

well_locations (dict[str, tuple[float, float]]) – Mapping of well ID to (x, y) coordinates.
clustering_result (ClusteringResult) – Clustering result with membership matrix.
grid (AppGrid | None) – Model grid for background mesh.
figsize (tuple[float, float]) – Figure size.

Returns:

The figure and axes.

Return type:

tuple[Figure, Axes]

pyiwfm.visualization.calibration_plots.plot_typical_hydrographs(result, figsize=(12, 6))[source]#

Plot typical hydrographs by cluster.

Parameters:

result (CalcTypHydResult) – Typical hydrograph computation result.
figsize (tuple[float, float]) – Figure size.

Returns:

The figure and axes.

Return type:

tuple[Figure, Axes]

PEST CLI #

Command-line interface for PEST++ calibration workflows (setup, run, analyze).

CLI subcommand for PEST++ parameter estimation workflows.

Usage:

pyiwfm pest setup --model-dir ./model [--output-dir ./pest] [--case-name iwfm_cal]
pyiwfm pest run --pst-file ./pest/iwfm_cal.pst [--executable pestpp-glm]
pyiwfm pest analyze --pst-file ./pest/iwfm_cal.pst [--format text]

pyiwfm.cli.pest.add_pest_parser(subparsers)[source]#

Register the pest subcommand with its sub-subcommands.

pyiwfm.cli.pest.run_pest(args)[source]#

Dispatch to the appropriate pest sub-command.