Runner and PEST++ Integration#

The runner modules provide subprocess execution, scenario management, and complete PEST++ calibration interface for IWFM models.

Runner Module #

The runner module provides the core subprocess interface for executing IWFM executables, including IWFMRunner and IWFMExecutables.

IWFM subprocess runner for executing IWFM executables.

class pyiwfm.runner.runner.IWFMExecutables(simulation=None, simulation_parallel=None, preprocessor=None, budget=None, zbudget=None)[source]#

Bases: object

Paths to IWFM executables.

Variables:

simulation (Path | None) – Path to the Simulation executable.
simulation_parallel (Path | None) – Path to the parallel Simulation executable.
preprocessor (Path | None) – Path to the PreProcessor executable.
budget (Path | None) – Path to the Budget post-processor executable.
zbudget (Path | None) – Path to the ZBudget post-processor executable.

simulation: Path | None = None#

simulation_parallel: Path | None = None#

preprocessor: Path | None = None#

budget: Path | None = None#

zbudget: Path | None = None#

__post_init__()[source]#

Validate and convert paths.

property available: list[str]#: List of available executables.

__repr__()[source]#

Return string representation.

__init__(simulation=None, simulation_parallel=None, preprocessor=None, budget=None, zbudget=None)#

pyiwfm.runner.runner.find_iwfm_executables(search_paths=None, env_var='IWFM_BIN')[source]#

Find IWFM executables on the system.

Searches for executables in the following order: 1. Paths provided in search_paths 2. Path from environment variable (default: IWFM_BIN) 3. Current working directory 4. System PATH

Parameters:

search_paths (list[Path] | None) – Additional paths to search for executables.
env_var (str) – Environment variable containing IWFM bin directory.

Returns:

Dataclass containing paths to found executables.

Return type:

IWFMExecutables

class pyiwfm.runner.runner.IWFMRunner(executables=None, working_dir=None)[source]#

Bases: object

Run IWFM executables via subprocess.

This class provides methods to run IWFM PreProcessor, Simulation, Budget, and ZBudget executables. It handles input/output redirection, working directory management, and result parsing.

Parameters:

executables (IWFMExecutables | None) – Paths to IWFM executables. If None, will auto-detect.
working_dir (Path | None) – Default working directory for runs. If None, uses the directory containing the input file.

Examples

>>> runner = IWFMRunner()
>>> result = runner.run_simulation("Simulation/Simulation.in")
>>> if result.success:
...     print(f"Completed {result.n_timesteps} timesteps")

__init__(executables=None, working_dir=None)[source]#

Initialize the IWFM runner.

run_preprocessor(main_file, working_dir=None, timeout=None)[source]#

Run the IWFM PreProcessor.

Parameters:

main_file (Path | str) – Path to the PreProcessor main input file.
working_dir (Path | str | None) – Working directory. Defaults to main_file’s directory.
timeout (float | None) – Timeout in seconds.

Returns:

Result object containing success status and outputs.

Return type:

PreprocessorResult

Raises:

FileNotFoundError – If the executable or main file is not found.

run_simulation(main_file, working_dir=None, timeout=None, parallel=False)[source]#

Run the IWFM Simulation.

Parameters:

main_file (Path | str) – Path to the Simulation main input file.
working_dir (Path | str | None) – Working directory. Defaults to main_file’s directory.
timeout (float | None) – Timeout in seconds.
parallel (bool) – Use parallel executable if available.

Returns:

Result object containing success status and outputs.

Return type:

SimulationResult

Raises:

FileNotFoundError – If the executable or main file is not found.

run_budget(budget_file, working_dir=None, timeout=None, instructions=None)[source]#

Run the IWFM Budget post-processor.

Parameters:

budget_file (Path | str) – Path to the budget binary file.
working_dir (Path | str | None) – Working directory. Defaults to budget_file’s directory.
timeout (float | None) – Timeout in seconds.
instructions (str | None) – Budget processing instructions (interactive responses).

Returns:

Result object containing success status and outputs.

Return type:

BudgetResult

Raises:

FileNotFoundError – If the executable or budget file is not found.

run_zbudget(zbudget_file, zone_file=None, working_dir=None, timeout=None, instructions=None)[source]#

Run the IWFM ZBudget post-processor.

Parameters:

zbudget_file (Path | str) – Path to the zone budget HDF5 file.
zone_file (Path | str | None) – Path to the zone definition file.
working_dir (Path | str | None) – Working directory. Defaults to zbudget_file’s directory.
timeout (float | None) – Timeout in seconds.
instructions (str | None) – ZBudget processing instructions (interactive responses).

Returns:

Result object containing success status and outputs.

Return type:

ZBudgetResult

Raises:

FileNotFoundError – If the executable or zbudget file is not found.

__repr__()[source]#

Return string representation.

Results Module #

Typed result classes for IWFM executable runs.

Result dataclasses for IWFM subprocess runs.

class pyiwfm.runner.results.RunResult(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>)[source]#

Bases: object

Base result class for IWFM executable runs.

Variables:

success (bool) – Whether the run completed successfully.
return_code (int) – Process return code (0 = success).
stdout (str) – Standard output from the process.
stderr (str) – Standard error from the process.
working_dir (Path) – Working directory where the run executed.
elapsed_time (timedelta) – Wall-clock time for the run.
log_file (Path | None) – Path to the IWFM log/message file if created.
log_content (str) – Content of the log file if available.
errors (list[str]) – List of error messages extracted from output/logs.
warnings (list[str]) – List of warning messages extracted from output/logs.

success: bool#

return_code: int#

stdout: str = ''#

stderr: str = ''#

working_dir: Path#

elapsed_time: timedelta#

log_file: Path | None = None#

log_content: str = ''#

errors: list[str]#

warnings: list[str]#

__post_init__()[source]#

Convert paths to Path objects.

property failed: bool#: Check if the run failed.

raise_on_error()[source]#

Raise an exception if the run failed.

Raises:: RuntimeError – If the run did not complete successfully.

__init__(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>)#

class pyiwfm.runner.results.PreprocessorResult(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, main_file=None, binary_output=None, n_nodes=0, n_elements=0, n_layers=0, n_subregions=0)[source]#

Bases: RunResult

Result from running the IWFM PreProcessor.

Variables:

main_file (Path | None) – Path to the preprocessor main input file.
binary_output (Path | None) – Path to the generated binary output file.
n_nodes (int) – Number of nodes in the model.
n_elements (int) – Number of elements in the model.
n_layers (int) – Number of layers in the model.
n_subregions (int) – Number of subregions.

main_file: Path | None = None#

binary_output: Path | None = None#

n_nodes: int = 0#

n_elements: int = 0#

n_layers: int = 0#

n_subregions: int = 0#

__post_init__()[source]#

Convert paths to Path objects.

__init__(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, main_file=None, binary_output=None, n_nodes=0, n_elements=0, n_layers=0, n_subregions=0)#

class pyiwfm.runner.results.SimulationResult(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, main_file=None, n_timesteps=0, start_date=None, end_date=None, budget_files=<factory>, hydrograph_files=<factory>, final_heads_file=None, convergence_failures=0, mass_balance_error=0.0)[source]#

Bases: RunResult

Result from running the IWFM Simulation.

Variables:

main_file (Path | None) – Path to the simulation main input file.
n_timesteps (int) – Number of timesteps completed.
start_date (datetime | None) – Simulation start date.
end_date (datetime | None) – Simulation end date.
budget_files (list[Path]) – List of budget output files generated.
hydrograph_files (list[Path]) – List of hydrograph output files generated.
final_heads_file (Path | None) – Path to final groundwater heads file.
convergence_failures (int) – Number of timesteps with convergence issues.
mass_balance_error (float) – Maximum mass balance error (if reported).

main_file: Path | None = None#

n_timesteps: int = 0#

start_date: datetime | None = None#

end_date: datetime | None = None#

budget_files: list[Path]#

hydrograph_files: list[Path]#

final_heads_file: Path | None = None#

convergence_failures: int = 0#

mass_balance_error: float = 0.0#

__post_init__()[source]#

Convert paths to Path objects.

__init__(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, main_file=None, n_timesteps=0, start_date=None, end_date=None, budget_files=<factory>, hydrograph_files=<factory>, final_heads_file=None, convergence_failures=0, mass_balance_error=0.0)#

class pyiwfm.runner.results.BudgetResult(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, budget_file=None, output_file=None, n_locations=0, n_timesteps=0, components=<factory>)[source]#

Bases: RunResult

Result from running the IWFM Budget post-processor.

Variables:

budget_file (Path | None) – Path to the budget binary file processed.
output_file (Path | None) – Path to the output file generated.
n_locations (int) – Number of locations in the budget.
n_timesteps (int) – Number of timesteps in the budget.
components (list[str]) – List of budget components.

budget_file: Path | None = None#

output_file: Path | None = None#

n_locations: int = 0#

n_timesteps: int = 0#

components: list[str]#

__post_init__()[source]#

Convert paths to Path objects.

__init__(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, budget_file=None, output_file=None, n_locations=0, n_timesteps=0, components=<factory>)#

class pyiwfm.runner.results.ZBudgetResult(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, zbudget_file=None, zone_file=None, output_file=None, n_zones=0, n_timesteps=0)[source]#

Bases: RunResult

Result from running the IWFM ZBudget post-processor.

Variables:

zbudget_file (Path | None) – Path to the zone budget HDF5 file processed.
zone_file (Path | None) – Path to the zone definition file used.
output_file (Path | None) – Path to the output file generated.
n_zones (int) – Number of zones processed.
n_timesteps (int) – Number of timesteps in the output.

zbudget_file: Path | None = None#

zone_file: Path | None = None#

output_file: Path | None = None#

n_zones: int = 0#

n_timesteps: int = 0#

__post_init__()[source]#

Convert paths to Path objects.

__init__(success, return_code, stdout='', stderr='', working_dir=<factory>, elapsed_time=<factory>, log_file=None, log_content='', errors=<factory>, warnings=<factory>, zbudget_file=None, zone_file=None, output_file=None, n_zones=0, n_timesteps=0)#

Scenario Module #

Scenario management for running and comparing multiple model configurations.

Scenario management for IWFM model runs.

class pyiwfm.runner.scenario.Scenario(name, description='', modifications=<factory>, modifier_func=None)[source]#

Bases: object

Definition of a model scenario.

A scenario represents a modified version of a baseline model run. It can include modifications to input files (pumping, diversions, land use, etc.) and is identified by a unique name.

Variables:

name (str) – Unique name identifying this scenario.
description (str) – Description of what this scenario represents.
modifications (dict[str, Any]) – Dictionary of modifications to apply to input files. Keys are file types (e.g., “pumping”, “diversion”), values are modification specifications.
modifier_func (Callable | None) – Optional function to apply custom modifications to the scenario directory. Function signature: (scenario_dir: Path, baseline_dir: Path) -> None

name: str#

description: str = ''#

modifications: dict[str, Any]#

modifier_func: Callable[[Path, Path], None] | None = None#

__post_init__()[source]#

Validate scenario name.

__repr__()[source]#

Return string representation.

__init__(name, description='', modifications=<factory>, modifier_func=None)#

class pyiwfm.runner.scenario.ScenarioResult(scenario, result, scenario_dir, differences=<factory>)[source]#

Bases: object

Result of a scenario run with comparison to baseline.

Variables:

scenario (Scenario) – The scenario that was run.
result (SimulationResult) – The simulation result for this scenario.
scenario_dir (Path) – Directory where the scenario was run.
differences (dict[str, Any]) – Computed differences from baseline (if available).

scenario: Scenario#

result: SimulationResult#

scenario_dir: Path#

differences: dict[str, Any]#

property success: bool#: Check if the scenario run succeeded.

__repr__()[source]#

Return string representation.

__init__(scenario, result, scenario_dir, differences=<factory>)#

class pyiwfm.runner.scenario.ScenarioManager(baseline_dir, scenarios_root=None, main_file_name='Simulation.in')[source]#

Bases: object

Manage and run multiple IWFM scenarios.

This class provides utilities for: - Creating scenario directories from a baseline - Applying modifications to scenario input files - Running scenarios in parallel - Comparing results to baseline

Parameters:

baseline_dir (Path | str) – Path to the baseline model directory.
scenarios_root (Path | str | None) – Root directory for scenario runs. If None, creates a ‘scenarios’ subdirectory in baseline_dir.
main_file_name (str) – Name of the simulation main file (relative to model dir).

Examples

>>> manager = ScenarioManager("C2VSim/Simulation")
>>> scenarios = [
...     Scenario("reduced_pumping", modifications={"pumping": 0.8}),
...     Scenario("no_diversions", modifications={"diversion": 0.0}),
... ]
>>> results = manager.run_scenarios(scenarios, parallel=4)

__init__(baseline_dir, scenarios_root=None, main_file_name='Simulation.in')[source]#

Initialize the scenario manager.

create_scenario_dir(scenario, copy_outputs=False)[source]#

Create a directory for a scenario run.

Copies the baseline model files to a new directory and applies the scenario modifications.

Parameters:

scenario (Scenario) – The scenario to create.
copy_outputs (bool) – If True, also copy output files from baseline.

Returns:

Path to the created scenario directory.

Return type:

Path

run_scenario(scenario, runner=None, timeout=None, cleanup_on_success=False)[source]#

Run a single scenario.

Parameters:

scenario (Scenario) – The scenario to run.
runner (IWFMRunner | None) – Runner to use. If None, creates a new one.
timeout (float | None) – Timeout in seconds for the simulation.
cleanup_on_success (bool) – If True, remove scenario directory after successful run.

Returns:

Result of the scenario run.

Return type:

ScenarioResult

run_scenarios(scenarios, runner=None, parallel=1, timeout=None, progress_callback=None)[source]#

Run multiple scenarios, optionally in parallel.

Parameters:

scenarios (list[Scenario]) – List of scenarios to run.
runner (IWFMRunner | None) – Runner to use. If None, creates a new one.
parallel (int) – Number of parallel workers. 1 = sequential.
timeout (float | None) – Timeout per scenario in seconds.
progress_callback (Callable | None) – Optional callback for progress updates. Signature: (scenario_name, completed, total) -> None

Returns:

Dictionary mapping scenario names to results.

Return type:

dict[str, ScenarioResult]

compare_to_baseline(baseline_result, scenario_results)[source]#

Compare scenario results to baseline.

This is a placeholder for comparison logic. Actual comparison depends on the specific outputs being analyzed (budgets, hydrographs, etc.).

Parameters:

baseline_result (SimulationResult) – Result from the baseline simulation.
scenario_results (dict[str, ScenarioResult]) – Results from scenario runs.

Returns:

Comparison results keyed by scenario name.

Return type:

dict[str, dict[str, Any]]

__repr__()[source]#

Return string representation.

PEST++ Interface #

Low-level PEST++ control file interface.

PEST++ integration for IWFM calibration and uncertainty analysis.

This module provides utilities for:

Creating PEST++ template (.tpl) and instruction (.ins) files
Writing PEST++ control files (.pst) in v1 or v2 (keyword/external) format
Running IWFM as a PEST++ model
Parsing PEST++ output

PEST++ suite includes:

pestpp-glm: Gauss-Levenberg-Marquardt parameter estimation
pestpp-ies: Iterative ensemble smoother (uncertainty analysis)
pestpp-opt: Optimization under uncertainty
pestpp-sen: Global sensitivity analysis
pestpp-sqp: Sequential quadratic programming

Control file formats:

v1 (traditional): Positional control data, inline parameter/observation data
v2 (keyword/external): * control data keyword section with key-value pairs, external CSV files for parameter data, observation data, and model I/O. Introduced in PEST++ 4.3.0. Counts (NPAR, NOBS, etc.) are inferred from the external files; SVD settings and ++ options are folded into the keyword section.

class pyiwfm.runner.pest.Parameter(name, initial_value, lower_bound, upper_bound, group='default', transform='none', change_limit='factor', scale=1.0, offset=0.0, dercom=1, tied_to='')[source]#

Bases: object

PEST++ parameter definition.

Variables:

name (str) – Parameter name (up to 200 chars in PEST++).
initial_value (float) – Initial parameter value.
lower_bound (float) – Lower bound for parameter.
upper_bound (float) – Upper bound for parameter.
group (str) – Parameter group name.
transform (str) – Transformation type: ‘none’, ‘log’, ‘fixed’, ‘tied’.
change_limit (str) – Parameter change limit type: ‘factor’, ‘relative’, ‘absolute’.
scale (float) – Scale factor for parameter.
offset (float) – Offset for parameter.
dercom (int) – Derivative command index (1-based).
tied_to (str) – Name of parent parameter if transform is ‘tied’.

name: str#

initial_value: float#

lower_bound: float#

upper_bound: float#

group: str = 'default'#

transform: str = 'none'#

change_limit: str = 'factor'#

scale: float = 1.0#

offset: float = 0.0#

dercom: int = 1#

tied_to: str = ''#

__post_init__()[source]#

Validate parameter.

to_pest_line()[source]#

Format as v1 PEST control file parameter line.

to_csv_dict()[source]#

Return dict for CSV export (v2 external format).

__init__(name, initial_value, lower_bound, upper_bound, group='default', transform='none', change_limit='factor', scale=1.0, offset=0.0, dercom=1, tied_to='')#

class pyiwfm.runner.pest.Observation(name, value, weight=1.0, group='default', extra_columns=None)[source]#

Bases: object

PEST++ observation definition.

Variables:

name (str) – Observation name (up to 200 chars in PEST++).
value (float) – Observed value.
weight (float) – Observation weight (inverse of standard deviation).
group (str) – Observation group name.

name: str#

value: float#

weight: float = 1.0#

group: str = 'default'#

extra_columns: dict[str, str] | None = None#

__post_init__()[source]#

Validate observation.

to_pest_line()[source]#

Format as PEST control file observation line.

to_csv_dict(extra_fieldnames=None)[source]#

Return dict for CSV export (v2 external format).

Parameters:: extra_fieldnames (list[str] | None) – If provided, include these extra columns from extra_columns. Columns not present in extra_columns are written as empty strings.

__init__(name, value, weight=1.0, group='default', extra_columns=None)#

class pyiwfm.runner.pest.ObservationGroup(name, observations=<factory>, covariance_matrix=None)[source]#

Bases: object

Group of observations with shared properties.

Variables:

name (str) – Group name.
observations (list[Observation]) – Observations in this group.
covariance_matrix (str | None) – Path to covariance matrix file for this group.

name: str#

observations: list[Observation]#

covariance_matrix: str | None = None#

add_observation(name, value, weight=1.0)[source]#

Add an observation to this group.

Parameters:

name (str) – Observation name.
value (float) – Observed value.
weight (float) – Observation weight.

Returns:

The created observation.

Return type:

Observation

__init__(name, observations=<factory>, covariance_matrix=None)#

class pyiwfm.runner.pest.TemplateFile(template_path, input_path, delimiter='#', parameters=<factory>)[source]#

Bases: object

PEST++ template file (.tpl) definition.

A template file is an input file with parameters marked by delimiters. PEST++ replaces these markers with parameter values.

Variables:

template_path (Path) – Path to the template file.
input_path (Path) – Path to the model input file to generate.
delimiter (str) – Delimiter character for parameter markers (default: ‘#’).
parameters (list[str]) – List of parameter names in this template.

template_path: Path#

input_path: Path#

delimiter: str = '#'#

parameters: list[str]#

__post_init__()[source]#

Convert paths.

classmethod create_from_file(input_file, template_file, parameters, delimiter='#')[source]#

Create a template file from an existing input file.

Parameters:

input_file (Path | str) – Path to the original input file.
template_file (Path | str) – Path where template will be written.
parameters (dict[str, float]) – Dictionary mapping parameter names to their current values in the input file. These values will be replaced with markers.
delimiter (str) – Delimiter character for parameter markers.

Returns:

The created template file object.

Return type:

TemplateFile

to_pest_line()[source]#

Format as PEST control file template line.

__init__(template_path, input_path, delimiter='#', parameters=<factory>)#

class pyiwfm.runner.pest.InstructionFile(instruction_path, output_path, marker='@', observations=<factory>)[source]#

Bases: object

PEST++ instruction file (.ins) definition.

An instruction file tells PEST++ how to read model output to extract simulated observation values.

Variables:

instruction_path (Path) – Path to the instruction file.
output_path (Path) – Path to the model output file to read.
marker (str) – Marker character for instructions (default: ‘@’).
observations (list[str]) – List of observation names extracted by this file.

instruction_path: Path#

output_path: Path#

marker: str = '@'#

observations: list[str]#

__post_init__()[source]#

Convert paths.

classmethod create_for_timeseries(output_file, instruction_file, observations, header_lines=0, marker='@')[source]#

Create instruction file for reading time series output.

Parameters:

output_file (Path | str) – Path to the model output file.
instruction_file (Path | str) – Path where instruction file will be written.
observations (list[tuple[str, int, int]]) – List of (obs_name, line_number, column_number) tuples. Line numbers are 1-based (after header).
header_lines (int) – Number of header lines to skip.
marker (str) – Marker character for instructions.

Returns:

The created instruction file object.

Return type:

InstructionFile

classmethod create_for_hydrograph(output_file, instruction_file, location_name, observation_times, header_lines=1, time_column=1, value_column=2, marker='@')[source]#

Create instruction file for reading hydrograph output.

This creates instructions to read specific time values from a hydrograph file by searching for timestamps.

Parameters:

output_file (Path | str) – Path to the hydrograph output file.
instruction_file (Path | str) – Path where instruction file will be written.
location_name (str) – Name prefix for observations.
observation_times (list[tuple[datetime, str]]) – List of (datetime, obs_suffix) tuples specifying which times to extract and their observation name suffix.
header_lines (int) – Number of header lines to skip.
time_column (int) – Column containing timestamp (1-based).
value_column (int) – Column containing value to read (1-based).
marker (str) – Marker character for instructions.

Returns:

The created instruction file object.

Return type:

InstructionFile

to_pest_line()[source]#

Format as PEST control file instruction line.

__init__(instruction_path, output_path, marker='@', observations=<factory>)#

class pyiwfm.runner.pest.PESTInterface(model_dir, case_name, pest_dir=None, parameters=<factory>, parameter_groups=<factory>, observations=<factory>, observation_groups=<factory>, template_files=<factory>, instruction_files=<factory>, model_command='python run_model.py', pestpp_options=<factory>, obs_csv_files=<factory>, control_data=<factory>, svd_settings=<factory>)[source]#

Bases: object

Interface for setting up and running PEST++ with IWFM.

This class manages the creation of PEST++ input files and coordinates running IWFM as a PEST++ model.

Parameters:

model_dir (Path) – Directory containing the IWFM model.
pest_dir (Path | None) – Directory for PEST++ files. Defaults to model_dir/pest.
case_name (str) – Base name for PEST++ files (e.g., “iwfm” -> iwfm.pst).

Examples

>>> pest = PESTInterface("C2VSim/Simulation", case_name="c2vsim_cal")
>>> pest.add_parameter("hk_zone1", 1.0, 0.01, 100.0, group="hk")
>>> pest.add_observation_group("heads", obs_data)
>>> pest.write_control_file()
>>> pest.run_pestpp_glm()

model_dir: Path#

case_name: str#

pest_dir: Path | None = None#

parameters: list[Parameter]#

parameter_groups: dict[str, dict[str, Any]]#

observations: list[Observation]#

observation_groups: dict[str, ObservationGroup]#

template_files: list[TemplateFile]#

instruction_files: list[InstructionFile]#

model_command: str = 'python run_model.py'#

pestpp_options: dict[str, Any]#

obs_csv_files: list[str]#

__post_init__()[source]#

Initialize paths.

add_parameter(name, initial_value, lower_bound, upper_bound, group='default', transform='none')[source]#

Add a parameter to the calibration.

Parameters:

name (str) – Parameter name.
initial_value (float) – Initial value.
lower_bound (float) – Lower bound.
upper_bound (float) – Upper bound.
group (str) – Parameter group name.
transform (str) – Transformation: ‘none’, ‘log’, ‘fixed’, ‘tied’.

Returns:

The created parameter.

Return type:

Parameter

add_parameter_group(name, inctyp='relative', derinc=0.01, **kwargs)[source]#

Add or configure a parameter group.

Parameters:

name (str) – Group name.
inctyp (str) – Increment type for derivatives: ‘relative’, ‘absolute’.
derinc (float) – Derivative increment.
**kwargs (Any) – Additional group options.

add_observation(name, value, weight=1.0, group='default')[source]#

Add an observation.

Parameters:

name (str) – Observation name.
value (float) – Observed value.
weight (float) – Observation weight.
group (str) – Observation group name.

Returns:

The created observation.

Return type:

Observation

add_observation_group(name, observations=None)[source]#

Add an observation group.

Parameters:

name (str) – Group name.
observations (list[tuple[str, float, float]] | None) – Optional list of (name, value, weight) tuples.

Returns:

The created observation group.

Return type:

ObservationGroup

add_template_file(template)[source]#

Add a template file.

add_instruction_file(instruction)[source]#

Add an instruction file.

set_model_command(command)[source]#

Set the model run command.

Parameters:: command (str) – Command to run the model (e.g., “python run_model.py”).

set_pestpp_option(option, value)[source]#

Set a PEST++ option.

Parameters:

option (str) – Option name (e.g., “svd_pack”, “ies_num_reals”).
value (Any) – Option value.

control_data: dict[str, Any]#

svd_settings: dict[str, Any]#

set_control_data(**kwargs)[source]#

Set control data values.

Common keywords: pestmode, noptmax, rlambda1, rlamfac, phiratsuf, phiredlam, numlam, relparmax, facparmax, facorig, phiredswh, phiredstp, nphistp, nphinored, relparstp, nrelpar, icov, icor, ieig.

set_svd(maxsing=1500, eigthresh=1e-12)[source]#

Configure singular value decomposition.

Parameters:

maxsing (int) – Maximum number of singular values.
eigthresh (float) – Eigenvalue ratio threshold.

write_control_file(filepath=None, *, version=1, external_dir=None)[source]#

Write the PEST++ control file (.pst).

Parameters:

filepath (Path | str | None) – Output path. Defaults to pest_dir / case_name.pst.
version ({1, 2}) –
Control file format version.
- 1 – Traditional PEST format with positional control data and inline parameter / observation data.
- 2 – PEST++ v2 keyword/external format. Control data is written as keyword-value pairs in a * control data keyword section. Parameter data, observation data, and model I/O are written to external CSV files referenced from the PST. SVD settings and ++ options are folded into the keyword section. Introduced in PEST++ 4.3.0.
external_dir (str | None) – Subdirectory (relative to the PST file) for external CSV files. Only used when version is 2. Defaults to "pest".

Returns:

Path to the written control file.

Return type:

Path

classmethod from_pst(filepath, model_dir=None)[source]#

Read a v1 PEST control file and populate a PESTInterface.

This enables round-tripping: read a v1 PST, then write it back as v2 (or v1 with modifications).

Parameters:

filepath (Path | str) – Path to the .pst file.
model_dir (Path | str | None) – Model directory. Defaults to the directory containing the PST.

Returns:

Populated interface instance.

Return type:

PESTInterface

write_model_runner(filepath=None)[source]#

Write a Python script to run IWFM for PEST++.

This creates a run_model.py script that PEST++ will call. The script reads the template-generated input files and runs the IWFM simulation.

Parameters:: filepath (Path | str | None) – Output path. Defaults to pest_dir/run_model.py.
Returns:: Path to the written script.
Return type:: Path

__repr__()[source]#

Return string representation.

__init__(model_dir, case_name, pest_dir=None, parameters=<factory>, parameter_groups=<factory>, observations=<factory>, observation_groups=<factory>, template_files=<factory>, instruction_files=<factory>, model_command='python run_model.py', pestpp_options=<factory>, obs_csv_files=<factory>, control_data=<factory>, svd_settings=<factory>)#

pyiwfm.runner.pest.write_pest_control_file(filepath, parameters, observations, template_files, instruction_files, model_command='python run_model.py', version=1, **pestpp_options)[source]#

Convenience function to write a PEST++ control file.

Parameters:

filepath (Path | str) – Output path for the control file.
parameters (list[Parameter]) – List of parameters.
observations (list[Observation]) – List of observations.
template_files (list[TemplateFile]) – List of template files.
instruction_files (list[InstructionFile]) – List of instruction files.
model_command (str) – Command to run the model.
version ({1, 2}) – Control file format version (1 = traditional, 2 = keyword/external).
**pestpp_options (Any) – PEST++ options.

Returns:

Path to the written control file.

Return type:

Path

Parameter Management #

IWFM-specific parameter types, transforms, and parameterization strategies.

Parameter Types and Strategies #

IWFM parameter types and parameterization strategies for PEST++.

This module provides the core parameter management classes for setting up highly parameterized PEST++ calibration of IWFM models.

class pyiwfm.runner.pest_params.IWFMParameterType(*values)[source]#

Bases: Enum

Types of parameters in IWFM models.

Each parameter type corresponds to a specific physical property or input that can be adjusted during calibration.

HORIZONTAL_K = 'hk'#

VERTICAL_K = 'vk'#

SPECIFIC_STORAGE = 'ss'#

SPECIFIC_YIELD = 'sy'#

POROSITY = 'por'#

STREAMBED_K = 'strk'#

STREAMBED_THICKNESS = 'strt'#

STREAM_WIDTH = 'strw'#

MANNING_N = 'mann'#

LAKEBED_K = 'lakk'#

LAKEBED_THICKNESS = 'lakt'#

CROP_COEFFICIENT = 'kc'#

IRRIGATION_EFFICIENCY = 'ie'#

ROOT_DEPTH = 'rd'#

FIELD_CAPACITY = 'fc'#

WILTING_POINT = 'wp'#

SOIL_AWC = 'awc'#

PUMPING_MULT = 'pump'#

RECHARGE_MULT = 'rech'#

DIVERSION_MULT = 'div'#

BYPASS_MULT = 'byp'#

PRECIP_MULT = 'ppt'#

ET_MULT = 'et'#

STREAM_INFLOW_MULT = 'infl'#

RETURN_FLOW_MULT = 'rtf'#

GHB_CONDUCTANCE = 'ghbc'#

GHB_HEAD = 'ghbh'#

SPECIFIED_HEAD = 'chd'#

SPECIFIED_FLOW = 'wel'#

ELASTIC_STORAGE = 'ske'#

INELASTIC_STORAGE = 'skv'#

PRECONSOLIDATION = 'pcs'#

property default_bounds: tuple[float, float]#: Get default parameter bounds based on type.

property default_transform: str#: Get default transform based on parameter type.

property is_multiplier: bool#: Check if this parameter type is a multiplier.

class pyiwfm.runner.pest_params.ParameterTransform(*values)[source]#

Bases: Enum

Parameter transformation types for PEST++.

NONE = 'none'#

LOG = 'log'#

FIXED = 'fixed'#

TIED = 'tied'#

class pyiwfm.runner.pest_params.ParameterGroup(name, inctyp='relative', derinc=0.01, derinclb=0.0, forcen='switch', derincmul=2.0, dermthd='parabolic', splitthresh=1e-05, splitreldiff=0.5)[source]#

Bases: object

Parameter group definition for PEST++.

Variables:

name (str) – Group name (max 12 characters for PEST compatibility).
inctyp (str) – Increment type for derivatives: ‘relative’ or ‘absolute’.
derinc (float) – Derivative increment value.
derinclb (float) – Lower bound for derivative increment.
forcen (str) – Force numerical derivatives: ‘switch’, ‘always_2’, ‘always_3’.
derincmul (float) – Multiplier for derivative increment.
dermthd (str) – Derivative method: ‘parabolic’, ‘outside_pts’, ‘best_fit’.
splitthresh (float) – Split threshold for parameter splitting.
splitreldiff (float) – Relative difference threshold for splitting.

name: str#

inctyp: str = 'relative'#

derinc: float = 0.01#

derinclb: float = 0.0#

forcen: str = 'switch'#

derincmul: float = 2.0#

dermthd: str = 'parabolic'#

splitthresh: float = 1e-05#

splitreldiff: float = 0.5#

__post_init__()[source]#

Validate group name length.

to_pest_line()[source]#

Format as PEST control file parameter group line.

__init__(name, inctyp='relative', derinc=0.01, derinclb=0.0, forcen='switch', derincmul=2.0, dermthd='parabolic', splitthresh=1e-05, splitreldiff=0.5)#

class pyiwfm.runner.pest_params.Parameter(name, initial_value, lower_bound, upper_bound, group='default', transform=ParameterTransform.NONE, scale=1.0, offset=0.0, param_type=None, layer=None, zone=None, location=None, tied_to=None, tied_ratio=1.0, metadata=<factory>)[source]#

Bases: object

Individual parameter definition for PEST++.

Variables:

name (str) – Parameter name (max 200 characters for PEST++).
initial_value (float) – Initial parameter value.
lower_bound (float) – Lower bound for parameter.
upper_bound (float) – Upper bound for parameter.
group (str) – Parameter group name.
transform (ParameterTransform) – Transformation type.
scale (float) – Scale factor for parameter.
offset (float) – Offset for parameter.
param_type (IWFMParameterType | None) – IWFM parameter type (for metadata).
layer (int | None) – Model layer (if applicable).
zone (int | None) – Zone ID (if applicable).
location (tuple[float, float] | None) – (x, y) location for pilot points.
tied_to (str | None) – Name of parent parameter if tied.
tied_ratio (float) – Ratio to parent parameter if tied.
metadata (dict[str, Any]) – Additional metadata.

name: str#

initial_value: float#

lower_bound: float#

upper_bound: float#

group: str = 'default'#

transform: ParameterTransform = 'none'#

scale: float = 1.0#

offset: float = 0.0#

param_type: IWFMParameterType | None = None#

layer: int | None = None#

zone: int | None = None#

location: tuple[float, float] | None = None#

tied_to: str | None = None#

tied_ratio: float = 1.0#

metadata: dict[str, Any]#

__post_init__()[source]#

Validate parameter.

property partrans: str#: Get PEST parameter transformation string.

property parval1: float#: Get initial parameter value for PEST.

property parlbnd: float#: Get lower bound for PEST.

property parubnd: float#: Get upper bound for PEST.

to_pest_line()[source]#

Format as PEST control file parameter data line.

__repr__()[source]#

Return string representation.

__init__(name, initial_value, lower_bound, upper_bound, group='default', transform=ParameterTransform.NONE, scale=1.0, offset=0.0, param_type=None, layer=None, zone=None, location=None, tied_to=None, tied_ratio=1.0, metadata=<factory>)#

class pyiwfm.runner.pest_params.ParameterizationStrategy(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None)[source]#

Bases: object

Base class for parameterization strategies.

A parameterization strategy defines how a specific type of parameter is distributed across the model domain.

param_type: IWFMParameterType#

transform: ParameterTransform = 'none'#

bounds: tuple[float, float] | None = None#

group_name: str | None = None#

__post_init__()[source]#

Set defaults from parameter type.

generate_parameters(model)[source]#

Generate parameters for this strategy.

Must be implemented by subclasses.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: List of generated parameters.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None)#

