Component Modules#

The component modules represent the major hydrologic components of an IWFM model. All components inherit from BaseComponent, which provides a consistent interface with validate() and n_items methods.

Groundwater Module#

The groundwater module contains classes for representing aquifer properties and groundwater simulation parameters.

Groundwater component classes for IWFM models.

This module provides classes for representing groundwater system components including wells, pumping, boundary conditions, tile drains, and aquifer parameters. It mirrors IWFM’s Package_AppGW.

class pyiwfm.components.groundwater.Well(id, x, y, element, name='', top_screen=0.0, bottom_screen=0.0, max_pump_rate=inf, pump_rate=0.0, layers=<factory>, radius=0.0, pump_column=0, pump_fraction=1.0, dist_method=0, dest_type=-1, dest_id=0, irig_frac_column=0, adjust_column=0, pump_max_column=0, pump_max_fraction=0.0)[source]#

Bases: object

A pumping well in the groundwater system.

Variables:

  • id (int) – Unique well identifier

  • name (str) – Well name/description

  • x (float) – X coordinate

  • y (float) – Y coordinate

  • element (int) – Element ID containing the well

  • top_screen (float) – Top of screen elevation

  • bottom_screen (float) – Bottom of screen elevation

  • max_pump_rate (float) – Maximum allowable pumping rate

  • pump_rate (float) – Current pumping rate

  • layers (list[int]) – List of layers the well screens across

  • radius (float) – Well radius

  • pump_column (int) – Column in time series pumping file

  • pump_fraction (float) – Fraction of pumping at this column

  • dist_method (int) – Distribution method (0-4)

  • dest_type (int) – Destination type (-1, 0, 1, 2, 3)

  • dest_id (int) – Destination ID (element, subregion, or group)

  • irig_frac_column (int) – Column for irrigation fraction

  • adjust_column (int) – Column for supply adjustment

  • pump_max_column (int) – Column for maximum pumping

  • pump_max_fraction (float) – Fraction of maximum pumping

id: int#
x: float#
y: float#
element: int#
name: str = ''#
top_screen: float = 0.0#
bottom_screen: float = 0.0#
max_pump_rate: float = inf#
pump_rate: float = 0.0#
layers: list[int]#
radius: float = 0.0#
pump_column: int = 0#
pump_fraction: float = 1.0#
dist_method: int = 0#
dest_type: int = -1#
dest_id: int = 0#
irig_frac_column: int = 0#
adjust_column: int = 0#
pump_max_column: int = 0#
pump_max_fraction: float = 0.0#
property screen_length: float#

Return the length of the well screen.

property n_layers: int#

Return number of layers the well screens across.

__init__(id, x, y, element, name='', top_screen=0.0, bottom_screen=0.0, max_pump_rate=inf, pump_rate=0.0, layers=<factory>, radius=0.0, pump_column=0, pump_fraction=1.0, dist_method=0, dest_type=-1, dest_id=0, irig_frac_column=0, adjust_column=0, pump_max_column=0, pump_max_fraction=0.0)#
class pyiwfm.components.groundwater.ElementPumping(element_id, layer, pump_rate, layer_fraction=1.0, pump_column=0, pump_fraction=1.0, dist_method=0, layer_factors=<factory>, dest_type=-1, dest_id=0, irig_frac_column=0, adjust_column=0, pump_max_column=0, pump_max_fraction=0.0)[source]#

Bases: object

Pumping applied to an element (distributed pumping).

Variables:

  • element_id (int) – Element ID

  • layer (int) – Layer number

  • pump_rate (float) – Total pumping rate

  • layer_fraction (float) – Fraction of pumping from this layer

  • pump_column (int) – Column in time series pumping file

  • pump_fraction (float) – Fraction of pumping at this column

  • dist_method (int) – Distribution method (0-4)

  • layer_factors (list[float]) – Layer distribution factors

  • dest_type (int) – Destination type (-1, 0, 1, 2, 3)

  • dest_id (int) – Destination ID

  • irig_frac_column (int) – Column for irrigation fraction

  • adjust_column (int) – Column for supply adjustment

  • pump_max_column (int) – Column for maximum pumping

  • pump_max_fraction (float) – Fraction of maximum pumping

element_id: int#
layer: int#
pump_rate: float#
layer_fraction: float = 1.0#
pump_column: int = 0#
pump_fraction: float = 1.0#
dist_method: int = 0#
layer_factors: list[float]#
dest_type: int = -1#
dest_id: int = 0#
irig_frac_column: int = 0#
adjust_column: int = 0#
pump_max_column: int = 0#
pump_max_fraction: float = 0.0#
property effective_rate: float#

Return effective pumping rate after applying fraction.

__init__(element_id, layer, pump_rate, layer_fraction=1.0, pump_column=0, pump_fraction=1.0, dist_method=0, layer_factors=<factory>, dest_type=-1, dest_id=0, irig_frac_column=0, adjust_column=0, pump_max_column=0, pump_max_fraction=0.0)#
class pyiwfm.components.groundwater.BoundaryCondition(id, bc_type, nodes, values, layer, conductance=<factory>, constraining_head=0.0, max_flow=0.0, ts_column=0, max_flow_ts_column=0)[source]#

Bases: object

A groundwater boundary condition.

Variables:

  • id (int) – Unique BC identifier

  • bc_type (str) – Type of boundary (‘specified_head’, ‘specified_flow’, ‘general_head’, ‘constrained_general_head’)

  • nodes (list[int]) – List of node IDs where BC is applied

  • values (list[float]) – BC values at each node

  • layer (int) – Layer number

  • conductance (list[float]) – Conductance values for general head BC

  • constraining_head (float) – Constraining head for constrained GH BC

  • max_flow (float) – Maximum flow for constrained GH BC

  • ts_column (int) – Time series column (0 = static)

  • max_flow_ts_column (int) – Time series column for max flow (0 = static)

id: int#
bc_type: str#
nodes: list[int]#
values: list[float]#
layer: int#
conductance: list[float]#
constraining_head: float = 0.0#
max_flow: float = 0.0#
ts_column: int = 0#
max_flow_ts_column: int = 0#
__post_init__()[source]#

Validate boundary condition data.

__init__(id, bc_type, nodes, values, layer, conductance=<factory>, constraining_head=0.0, max_flow=0.0, ts_column=0, max_flow_ts_column=0)#
class pyiwfm.components.groundwater.TileDrain(id, element, elevation, conductance, destination_type='outside', destination_id=None)[source]#

Bases: object

A tile drain in the groundwater system.

Variables:

  • id (int) – Unique drain identifier

  • element (int) – Element ID containing the drain

  • elevation (float) – Drain elevation

  • conductance (float) – Drain conductance

  • destination_type (str) – Where drain water goes (‘stream’, ‘outside’, etc.)

  • destination_id (int | None) – ID of destination (e.g., stream node)

id: int#
element: int#
elevation: float#
conductance: float#
destination_type: str = 'outside'#
destination_id: int | None = None#
__init__(id, element, elevation, conductance, destination_type='outside', destination_id=None)#
class pyiwfm.components.groundwater.Subsidence(element, layer, elastic_storage, inelastic_storage, preconsolidation_head, interbed_thick=0.0, interbed_thick_min=0.0)[source]#

Bases: object

Subsidence parameters for a node/layer.

Variables:

  • element (int) – Node ID (historical name; actually per-node in IWFM)

  • layer (int) – Layer number

  • elastic_storage (float) – Elastic skeletal storage coefficient

  • inelastic_storage (float) – Inelastic skeletal storage coefficient

  • preconsolidation_head (float) – Preconsolidation head

  • interbed_thick (float) – Interbed thickness

  • interbed_thick_min (float) – Minimum interbed thickness

element: int#
layer: int#
elastic_storage: float#
inelastic_storage: float#
preconsolidation_head: float#
interbed_thick: float = 0.0#
interbed_thick_min: float = 0.0#
property node: int#

Alias for element (subsidence is actually per-node).

__init__(element, layer, elastic_storage, inelastic_storage, preconsolidation_head, interbed_thick=0.0, interbed_thick_min=0.0)#
class pyiwfm.components.groundwater.NodeSubsidence(node_id, elastic_sc=<factory>, inelastic_sc=<factory>, interbed_thick=<factory>, interbed_thick_min=<factory>, precompact_head=<factory>, kv_sub=<factory>, n_eq=<factory>)[source]#

Bases: object

Subsidence parameters for a node across all layers.

Mirrors the per-node subsidence data from IWFM’s subsidence file.

Variables:

  • node_id (int) – GW node ID (1-based)

  • elastic_sc (list[float]) – Elastic storage coefficient per layer

  • inelastic_sc (list[float]) – Inelastic storage coefficient per layer

  • interbed_thick (list[float]) – Interbed thickness per layer

  • interbed_thick_min (list[float]) – Minimum interbed thickness per layer

  • precompact_head (list[float]) – Pre-compaction head per layer

  • kv_sub (list[float]) – Vertical conductivity of interbeds per layer (v5.0 only)

  • n_eq (list[float]) – Number of equivalent delay interbeds per layer (v5.0 only)

node_id: int#
elastic_sc: list[float]#
inelastic_sc: list[float]#
interbed_thick: list[float]#
interbed_thick_min: list[float]#
precompact_head: list[float]#
kv_sub: list[float]#
n_eq: list[float]#
__init__(node_id, elastic_sc=<factory>, inelastic_sc=<factory>, interbed_thick=<factory>, interbed_thick_min=<factory>, precompact_head=<factory>, kv_sub=<factory>, n_eq=<factory>)#
class pyiwfm.components.groundwater.SubIrrigation(id, gw_node, elevation, conductance)[source]#

Bases: object

A sub-irrigation location in the groundwater system.

Variables:

  • id (int) – Unique identifier

  • gw_node (int) – GW node ID

  • elevation (float) – Sub-irrigation elevation

  • conductance (float) – Conductance value

id: int#
gw_node: int#
elevation: float#
conductance: float#
__init__(id, gw_node, elevation, conductance)#
class pyiwfm.components.groundwater.AquiferParameters(n_nodes, n_layers, kh=None, kv=None, specific_storage=None, specific_yield=None, aquitard_kv=None)[source]#

Bases: object

Aquifer parameters for the groundwater system.

Variables:
n_nodes: int#
n_layers: int#
kh: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
kv: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
specific_storage: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
specific_yield: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
aquitard_kv: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
get_layer_kh(layer)[source]#

Get horizontal K for a specific layer.

get_layer_kv(layer)[source]#

Get vertical K for a specific layer.

__init__(n_nodes, n_layers, kh=None, kv=None, specific_storage=None, specific_yield=None, aquitard_kv=None)#
class pyiwfm.components.groundwater.HydrographLocation(node_id, layer, x, y, name='')[source]#

Bases: object

A groundwater hydrograph output location (observation point).

These are locations in the model where groundwater heads are tracked over time for comparison with field observations.

Variables:

  • node_id (int) – Mesh node ID for this observation

  • layer (int) – Aquifer layer (1-based)

  • x (float) – X coordinate

  • y (float) – Y coordinate

  • name (str) – Optional name/description

node_id: int#
layer: int#
x: float#
y: float#
name: str = ''#
__init__(node_id, layer, x, y, name='')#
class pyiwfm.components.groundwater.AppGW(n_nodes, n_layers, n_elements, wells=<factory>, boundary_conditions=<factory>, tile_drains=<factory>, element_pumping=<factory>, subsidence=<factory>, node_subsidence=<factory>, sub_irrigations=<factory>, aquifer_params=None, heads=None, hydrograph_locations=<factory>, face_flow_specs=<factory>, kh_anomalies=<factory>, return_flow_destinations=<factory>, parametric_groups=<factory>, subsidence_config=None, bc_config=None, gw_main_config=None, pumping_ts_file=None, bc_ts_file=None, n_bc_output_nodes=0, bc_output_file='', bc_output_file_raw='', bc_output_specs=<factory>, td_elev_factor=1.0, td_cond_factor=1.0, td_time_unit='1DAY', si_elev_factor=1.0, si_cond_factor=1.0, si_time_unit='1MON', td_n_hydro=0, td_hydro_volume_factor=1.0, td_hydro_volume_unit='', td_output_file_raw='', td_hydro_specs=<factory>)[source]#

Bases: BaseComponent

Groundwater application component.

This class manages the groundwater system including wells, pumping, boundary conditions, and aquifer parameters. It mirrors IWFM’s Package_AppGW.

Variables:
n_nodes: int#
n_layers: int#
n_elements: int#
wells: dict[int, Well]#
boundary_conditions: list[BoundaryCondition]#
tile_drains: dict[int, TileDrain]#
element_pumping: list[ElementPumping]#
subsidence: list[Subsidence]#
node_subsidence: list[NodeSubsidence]#
sub_irrigations: list[SubIrrigation]#
aquifer_params: AquiferParameters | None = None#
heads: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
hydrograph_locations: list[HydrographLocation]#
face_flow_specs: list[str]#
kh_anomalies: list[str]#
return_flow_destinations: dict[int, tuple[int, int]]#
parametric_groups: list[Any]#
subsidence_config: Any = None#
bc_config: Any = None#
gw_main_config: Any = None#
pumping_ts_file: Any = None#
bc_ts_file: Any = None#
n_bc_output_nodes: int = 0#
bc_output_file: str = ''#
bc_output_file_raw: str = ''#
bc_output_specs: list[Any]#
td_elev_factor: float = 1.0#
td_cond_factor: float = 1.0#
td_time_unit: str = '1DAY'#
si_elev_factor: float = 1.0#
si_cond_factor: float = 1.0#
si_time_unit: str = '1MON'#
td_n_hydro: int = 0#
td_hydro_volume_factor: float = 1.0#
td_hydro_volume_unit: str = ''#
td_output_file_raw: str = ''#
td_hydro_specs: list[dict[str, Any]]#
property n_items: int#

Return number of wells (primary entities).

property n_wells: int#

Return number of wells.

property n_hydrograph_locations: int#

Return number of hydrograph output locations.

property n_boundary_conditions: int#

Return number of boundary conditions.

property n_tile_drains: int#

Return number of tile drains.

add_well(well)[source]#

Add a well to the groundwater system.

add_boundary_condition(bc)[source]#

Add a boundary condition.

add_tile_drain(drain)[source]#

Add a tile drain.

add_element_pumping(pumping)[source]#

Add element-based pumping.

add_subsidence(sub)[source]#

Add subsidence parameters.

add_node_subsidence(ns)[source]#

Add per-node subsidence parameters.

add_sub_irrigation(si)[source]#

Add a sub-irrigation location.

property n_sub_irrigations: int#

Return number of sub-irrigation locations.

property n_node_subsidence: int#

Return number of nodes with subsidence data.

add_hydrograph_location(location)[source]#

Add a groundwater hydrograph output location.

get_well(well_id)[source]#

Get a well by ID.

get_tile_drain(drain_id)[source]#

Get a tile drain by ID.

set_aquifer_parameters(params)[source]#

Set aquifer parameters.

set_heads(heads)[source]#

Set the current heads array.

get_head(node, layer)[source]#

Get head at a specific node and layer.

get_wells_in_element(element_id)[source]#

Get all wells in an element.

get_total_pumping()[source]#

Calculate total pumping from all wells.

get_total_element_pumping()[source]#

Calculate total element-based pumping.

iter_wells()[source]#

Iterate over wells in ID order.

validate()[source]#

Validate the groundwater component.

Raises:

ComponentError – If component is invalid

to_arrays()[source]#

Convert groundwater data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[float64]]]

classmethod from_arrays(n_nodes, n_layers, n_elements, heads=None, kh=None, kv=None, specific_storage=None, specific_yield=None)[source]#

Create groundwater component from arrays.

Parameters:
Returns:

AppGW instance

Return type:

AppGW

__init__(n_nodes, n_layers, n_elements, wells=<factory>, boundary_conditions=<factory>, tile_drains=<factory>, element_pumping=<factory>, subsidence=<factory>, node_subsidence=<factory>, sub_irrigations=<factory>, aquifer_params=None, heads=None, hydrograph_locations=<factory>, face_flow_specs=<factory>, kh_anomalies=<factory>, return_flow_destinations=<factory>, parametric_groups=<factory>, subsidence_config=None, bc_config=None, gw_main_config=None, pumping_ts_file=None, bc_ts_file=None, n_bc_output_nodes=0, bc_output_file='', bc_output_file_raw='', bc_output_specs=<factory>, td_elev_factor=1.0, td_cond_factor=1.0, td_time_unit='1DAY', si_elev_factor=1.0, si_cond_factor=1.0, si_time_unit='1MON', td_n_hydro=0, td_hydro_volume_factor=1.0, td_hydro_volume_unit='', td_output_file_raw='', td_hydro_specs=<factory>)#