class pyiwfm.runner.pest_params.ZoneParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, zones='subregions', layer=None, initial_values=1.0, zone_names=None)[source]#

Bases: ParameterizationStrategy

Zone-based parameterization strategy.

Creates one parameter per zone for a given parameter type. Zones can be model subregions, custom zone definitions, or any spatial grouping of elements.

Variables:

zones (list[int] | str) – List of zone IDs or “subregions” to use model subregions.
layer (int | None) – Model layer (for layered parameters like K).
initial_values (float | dict[int, float]) – Initial value(s). If float, used for all zones. If dict, maps zone ID to value.
zone_names (dict[int, str] | None) – Optional zone names for parameter naming.

zones: list[int] | str = 'subregions'#

layer: int | None = None#

initial_values: float | dict[int, float] = 1.0#

zone_names: dict[int, str] | None = None#

generate_parameters(model)[source]#

Generate zone-based parameters.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: One parameter per zone.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, zones='subregions', layer=None, initial_values=1.0, zone_names=None)#

class pyiwfm.runner.pest_params.MultiplierParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, spatial_extent='global', temporal_extent='constant', zones=None, initial_value=1.0, target_file=None)[source]#

Bases: ParameterizationStrategy

Multiplier parameterization strategy.

Creates parameters that act as multipliers on existing model values. Useful for adjusting time series inputs like pumping, recharge, etc.

Variables:

spatial_extent (str) – Spatial scope: ‘global’, ‘zone’, or ‘element’.
temporal_extent (str) – Temporal scope: ‘constant’, ‘seasonal’, ‘monthly’, ‘annual’.
zones (list[int] | None) – Zone IDs for zone-based multipliers.
initial_value (float) – Initial multiplier value (typically 1.0).
target_file (Path | None) – File containing base values to multiply.

spatial_extent: str = 'global'#

temporal_extent: str = 'constant'#

zones: list[int] | None = None#

initial_value: float = 1.0#

target_file: Path | None = None#

__post_init__()[source]#

Validate and set defaults.

generate_parameters(model)[source]#

Generate multiplier parameters.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: Multiplier parameters.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, spatial_extent='global', temporal_extent='constant', zones=None, initial_value=1.0, target_file=None)#

class pyiwfm.runner.pest_params.PilotPointParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, points=None, spacing=None, layer=1, initial_value=1.0, variogram=None, kriging_type='ordinary', search_radius=None, min_points=1, max_points=20)[source]#

Bases: ParameterizationStrategy

Pilot point parameterization strategy.

Creates spatially distributed parameters at pilot point locations. Values at model nodes/elements are interpolated using kriging.

Variables:

points (list[tuple[float, float]] | None) – Explicit pilot point (x, y) coordinates.
spacing (float | None) – Regular grid spacing (if points not specified).
layer (int) – Model layer for these parameters.
initial_value (float | NDArray[np.float64]) – Initial value(s) at pilot points.
variogram (dict | None) – Variogram specification for kriging.
kriging_type (str) – Type of kriging: ‘ordinary’, ‘simple’, ‘universal’.
search_radius (float | None) – Search radius for kriging interpolation.
min_points (int) – Minimum pilot points in search neighborhood.
max_points (int) – Maximum pilot points in search neighborhood.

points: list[tuple[float, float]] | None = None#

spacing: float | None = None#

layer: int = 1#

initial_value: float | ndarray[tuple[Any, ...], dtype[float64]] = 1.0#

variogram: dict[str, Any] | None = None#

kriging_type: str = 'ordinary'#

search_radius: float | None = None#

min_points: int = 1#

max_points: int = 20#

__post_init__()[source]#

Validate configuration.

generate_pilot_point_grid(model, buffer=0.0)[source]#

Generate regular grid of pilot points within model domain.

Parameters:

model (IWFMModel) – Model with grid information.
buffer (float) – Buffer distance inside domain boundary.

Returns:

List of (x, y) pilot point coordinates.

Return type:

list[tuple[float, float]]

generate_parameters(model)[source]#

Generate pilot point parameters.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: One parameter per pilot point.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, points=None, spacing=None, layer=1, initial_value=1.0, variogram=None, kriging_type='ordinary', search_radius=None, min_points=1, max_points=20)#

class pyiwfm.runner.pest_params.DirectParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, name='', initial_value=1.0, layer=None, location_id=None)[source]#

Bases: ParameterizationStrategy

Direct parameterization strategy.

Creates a single parameter for direct adjustment of a model value. Useful for scalar parameters or specific locations.

Variables:

name (str) – Parameter name.
initial_value (float) – Initial parameter value.
layer (int | None) – Model layer (if applicable).
location_id (int | None) – Location ID (element, node, reach, etc.).

name: str = ''#

initial_value: float = 1.0#

layer: int | None = None#

location_id: int | None = None#

__post_init__()[source]#

Set defaults.

generate_parameters(model)[source]#

Generate a single direct parameter.

Parameters:: model (IWFMModel) – The IWFM model (not used for direct params).
Returns:: Single parameter.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, name='', initial_value=1.0, layer=None, location_id=None)#

class pyiwfm.runner.pest_params.StreamParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, reaches='all', by_node=False, initial_values=1.0)[source]#

Bases: ParameterizationStrategy

Stream-specific parameterization strategy.

Creates parameters for stream properties by reach or node.

Variables:

reaches (list[int] | str) – Reach IDs or “all” for all reaches.
by_node (bool) – If True, create parameters by stream node instead of reach.
initial_values (float | dict[int, float]) – Initial values by reach/node ID.

reaches: list[int] | str = 'all'#

by_node: bool = False#

initial_values: float | dict[int, float] = 1.0#

generate_parameters(model)[source]#

Generate stream parameters.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: Stream parameters by reach or node.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, reaches='all', by_node=False, initial_values=1.0)#

class pyiwfm.runner.pest_params.RootZoneParameterization(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, land_use_types='all', crop_ids=None, initial_values=1.0)[source]#

Bases: ParameterizationStrategy

Root zone parameterization strategy.

Creates parameters for root zone properties by land use type.

Variables:

land_use_types (list[str] | str) – Land use type names or “all”.
crop_ids (list[int] | None) – Crop/land use IDs if not using names.
initial_values (float | dict[str, float]) – Initial values by land use type.

land_use_types: list[str] | str = 'all'#

crop_ids: list[int] | None = None#

initial_values: float | dict[str, float] = 1.0#

generate_parameters(model)[source]#

Generate root zone parameters.

Parameters:: model (IWFMModel) – The IWFM model to parameterize.
Returns:: Root zone parameters by land use type.
Return type:: list[Parameter]

__init__(param_type, transform=ParameterTransform.NONE, bounds=None, group_name=None, land_use_types='all', crop_ids=None, initial_values=1.0)#

Parameter Manager #

IWFM Parameter Manager for PEST++ setup.

This module provides the IWFMParameterManager class that coordinates parameter generation and management for PEST++ calibration.

class pyiwfm.runner.pest_manager.IWFMParameterManager(model=None)[source]#

Bases: object

Manages all parameters for an IWFM PEST++ setup.

This class provides methods to add various types of parameters (zone-based, pilot points, multipliers, etc.) and generates the appropriate PEST++ parameter definitions.

Parameters:: model (IWFMModel | None) – The IWFM model to parameterize. If None, some features will be limited.

Examples

>>> from pyiwfm import IWFMModel
>>> from pyiwfm.runner.pest_manager import IWFMParameterManager

>>> model = IWFMModel.from_preprocessor("model/PreProcessor.in")
>>> pm = IWFMParameterManager(model)

>>> # Add zone-based hydraulic conductivity
>>> pm.add_zone_parameters(
...     IWFMParameterType.HORIZONTAL_K,
...     zones="subregions",
...     layer=1,
...     bounds=(0.1, 1000.0),
... )

>>> # Add global pumping multiplier
>>> pm.add_multiplier_parameters(
...     IWFMParameterType.PUMPING_MULT,
...     spatial="global",
...     bounds=(0.8, 1.2),
... )

>>> # Get all parameters
>>> params = pm.get_all_parameters()
>>> print(f"Total parameters: {len(params)}")

__init__(model=None)[source]#

Initialize the parameter manager.

Parameters:: model (IWFMModel | None) – The IWFM model to parameterize.

add_zone_parameters(param_type, zones='subregions', layer=None, initial_values=1.0, bounds=None, transform='auto', group=None, zone_names=None)[source]#

Add zone-based parameters.

Creates one parameter per zone for the specified property. Zones can be subregions or custom zone definitions.

Parameters:

param_type (IWFMParameterType | str) – Parameter type to add.
zones (list[int] | str) – Zone IDs or “subregions” to use model subregions.
layer (int | None) – Model layer for layered parameters.
initial_values (float | dict[int, float]) – Initial value(s). Float for uniform, dict for zone-specific.
bounds (tuple[float, float] | None) – Parameter bounds. None uses type defaults.
transform (str | ParameterTransform) – Transformation: ‘none’, ‘log’, ‘auto’ (uses type default).
group (str | None) – Parameter group name. None uses type abbreviation.
zone_names (dict[int, str] | None) – Optional zone names for parameter naming.

Returns:

List of created parameters.

Return type:

list[Parameter]

Examples

>>> pm.add_zone_parameters(
...     IWFMParameterType.HORIZONTAL_K,
...     zones="subregions",
...     layer=1,
...     bounds=(0.1, 1000.0),
... )

add_multiplier_parameters(param_type, spatial='global', temporal='constant', zones=None, initial_value=1.0, bounds=None, transform='none', group=None, target_file=None)[source]#

Add multiplier parameters.

Multipliers adjust existing model values rather than replacing them directly.

Parameters:

param_type (IWFMParameterType | str) – Parameter type (typically a _MULT type).
spatial (str) – Spatial scope: ‘global’, ‘zone’, or ‘element’.
temporal (str) – Temporal scope: ‘constant’, ‘seasonal’, ‘monthly’.
zones (list[int] | None) – Zone IDs for zone-based multipliers.
initial_value (float) – Initial multiplier value (typically 1.0).
bounds (tuple[float, float] | None) – Parameter bounds. None uses type defaults.
transform (str | ParameterTransform) – Transformation (typically ‘none’ for multipliers).
group (str | None) – Parameter group name.
target_file (Path | str | None) – File containing base values to multiply.

Returns:

List of created multiplier parameters.

Return type:

list[Parameter]

Examples

>>> # Global pumping multiplier
>>> pm.add_multiplier_parameters(
...     IWFMParameterType.PUMPING_MULT,
...     spatial="global",
...     bounds=(0.8, 1.2),
... )

>>> # Seasonal ET multipliers
>>> pm.add_multiplier_parameters(
...     IWFMParameterType.ET_MULT,
...     temporal="seasonal",
...     bounds=(0.9, 1.1),
... )

add_pilot_points(param_type, spacing=None, points=None, layer=1, initial_value=1.0, bounds=None, transform='auto', group=None, variogram=None, kriging_type='ordinary', prefix=None)[source]#

Add pilot point parameters.

Pilot points enable highly parameterized spatial heterogeneity. Values at model nodes/elements are interpolated using kriging.

Parameters:

param_type (IWFMParameterType | str) – Parameter type (e.g., HORIZONTAL_K).
spacing (float | None) – Regular grid spacing. If None, must provide points.
points (list[tuple[float, float]] | None) – Explicit pilot point (x, y) coordinates.
layer (int) – Model layer for these parameters.
initial_value (float | NDArray[np.float64]) – Initial value(s) at pilot points.
bounds (tuple[float, float] | None) – Parameter bounds.
transform (str | ParameterTransform) – Transformation (‘auto’ uses type default).
group (str | None) – Parameter group name.
variogram (dict | None) – Variogram specification: {‘type’: ‘exponential’, ‘a’: 10000, …}
kriging_type (str) – Kriging type: ‘ordinary’, ‘simple’, ‘universal’.
prefix (str | None) – Custom prefix for parameter names.

Returns:

List of pilot point parameters.

Return type:

list[Parameter]

Examples

>>> # Regular grid of pilot points
>>> pm.add_pilot_points(
...     IWFMParameterType.HORIZONTAL_K,
...     spacing=5000.0,
...     layer=1,
...     variogram={'type': 'exponential', 'a': 10000, 'sill': 1.0},
... )

add_stream_parameters(param_type, reaches='all', initial_values=1.0, bounds=None, transform='auto', group=None)[source]#

Add stream-related parameters by reach.

Parameters:

param_type (IWFMParameterType | str) – Stream parameter type (STREAMBED_K, etc.).
reaches (list[int] | str) – Reach IDs or “all” for all reaches.
initial_values (float | dict[int, float]) – Initial values by reach ID.
bounds (tuple[float, float] | None) – Parameter bounds.
transform (str | ParameterTransform) – Transformation.
group (str | None) – Parameter group name.

Returns:

List of stream parameters.

Return type:

list[Parameter]

add_rootzone_parameters(param_type, land_use_types='all', initial_values=1.0, bounds=None, transform='none', group=None)[source]#

Add root zone parameters by land use type.

Parameters:

param_type (IWFMParameterType | str) – Root zone parameter type (CROP_COEFFICIENT, etc.).
land_use_types (list[str] | str) – Land use type names or “all”.
initial_values (float | dict[str, float]) – Initial values by land use type name.
bounds (tuple[float, float] | None) – Parameter bounds.
transform (str | ParameterTransform) – Transformation.
group (str | None) – Parameter group name.

Returns:

List of root zone parameters.

Return type:

list[Parameter]

add_parameter(name, param_type, initial_value, bounds=None, transform='auto', group=None, layer=None, **metadata)[source]#

Add a single direct parameter.

Parameters:

name (str) – Parameter name.
param_type (IWFMParameterType | str) – Parameter type.
initial_value (float) – Initial parameter value.
bounds (tuple[float, float] | None) – Parameter bounds.
transform (str | ParameterTransform) – Transformation.
group (str | None) – Parameter group name.
layer (int | None) – Model layer (if applicable).
**metadata (Any) – Additional metadata.

Returns:

The created parameter.

Return type:

Parameter

tie_parameters(parent, children, ratios=1.0)[source]#

Set up tied parameters (children follow parent).

Tied parameters are adjusted as a ratio of their parent parameter, reducing the effective number of adjustable parameters while maintaining relationships.

Parameters:

parent (str) – Parent parameter name.
children (list[str]) – Child parameter names.
ratios (float | list[float]) – Ratio(s) to parent. If float, used for all children.

fix_parameter(name)[source]#

Fix a parameter (no adjustment during calibration).

Parameters:: name (str) – Parameter name to fix.

unfix_parameter(name, transform='auto')[source]#

Unfix a parameter.

Parameters:

name (str) – Parameter name to unfix.
transform (str | ParameterTransform) – Transform to apply after unfixing.

add_parameter_group(name, inctyp='relative', derinc=0.01, **kwargs)[source]#

Add or update a parameter group.

Parameters:

name (str) – Group name (max 12 characters).
inctyp (str) – Increment type: ‘relative’ or ‘absolute’.
derinc (float) – Derivative increment.
**kwargs (Any) – Additional group settings.

Returns:

The created or updated group.

Return type:

ParameterGroup

get_parameter_group(name)[source]#

Get a parameter group by name.

get_parameter(name)[source]#

Get a parameter by name.

get_parameters_by_type(param_type)[source]#

Get all parameters of a specific type.

get_parameters_by_group(group)[source]#

Get all parameters in a specific group.

get_parameters_by_layer(layer)[source]#

Get all parameters for a specific layer.

get_pilot_point_parameters()[source]#

Get all pilot point parameters.

get_all_parameters()[source]#

Get all parameters as a list.

get_adjustable_parameters()[source]#

Get all adjustable (non-fixed, non-tied) parameters.

get_all_groups()[source]#

Get all parameter groups that have parameters.

to_dataframe()[source]#

Export parameters to a pandas DataFrame.

Returns:: DataFrame with parameter information.
Return type:: pd.DataFrame
Raises:: ImportError – If pandas is not available.

from_dataframe(df)[source]#

Load parameter values from a DataFrame.

Useful for loading calibrated parameter values.

Parameters:: df (pd.DataFrame) – DataFrame with ‘name’ and ‘initial_value’ columns.

write_parameter_file(filepath)[source]#

Write parameter values to a file.

Parameters:: filepath (Path | str) – Output file path.

read_parameter_file(filepath)[source]#

Read parameter values from a file.

Updates initial values for existing parameters.

Parameters:: filepath (Path | str) – Input file path.

property n_parameters: int#: Total number of parameters.

property n_adjustable: int#: Number of adjustable parameters.

property n_groups: int#: Number of parameter groups in use.

summary()[source]#

Get a summary of parameters.

Returns:: Summary string.
Return type:: str

__iter__()[source]#

Iterate over all parameters.

__len__()[source]#

Return number of parameters.

__repr__()[source]#

Return string representation.

Observation Management #

Observation types, weights, and management for PEST++ calibration targets.

Observation Types #

PEST++ observation types and classes for IWFM models.

This module provides IWFM-specific observation types and enhanced observation classes for use with PEST++ calibration, uncertainty analysis, and optimization.

The observation types cover: - Groundwater: head, drawdown, head differences, vertical gradients - Streams: flow, stage, gain/loss - Lakes: level, storage - Budgets: GW, stream, root zone components - Land subsidence: subsidence, compaction

class pyiwfm.runner.pest_observations.IWFMObservationType(*values)[source]#

Bases: Enum

Types of observations in IWFM models.

Each observation type has default properties for weight calculation and transformation strategies commonly used in calibration.

Categories#

Groundwater observations:: HEAD, DRAWDOWN, HEAD_DIFFERENCE, VERTICAL_GRADIENT
Stream observations:: STREAM_FLOW, STREAM_STAGE, STREAM_GAIN_LOSS
Lake observations:: LAKE_LEVEL, LAKE_STORAGE
Budget observations:: GW_BUDGET, STREAM_BUDGET, ROOTZONE_BUDGET, LAKE_BUDGET
Land subsidence observations:: SUBSIDENCE, COMPACTION

HEAD = 'head'#

DRAWDOWN = 'drawdown'#

HEAD_DIFFERENCE = 'hdiff'#

VERTICAL_GRADIENT = 'vgrad'#

STREAM_FLOW = 'flow'#

STREAM_STAGE = 'stage'#

STREAM_GAIN_LOSS = 'sgl'#

LAKE_LEVEL = 'lake'#

LAKE_STORAGE = 'lsto'#

GW_BUDGET = 'gwbud'#

STREAM_BUDGET = 'strbud'#

ROOTZONE_BUDGET = 'rzbud'#

LAKE_BUDGET = 'lakbud'#

SUBSIDENCE = 'sub'#

COMPACTION = 'comp'#

property default_transform: str#

Get default transformation for this observation type.

Returns:: ‘none’, ‘log’, or ‘sqrt’ depending on observation type.
Return type:: str

property typical_error: float#

Get typical measurement error for this observation type.

Returns:: Typical standard deviation of measurement error. Units depend on observation type.
Return type:: float

property is_relative_error: bool#

Check if typical error is relative (fraction) or absolute.

Returns:: True if typical_error is a relative error (0-1 range).
Return type:: bool

property group_prefix: str#

Get default group name prefix for this observation type.

Returns:: Prefix for observation group names.
Return type:: str

class pyiwfm.runner.pest_observations.WeightStrategy(*values)[source]#

Bases: Enum

Strategies for calculating observation weights.

Weight calculation is critical for proper calibration. Different strategies are appropriate for different situations.

EQUAL = 'equal'#: All observations have weight = 1.

INVERSE_VARIANCE = 'inverse_variance'#: Weight = 1/variance. Requires measurement error estimates.

GROUP_CONTRIBUTION = 'group_contribution'#: Weights adjusted so each group contributes equally to objective function.

TEMPORAL_DECAY = 'temporal_decay'#: Recent observations weighted higher than older ones.

MAGNITUDE_BASED = 'magnitude_based'#: Weight scales with observation magnitude (for relative errors).

CUSTOM = 'custom'#: User-specified weights.

class pyiwfm.runner.pest_observations.ObservationLocation(x, y, z=None, node_id=None, element_id=None, layer=None, reach_id=None, lake_id=None)[source]#

Bases: object

Location information for an observation point.

Variables:

x (float) – X coordinate.
y (float) – Y coordinate.
z (float | None) – Z coordinate (elevation or depth).
node_id (int | None) – Associated model node ID.
element_id (int | None) – Associated model element ID.
layer (int | None) – Model layer.
reach_id (int | None) – Stream reach ID (for stream observations).
lake_id (int | None) – Lake ID (for lake observations).

x: float#

y: float#

z: float | None = None#

node_id: int | None = None#

element_id: int | None = None#

layer: int | None = None#

reach_id: int | None = None#

lake_id: int | None = None#

to_dict()[source]#

Convert to dictionary.

__init__(x, y, z=None, node_id=None, element_id=None, layer=None, reach_id=None, lake_id=None)#

class pyiwfm.runner.pest_observations.IWFMObservation(name, value, weight=1.0, group='default', obs_type=None, datetime=None, location=None, simulated_name=None, error_std=None, transform='none', metadata=<factory>)[source]#

Bases: object

Enhanced observation class with IWFM-specific attributes.

This extends the basic Observation class with additional metadata useful for IWFM calibration and post-processing.

Variables:

name (str) – Observation name (up to 200 chars for PEST++).
value (float) – Observed value.
weight (float) – Observation weight (inverse of standard deviation).
group (str) – Observation group name.
obs_type (IWFMObservationType | None) – Type of observation.
datetime (datetime | None) – Time of observation.
location (ObservationLocation | None) – Spatial location of observation.
simulated_name (str | None) – Name used in model output (if different from obs name).
error_std (float | None) – Estimated measurement error standard deviation.
transform (str) – Transformation applied: ‘none’, ‘log’, ‘sqrt’.
metadata (dict) – Additional metadata.

name: str#

value: float#

weight: float = 1.0#

group: str = 'default'#

obs_type: IWFMObservationType | None = None#

datetime: datetime | None = None#

location: ObservationLocation | None = None#

simulated_name: str | None = None#

error_std: float | None = None#

transform: str = 'none'#

metadata: dict[str, Any]#

__post_init__()[source]#

Validate observation.

property transformed_value: float#

Get transformed observation value.

Returns:: Value after applying transform.
Return type:: float

calculate_weight(strategy=WeightStrategy.INVERSE_VARIANCE, **kwargs)[source]#

Calculate observation weight using specified strategy.

Parameters:

strategy (WeightStrategy) – Weight calculation strategy.
**kwargs (Any) – Strategy-specific parameters.

Returns:

Calculated weight.

Return type:

float

to_pest_line()[source]#

Format as PEST control file observation line.

Returns:: Formatted line for PEST control file.
Return type:: str

to_dict()[source]#

Convert to dictionary.

Returns:: Dictionary representation.
Return type:: dict

__repr__()[source]#

Return string representation.

__init__(name, value, weight=1.0, group='default', obs_type=None, datetime=None, location=None, simulated_name=None, error_std=None, transform='none', metadata=<factory>)#

class pyiwfm.runner.pest_observations.IWFMObservationGroup(name, obs_type=None, observations=<factory>, target_contribution=None, covariance_matrix=None)[source]#

Bases: object

Group of observations with shared properties.

Observation groups allow setting common properties and calculating weights to achieve target contributions to the objective function.

Variables:

name (str) – Group name (up to 200 chars for PEST++).
obs_type (IWFMObservationType | None) – Type of observations in this group.
observations (list[IWFMObservation]) – Observations in this group.
target_contribution (float | None) – Target contribution to objective function (0-1).
covariance_matrix (NDArray | None) – Observation error covariance matrix.

name: str#

obs_type: IWFMObservationType | None = None#

observations: list[IWFMObservation]#

target_contribution: float | None = None#

covariance_matrix: ndarray[tuple[Any, ...], dtype[float64]] | None = None#

__post_init__()[source]#

Validate group.

property n_observations: int#: Number of observations in group.

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

Get array of observation values.

Returns:: Array of observation values.
Return type:: NDArray

property weights: ndarray[tuple[Any, ...], dtype[float64]]#

Get array of observation weights.

Returns:: Array of observation weights.
Return type:: NDArray

property contribution: float#

Calculate current contribution to objective function.

Assumes residuals of 1 (equal to weights) for estimation. Actual contribution depends on simulated values.

Returns:: Estimated contribution (sum of squared weighted values).
Return type:: float

add_observation(name, value, weight=1.0, **kwargs)[source]#

Add an observation to this group.

Parameters:

name (str) – Observation name.
value (float) – Observed value.
weight (float) – Observation weight.
**kwargs (Any) – Additional observation attributes.

Returns:

The created observation.

Return type:

IWFMObservation

set_weights(strategy=WeightStrategy.EQUAL, **kwargs)[source]#

Set weights for all observations in group.

Parameters:

strategy (WeightStrategy) – Weight calculation strategy.
**kwargs (Any) – Strategy-specific parameters.

scale_weights(factor)[source]#

Scale all weights by a factor.

Parameters:: factor (float) – Multiplicative scaling factor.

normalize_weights(target_sum=1.0)[source]#

Normalize weights to sum to target value.

Parameters:: target_sum (float) – Target sum of weights.

get_observations_by_time(start_date=None, end_date=None)[source]#

Get observations within a time range.

Parameters:

start_date (datetime | None) – Start of time range.
end_date (datetime | None) – End of time range.

Returns:

Observations within the specified range.

Return type:

list[IWFMObservation]

summary()[source]#

Get summary statistics for this group.

Returns:: Summary statistics.
Return type:: dict

__repr__()[source]#

Return string representation.

__len__()[source]#

Return number of observations.

__iter__()[source]#

Iterate over observations.

__init__(name, obs_type=None, observations=<factory>, target_contribution=None, covariance_matrix=None)#

class pyiwfm.runner.pest_observations.DerivedObservation(name, expression, source_observations, target_value=0.0, weight=1.0, group='derived')[source]#

Bases: object

Observation derived from other observations via expression.

Derived observations allow creating constraints and composite targets from model outputs. Common uses include: - Mass balance checks - Head differences - Flow ratios

Variables:

name (str) – Derived observation name.
expression (str) – Mathematical expression using observation names.
source_observations (list[str]) – Names of observations used in the expression.
target_value (float) – Target value for the derived quantity.
weight (float) – Observation weight.
group (str) – Observation group name.

name: str#

expression: str#

source_observations: list[str]#

__init__(name, expression, source_observations, target_value=0.0, weight=1.0, group='derived')#

target_value: float = 0.0#

weight: float = 1.0#

group: str = 'derived'#

evaluate(obs_values)[source]#

Evaluate the expression with given observation values.