Stream Module#

The stream module contains classes for representing stream networks, including stream nodes and reaches.

Stream network classes for IWFM models.

This module provides classes for representing stream networks, including stream nodes, reaches, diversions, and bypasses. It mirrors IWFM’s Package_AppStream.

pyiwfm.components.stream.segments_intersect(p1, p2, p3, p4, tolerance=1e-10)[source]#

Check if two line segments intersect (excluding shared endpoints).

Segment 1: p1 to p2 Segment 2: p3 to p4

Parameters:

  • p1 (tuple of float) – Endpoints of first segment (x, y).

  • p2 (tuple of float) – Endpoints of first segment (x, y).

  • p3 (tuple of float) – Endpoints of second segment (x, y).

  • p4 (tuple of float) – Endpoints of second segment (x, y).

  • tolerance (float) – Distance tolerance for considering points as shared endpoints.

Returns:

True if segments intersect at a point that is not a shared endpoint.

Return type:

bool

Examples

>>> segments_intersect((0, 0), (2, 2), (0, 2), (2, 0))  # X crossing
True
>>> segments_intersect((0, 0), (1, 1), (1, 1), (2, 2))  # Shared endpoint
False
>>> segments_intersect((0, 0), (1, 0), (2, 0), (3, 0))  # Parallel, no overlap
False
class pyiwfm.components.stream.ReachCrossing(segment1, segment2, intersection_point=None)[source]#

Bases: object

Represents a detected crossing between two stream reach segments.

Variables:

  • segment1 (tuple) – First segment as ((x1, y1), (x2, y2), reach_id, segment_index).

  • segment2 (tuple) – Second segment as ((x1, y1), (x2, y2), reach_id, segment_index).

  • intersection_point (tuple of float, optional) – Approximate intersection point (x, y) if computed.

segment1: tuple[tuple[float, float], tuple[float, float], int, int]#
segment2: tuple[tuple[float, float], tuple[float, float], int, int]#
intersection_point: tuple[float, float] | None = None#
property reach1_id: int#

ID of first reach involved in crossing.

property reach2_id: int#

ID of second reach involved in crossing.

__init__(segment1, segment2, intersection_point=None)#
class pyiwfm.components.stream.CrossSectionData(bottom_elev=0.0, B0=0.0, s=0.0, n=0.04, max_flow_depth=10.0)[source]#

Bases: object

v5.0 hydraulic cross-section parameters for Manning’s equation.

Variables:

  • bottom_elev (float) – Stream bed bottom elevation

  • B0 (float) – Bottom width (0 = triangular channel)

  • s (float) – Inverse side gradient (0 = rectangular channel)

  • n (float) – Manning’s roughness coefficient

  • max_flow_depth (float) – Maximum flow depth

bottom_elev: float = 0.0#
B0: float = 0.0#
s: float = 0.0#
n: float = 0.04#
max_flow_depth: float = 10.0#
__init__(bottom_elev=0.0, B0=0.0, s=0.0, n=0.04, max_flow_depth=10.0)#
class pyiwfm.components.stream.StrmEvapNodeSpec(node_id, et_column=0, area_column=0)[source]#

Bases: object

Per-node stream evaporation column pointers.

Variables:

  • node_id (int) – Stream node ID

  • et_column (int) – Column index in ET data file

  • area_column (int) – Column index in stream surface area file

node_id: int#
et_column: int = 0#
area_column: int = 0#
__init__(node_id, et_column=0, area_column=0)#
class pyiwfm.components.stream.StreamRating(stages, flows)[source]#

Bases: object

Stage-discharge rating curve for a stream node.

Variables:
stages: ndarray[tuple[Any, ...], dtype[float64]]#
flows: ndarray[tuple[Any, ...], dtype[float64]]#
__post_init__()[source]#

Validate rating curve data.

get_flow(stage)[source]#

Interpolate flow from stage.

Parameters:

stage (float) – Water level

Returns:

Interpolated flow rate

Return type:

float

get_stage(flow)[source]#

Interpolate stage from flow.

Parameters:

flow (float) – Flow rate

Returns:

Interpolated water level

Return type:

float

__init__(stages, flows)#
class pyiwfm.components.stream.StrmNode(id, x, y, reach_id=0, gw_node=None, bottom_elev=0.0, wetted_perimeter=0.0, upstream_node=None, downstream_node=None, rating=None, conductivity=0.0, bed_thickness=0.0, cross_section=None, initial_condition=0.0)[source]#

Bases: object

A stream node representing a point in the stream network.

Variables:

  • id (int) – Unique node identifier

  • x (float) – X coordinate

  • y (float) – Y coordinate

  • reach_id (int) – ID of the reach this node belongs to

  • gw_node (int | None) – ID of the associated groundwater node (for stream-aquifer interaction)

  • bottom_elev (float) – Stream bed elevation

  • wetted_perimeter (float) – Wetted perimeter for flow calculations

  • upstream_node (int | None) – ID of upstream node

  • downstream_node (int | None) – ID of downstream node

  • rating (pyiwfm.components.stream.StreamRating | None) – Stage-discharge rating curve

id: int#
x: float#
y: float#
reach_id: int = 0#
gw_node: int | None = None#
bottom_elev: float = 0.0#
wetted_perimeter: float = 0.0#
upstream_node: int | None = None#
downstream_node: int | None = None#
rating: StreamRating | None = None#
conductivity: float = 0.0#
bed_thickness: float = 0.0#
cross_section: CrossSectionData | None = None#
initial_condition: float = 0.0#
distance_to(other)[source]#

Calculate distance to another stream node.

__init__(id, x, y, reach_id=0, gw_node=None, bottom_elev=0.0, wetted_perimeter=0.0, upstream_node=None, downstream_node=None, rating=None, conductivity=0.0, bed_thickness=0.0, cross_section=None, initial_condition=0.0)#
class pyiwfm.components.stream.StrmReach(id, upstream_node, downstream_node, name='', nodes=<factory>, outflow_destination=None)[source]#

Bases: object

A stream reach representing a segment of the stream network.

Variables:

  • id (int) – Unique reach identifier

  • name (str) – Descriptive name

  • upstream_node (int) – ID of the most upstream node

  • downstream_node (int) – ID of the most downstream node

  • nodes (list[int]) – List of node IDs in this reach (upstream to downstream order)

  • outflow_destination (tuple[str, int] | None) – Where outflow goes - tuple of (type, id) type: ‘reach’, ‘lake’, ‘boundary’, etc. id: destination ID

id: int#
upstream_node: int#
downstream_node: int#
name: str = ''#
nodes: list[int]#
outflow_destination: tuple[str, int] | None = None#
property n_nodes: int#

Return number of nodes in this reach.

__init__(id, upstream_node, downstream_node, name='', nodes=<factory>, outflow_destination=None)#
class pyiwfm.components.stream.Diversion(id, source_node, destination_type, destination_id, name='', max_rate=inf, priority=99, reach_id=0, max_div_column=0, max_div_fraction=1.0, recoverable_loss_column=0, recoverable_loss_fraction=0.0, non_recoverable_loss_column=0, non_recoverable_loss_fraction=0.0, spill_column=0, spill_fraction=0.0, delivery_dest_type=0, delivery_dest_id=0, delivery_column=0, delivery_fraction=1.0, irrigation_fraction_column=0, adjustment_column=0, element_groups=<factory>, recharge_zones=<factory>, spill_locations=<factory>)[source]#

Bases: object

A stream diversion that removes water from the stream.

Variables:

  • id (int) – Unique diversion identifier

  • name (str) – Descriptive name

  • source_node (int) – Stream node ID where water is diverted

  • destination_type (str) – Type of destination (‘element’, ‘stream_node’, ‘outside’)

  • destination_id (int) – ID of destination

  • max_rate (float) – Maximum diversion rate

  • priority (int) – Allocation priority (lower = higher priority)

id: int#
source_node: int#
destination_type: str#
destination_id: int#
name: str = ''#
max_rate: float = inf#
priority: int = 99#
reach_id: int = 0#
max_div_column: int = 0#
max_div_fraction: float = 1.0#
recoverable_loss_column: int = 0#
recoverable_loss_fraction: float = 0.0#
non_recoverable_loss_column: int = 0#
non_recoverable_loss_fraction: float = 0.0#
spill_column: int = 0#
spill_fraction: float = 0.0#
delivery_dest_type: int = 0#
delivery_dest_id: int = 0#
delivery_column: int = 0#
delivery_fraction: float = 1.0#
irrigation_fraction_column: int = 0#
adjustment_column: int = 0#
element_groups: list[Any]#
recharge_zones: list[Any]#
spill_locations: list[Any]#
property n_delivery_locs: int#

Number of delivery locations (element groups).

__init__(id, source_node, destination_type, destination_id, name='', max_rate=inf, priority=99, reach_id=0, max_div_column=0, max_div_fraction=1.0, recoverable_loss_column=0, recoverable_loss_fraction=0.0, non_recoverable_loss_column=0, non_recoverable_loss_fraction=0.0, spill_column=0, spill_fraction=0.0, delivery_dest_type=0, delivery_dest_id=0, delivery_column=0, delivery_fraction=1.0, irrigation_fraction_column=0, adjustment_column=0, element_groups=<factory>, recharge_zones=<factory>, spill_locations=<factory>)#
class pyiwfm.components.stream.Bypass(id, source_node, destination_node, dest_type=0, name='', capacity=inf, flow_factor=1.0, flow_time_unit='1DAY', spill_factor=1.0, spill_time_unit='1DAY', diversion_column=0, recoverable_loss_fraction=0.0, non_recoverable_loss_fraction=0.0, rating_table_flows=<factory>, rating_table_spills=<factory>, seepage_locations=<factory>)[source]#

Bases: object

A stream bypass that routes water around a section.

Variables:

  • id (int) – Unique bypass identifier

  • name (str) – Descriptive name

  • source_node (int) – Stream node ID where bypass starts

  • destination_node (int) – Stream node ID where bypass ends

  • capacity (float) – Maximum bypass flow capacity

id: int#
source_node: int#
destination_node: int#
dest_type: int = 0#
name: str = ''#
capacity: float = inf#
flow_factor: float = 1.0#
flow_time_unit: str = '1DAY'#
spill_factor: float = 1.0#
spill_time_unit: str = '1DAY'#
diversion_column: int = 0#
recoverable_loss_fraction: float = 0.0#
non_recoverable_loss_fraction: float = 0.0#
rating_table_flows: list[float]#
rating_table_spills: list[float]#
seepage_locations: list[Any]#
property n_seepage_locs: int#

Number of seepage locations.

property has_rating_table: bool#

Whether this bypass uses a rating table (negative IDIVC).

property n_rating_points: int#

Number of rating table points.

__init__(id, source_node, destination_node, dest_type=0, name='', capacity=inf, flow_factor=1.0, flow_time_unit='1DAY', spill_factor=1.0, spill_time_unit='1DAY', diversion_column=0, recoverable_loss_fraction=0.0, non_recoverable_loss_fraction=0.0, rating_table_flows=<factory>, rating_table_spills=<factory>, seepage_locations=<factory>)#
class pyiwfm.components.stream.AppStream(nodes=<factory>, reaches=<factory>, diversions=<factory>, bypasses=<factory>, conductivity_factor=1.0, conductivity_time_unit='', length_factor=1.0, interaction_type=1, evap_area_file='', evap_node_specs=<factory>, ic_type=0, ic_time_unit='', ic_factor=1.0, final_flow_file='', budget_node_count=0, budget_output_file='', budget_node_ids=<factory>, roughness_factor=1.0, cross_section_length_factor=1.0, diversion_element_groups=<factory>, diversion_recharge_zones=<factory>, diversion_spill_zones=<factory>, diversion_has_spills=False, _downstream_map=<factory>, _upstream_map=<factory>)[source]#

Bases: BaseComponent

Stream network application class.

This class manages the complete stream network including nodes, reaches, diversions, and bypasses. It mirrors IWFM’s Package_AppStream.

Variables:
nodes: dict[int, StrmNode]#
reaches: dict[int, StrmReach]#
diversions: dict[int, Diversion]#
bypasses: dict[int, Bypass]#
conductivity_factor: float = 1.0#
conductivity_time_unit: str = ''#
length_factor: float = 1.0#
interaction_type: int = 1#
evap_area_file: str = ''#
evap_node_specs: list[StrmEvapNodeSpec]#
ic_type: int = 0#
ic_time_unit: str = ''#
ic_factor: float = 1.0#
final_flow_file: str = ''#
budget_node_count: int = 0#
budget_output_file: str = ''#
budget_node_ids: list[int]#
roughness_factor: float = 1.0#
cross_section_length_factor: float = 1.0#
diversion_element_groups: list[Any]#
diversion_recharge_zones: list[Any]#
diversion_spill_zones: list[Any]#
diversion_has_spills: bool = False#
property n_items: int#

Return number of stream nodes (primary entities).

property n_nodes: int#

Return number of stream nodes.

property n_reaches: int#

Return number of reaches.

property n_diversions: int#

Return number of diversions.

property n_bypasses: int#

Return number of bypasses.

add_node(node)[source]#

Add a stream node to the network.

add_reach(reach)[source]#

Add a reach to the network.

add_diversion(diversion)[source]#

Add a diversion to the network.

add_bypass(bypass)[source]#

Add a bypass to the network.

get_node(node_id)[source]#

Get a stream node by ID.

get_reach(reach_id)[source]#

Get a reach by ID.

get_diversion(div_id)[source]#

Get a diversion by ID.

get_bypass(bypass_id)[source]#

Get a bypass by ID.

get_nodes_in_reach(reach_id)[source]#

Get all nodes in a reach.

build_connectivity()[source]#

Build node connectivity maps from node data.

get_downstream_node(node_id)[source]#

Get the downstream node ID for a given node.

get_upstream_node(node_id)[source]#

Get the upstream node ID for a given node.

get_reach_length(reach_id)[source]#

Calculate the length of a reach.

Parameters:

reach_id (int) – Reach ID

Returns:

Total length of the reach

Return type:

float

get_total_length()[source]#

Calculate total length of all reaches.

iter_nodes()[source]#

Iterate over nodes in ID order.

iter_reaches()[source]#

Iterate over reaches in ID order.

validate()[source]#

Validate the stream network.

Raises:

ComponentError – If network is invalid

get_reach_segments(reach_id)[source]#

Get all line segments for a reach.

Parameters:

reach_id (int) – The reach ID.

Returns:

List of segments, each as ((x1, y1), (x2, y2)).

Return type:

list of tuple

detect_crossings()[source]#

Detect all crossing (intersecting) reach segments in the network.

Stream reaches can share endpoints (confluences/junctions) but cannot cross each other geometrically. This method identifies all such invalid crossings.

Returns:

List of detected crossings. Empty if no crossings found.

Return type:

list of ReachCrossing

Examples

>>> stream = AppStream()
>>> # ... add nodes and reaches ...
>>> crossings = stream.detect_crossings()
>>> if crossings:
...     print(f"Found {len(crossings)} crossing(s)")
...     for c in crossings:
...         print(f"  {c}")
validate_no_crossings(raise_on_error=True)[source]#

Validate that no stream reaches cross each other.

Stream reaches can share endpoints (confluences) but cannot cross (intersect at non-endpoint locations).

Parameters:

raise_on_error (bool, default True) – If True, raise ComponentError when crossings are found. If False, just return the list of crossings.

Returns:

List of detected crossings.

Return type:

list of ReachCrossing

Raises:

ComponentError – If raise_on_error is True and crossings are found.

Examples

>>> stream.validate_no_crossings()  # Raises if crossings found

>>> crossings = stream.validate_no_crossings(raise_on_error=False)
>>> if crossings:
...     print(f"Found {len(crossings)} invalid crossing(s)")
to_arrays()[source]#

Convert stream network to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[float64]]]

classmethod from_arrays(node_ids, x, y, reach_ids, gw_nodes=None)[source]#

Create stream network from arrays.

Parameters:
Returns:

AppStream instance

Return type:

AppStream