Parameters:: obs_values (dict[str, float]) – Dictionary mapping observation names to values.
Returns:: Result of expression evaluation.
Return type:: float
Raises:: ValueError – If required observations are missing.

to_prior_equation()[source]#

Format as PEST prior information equation.

Returns:: PEST-format prior information line.
Return type:: str

__repr__()[source]#

Return string representation.

Observation Manager #

PEST++ observation manager for IWFM models.

This module provides the IWFMObservationManager class for managing observations in PEST++ calibration, uncertainty analysis, and optimization setups.

The manager supports: - Groundwater head and drawdown observations - Stream flow and stage observations - Lake level and storage observations - Water budget component observations - Derived observations from expressions - Flexible weight calculation strategies

class pyiwfm.runner.pest_obs_manager.WellInfo(well_id, x, y, screen_top=None, screen_bottom=None, layer=None, node_id=None, name=None)[source]#

Bases: object

Information about an observation well.

Variables:

well_id (str) – Unique well identifier.
x (float) – X coordinate.
y (float) – Y coordinate.
screen_top (float | None) – Top of well screen elevation.
screen_bottom (float | None) – Bottom of well screen elevation.
layer (int | None) – Model layer (if known).
node_id (int | None) – Associated model node.
name (str | None) – Well name.

well_id: str#

x: float#

y: float#

screen_top: float | None = None#

screen_bottom: float | None = None#

layer: int | None = None#

node_id: int | None = None#

name: str | None = None#

to_location()[source]#

Convert to ObservationLocation.

Returns:: Location object for this well.
Return type:: ObservationLocation

__init__(well_id, x, y, screen_top=None, screen_bottom=None, layer=None, node_id=None, name=None)#

class pyiwfm.runner.pest_obs_manager.GageInfo(gage_id, reach_id=None, node_id=None, x=None, y=None, name=None)[source]#

Bases: object

Information about a stream gage.

Variables:

gage_id (str) – Unique gage identifier.
reach_id (int | None) – Stream reach ID.
node_id (int | None) – Stream node ID.
x (float | None) – X coordinate.
y (float | None) – Y coordinate.
name (str | None) – Gage name.

gage_id: str#

reach_id: int | None = None#

node_id: int | None = None#

x: float | None = None#

y: float | None = None#

name: str | None = None#

to_location()[source]#

Convert to ObservationLocation.

Returns:: Location object for this gage, or None if no coordinates.
Return type:: ObservationLocation | None

__init__(gage_id, reach_id=None, node_id=None, x=None, y=None, name=None)#

class pyiwfm.runner.pest_obs_manager.IWFMObservationManager(model=None)[source]#

Bases: object

Manages all observations for an IWFM PEST++ setup.

This class provides methods for adding various types of observations, managing observation weights, and exporting to PEST++ format.

Parameters:: model (Any) – IWFM model instance (optional, for auto-detection of locations).

Examples

>>> om = IWFMObservationManager()
>>> # Add head observations from files
>>> om.add_head_observations(
...     wells="observation_wells.csv",
...     observed_data="head_timeseries.csv",
...     weight_strategy="inverse_variance",
... )
>>> # Add streamflow observations
>>> om.add_streamflow_observations(
...     gages="stream_gages.csv",
...     observed_data="flow_timeseries.csv",
...     transform="log",
... )
>>> # Balance weights
>>> om.balance_observation_groups({"head": 0.5, "flow": 0.5})
>>> # Export to dataframe
>>> df = om.to_dataframe()

__init__(model=None)[source]#

Initialize the observation manager.

Parameters:: model (Any) – IWFM model instance (optional).

add_head_observations(wells, observed_data, layers='auto', weight_strategy=WeightStrategy.EQUAL, start_date=None, end_date=None, frequency=None, group_by='well', group_name=None, obs_name_format='{well}_{date}', error_std=None)[source]#

Add groundwater head observations.

Parameters:

wells (pd.DataFrame | Path | str | list[WellInfo]) – Well information with columns: well_id, x, y, (optional) screen_top, screen_bottom, layer. Or list of WellInfo objects.
observed_data (pd.DataFrame | Path | str) – Observed head data with columns: well_id, datetime, head. Or path to CSV file.
layers (int | list[int] | str) – Layer(s) for observations. “auto” determines from screen depth.
weight_strategy (str | WeightStrategy) – Weight calculation strategy.
start_date (datetime | None) – Filter observations to start at this date.
end_date (datetime | None) – Filter observations to end at this date.
frequency (str | None) – Resample frequency (e.g., “MS” for monthly start).
group_by (str) – How to group observations: “well”, “layer”, “time”, “all”.
group_name (str | None) – Custom group name. Defaults based on group_by.
obs_name_format (str) – Format string for observation names.
error_std (float | None) – Measurement error standard deviation.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_drawdown_observations(wells, observed_data, reference_date=None, reference_values=None, **kwargs)[source]#

Add drawdown observations (change from reference).

Parameters:

wells (pd.DataFrame | Path | str | list[WellInfo]) – Well information.
observed_data (pd.DataFrame | Path | str) – Observed head data.
reference_date (datetime | None) – Date to use as reference (drawdown = head - head_at_reference).
reference_values (dict[str, float] | None) – Dictionary of well_id -> reference head values.
**kwargs (Any) – Additional arguments passed to add_head_observations.

Returns:

List of created drawdown observations.

Return type:

list[IWFMObservation]

add_head_difference_observations(well_pairs, observed_data, weight=1.0, group_name='hdiff', **kwargs)[source]#

Add head difference observations between well pairs.

Parameters:

well_pairs (list[tuple[str, str]]) – List of (well_id_1, well_id_2) pairs. Difference = head1 - head2.
observed_data (pd.DataFrame | Path | str) – Observed head data with columns: well_id, datetime, head.
weight (float) – Observation weight.
group_name (str) – Observation group name.
**kwargs (Any) – Additional arguments.

Returns:

List of created head difference observations.

Return type:

list[IWFMObservation]

add_streamflow_observations(gages, observed_data, weight_strategy=WeightStrategy.EQUAL, transform='none', start_date=None, end_date=None, frequency=None, group_name=None, obs_name_format='{gage}_{date}', error_std=None)[source]#

Add stream discharge observations.

Parameters:

gages (pd.DataFrame | Path | str | list[GageInfo]) – Gage information with columns: gage_id, reach_id or node_id.
observed_data (pd.DataFrame | Path | str) – Observed flow data with columns: gage_id, datetime, flow.
weight_strategy (str | WeightStrategy) – Weight calculation strategy.
transform (str) – Transform for flow: ‘none’, ‘log’, ‘sqrt’.
start_date (datetime | None) – Filter start date.
end_date (datetime | None) – Filter end date.
frequency (str | None) – Resample frequency.
group_name (str | None) – Custom group name.
obs_name_format (str) – Format for observation names.
error_std (float | None) – Measurement error.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_stream_stage_observations(gages, observed_data, **kwargs)[source]#

Add stream stage observations.

Parameters:

gages (pd.DataFrame | Path | str | list[GageInfo]) – Gage information.
observed_data (pd.DataFrame | Path | str) – Observed stage data with columns: gage_id, datetime, stage.
**kwargs (Any) – Additional arguments.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_gain_loss_observations(reaches, observed_data, weight=1.0, group_name='gain_loss', **kwargs)[source]#

Add stream gain/loss observations.

Parameters:

reaches (list[int]) – List of reach IDs.
observed_data (pd.DataFrame | Path | str) – Observed data with columns: reach_id, datetime, gain_loss.
weight (float) – Observation weight.
group_name (str) – Observation group name.
**kwargs (Any) – Additional arguments.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_lake_observations(lakes='all', observed_data=None, obs_type='level', weight=1.0, group_name=None, **kwargs)[source]#

Add lake level or storage observations.

Parameters:

lakes (list[int] | str) – Lake IDs or “all”.
observed_data (pd.DataFrame | Path | str | None) – Observed data with columns: lake_id, datetime, value.
obs_type (str) – “level” or “storage”.
weight (float) – Observation weight.
group_name (str | None) – Custom group name.
**kwargs (Any) – Additional arguments.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_budget_observations(budget_type, components=None, locations='all', aggregate='sum', observed_data=None, weight=1.0, group_name=None, **kwargs)[source]#

Add water budget component observations.

Parameters:

budget_type (str) – Budget type: “gw”, “stream”, “rootzone”, “lake”.
components (list[str] | None) – Budget components to observe.
locations (list[int] | str) – Location IDs or “all”.
aggregate (str) – Aggregation method: “sum”, “mean”, “by_location”.
observed_data (pd.DataFrame | Path | str | None) – Observed budget data.
weight (float) – Observation weight.
group_name (str | None) – Custom group name.
**kwargs (Any) – Additional arguments.

Returns:

List of created observations.

Return type:

list[IWFMObservation]

add_derived_observation(expression, obs_names, result_name, target_value=0.0, weight=1.0, group='derived')[source]#

Add derived observation from expression.

Parameters:

expression (str) – Mathematical expression using observation names.
obs_names (list[str]) – Names of observations used in the expression.
result_name (str) – Name for the derived observation.
target_value (float) – Target value for the derived quantity.
weight (float) – Observation weight.
group (str) – Observation group name.

Returns:

The created derived observation.

Return type:

DerivedObservation

Examples

>>> # Mass balance closure
>>> om.add_derived_observation(
...     expression="inflow - outflow - storage_change",
...     obs_names=["total_inflow", "total_outflow", "delta_storage"],
...     result_name="mass_balance_error",
...     target_value=0.0,
...     weight=10.0,
... )

set_group_weights(group, weight='auto', contribution=None)[source]#

Set weights for an observation group.

Parameters:

group (str) – Group name.
weight (float | str) – Weight value or “auto” to calculate from contribution.
contribution (float | None) – Target contribution to objective function (0-1).

balance_observation_groups(target_contributions=None)[source]#

Balance weights so groups contribute equally or as specified.

Parameters:: target_contributions (dict[str, float] | None) – Dictionary mapping group names to target contributions (0-1). If None, groups contribute equally.

apply_temporal_weights(decay_factor=0.95, reference_date=None)[source]#

Apply temporal decay to observation weights.

Recent observations are weighted higher than older ones.

Parameters:

decay_factor (float) – Annual decay factor (0-1). Weight = decay_factor^(years_from_reference).
reference_date (datetime | None) – Reference date. If None, uses most recent observation date.

get_observation(name)[source]#

Get observation by name.

Parameters:: name (str) – Observation name.
Returns:: The observation, or None if not found.
Return type:: IWFMObservation | None

get_observations_by_type(obs_type)[source]#

Get all observations of a specific type.

Parameters:: obs_type (IWFMObservationType) – Observation type.
Returns:: Matching observations.
Return type:: list[IWFMObservation]

get_observations_by_group(group)[source]#

Get all observations in a group.

Parameters:: group (str) – Group name.
Returns:: Observations in the group.
Return type:: list[IWFMObservation]

get_all_observations()[source]#

Get all observations.

Returns:: All observations.
Return type:: list[IWFMObservation]

get_observation_group(name)[source]#

Get observation group by name.

Parameters:: name (str) – Group name.
Returns:: The group, or None if not found.
Return type:: IWFMObservationGroup | None

get_all_groups()[source]#

Get all observation groups with observations.

Returns:: All groups that have at least one observation.
Return type:: list[IWFMObservationGroup]

to_dataframe()[source]#

Export all observations to a DataFrame.

Returns:: DataFrame with observation data.
Return type:: pd.DataFrame
Raises:: ImportError – If pandas is not available.

from_dataframe(df)[source]#

Load observations from a DataFrame.

Parameters:: df (pd.DataFrame) – DataFrame with observation data.

write_observation_file(filepath)[source]#

Write observations to a CSV file.

Parameters:: filepath (Path | str) – Output file path.

read_observation_file(filepath)[source]#

Read observations from a CSV file.

Parameters:: filepath (Path | str) – Input file path.

property n_observations: int#: Total number of observations.

property n_groups: int#: Number of observation groups with observations.

summary()[source]#

Get summary of all observations.

Returns:: Summary statistics.
Return type:: dict

__iter__()[source]#

Iterate over observations.

__len__()[source]#

Return number of observations.

__repr__()[source]#

Return string representation.

Template and Instruction Files #

Automatic generation of PEST++ template (.tpl) and instruction (.ins) files from IWFM input/output files.

Template Manager #

PEST++ template file generation for IWFM models.

This module provides IWFM-aware template file generation for PEST++ calibration and uncertainty analysis. It understands IWFM file formats and generates appropriate template files for different parameter types.

Template files (.tpl) contain parameter markers that PEST++ replaces with parameter values during model runs.

class pyiwfm.runner.pest_templates.TemplateMarker(parameter_name, line_number, column_start, column_end, original_value)[source]#

Bases: object

A parameter marker in a template file.

Variables:

parameter_name (str) – Name of the PEST parameter.
line_number (int) – Line number in the file (1-based).
column_start (int) – Starting column position.
column_end (int) – Ending column position.
original_value (str) – Original value that was replaced.

parameter_name: str#

line_number: int#

column_start: int#

column_end: int#

original_value: str#

__init__(parameter_name, line_number, column_start, column_end, original_value)#

class pyiwfm.runner.pest_templates.IWFMFileSection(name, start_line, end_line, data_columns=<factory>)[source]#

Bases: object

Represents a section of an IWFM input file.

IWFM files typically have structured sections with: - Comment lines (starting with C or *) - Data lines with fixed-format columns - Section delimiters

Variables:

name (str) – Section name or identifier.
start_line (int) – Starting line number (1-based).
end_line (int) – Ending line number (1-based).
data_columns (dict[str, int]) – Mapping of data field names to column indices.

name: str#

start_line: int#

end_line: int#

data_columns: dict[str, int]#

__init__(name, start_line, end_line, data_columns=<factory>)#

class pyiwfm.runner.pest_templates.IWFMTemplateManager(model=None, parameter_manager=None, output_dir=None, delimiter='#')[source]#

Bases: object

Generates PEST++ template files for IWFM input files.

This class understands IWFM file formats and generates appropriate template files for different parameter types. It supports:

Aquifer parameter templates (K, Ss, Sy by layer/zone)
Stream parameter templates (streambed K, thickness)
Multiplier templates (pumping, recharge, ET)
Pilot point templates (separate files for kriging)

Parameters:

model (Any) – IWFM model instance (optional, for auto-detection).
parameter_manager (IWFMParameterManager) – Parameter manager containing parameters to template.
output_dir (Path | str) – Directory for output template files.
delimiter (str) – Delimiter character for parameter markers (default: ‘#’).

Examples

>>> tm = IWFMTemplateManager(parameter_manager=pm, output_dir="pest/templates")
>>> tpl = tm.generate_aquifer_template(
...     input_file="Groundwater.dat",
...     param_type=IWFMParameterType.HORIZONTAL_K,
...     layer=1,
... )

__init__(model=None, parameter_manager=None, output_dir=None, delimiter='#')[source]#

Initialize the template manager.

Parameters:

model (Any) – IWFM model instance (optional).
parameter_manager (IWFMParameterManager | None) – Parameter manager with defined parameters.
output_dir (Path | str | None) – Output directory for templates.
delimiter (str) – PEST template delimiter character.

generate_aquifer_template(input_file, param_type, layer=None, parameters=None, output_template=None)[source]#

Generate template for aquifer parameter file.

Creates a template file for aquifer properties like hydraulic conductivity, specific storage, or specific yield.

Parameters:

input_file (Path | str) – Path to the IWFM aquifer parameter file.
param_type (IWFMParameterType | str) – Type of parameter (e.g., HORIZONTAL_K, SPECIFIC_YIELD).
layer (int | None) – Model layer for these parameters. If None, applies to all layers.
parameters (list[Parameter] | None) – Parameters to include. If None, uses parameters from manager.
output_template (Path | str | None) – Output template path. If None, auto-generates name.

Returns:

The created template file.

Return type:

TemplateFile

generate_aquifer_template_by_zone(input_file, param_type, zone_column, value_column, layer=None, header_lines=0, output_template=None)[source]#

Generate template for zone-based aquifer parameters.

For files where each row represents a zone with a parameter value.

Parameters:

input_file (Path | str) – Path to the input file.
param_type (IWFMParameterType | str) – Type of parameter.
zone_column (int) – Column containing zone ID (1-based).
value_column (int) – Column containing parameter value (1-based).
layer (int | None) – Model layer.
header_lines (int) – Number of header lines to skip.
output_template (Path | str | None) – Output template path.

Returns:

The created template file.

Return type:

TemplateFile

generate_stream_template(input_file, param_type, reach_column=1, value_column=2, header_lines=0, output_template=None)[source]#

Generate template for stream parameter file.

Creates a template for stream parameters like streambed K, thickness, or width by reach.

Parameters:

input_file (Path | str) – Path to stream parameter file.
param_type (IWFMParameterType | str) – Type of parameter (e.g., STREAMBED_K).
reach_column (int) – Column containing reach ID (1-based).
value_column (int) – Column containing parameter value (1-based).
header_lines (int) – Number of header lines to skip.
output_template (Path | str | None) – Output template path.

Returns:

The created template file.

Return type:

TemplateFile

generate_multiplier_template(param_type, output_template=None, format_width=15)[source]#

Generate template for multiplier parameter file.

Creates a simple multiplier file with parameter markers. Multipliers are applied to base values by a preprocessor.

Parameters:

param_type (IWFMParameterType | str) – Type of multiplier parameter.
output_template (Path | str | None) – Output template path.
format_width (int) – Width for parameter markers.

Returns:

The created template file.

Return type:

TemplateFile

generate_zone_multiplier_template(param_type, zones=None, output_template=None)[source]#

Generate template for zone-based multipliers.

Creates a multiplier file with one value per zone.

Parameters:

param_type (IWFMParameterType | str) – Type of multiplier parameter.
zones (list[int] | None) – Zone IDs. If None, determines from parameters.
output_template (Path | str | None) – Output template path.

Returns:

The created template file.

Return type:

TemplateFile

generate_pilot_point_template(param_type, layer=1, output_template=None)[source]#

Generate template for pilot point parameter file.

Pilot points are separate from IWFM input files. They are interpolated to model nodes using kriging by a preprocessor.

Parameters:

param_type (IWFMParameterType | str) – Type of parameter.
layer (int) – Model layer.
output_template (Path | str | None) – Output template path.

Returns:

The created template file.

Return type:

TemplateFile

generate_rootzone_template(input_file, param_type, land_use_column=1, value_column=2, header_lines=0, output_template=None)[source]#

Generate template for root zone parameter file.

Creates a template for root zone parameters like crop coefficients, irrigation efficiency, etc. by land use type.

Parameters:

input_file (Path | str) – Path to root zone parameter file.
param_type (IWFMParameterType | str) – Type of parameter.
land_use_column (int) – Column containing land use type (1-based).
value_column (int) – Column containing parameter value (1-based).
header_lines (int) – Number of header lines to skip.
output_template (Path | str | None) – Output template path.

Returns:

The created template file.

Return type:

TemplateFile

generate_all_templates(input_files=None)[source]#

Generate all required template files based on parameters.

Automatically generates templates for all parameters in the parameter manager.

Parameters:: input_files (dict[str, Path | str] | None) – Mapping of parameter type values to input file paths. E.g., {“hk”: “Groundwater.dat”, “strk”: “Stream.dat”}
Returns:: List of created template files.
Return type:: list[TemplateFile]

get_all_templates()[source]#

Get all created template files.

Returns:: All template files created by this manager.
Return type:: list[TemplateFile]

clear_templates()[source]#

Clear all created templates.

__repr__()[source]#

Return string representation.

Instruction Manager #

PEST++ instruction file generation for IWFM models.

This module provides IWFM-aware instruction file generation for PEST++ calibration and uncertainty analysis. It understands IWFM output file formats and generates appropriate instruction files for extracting simulated values.

Instruction files (.ins) tell PEST++ how to read model output files to extract simulated observation values.

class pyiwfm.runner.pest_instructions.OutputFileFormat(name, header_lines=1, time_column=1, time_format='%m/%d/%Y', value_columns=<factory>, delimiter='whitespace')[source]#

Bases: object

Describes the format of an IWFM output file.

Variables:

name (str) – Format name (e.g., “hydrograph”, “budget”).
header_lines (int) – Number of header lines to skip.
time_column (int) – Column containing timestamp (1-based).
time_format (str) – strftime format for parsing timestamps.
value_columns (dict[str, int]) – Mapping of variable names to column indices.
delimiter (str) – Column delimiter (whitespace, comma, etc.).

name: str#

header_lines: int = 1#

time_column: int = 1#

time_format: str = '%m/%d/%Y'#

value_columns: dict[str, int]#

delimiter: str = 'whitespace'#

__init__(name, header_lines=1, time_column=1, time_format='%m/%d/%Y', value_columns=<factory>, delimiter='whitespace')#

class pyiwfm.runner.pest_instructions.IWFMInstructionManager(model=None, observation_manager=None, output_dir=None, marker='@')[source]#

Bases: object

Generates PEST++ instruction files for IWFM output files.

This class understands IWFM output file formats and generates appropriate instruction files for extracting simulated values.

Supports: - Head hydrograph files - Stream flow/stage hydrographs - Budget output files (GW, stream, lake, root zone) - Subsidence output files

Parameters:

model (Any) – IWFM model instance (optional).
observation_manager (IWFMObservationManager | None) – Observation manager containing observation definitions.
output_dir (Path | str) – Directory for output instruction files.
marker (str) – Marker character for instructions (default: ‘@’).

Examples

>>> im = IWFMInstructionManager(
...     observation_manager=om,
...     output_dir="pest/instructions",
... )
>>> ins = im.generate_head_instructions(
...     output_file="GW_Heads.out",
...     wells=["W1", "W2", "W3"],
... )

__init__(model=None, observation_manager=None, output_dir=None, marker='@')[source]#

Initialize the instruction manager.

Parameters:

model (Any) – IWFM model instance (optional).
observation_manager (IWFMObservationManager | None) – Observation manager with defined observations.
output_dir (Path | str | None) – Output directory for instructions.
marker (str) – PEST instruction marker character.

generate_head_instructions(output_file, wells=None, observations=None, times=None, header_lines=3, time_column=1, value_column=2, time_format='%m/%d/%Y_%H:%M', instruction_file=None)[source]#

Generate instructions for head hydrograph output.

Creates an instruction file for reading groundwater head values from an IWFM hydrograph output file.

Parameters:

output_file (Path | str) – Path to the IWFM head output file.
wells (list[str] | None) – Well IDs to extract. If None, uses observations.
observations (list[IWFMObservation] | None) – Observations to extract. If None, uses observation manager.
times (list[datetime] | None) – Times to extract. If None, uses observation times.
header_lines (int) – Number of header lines in output file.
time_column (int) – Column containing timestamp (1-based).
value_column (int) – Column containing head values (1-based).
time_format (str) – strftime format for timestamps in output file.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_head_instructions_by_well(output_files, observations=None, header_lines=3, value_column=2, time_format='%m/%d/%Y_%H:%M')[source]#

Generate instructions for per-well head output files.

IWFM can output head hydrographs to separate files per well. This method handles that case.

Parameters:

output_files (dict[str, Path | str]) – Mapping of well IDs to output file paths.
observations (list[IWFMObservation] | None) – Observations to extract.
header_lines (int) – Number of header lines.
value_column (int) – Column containing values.
time_format (str) – Timestamp format.

Returns:

List of created instruction files.

Return type:

list[InstructionFile]

generate_flow_instructions(output_file, gages=None, observations=None, variable='flow', header_lines=3, time_column=1, value_column=2, time_format='%m/%d/%Y_%H:%M', instruction_file=None)[source]#

Generate instructions for streamflow output.

Creates an instruction file for reading stream flow or stage values from an IWFM stream hydrograph output file.

Parameters:

output_file (Path | str) – Path to the IWFM stream output file.
gages (list[str] | None) – Gage IDs to extract.
observations (list[IWFMObservation] | None) – Observations to extract.
variable (str) – Variable to extract: “flow” or “stage”.
header_lines (int) – Number of header lines.
time_column (int) – Column containing timestamp.
value_column (int) – Column containing values.
time_format (str) – Timestamp format.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_gain_loss_instructions(output_file, reaches=None, observations=None, header_lines=4, time_column=1, reach_column=2, value_column=3, time_format='%m/%d/%Y', instruction_file=None)[source]#

Generate instructions for stream gain/loss output.

Parameters:

output_file (Path | str) – Path to gain/loss output file.
reaches (list[int] | None) – Reach IDs to extract.
observations (list[IWFMObservation] | None) – Observations to extract.
header_lines (int) – Number of header lines.
time_column (int) – Column containing timestamp.
reach_column (int) – Column containing reach ID.
value_column (int) – Column containing values.
time_format (str) – Timestamp format.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_budget_instructions(budget_file, budget_type, components=None, locations=None, observations=None, header_lines=4, time_column=1, time_format='%m/%d/%Y', instruction_file=None)[source]#

Generate instructions for budget output.

Creates instruction file for reading water budget components from IWFM budget output files.

Parameters:

budget_file (Path | str) – Path to budget output file.
budget_type (str) – Budget type: “gw”, “stream”, “lake”, “rootzone”.
components (list[str] | None) – Budget components to extract.
locations (list[int] | None) – Location IDs (subregions, reaches, etc.).
observations (list[IWFMObservation] | None) – Observations to extract.
header_lines (int) – Number of header lines.
time_column (int) – Column containing timestamp.
time_format (str) – Timestamp format.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_lake_instructions(output_file, lakes=None, observations=None, variable='level', header_lines=3, time_column=1, value_column=2, time_format='%m/%d/%Y_%H:%M', instruction_file=None)[source]#

Generate instructions for lake output.

Parameters:

output_file (Path | str) – Path to lake output file.
lakes (list[int] | None) – Lake IDs to extract.
observations (list[IWFMObservation] | None) – Observations to extract.
variable (str) – Variable: “level” or “storage”.
header_lines (int) – Number of header lines.
time_column (int) – Column containing timestamp.
value_column (int) – Column containing values.
time_format (str) – Timestamp format.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_subsidence_instructions(output_file, observations=None, header_lines=3, time_column=1, value_column=2, time_format='%m/%d/%Y_%H:%M', instruction_file=None)[source]#

Generate instructions for subsidence output.

Parameters:

output_file (Path | str) – Path to subsidence output file.
observations (list[IWFMObservation] | None) – Observations to extract.
header_lines (int) – Number of header lines.
time_column (int) – Column containing timestamp.
value_column (int) – Column containing values.
time_format (str) – Timestamp format.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_custom_instructions(output_file, observations, header_lines=0, instruction_file=None)[source]#

Generate custom instructions for any output file.

This method allows creating instructions for non-standard output file formats.

Parameters:

output_file (Path | str) – Path to output file.
observations (list[tuple[str, str, int]]) – List of (obs_name, search_string, value_column) tuples. search_string is a marker to search for in the file.
header_lines (int) – Number of header lines to skip.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_fixed_format_instructions(output_file, observations, header_lines=0, instruction_file=None)[source]#

Generate instructions for fixed-format output file.

For output files with fixed-width columns rather than delimiter-separated values.

Parameters:

output_file (Path | str) – Path to output file.
observations (list[tuple[str, int, int, int]]) – List of (obs_name, line_number, start_column, end_column) tuples. Line numbers are 1-based (after header). Columns are 1-based character positions.
header_lines (int) – Number of header lines.
instruction_file (Path | str | None) – Output instruction file path.

Returns:

The created instruction file.

Return type:

InstructionFile

generate_all_instructions(output_files=None)[source]#

Generate all required instruction files based on observations.

Automatically generates instructions for all observations in the observation manager.

Parameters:: output_files (dict[str, Path | str] | None) – Mapping of observation type values to output file paths. E.g., {“head”: “GW_Heads.out”, “flow”: “StreamFlow.out”}
Returns:: List of created instruction files.
Return type:: list[InstructionFile]

get_all_instructions()[source]#

Get all created instruction files.

Returns:: All instruction files created by this manager.
Return type:: list[InstructionFile]

clear_instructions()[source]#

Clear all created instructions.

__repr__()[source]#

Return string representation.

Geostatistics #

Variogram modeling and spatial correlation for pilot point and ensemble parameterization.

Geostatistics module for PEST++ pilot point parameterization.

This module provides geostatistical tools for: - Variogram modeling (spherical, exponential, gaussian, matern) - Kriging interpolation (ordinary, simple) - Covariance matrix computation - Geostatistical realization generation - Prior ensemble generation with spatial correlation

These tools support highly parameterized inversion with pilot points, where parameter values at scattered pilot points are interpolated to model nodes using kriging.

class pyiwfm.runner.pest_geostat.VariogramType(*values)[source]#

Bases: Enum

Types of variogram models.

SPHERICAL = 'spherical'#

EXPONENTIAL = 'exponential'#

GAUSSIAN = 'gaussian'#

MATERN = 'matern'#

LINEAR = 'linear'#

POWER = 'power'#

NUGGET = 'nugget'#

class pyiwfm.runner.pest_geostat.Variogram(variogram_type, a, sill=1.0, nugget=0.0, anisotropy_ratio=1.0, anisotropy_angle=0.0, power=1.0)[source]#

Bases: object

Variogram model for geostatistical analysis.

A variogram describes the spatial correlation structure of a variable. It quantifies how dissimilarity between values increases with distance.

Parameters:

variogram_type (str | VariogramType) – Type of variogram model.
a (float) – Range parameter - distance at which correlation approaches zero.
sill (float) – Sill - variance at which the variogram levels off.
nugget (float) – Nugget effect - discontinuity at the origin (measurement error + micro-scale variation).
anisotropy_ratio (float) – Ratio of major to minor range (1.0 = isotropic).
anisotropy_angle (float) – Angle of major axis in degrees from east (counterclockwise).
power (float) – Power parameter for power variogram model.

Examples

>>> # Exponential variogram with range=10000m, sill=1.0
>>> vario = Variogram("exponential", a=10000, sill=1.0, nugget=0.1)
>>> # Evaluate at distances
>>> gamma = vario.evaluate(np.array([0, 1000, 5000, 10000]))

variogram_type: str | VariogramType#

a: float#

sill: float = 1.0#

nugget: float = 0.0#

anisotropy_ratio: float = 1.0#

anisotropy_angle: float = 0.0#

power: float = 1.0#

__post_init__()[source]#

Validate and convert variogram type.

property total_sill: float#: Total sill (nugget + partial sill).

property effective_range: float#

Effective range (distance at 95% of sill).

For exponential: ~3*a For gaussian: ~sqrt(3)*a For spherical: a

evaluate(h)[source]#

Evaluate variogram at lag distances h.

Parameters:: h (NDArray[np.float64] | float) – Lag distance(s).
Returns:: Variogram value(s) gamma(h).
Return type:: NDArray[np.float64] | float

covariance(h)[source]#

Compute covariance from variogram.

C(h) = sill - gamma(h) for stationary variograms.

Parameters:: h (NDArray[np.float64] | float) – Lag distance(s).
Returns:: Covariance value(s).
Return type:: NDArray[np.float64] | float

transform_coordinates(x, y)[source]#

Transform coordinates for anisotropy.

Parameters:

x (NDArray) – X coordinates.
y (NDArray) – Y coordinates.

Returns:

Transformed coordinates.

Return type:

tuple[NDArray, NDArray]

compute_distance_matrix(x1, y1, x2=None, y2=None)[source]#

Compute distance matrix with anisotropy.

Parameters:

x1 (NDArray) – First set of coordinates.
y1 (NDArray) – First set of coordinates.
x2 (NDArray | None) – Second set of coordinates. If None, compute self-distances.
y2 (NDArray | None) – Second set of coordinates. If None, compute self-distances.

Returns:

Distance matrix.

Return type:

NDArray

classmethod from_data(x, y, values, variogram_type='exponential', n_lags=15, max_lag=None)[source]#

Fit variogram to data using empirical variogram.

Parameters:

x (NDArray) – Coordinates of data points.
y (NDArray) – Coordinates of data points.
values (NDArray) – Values at data points.
variogram_type (str) – Type of variogram model to fit.
n_lags (int) – Number of lag bins for empirical variogram.
max_lag (float | None) – Maximum lag distance. If None, uses half of maximum distance.

Returns:

Fitted variogram model.

Return type:

Variogram

Raises:

ImportError – If scipy is not available.

to_dict()[source]#

Convert to dictionary.

classmethod from_dict(d)[source]#

Create from dictionary.

__repr__()[source]#

Return string representation.

__init__(variogram_type, a, sill=1.0, nugget=0.0, anisotropy_ratio=1.0, anisotropy_angle=0.0, power=1.0)#

class pyiwfm.runner.pest_geostat.GeostatManager(model=None)[source]#

Bases: object

Manages geostatistical operations for pilot point parameterization.

This class provides methods for: - Computing covariance matrices between points - Kriging interpolation from pilot points to model nodes - Generating geostatistically correlated realizations - Writing kriging factors for use in PEST++ preprocessing

Parameters:: model (Any) – IWFM model instance (optional, for mesh information).

Examples

>>> gm = GeostatManager()
>>> # Compute covariance matrix
>>> cov = gm.compute_covariance_matrix(pp_x, pp_y, variogram)
>>> # Krige to model nodes
>>> node_values = gm.krige(pp_x, pp_y, pp_values, node_x, node_y, variogram)

__init__(model=None)[source]#

Initialize the geostat manager.

Parameters:: model (Any) – IWFM model instance (optional).

compute_covariance_matrix(x, y, variogram)[source]#

Compute covariance matrix between points.

Parameters:

x (NDArray) – Coordinates of points.
y (NDArray) – Coordinates of points.
variogram (Variogram) – Variogram model.

Returns:

Covariance matrix (n x n).

Return type:

NDArray

compute_variogram_matrix(x, y, variogram)[source]#

Compute variogram matrix between points.

Parameters:

x (NDArray) – Coordinates of points.
y (NDArray) – Coordinates of points.
variogram (Variogram) – Variogram model.

Returns:

Variogram matrix (n x n).

Return type:

NDArray

krige(pilot_x, pilot_y, pilot_values, target_x, target_y, variogram, kriging_type='ordinary', return_variance=False)[source]#

Interpolate values using kriging.

Parameters:

pilot_x (NDArray) – Coordinates of pilot points.
pilot_y (NDArray) – Coordinates of pilot points.
pilot_values (NDArray) – Values at pilot points.
target_x (NDArray) – Coordinates of target points.
target_y (NDArray) – Coordinates of target points.
variogram (Variogram) – Variogram model.
kriging_type (str) – Type of kriging: “ordinary” or “simple”.
return_variance (bool) – If True, also return kriging variance.

Returns:

Interpolated values, optionally with variance.

Return type:

NDArray | tuple[NDArray, NDArray]

Raises:

ImportError – If scipy is not available.

compute_kriging_factors(pilot_x, pilot_y, target_x, target_y, variogram, kriging_type='ordinary')[source]#

Compute kriging interpolation factors.

These factors can be saved and applied multiple times without re-solving the kriging system.

Parameters:

pilot_x (NDArray) – Coordinates of pilot points.
pilot_y (NDArray) – Coordinates of pilot points.
target_x (NDArray) – Coordinates of target points.
target_y (NDArray) – Coordinates of target points.
variogram (Variogram) – Variogram model.
kriging_type (str) – Type of kriging.

Returns:

Kriging factors matrix (n_target x n_pilot).

Return type:

NDArray

generate_realizations(x, y, variogram, n_realizations=100, mean=0.0, conditioning_data=None, seed=None)[source]#

Generate geostatistical realizations.

Generates spatially correlated random fields using the covariance structure defined by the variogram.

Parameters:

x (NDArray) – Coordinates where realizations are generated.
y (NDArray) – Coordinates where realizations are generated.
variogram (Variogram) – Variogram model defining spatial correlation.
n_realizations (int) – Number of realizations to generate.
mean (float) – Mean value of the field.
conditioning_data (tuple[NDArray, NDArray, NDArray] | None) – Optional conditioning data as (x, y, values).
seed (int | None) – Random seed for reproducibility.

Returns:

Realizations array (n_realizations x n_points).

Return type:

NDArray

Raises:

ImportError – If scipy is not available.

generate_prior_ensemble(parameters, n_realizations=100, variogram=None, seed=None, method='lhs')[source]#

Generate prior parameter ensemble.

For parameters with spatial locations (pilot points), generates spatially correlated realizations. For non-spatial parameters, uses Latin Hypercube Sampling or uniform sampling.

Parameters:

parameters (list) – List of Parameter objects.
n_realizations (int) – Number of ensemble members.
variogram (Variogram | None) – Variogram for spatial correlation. If None, assumes uncorrelated.
seed (int | None) – Random seed.
method (str) – Sampling method: “lhs” for Latin Hypercube, “uniform” for uniform.

Returns:

Ensemble array (n_realizations x n_parameters).

Return type:

NDArray

write_kriging_factors(pilot_x, pilot_y, pilot_names, target_x, target_y, target_ids, variogram, filepath, kriging_type='ordinary', format='pest')[source]#

Write kriging interpolation factors to file.

Parameters:

pilot_x (NDArray) – Coordinates of pilot points.
pilot_y (NDArray) – Coordinates of pilot points.
pilot_names (list[str]) – Names of pilot point parameters.
target_x (NDArray) – Coordinates of target points.
target_y (NDArray) – Coordinates of target points.
target_ids (list[int | str]) – IDs of target points (nodes, elements).
variogram (Variogram) – Variogram model.
filepath (Path | str) – Output file path.
kriging_type (str) – Type of kriging.
format (str) – Output format: “pest” for PEST pp_factors, “csv” for CSV.

Returns:

Path to written file.

Return type:

Path

write_structure_file(variogram, filepath, name='structure1')[source]#

Write PEST++ structure file for variogram.

Parameters:

variogram (Variogram) – Variogram model.
filepath (Path | str) – Output file path.
name (str) – Structure name.

Returns:

Path to written file.

Return type:

Path

__repr__()[source]#

Return string representation.

pyiwfm.runner.pest_geostat.compute_empirical_variogram(x, y, values, n_lags=15, max_lag=None)[source]#

Compute empirical variogram from data.

Parameters:

x (NDArray) – Coordinates of data points.
y (NDArray) – Coordinates of data points.
values (NDArray) – Values at data points.
n_lags (int) – Number of lag bins.
max_lag (float | None) – Maximum lag distance.

Returns:

Lag centers, semivariance values, and pair counts.

Return type:

tuple[NDArray, NDArray, NDArray]

Main Helper Interface #

The IWFMPestHelper class is the primary entry point for setting up PEST++ calibration. It coordinates parameters, observations, templates, instructions, and control file generation.

Main PEST++ helper interface for IWFM models.

This module provides the IWFMPestHelper class - the primary high-level interface for setting up PEST++ calibration, uncertainty analysis, and optimization for IWFM models.

It coordinates all PEST++ components: - Parameter management - Observation management - Template file generation - Instruction file generation - Geostatistics - Control file writing - Execution of PEST++ programs

Examples

>>> helper = IWFMPestHelper(pest_dir="pest_setup", case_name="c2vsim_cal")
>>> helper.add_zone_parameters("hk", zones=[1, 2, 3], layer=1)
>>> helper.add_multiplier("pumping", bounds=(0.8, 1.2))
>>> helper.build()

class pyiwfm.runner.pest_helper.RegularizationType(*values)[source]#

Bases: Enum

Types of regularization for PEST++.

PREFERRED_HOMOGENEITY = 'preferred_homogeneity'#

PREFERRED_VALUE = 'preferred_value'#

TIKHONOV = 'tikhonov'#

NONE = 'none'#

class pyiwfm.runner.pest_helper.SVDConfig(maxsing=100, eigthresh=1e-06)[source]#

Bases: object

SVD configuration for PEST++.

Parameters:

maxsing (int) – Maximum number of singular values to use.
eigthresh (float) – Eigenvalue ratio threshold for truncation.

maxsing: int = 100#

eigthresh: float = 1e-06#

to_dict()[source]#

Convert to PEST++ options dict.

__init__(maxsing=100, eigthresh=1e-06)#

class pyiwfm.runner.pest_helper.RegularizationConfig(reg_type=RegularizationType.PREFERRED_HOMOGENEITY, weight=1.0, preferred_value=None)[source]#

Bases: object

Regularization configuration.

Parameters:

reg_type (RegularizationType) – Type of regularization.
weight (float) – Regularization weight multiplier.
preferred_value (float | None) – Preferred parameter value (for preferred_value type).

reg_type: RegularizationType = 'preferred_homogeneity'#

weight: float = 1.0#

preferred_value: float | None = None#

__init__(reg_type=RegularizationType.PREFERRED_HOMOGENEITY, weight=1.0, preferred_value=None)#

class pyiwfm.runner.pest_helper.IWFMPestHelper(pest_dir, case_name='iwfm_cal', model_dir=None, model=None)[source]#

Bases: object

Main interface for IWFM PEST++ setup.

This class provides a high-level interface for setting up PEST++ calibration, uncertainty analysis, and optimization for IWFM models. It coordinates parameter management, observation management, template/instruction generation, geostatistics, and control file writing.

Parameters:

pest_dir (Path | str) – Directory for PEST++ files.
case_name (str) – Base name for PEST++ files (e.g., “iwfm_cal” -> iwfm_cal.pst).
model_dir (Path | str | None) – Directory containing the IWFM model. If None, uses pest_dir.
model (Any) – IWFM model instance (optional, for mesh/structure queries).

Examples

>>> helper = IWFMPestHelper(pest_dir="pest_setup", case_name="c2vsim_cal")
>>> helper.add_zone_parameters("hk", zones=[1, 2, 3], layer=1)
>>> helper.add_head_observations(wells, head_data)
>>> helper.build()

__init__(pest_dir, case_name='iwfm_cal', model_dir=None, model=None)[source]#

Initialize the PEST++ helper.

Parameters:

pest_dir (Path | str) – Directory for PEST++ files.
case_name (str) – Base name for PEST++ files.
model_dir (Path | str | None) – Directory containing the IWFM model.
model (Any) – IWFM model instance.

add_zone_parameters(param_type, zones, layer=None, initial_value=1.0, bounds=(0.01, 100.0), transform='log', group=None)[source]#

Add zone-based parameters.

Creates one parameter per zone for the specified property type.

Parameters:

param_type (str | IWFMParameterType) – Parameter type (e.g., “hk”, IWFMParameterType.HORIZONTAL_K).
zones (list[int]) – Zone IDs.
layer (int | None) – Model layer.
initial_value (float) – Initial value for all zone parameters.
bounds (tuple[float, float]) – (lower_bound, upper_bound).
transform (str) – Parameter transform: “log”, “none”, “fixed”.
group (str | None) – Parameter group name. Auto-generated if None.

Returns:

Created parameters.

Return type:

list[Parameter]

add_pilot_points(param_type, points, layer=1, initial_value=1.0, bounds=(0.01, 100.0), variogram=None, transform='log', prefix=None)[source]#

Add pilot point parameters.

Parameters:

param_type (str | IWFMParameterType) – Parameter type.
points (list[tuple[float, float]]) – Pilot point (x, y) coordinates.
layer (int) – Model layer.
initial_value (float) – Initial value.
bounds (tuple[float, float]) – (lower_bound, upper_bound).
variogram (Variogram | dict | None) – Variogram for kriging. If dict, creates Variogram from it.
transform (str) – Parameter transform.
prefix (str | None) – Name prefix.

Returns:

Created parameters.

Return type:

list[Parameter]

add_multiplier(param_type, spatial='global', zones=None, initial_value=1.0, bounds=(0.5, 2.0), transform='none')[source]#

Add multiplier parameters.

Parameters:

param_type (str | IWFMParameterType) – Parameter type.
spatial (str) – Spatial extent: “global”, “zone”.
zones (list[int] | None) – Zone IDs (required if spatial=”zone”).
initial_value (float) – Initial multiplier value.
bounds (tuple[float, float]) – (lower_bound, upper_bound).
transform (str) – Parameter transform.

Returns:

Created parameters.

Return type:

list[Parameter]

add_stream_parameters(param_type, reaches, initial_value=1.0, bounds=(0.001, 10.0), transform='log')[source]#

Add stream-related parameters.

Parameters:

param_type (str | IWFMParameterType) – Parameter type (e.g., STREAMBED_K).
reaches (list[int]) – Stream reach IDs.
initial_value (float) – Initial value.
bounds (tuple[float, float]) – (lower_bound, upper_bound).
transform (str) – Parameter transform.

Returns:

Created parameters.

Return type:

list[Parameter]

add_rootzone_parameters(param_type, land_use_types, initial_value=1.0, bounds=(0.5, 1.5), transform='none')[source]#

Add root zone parameters by land use type.

Parameters:

param_type (str | IWFMParameterType) – Parameter type (e.g., CROP_COEFFICIENT).
land_use_types (list[str]) – Land use type names.
initial_value (float) – Initial value.
bounds (tuple[float, float]) – (lower_bound, upper_bound).
transform (str) – Parameter transform.

Returns:

Created parameters.

Return type:

list[Parameter]

add_head_observations(well_id, x, y, times, values, layer=1, weight=1.0, group=None)[source]#

Add groundwater head observations for a well.

This is a simplified convenience method that creates observations directly. For more control (DataFrames, weight strategies, etc.), use self.observations.add_head_observations() with WellInfo objects.

Parameters:

well_id (str) – Well identifier.
x (float) – Well coordinates.
y (float) – Well coordinates.
times (list[datetime]) – Observation timestamps.
values (list[float]) – Observed head values.
layer (int) – Model layer.
weight (float) – Observation weight.
group (str | None) – Observation group name.

Returns:

Created observations.

Return type:

list[IWFMObservation]

add_streamflow_observations(gage_id, reach_id, times, values, weight=1.0, transform='none', group=None)[source]#

Add streamflow observations for a gage.

This is a simplified convenience method. For more control, use self.observations.add_streamflow_observations() with GageInfo objects.

Parameters:

gage_id (str) – Gage identifier.
reach_id (int) – Stream reach ID.
times (list[datetime]) – Observation timestamps.
values (list[float]) – Observed flow values.
weight (float) – Observation weight.
transform (str) – Transform: “none”, “log”.
group (str | None) – Observation group name.

Returns:

Created observations.

Return type:

list[IWFMObservation]

set_svd(maxsing=100, eigthresh=1e-06)[source]#

Configure SVD truncation for parameter estimation.

Parameters:

maxsing (int) – Maximum number of singular values.
eigthresh (float) – Eigenvalue ratio threshold.

set_regularization(reg_type='preferred_homogeneity', weight=1.0, preferred_value=None)[source]#

Configure regularization for pilot points.

Parameters:

reg_type (str) – Type: “preferred_homogeneity”, “preferred_value”, “tikhonov”.
weight (float) – Regularization weight multiplier.
preferred_value (float | None) – Preferred parameter value.

set_model_command(command)[source]#

Set the model forward run command.

Parameters:: command (str) – Command string that PEST++ will execute.

set_pestpp_options(**options)[source]#

Set PEST++ specific options.

Parameters:: **options (Any) – Option name-value pairs (e.g., ies_num_reals=100).

get_pestpp_option(key, default=None)[source]#

Get a PEST++ option value.

Parameters:

key (str) – Option name.
default (Any) – Default value if not set.

Returns:

Option value.

Return type:

Any

balance_observation_weights(contributions=None)[source]#

Balance weights across observation groups.

Parameters:: contributions (dict[str, float] | None) – Target contributions per group (e.g., {“head”: 0.5, “flow”: 0.5}). If None, equalizes contributions.

build(pst_file=None)[source]#

Build complete PEST++ setup.

Creates: - Control file (.pst) - Template files (.tpl) - Instruction files (.ins) - Forward run script

Parameters:: pst_file (Path | str | None) – Path for control file. Defaults to pest_dir/case_name.pst.
Returns:: Path to the control file.
Return type:: Path
Raises:: ValueError – If no parameters or observations are defined.

add_template(template)[source]#

Register a template file for the build.

Parameters:: template (TemplateFile) – Template file to register.

add_instruction(instruction)[source]#

Register an instruction file for the build.

Parameters:: instruction (InstructionFile) – Instruction file to register.

write_forward_run_script(filepath=None)[source]#

Write script that runs IWFM for PEST++.

Parameters:: filepath (Path | str | None) – Output path. Defaults to pest_dir/forward_run.py.
Returns:: Path to the written script.
Return type:: Path

write_pp_interpolation_script(filepath=None)[source]#

Write pilot point interpolation script.

This script is run before the model to interpolate pilot point values to model nodes/elements using kriging.

Parameters:: filepath (Path | str | None) – Output path.
Returns:: Path to the written script.
Return type:: Path

run_pestpp(program='pestpp-glm', n_workers=1, extra_args=None)[source]#

Run a PEST++ program.

Parameters:

program (str) – PEST++ program: “pestpp-glm”, “pestpp-ies”, “pestpp-sen”, etc.
n_workers (int) – Number of parallel workers.
extra_args (list[str] | None) – Additional command-line arguments.

Returns:

Process result.

Return type:

subprocess.CompletedProcess[str]

Raises:

FileNotFoundError – If the PEST++ executable or control file is not found.
RuntimeError – If the setup has not been built yet.

run_pestpp_glm(n_workers=1, **kwargs)[source]#

Run pestpp-glm for parameter estimation.

Parameters:

n_workers (int) – Number of parallel workers.
**kwargs (Any) – Additional PEST++ options set before running.

Returns:

Process result.

Return type:

subprocess.CompletedProcess[str]

run_pestpp_ies(n_realizations=100, n_workers=1, **kwargs)[source]#

Run pestpp-ies for ensemble calibration.

Parameters:

n_realizations (int) – Number of ensemble members.
n_workers (int) – Number of parallel workers.
**kwargs (Any) – Additional PEST++ options.

Returns:

Process result.

Return type:

subprocess.CompletedProcess[str]

run_pestpp_sen(method='sobol', n_samples=1000, **kwargs)[source]#

Run pestpp-sen for sensitivity analysis.

Parameters:

method (str) – Sensitivity method: “sobol”, “morris”.
n_samples (int) – Number of samples.
**kwargs (Any) – Additional PEST++ options.

Returns:

Process result.

Return type:

subprocess.CompletedProcess[str]

property n_parameters: int#: Number of defined parameters.

property n_observations: int#: Number of defined observations.

property parameter_groups: list[str]#: List of parameter group names.

property observation_groups: list[str]#: List of observation group names.

summary()[source]#

Get a summary of the PEST++ setup.

Returns:: Summary information.
Return type:: dict[str, Any]

__repr__()[source]#

Return string representation.

Ensemble Management #

Prior and posterior ensemble generation for pestpp-ies iterative ensemble smoother workflows.

Ensemble management for PEST++ IES workflows.

This module provides the IWFMEnsembleManager class for generating, writing, and analyzing parameter and observation ensembles for pestpp-ies (Iterative Ensemble Smoother) workflows.

Features: - Prior parameter ensemble generation (LHS, Gaussian, uniform) - Spatially correlated ensembles via geostatistics - Observation noise ensemble generation - Ensemble file I/O (CSV format for PEST++) - Posterior ensemble loading and analysis - Ensemble statistics computation

class pyiwfm.runner.pest_ensemble.EnsembleStatistics(mean, std, median, q05, q95, n_realizations, n_parameters, parameter_names)[source]#

Bases: object

Summary statistics for a parameter ensemble.

Variables:

mean (NDArray) – Mean values for each parameter.
std (NDArray) – Standard deviations for each parameter.
median (NDArray) – Median values for each parameter.
q05 (NDArray) – 5th percentile for each parameter.
q95 (NDArray) – 95th percentile for each parameter.
n_realizations (int) – Number of realizations in the ensemble.
n_parameters (int) – Number of parameters.
parameter_names (list[str]) – Parameter names.

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

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

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

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

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

n_realizations: int#

n_parameters: int#

parameter_names: list[str]#

to_dict()[source]#

Convert to dictionary.

__init__(mean, std, median, q05, q95, n_realizations, n_parameters, parameter_names)#

class pyiwfm.runner.pest_ensemble.IWFMEnsembleManager(parameters=None, geostat=None)[source]#

Bases: object

Manages ensemble generation and analysis for IWFM PEST++ workflows.

This class provides methods for creating prior parameter ensembles, observation noise ensembles, and analyzing posterior ensembles from pestpp-ies runs.

Parameters:

parameters (list[Parameter]) – List of parameters for ensemble generation.
geostat (GeostatManager | None) – Geostatistical manager for spatially correlated ensembles.

Examples

>>> em = IWFMEnsembleManager(parameters=params)
>>> prior = em.generate_prior_ensemble(n_realizations=100)
>>> em.write_parameter_ensemble(prior, "prior_ensemble.csv")

__init__(parameters=None, geostat=None)[source]#

Initialize the ensemble manager.

Parameters:

parameters (list[Parameter] | None) – List of parameters.
geostat (GeostatManager | None) – Geostatistical manager.

property parameter_names: list[str]#: Get parameter names.

property n_parameters: int#: Number of parameters.

generate_prior_ensemble(n_realizations=100, method='lhs', variogram=None, seed=None)[source]#

Generate prior parameter ensemble.

For parameters with spatial locations (pilot points), generates spatially correlated realizations using the variogram. For non-spatial parameters, uses Latin Hypercube Sampling or uniform random sampling.

Parameters:

n_realizations (int) – Number of ensemble members.
method (str) – Sampling method: “lhs” (Latin Hypercube), “uniform”, “gaussian”.
variogram (Variogram | None) – Variogram for spatial correlation.
seed (int | None) – Random seed for reproducibility.

Returns:

Ensemble array (n_realizations x n_parameters).

Return type:

NDArray

generate_observation_ensemble(observation_values, observation_weights, n_realizations=100, noise_type='gaussian', seed=None)[source]#

Generate observation noise ensemble.

Adds noise to observation values based on their weights (weight = 1/standard_deviation).

Parameters:

observation_values (NDArray) – Observed values.
observation_weights (NDArray) – Observation weights (inverse of std deviation).
n_realizations (int) – Number of ensemble members.
noise_type (str) – Noise distribution: “gaussian”.
seed (int | None) – Random seed.

Returns:

Observation ensemble (n_realizations x n_observations).

Return type:

NDArray

write_parameter_ensemble(ensemble, filepath)[source]#

Write parameter ensemble to CSV file.

Format compatible with PEST++ pestpp-ies input.

Parameters:

ensemble (NDArray) – Ensemble array (n_realizations x n_parameters).
filepath (Path | str) – Output file path.

Returns:

Path to written file.

Return type:

Path

write_observation_ensemble(ensemble, observation_names, filepath)[source]#

Write observation ensemble to CSV file.

Parameters:

ensemble (NDArray) – Observation ensemble (n_realizations x n_observations).
observation_names (list[str]) – Observation names.
filepath (Path | str) – Output file path.

Returns:

Path to written file.

Return type:

Path

load_posterior_ensemble(filepath)[source]#

Load posterior parameter ensemble from pestpp-ies output.

Parameters:: filepath (Path | str) – Path to posterior ensemble CSV file.
Returns:: Posterior ensemble (n_realizations x n_parameters).
Return type:: NDArray
Raises:: FileNotFoundError – If the file does not exist.

analyze_ensemble(ensemble)[source]#

Compute ensemble statistics.

Parameters:: ensemble (NDArray) – Ensemble array (n_realizations x n_parameters).
Returns:: Computed statistics.
Return type:: EnsembleStatistics

compute_reduction_factor(prior, posterior)[source]#

Compute uncertainty reduction factor.

Measures how much the posterior uncertainty is reduced relative to the prior.

Parameters:

prior (NDArray) – Prior ensemble.
posterior (NDArray) – Posterior ensemble.

Returns:

Reduction factor per parameter (0 = no reduction, 1 = complete).

Return type:

NDArray

get_best_realization(ensemble, objective_values)[source]#

Get the best realization based on objective function.

Parameters:

ensemble (NDArray) – Parameter ensemble.
objective_values (NDArray) – Objective function values for each realization.

Returns:

Index and parameter values of best realization.

Return type:

tuple[int, NDArray]

__repr__()[source]#

Return string representation.

Post-Processing #

Load and analyze PEST++ output files for calibration assessment.

Post-processing module for PEST++ results.

This module provides the PestPostProcessor class for loading, analyzing, and summarizing PEST++ calibration results from pestpp-glm, pestpp-ies, and pestpp-sen output files.

Features: - Load PEST++ residual files (.rei, .res) - Parse parameter sensitivity files (.sen) - Load iteration records (.iobj, .isen) - Compute goodness-of-fit statistics (RMSE, NSE, R², bias) - Parameter identifiability analysis - Export calibrated parameter values

class pyiwfm.runner.pest_postprocessor.ResidualData(names, groups, observed, simulated, residuals, weights)[source]#

Bases: object

Observation residual data.

Variables:

names (list[str]) – Observation names.
groups (list[str]) – Observation group names.
observed (NDArray) – Observed values.
simulated (NDArray) – Simulated values.
residuals (NDArray) – Residuals (observed - simulated).
weights (NDArray) – Observation weights.

names: list[str]#

groups: list[str]#

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

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

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

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

property n_observations: int#: Number of observations.

property weighted_residuals: ndarray[tuple[Any, ...], dtype[float64]]#: Weighted residuals.

property phi: float#: Total objective function (sum of squared weighted residuals).

group_phi()[source]#

Objective function contribution by group.

__init__(names, groups, observed, simulated, residuals, weights)#

class pyiwfm.runner.pest_postprocessor.SensitivityData(parameter_names, composite_sensitivities)[source]#

Bases: object

Parameter sensitivity data.

Variables:

parameter_names (list[str]) – Parameter names.
composite_sensitivities (NDArray) – Composite scaled sensitivities.

parameter_names: list[str]#

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

property n_parameters: int#: Number of parameters.

most_sensitive(n=10)[source]#

Get most sensitive parameters.

Parameters:: n (int) – Number of parameters to return.
Returns:: Sorted (name, sensitivity) pairs.
Return type:: list[tuple[str, float]]

least_sensitive(n=10)[source]#

Get least sensitive parameters.

Parameters:: n (int) – Number of parameters to return.
Returns:: Sorted (name, sensitivity) pairs.
Return type:: list[tuple[str, float]]

__init__(parameter_names, composite_sensitivities)#

class pyiwfm.runner.pest_postprocessor.CalibrationResults(case_name, n_iterations=0, final_phi=0.0, iteration_phi=<factory>, residuals=None, sensitivities=None, calibrated_values=<factory>)[source]#

Bases: object

Summary of PEST++ calibration results.

Variables:

case_name (str) – PEST++ case name.
n_iterations (int) – Number of iterations completed.
final_phi (float) – Final objective function value.
iteration_phi (list[float]) – Objective function values per iteration.
residuals (ResidualData | None) – Final residual data.
sensitivities (SensitivityData | None) – Parameter sensitivity data.
calibrated_values (dict[str, float]) – Calibrated parameter values.

case_name: str#

n_iterations: int = 0#

final_phi: float = 0.0#

iteration_phi: list[float]#

residuals: ResidualData | None = None#

sensitivities: SensitivityData | None = None#

calibrated_values: dict[str, float]#

fit_statistics(group=None)[source]#

Compute goodness-of-fit statistics.

Parameters:: group (str | None) – Observation group. If None, uses all observations.
Returns:: Statistics including RMSE, MAE, R², NSE, bias.
Return type:: dict[str, float]

__init__(case_name, n_iterations=0, final_phi=0.0, iteration_phi=<factory>, residuals=None, sensitivities=None, calibrated_values=<factory>)#

pyiwfm.runner.pest_postprocessor.read_pest_res(filepath, prefix_filter=None, group_filter=None)[source]#

Read a PEST++ residual file (.res or .rei).

Returns a DataFrame with columns: name, group, observed, simulated, residual, weight. Optionally filters by observation name prefix and/or group.

Parameters:

filepath (str or Path) – Path to .res or .rei file.
prefix_filter (str | None) – If given, keep only observations whose name starts with this prefix.
group_filter (str | None) – If given, keep only observations in this group.

Returns:

Parsed residual data.

Return type:

pd.DataFrame

class pyiwfm.runner.pest_postprocessor.PestPostProcessor(pest_dir, case_name)[source]#

Bases: object

Post-processor for PEST++ calibration results.

Loads and analyzes results from pestpp-glm, pestpp-ies, and pestpp-sen output files.

Parameters:

pest_dir (Path | str) – Directory containing PEST++ output files.
case_name (str) – PEST++ case name (base name of .pst file).

Examples

>>> pp = PestPostProcessor("pest_output", "c2vsim_cal")
>>> results = pp.load_results()
>>> stats = results.fit_statistics()

__init__(pest_dir, case_name)[source]#

Initialize the post-processor.

Parameters:

pest_dir (Path | str) – PEST++ output directory.
case_name (str) – Case name.

load_results()[source]#

Load all available PEST++ results.

Returns:: Calibration results summary.
Return type:: CalibrationResults

export_calibrated_parameters(filepath, format='csv')[source]#

Export calibrated parameter values.

Parameters:

filepath (Path | str) – Output file path.
format (str) – Output format: “csv” or “pest”.

Returns:

Path to written file.

Return type:

Path

compute_identifiability()[source]#

Compute parameter identifiability from sensitivity data.

Returns identifiability index (0-1) for each parameter, where 1 means fully identifiable and 0 means unidentifiable.

Returns:: Parameter identifiability values, or None if data unavailable.
Return type:: dict[str, float] | None

summary_report()[source]#

Generate a text summary report.

Returns:: Formatted summary report.
Return type:: str

__repr__()[source]#

Return string representation.