__init__(nodes=<factory>, reaches=<factory>, diversions=<factory>, bypasses=<factory>, conductivity_factor=1.0, conductivity_time_unit='', length_factor=1.0, interaction_type=1, evap_area_file='', evap_node_specs=<factory>, ic_type=0, ic_time_unit='', ic_factor=1.0, final_flow_file='', budget_node_count=0, budget_output_file='', budget_node_ids=<factory>, roughness_factor=1.0, cross_section_length_factor=1.0, diversion_element_groups=<factory>, diversion_recharge_zones=<factory>, diversion_spill_zones=<factory>, diversion_has_spills=False, _downstream_map=<factory>, _upstream_map=<factory>)#

Lake Module#

The lake module contains classes for representing lakes and reservoirs.

Lake component classes for IWFM models.

This module provides classes for representing lakes, including lake elements, outflows, rating curves, and the main lake application class. It mirrors IWFM’s Package_AppLake.

class pyiwfm.components.lake.LakeRating(elevations, areas, volumes)[source]#

Bases: object

Elevation-area-volume rating curve for a lake.

Variables:
elevations: ndarray[tuple[Any, ...], dtype[float64]]#
areas: ndarray[tuple[Any, ...], dtype[float64]]#
volumes: ndarray[tuple[Any, ...], dtype[float64]]#
__post_init__()[source]#

Validate rating curve data.

get_area(elevation)[source]#

Interpolate surface area from elevation.

Parameters:

elevation (float) – Water surface elevation

Returns:

Interpolated surface area

Return type:

float

get_volume(elevation)[source]#

Interpolate storage volume from elevation.

Parameters:

elevation (float) – Water surface elevation

Returns:

Interpolated storage volume

Return type:

float

get_elevation(volume)[source]#

Interpolate elevation from storage volume.

Parameters:

volume (float) – Storage volume

Returns:

Interpolated water surface elevation

Return type:

float

__init__(elevations, areas, volumes)#
class pyiwfm.components.lake.LakeElement(element_id, lake_id, fraction=1.0)[source]#

Bases: object

An element that is part of a lake.

Variables:

  • element_id (int) – Element ID

  • lake_id (int) – ID of the lake this element belongs to

  • fraction (float) – Fraction of element covered by lake (0-1)

element_id: int#
lake_id: int#
fraction: float = 1.0#
__init__(element_id, lake_id, fraction=1.0)#
class pyiwfm.components.lake.LakeOutflow(lake_id, destination_type, destination_id, max_rate=inf)[source]#

Bases: object

Lake outflow configuration.

Variables:

  • lake_id (int) – ID of the source lake

  • destination_type (str) – Type of destination (‘stream’, ‘lake’, ‘outside’)

  • destination_id (int) – ID of destination (stream node or lake)

  • max_rate (float) – Maximum outflow rate

lake_id: int#
destination_type: str#
destination_id: int#
max_rate: float = inf#
__init__(lake_id, destination_type, destination_id, max_rate=inf)#
class pyiwfm.components.lake.Lake(id, name='', max_elevation=inf, initial_storage=0.0, initial_elevation=280.0, elements=<factory>, gw_nodes=<factory>, rating=None, outflow=None, bed_conductivity=2.0, bed_thickness=1.0, et_column=7, precip_column=2, max_elev_column=0, outflow_rating_elevations=<factory>, outflow_rating_flows=<factory>)[source]#

Bases: object

A lake in the model domain.

Variables:

  • id (int) – Unique lake identifier

  • name (str) – Descriptive name

  • max_elevation (float) – Maximum water surface elevation

  • initial_storage (float) – Initial storage volume

  • elements (list[int]) – List of element IDs that make up the lake

  • gw_nodes (list[int]) – List of groundwater node IDs for lake-aquifer interaction

  • rating (pyiwfm.components.lake.LakeRating | None) – Elevation-area-volume rating curve

  • outflow (pyiwfm.components.lake.LakeOutflow | None) – Outflow configuration

id: int#
name: str = ''#
max_elevation: float = inf#
initial_storage: float = 0.0#
initial_elevation: float = 280.0#
elements: list[int]#
gw_nodes: list[int]#
rating: LakeRating | None = None#
outflow: LakeOutflow | None = None#
bed_conductivity: float = 2.0#
bed_thickness: float = 1.0#
et_column: int = 7#
precip_column: int = 2#
max_elev_column: int = 0#
outflow_rating_elevations: list[float]#
outflow_rating_flows: list[float]#
property n_elements: int#

Return number of elements in this lake.

__init__(id, name='', max_elevation=inf, initial_storage=0.0, initial_elevation=280.0, elements=<factory>, gw_nodes=<factory>, rating=None, outflow=None, bed_conductivity=2.0, bed_thickness=1.0, et_column=7, precip_column=2, max_elev_column=0, outflow_rating_elevations=<factory>, outflow_rating_flows=<factory>)#
class pyiwfm.components.lake.AppLake(lakes=<factory>, lake_elements=<factory>, current_elevations=<factory>)[source]#

Bases: BaseComponent

Lake application component.

This class manages all lakes in the model domain including lake elements, outflows, and rating curves. It mirrors IWFM’s Package_AppLake.

Variables:
lakes: dict[int, Lake]#
lake_elements: list[LakeElement]#
current_elevations: dict[int, float]#
property n_items: int#

Return number of lakes (primary entities).

property n_lakes: int#

Return number of lakes.

property n_lake_elements: int#

Return number of lake elements.

add_lake(lake)[source]#

Add a lake to the component.

add_lake_element(elem)[source]#

Add a lake element.

get_lake(lake_id)[source]#

Get a lake by ID.

get_elements_for_lake(lake_id)[source]#

Get all elements for a specific lake.

set_elevation(lake_id, elevation)[source]#

Set the current water surface elevation for a lake.

get_elevation(lake_id)[source]#

Get the current water surface elevation for a lake.

get_area(lake_id)[source]#

Get the current surface area for a lake.

get_volume(lake_id)[source]#

Get the current storage volume for a lake.

get_total_area()[source]#

Calculate total surface area of all lakes.

get_total_volume()[source]#

Calculate total storage volume of all lakes.

iter_lakes()[source]#

Iterate over lakes in ID order.

validate()[source]#

Validate the lake component.

Raises:

ComponentError – If component is invalid

to_arrays()[source]#

Convert lake data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[float64]]]

classmethod from_arrays(lake_ids, names, max_elevations=None)[source]#

Create lake component from arrays.

Parameters:
Returns:

AppLake instance

Return type:

AppLake

__init__(lakes=<factory>, lake_elements=<factory>, current_elevations=<factory>)#

Root Zone Module#

The root zone module contains classes for representing land use and root zone processes.

Root zone component classes for IWFM models.

This module provides classes for representing the root zone including land use types, crop parameters, soil properties, and water budget calculations. It mirrors IWFM’s Package_RootZone.

class pyiwfm.components.rootzone.LandUseType(*values)[source]#

Bases: Enum

Land use type enumeration.

AGRICULTURAL = 'agricultural'#
URBAN = 'urban'#
NATIVE_RIPARIAN = 'native_riparian'#
WATER = 'water'#
class pyiwfm.components.rootzone.CropType(id, name='', root_depth=0.0, kc=1.0, monthly_kc=None)[source]#

Bases: object

Crop type definition with water requirements.

Variables:

  • id (int) – Unique crop identifier

  • name (str) – Crop name

  • root_depth (float) – Root zone depth (length units)

  • kc (float) – Crop coefficient for ET calculation

  • monthly_kc (numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.float64]] | None) – Optional monthly crop coefficients (12 values)

id: int#
name: str = ''#
root_depth: float = 0.0#
kc: float = 1.0#
monthly_kc: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
get_kc(month=0)[source]#

Get crop coefficient for a month.

Parameters:

month (int) – Month (1-12), or 0 for annual kc

Returns:

Crop coefficient

Return type:

float

__init__(id, name='', root_depth=0.0, kc=1.0, monthly_kc=None)#
class pyiwfm.components.rootzone.SoilParameters(porosity, field_capacity, wilting_point, saturated_kv, lambda_param=0.5, kunsat_method=2, k_ponded=-1.0, capillary_rise=0.0, precip_column=1, precip_factor=1.0, generic_moisture_column=0)[source]#

Bases: object

Soil hydraulic parameters.

Variables:

  • porosity (float) – Total porosity (volume fraction)

  • field_capacity (float) – Water content at field capacity

  • wilting_point (float) – Water content at wilting point

  • saturated_kv (float) – Saturated vertical hydraulic conductivity

  • lambda_param (float) – Pore size distribution index

  • kunsat_method (int) – Unsaturated K method (1=Campbell, 2=van Genuchten-Mualem)

  • k_ponded (float) – Ponded hydraulic conductivity (-1 = same as saturated_kv)

  • capillary_rise (float) – Capillary rise depth (v4.1+)

  • precip_column (int) – Column pointer to precipitation data

  • precip_factor (float) – Precipitation scaling factor

  • generic_moisture_column (int) – Column pointer to generic moisture data

porosity: float#
field_capacity: float#
wilting_point: float#
saturated_kv: float#
lambda_param: float = 0.5#
kunsat_method: int = 2#
k_ponded: float = -1.0#
capillary_rise: float = 0.0#
precip_column: int = 1#
precip_factor: float = 1.0#
generic_moisture_column: int = 0#
property available_water: float#

Return available water capacity (field_capacity - wilting_point).

property drainable_porosity: float#

Return drainable porosity (porosity - field_capacity).

__init__(porosity, field_capacity, wilting_point, saturated_kv, lambda_param=0.5, kunsat_method=2, k_ponded=-1.0, capillary_rise=0.0, precip_column=1, precip_factor=1.0, generic_moisture_column=0)#
class pyiwfm.components.rootzone.ElementLandUse(element_id, land_use_type, area, crop_fractions=<factory>, impervious_fraction=0.0)[source]#

Bases: object

Land use assignment for an element.

Variables:

  • element_id (int) – Element ID

  • land_use_type (pyiwfm.components.rootzone.LandUseType) – Type of land use

  • area (float) – Area of this land use in the element

  • crop_fractions (dict[int, float]) – For agricultural, mapping of crop_id to area fraction

  • impervious_fraction (float) – For urban, fraction of impervious surface

element_id: int#
land_use_type: LandUseType#
area: float#
crop_fractions: dict[int, float]#
impervious_fraction: float = 0.0#
__init__(element_id, land_use_type, area, crop_fractions=<factory>, impervious_fraction=0.0)#
class pyiwfm.components.rootzone.RootZone(n_elements, n_layers, crop_types=<factory>, soil_params=<factory>, element_landuse=<factory>, soil_moisture=None, nonponded_config=None, ponded_config=None, urban_config=None, native_riparian_config=None, surface_flow_destinations=<factory>, nonponded_area_file=None, ponded_area_file=None, urban_area_file=None, native_area_file=None, surface_flow_dest_ag=<factory>, surface_flow_dest_urban_in=<factory>, surface_flow_dest_urban_out=<factory>, surface_flow_dest_nvrv=<factory>)[source]#

Bases: BaseComponent

Root zone application component.

This class manages land use, soil parameters, and soil moisture for the model domain. It mirrors IWFM’s Package_RootZone.

Variables:
n_elements: int#
n_layers: int#
crop_types: dict[int, CropType]#
soil_params: dict[int, SoilParameters]#
element_landuse: list[ElementLandUse]#
soil_moisture: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
nonponded_config: Any = None#
ponded_config: Any = None#
urban_config: Any = None#
native_riparian_config: Any = None#
surface_flow_destinations: dict[int, tuple[int, int]]#
nonponded_area_file: Path | None = None#
ponded_area_file: Path | None = None#
urban_area_file: Path | None = None#
native_area_file: Path | None = None#
surface_flow_dest_ag: dict[int, tuple[int, int]]#
surface_flow_dest_urban_in: dict[int, tuple[int, int]]#
surface_flow_dest_urban_out: dict[int, tuple[int, int]]#
surface_flow_dest_nvrv: dict[int, tuple[int, int]]#
property n_items: int#

Return number of elements (primary entities).

property n_crop_types: int#

Return number of crop types.

add_crop_type(crop)[source]#

Add a crop type definition.

get_crop_type(crop_id)[source]#

Get a crop type by ID.

set_soil_parameters(element_id, params)[source]#

Set soil parameters for an element.

get_soil_parameters(element_id)[source]#

Get soil parameters for an element.

add_element_landuse(elu)[source]#

Add an element land use assignment.

get_landuse_for_element(element_id)[source]#

Get all land use assignments for an element.

get_total_area(land_use_type)[source]#

Calculate total area for a land use type.

set_soil_moisture(moisture)[source]#

Set the soil moisture array.

get_soil_moisture(element, layer)[source]#

Get soil moisture at a specific element and layer.

iter_elements_with_landuse()[source]#

Iterate over element IDs that have land use assignments.

load_land_use_from_arrays(snapshot)[source]#

Populate element_landuse from pre-aggregated snapshot data.

This is used by the web viewer when the HDF5 area manager has already loaded and aggregated the area data.

Parameters:

snapshot (dict[int, dict[str, Any]]) – Dict mapping element_id to a dict with keys fractions (dict of land-use type -> fraction), total_area (float), and dominant (str).

load_land_use_snapshot(timestep=0)[source]#

Read land use areas for a single timestep from area data files.

Populates element_landuse from the IWFM area time-series files referenced by nonponded_area_file, ponded_area_file, urban_area_file, and native_area_file.

Parameters:

timestep (int) – Zero-based timestep index to read.

validate()[source]#

Validate the root zone component.

Raises:

ComponentError – If component is invalid

to_arrays()[source]#

Convert root zone data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[float64]]]

classmethod from_arrays(n_elements, n_layers, soil_moisture=None)[source]#

Create root zone component from arrays.

Parameters:

  • n_elements (int) – Number of elements

  • n_layers (int) – Number of soil layers

  • soil_moisture (ndarray[tuple[Any, ...], dtype[float64]] | None) – Soil moisture array (optional)

Returns:

RootZone instance

Return type:

RootZone

__init__(n_elements, n_layers, crop_types=<factory>, soil_params=<factory>, element_landuse=<factory>, soil_moisture=None, nonponded_config=None, ponded_config=None, urban_config=None, native_riparian_config=None, surface_flow_destinations=<factory>, nonponded_area_file=None, ponded_area_file=None, urban_area_file=None, native_area_file=None, surface_flow_dest_ag=<factory>, surface_flow_dest_urban_in=<factory>, surface_flow_dest_urban_out=<factory>, surface_flow_dest_nvrv=<factory>)#

Small Watershed Module#

The small watershed module contains classes for representing small watershed units with integrated root zone and aquifer processes.

Small Watershed component classes for IWFM models.

This module provides classes for representing small watersheds, including watershed units with root zone and aquifer parameters, and the main application class. It mirrors IWFM’s Package_AppSmallWatershed.

class pyiwfm.components.small_watershed.WatershedGWNode(gw_node_id=0, max_perc_rate=0.0, is_baseflow=False, layer=0)[source]#

Bases: object

Groundwater node connection for a small watershed.

Variables:

  • gw_node_id (int) – Groundwater node ID (1-based)

  • max_perc_rate (float) – Maximum percolation rate (positive) or baseflow layer (negative)

  • is_baseflow (bool) – Whether this is a baseflow node

  • layer (int) – Baseflow layer (if is_baseflow)

gw_node_id: int = 0#
max_perc_rate: float = 0.0#
is_baseflow: bool = False#
layer: int = 0#
__init__(gw_node_id=0, max_perc_rate=0.0, is_baseflow=False, layer=0)#
class pyiwfm.components.small_watershed.WatershedUnit(id=0, area=0.0, dest_stream_node=0, gw_nodes=<factory>, precip_col=0, precip_factor=1.0, et_col=0, wilting_point=0.0, field_capacity=0.0, total_porosity=0.0, lambda_param=0.0, root_depth=0.0, hydraulic_cond=0.0, kunsat_method=0, curve_number=0.0, gw_threshold=0.0, max_gw_storage=0.0, surface_flow_coeff=0.0, baseflow_coeff=0.0, initial_soil_moisture=0.0, initial_gw_storage=0.0)[source]#

Bases: object

A single small watershed unit.

Variables:

  • id (int) – Watershed unit ID (1-based)

  • area (float) – Watershed area

  • dest_stream_node (int) – Destination stream node for outflow (1-based)

  • gw_nodes (list[pyiwfm.components.small_watershed.WatershedGWNode]) – Connected groundwater nodes

  • precip_col (int) – Precipitation time-series column index

  • precip_factor (float) – Precipitation conversion factor

  • et_col (int) – ET time-series column index

  • wilting_point (float) – Soil wilting point

  • field_capacity (float) – Soil field capacity

  • total_porosity (float) – Total porosity

  • lambda_param (float) – Pore size distribution parameter

  • root_depth (float) – Root zone depth

  • hydraulic_cond (float) – Hydraulic conductivity

  • kunsat_method (int) – Unsaturated K method code

  • curve_number (float) – SCS curve number

  • gw_threshold (float) – Groundwater storage threshold

  • max_gw_storage (float) – Maximum groundwater storage

  • surface_flow_coeff (float) – Surface flow recession coefficient

  • baseflow_coeff (float) – Baseflow recession coefficient

id: int = 0#
area: float = 0.0#
dest_stream_node: int = 0#
gw_nodes: list[WatershedGWNode]#
precip_col: int = 0#
precip_factor: float = 1.0#
et_col: int = 0#
wilting_point: float = 0.0#
field_capacity: float = 0.0#
total_porosity: float = 0.0#
lambda_param: float = 0.0#
root_depth: float = 0.0#
hydraulic_cond: float = 0.0#
kunsat_method: int = 0#
curve_number: float = 0.0#
gw_threshold: float = 0.0#
max_gw_storage: float = 0.0#
surface_flow_coeff: float = 0.0#
baseflow_coeff: float = 0.0#
initial_soil_moisture: float = 0.0#
initial_gw_storage: float = 0.0#
property n_gw_nodes: int#

Return number of connected groundwater nodes.

__init__(id=0, area=0.0, dest_stream_node=0, gw_nodes=<factory>, precip_col=0, precip_factor=1.0, et_col=0, wilting_point=0.0, field_capacity=0.0, total_porosity=0.0, lambda_param=0.0, root_depth=0.0, hydraulic_cond=0.0, kunsat_method=0, curve_number=0.0, gw_threshold=0.0, max_gw_storage=0.0, surface_flow_coeff=0.0, baseflow_coeff=0.0, initial_soil_moisture=0.0, initial_gw_storage=0.0)#
class pyiwfm.components.small_watershed.AppSmallWatershed(watersheds=<factory>, area_factor=1.0, flow_factor=1.0, flow_time_unit='', rz_solver_tolerance=1e-08, rz_max_iterations=2000, rz_length_factor=1.0, rz_cn_factor=1.0, rz_k_factor=1.0, rz_k_time_unit='', aq_gw_factor=1.0, aq_time_factor=1.0, aq_time_unit='', ic_factor=1.0, budget_output_file='', final_results_file='')[source]#

Bases: BaseComponent

Small Watershed application component.

This class manages all small watersheds in the model domain including watershed units with root zone and aquifer parameters. It mirrors IWFM’s Package_AppSmallWatershed.

Variables:

  • watersheds (dict[int, pyiwfm.components.small_watershed.WatershedUnit]) – Dictionary mapping watershed ID to WatershedUnit

  • area_factor (float) – Area conversion factor

  • flow_factor (float) – Flow rate conversion factor

  • flow_time_unit (str) – Time unit for flow rates

  • rz_solver_tolerance (float) – Root zone solver tolerance

  • rz_max_iterations (int) – Root zone solver max iterations

  • rz_length_factor (float) – Root zone length conversion factor

  • rz_cn_factor (float) – Curve number conversion factor

  • rz_k_factor (float) – Hydraulic conductivity conversion factor

  • rz_k_time_unit (str) – Time unit for hydraulic conductivity

  • aq_gw_factor (float) – GW conversion factor

  • aq_time_factor (float) – Time conversion factor

  • aq_time_unit (str) – Time unit for recession coefficients

  • budget_output_file (str) – Path to budget output file

  • final_results_file (str) – Path to final simulation results file

watersheds: dict[int, WatershedUnit]#
area_factor: float = 1.0#
flow_factor: float = 1.0#
flow_time_unit: str = ''#
rz_solver_tolerance: float = 1e-08#
rz_max_iterations: int = 2000#
rz_length_factor: float = 1.0#
rz_cn_factor: float = 1.0#
rz_k_factor: float = 1.0#
rz_k_time_unit: str = ''#
aq_gw_factor: float = 1.0#
aq_time_factor: float = 1.0#
aq_time_unit: str = ''#
ic_factor: float = 1.0#
budget_output_file: str = ''#
final_results_file: str = ''#
property n_items: int#

Return number of watersheds (primary entities).

property n_watersheds: int#

Return number of watersheds.

add_watershed(ws)[source]#

Add a watershed unit to the component.

get_watershed(ws_id)[source]#

Get a watershed unit by ID.

iter_watersheds()[source]#

Iterate over watersheds in ID order.

validate()[source]#

Validate the small watershed component.

Raises:

ComponentError – If component is invalid

classmethod from_config(config)[source]#

Create component from a parsed SmallWatershedMainConfig.

Parameters:

config (SmallWatershedMainConfig) – Parsed configuration from the reader

Returns:

AppSmallWatershed instance

Return type:

AppSmallWatershed

__init__(watersheds=<factory>, area_factor=1.0, flow_factor=1.0, flow_time_unit='', rz_solver_tolerance=1e-08, rz_max_iterations=2000, rz_length_factor=1.0, rz_cn_factor=1.0, rz_k_factor=1.0, rz_k_time_unit='', aq_gw_factor=1.0, aq_time_factor=1.0, aq_time_unit='', ic_factor=1.0, budget_output_file='', final_results_file='')#

Unsaturated Zone Module#

The unsaturated zone module contains classes for representing unsaturated zone layers and element-level soil moisture processes.

Unsaturated Zone component classes for IWFM models.

This module provides classes for representing the unsaturated (vadose) zone, including per-element layer properties and initial moisture conditions. It mirrors IWFM’s Package_AppUnsatZone.

class pyiwfm.components.unsaturated_zone.UnsatZoneLayer(thickness_max=0.0, total_porosity=0.0, lambda_param=0.0, hyd_cond=0.0, kunsat_method=0)[source]#

Bases: object

Per-layer unsaturated zone properties for a single element.

Variables:

  • thickness_max (float) – Maximum layer thickness

  • total_porosity (float) – Total porosity

  • lambda_param (float) – Pore size distribution parameter

  • hyd_cond (float) – Saturated hydraulic conductivity

  • kunsat_method (int) – Unsaturated K method code

thickness_max: float = 0.0#
total_porosity: float = 0.0#
lambda_param: float = 0.0#
hyd_cond: float = 0.0#
kunsat_method: int = 0#
__init__(thickness_max=0.0, total_porosity=0.0, lambda_param=0.0, hyd_cond=0.0, kunsat_method=0)#
class pyiwfm.components.unsaturated_zone.UnsatZoneElement(element_id=0, layers=<factory>, initial_moisture=None)[source]#

Bases: object

Per-element unsaturated zone data.

Variables:
element_id: int = 0#
layers: list[UnsatZoneLayer]#
initial_moisture: ndarray[tuple[Any, ...], dtype[float64]] | None = None#
property n_layers: int#

Return number of unsaturated zone layers.

__init__(element_id=0, layers=<factory>, initial_moisture=None)#
class pyiwfm.components.unsaturated_zone.AppUnsatZone(n_layers=0, elements=<factory>, solver_tolerance=1e-08, max_iterations=2000, coord_factor=1.0, thickness_factor=1.0, hyd_cond_factor=1.0, time_unit='', n_parametric_grids=0, budget_file='', zbudget_file='', final_results_file='')[source]#

Bases: BaseComponent

Unsaturated Zone application component.

This class manages the vadose zone modeling parameters for all elements. It mirrors IWFM’s Package_AppUnsatZone.

Variables:

  • n_layers (int) – Number of unsaturated zone layers

  • elements (dict[int, pyiwfm.components.unsaturated_zone.UnsatZoneElement]) – Dictionary mapping element ID to UnsatZoneElement

  • solver_tolerance (float) – Solver convergence tolerance

  • max_iterations (int) – Maximum solver iterations

  • coord_factor (float) – Conversion factor for x-y coordinates

  • thickness_factor (float) – Conversion factor for layer thickness

  • hyd_cond_factor (float) – Conversion factor for hydraulic conductivity

  • time_unit (str) – Time unit for hydraulic conductivity

  • n_parametric_grids (int) – Number of parametric grids (0=direct input)

  • budget_file (str) – Path to HDF5 budget output file

  • zbudget_file (str) – Path to HDF5 zone budget output file

  • final_results_file (str) – Path to final simulation results file

n_layers: int = 0#
elements: dict[int, UnsatZoneElement]#
solver_tolerance: float = 1e-08#
max_iterations: int = 2000#
coord_factor: float = 1.0#
thickness_factor: float = 1.0#
hyd_cond_factor: float = 1.0#
time_unit: str = ''#
n_parametric_grids: int = 0#
budget_file: str = ''#
zbudget_file: str = ''#
final_results_file: str = ''#
property n_items: int#

Return number of elements (primary entities).

property n_elements: int#

Return number of elements with unsaturated zone data.

add_element(elem)[source]#

Add an element to the component.

get_element(element_id)[source]#

Get an element by ID.

iter_elements()[source]#

Iterate over elements in ID order.

validate()[source]#

Validate the unsaturated zone component.

Raises:

ComponentError – If component is invalid

classmethod from_config(config)[source]#

Create component from a parsed UnsatZoneMainConfig.

Parameters:

config (UnsatZoneMainConfig) – Parsed configuration from the reader

Returns:

AppUnsatZone instance

Return type:

AppUnsatZone

__init__(n_layers=0, elements=<factory>, solver_tolerance=1e-08, max_iterations=2000, coord_factor=1.0, thickness_factor=1.0, hyd_cond_factor=1.0, time_unit='', n_parametric_grids=0, budget_file='', zbudget_file='', final_results_file='')#

Connectors Module#

The connectors module contains classes for representing interactions between different model components.

Component connectors for IWFM models.

This module provides classes for representing connections between model components: stream-groundwater, lake-groundwater, and stream-lake interactions. It mirrors IWFM’s Package_ComponentConnectors.

class pyiwfm.components.connectors.StreamGWConnection(stream_node_id, gw_node_id, layer=1, conductance=0.0, stream_bed_elev=0.0, stream_bed_thickness=0.0)[source]#

Bases: object

A connection between a stream node and groundwater node.

Variables:

  • stream_node_id (int) – Stream node ID

  • gw_node_id (int) – Groundwater node ID

  • layer (int) – Aquifer layer for interaction

  • conductance (float) – Streambed conductance

  • stream_bed_elev (float) – Stream bed elevation

  • stream_bed_thickness (float) – Stream bed thickness

stream_node_id: int#
gw_node_id: int#
layer: int = 1#
conductance: float = 0.0#
stream_bed_elev: float = 0.0#
stream_bed_thickness: float = 0.0#
__init__(stream_node_id, gw_node_id, layer=1, conductance=0.0, stream_bed_elev=0.0, stream_bed_thickness=0.0)#
class pyiwfm.components.connectors.StreamGWConnector(connections=<factory>)[source]#

Bases: object

Connector managing stream-groundwater interactions.

This class manages all connections between stream nodes and groundwater nodes, calculating exchange flows based on head differences and conductances.

Variables:

connections (list[pyiwfm.components.connectors.StreamGWConnection]) – List of stream-groundwater connections

connections: list[StreamGWConnection]#
property n_connections: int#

Return number of connections.

add_connection(conn)[source]#

Add a connection.

get_connections_for_stream_node(stream_node_id)[source]#

Get all connections for a stream node.

get_connections_for_gw_node(gw_node_id)[source]#

Get all connections for a groundwater node.

calculate_flow(stream_node_id, stream_stage, gw_head)[source]#

Calculate stream-aquifer exchange flow for a stream node.

Positive flow = stream to aquifer (losing stream) Negative flow = aquifer to stream (gaining stream)

Parameters:

  • stream_node_id (int) – Stream node ID

  • stream_stage (float) – Water surface elevation in stream

  • gw_head (float) – Groundwater head at connected node

Returns:

Exchange flow rate

Return type:

float

calculate_total_exchange(stream_stages, gw_heads)[source]#

Calculate total stream-aquifer exchange.

Parameters:

  • stream_stages (dict[int, float]) – Dictionary of stream node ID to stage

  • gw_heads (dict[int, float]) – Dictionary of GW node ID to head

Returns:

Total exchange flow (positive = net to aquifer)

Return type:

float

validate()[source]#

Validate the connector.

to_arrays()[source]#

Convert connector data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[Any]]]

__init__(connections=<factory>)#
class pyiwfm.components.connectors.LakeGWConnection(lake_id, gw_node_id, layer=1, conductance=0.0, lake_bed_elev=0.0, lake_bed_thickness=0.0)[source]#

Bases: object

A connection between a lake and groundwater node.

Variables:

  • lake_id (int) – Lake ID

  • gw_node_id (int) – Groundwater node ID

  • layer (int) – Aquifer layer for interaction

  • conductance (float) – Lake bed conductance

  • lake_bed_elev (float) – Lake bed elevation

  • lake_bed_thickness (float) – Lake bed thickness

lake_id: int#
gw_node_id: int#
layer: int = 1#
conductance: float = 0.0#
lake_bed_elev: float = 0.0#
lake_bed_thickness: float = 0.0#
__init__(lake_id, gw_node_id, layer=1, conductance=0.0, lake_bed_elev=0.0, lake_bed_thickness=0.0)#
class pyiwfm.components.connectors.LakeGWConnector(connections=<factory>)[source]#

Bases: object

Connector managing lake-groundwater interactions.

This class manages all connections between lakes and groundwater nodes, calculating exchange flows based on head differences.

Variables:

connections (list[pyiwfm.components.connectors.LakeGWConnection]) – List of lake-groundwater connections

connections: list[LakeGWConnection]#
property n_connections: int#

Return number of connections.

add_connection(conn)[source]#

Add a connection.

get_connections_for_lake(lake_id)[source]#

Get all connections for a lake.

get_connections_for_gw_node(gw_node_id)[source]#

Get all connections for a groundwater node.

calculate_flow(lake_id, lake_stage, gw_head)[source]#

Calculate lake-aquifer exchange flow.

Positive flow = lake to aquifer (losing lake) Negative flow = aquifer to lake (gaining lake)

Parameters:

  • lake_id (int) – Lake ID

  • lake_stage (float) – Water surface elevation in lake

  • gw_head (float) – Groundwater head at connected node

Returns:

Exchange flow rate

Return type:

float

calculate_total_exchange(lake_stages, gw_heads)[source]#

Calculate total lake-aquifer exchange.

Parameters:

  • lake_stages (dict[int, float]) – Dictionary of lake ID to stage

  • gw_heads (dict[int, float]) – Dictionary of GW node ID to head

Returns:

Total exchange flow (positive = net to aquifer)

Return type:

float

validate()[source]#

Validate the connector.

to_arrays()[source]#

Convert connector data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[Any]]]

__init__(connections=<factory>)#
class pyiwfm.components.connectors.StreamLakeConnection(stream_node_id, lake_id, connection_type, max_flow=inf)[source]#

Bases: object

A connection between a stream node and a lake.

Variables:

  • stream_node_id (int) – Stream node ID

  • lake_id (int) – Lake ID

  • connection_type (str) – Type of connection (‘inflow’ or ‘outflow’)

  • max_flow (float) – Maximum flow rate for outflow connections

stream_node_id: int#
lake_id: int#
connection_type: str#
max_flow: float = inf#
__post_init__()[source]#

Validate connection type.

__init__(stream_node_id, lake_id, connection_type, max_flow=inf)#
class pyiwfm.components.connectors.StreamLakeConnector(connections=<factory>)[source]#

Bases: object

Connector managing stream-lake interactions.

This class manages connections where streams flow into or out of lakes.

Variables:

connections (list[pyiwfm.components.connectors.StreamLakeConnection]) – List of stream-lake connections

connections: list[StreamLakeConnection]#
property n_connections: int#

Return number of connections.

add_connection(conn)[source]#

Add a connection.

get_connections_for_lake(lake_id)[source]#

Get all connections for a lake.

get_connections_for_stream_node(stream_node_id)[source]#

Get all connections for a stream node.

get_inflows_for_lake(lake_id)[source]#

Get inflow connections for a lake.

get_outflows_for_lake(lake_id)[source]#

Get outflow connections for a lake.

validate()[source]#

Validate the connector.

to_arrays()[source]#

Convert connector data to numpy arrays.

Returns:

Dictionary of arrays

Return type:

dict[str, ndarray[tuple[Any, …], dtype[Any]]]

__init__(connections=<factory>)#
On this page