easy_vic_build.tools.dpc_func.dpc_subclass

Concrete data-processing pipelines built on dpc_base.

This module defines level-specific subclasses of easy_vic_build.tools.dpc_func.dpc_base.dataProcess_base. Each subclass registers decorated loading steps that pull source datasets and cache them into the common processing state.

Classes

dataProcess_VIC_level0([load_path, ...])	Load VIC level-0 static inputs on the grid domain.
dataProcess_VIC_level1([load_path, ...])	Load VIC level-1 eco-hydrological attributes on grids.
dataProcess_VIC_level2([load_path, ...])	Load VIC level-2 meteorological forcing data.
dataProcess_VIC_level3([load_path, ...])	Load VIC level-3 basin-level observations and attributes.

class easy_vic_build.tools.dpc_func.dpc_subclass.dataProcess_VIC_level0(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Bases: dataProcess_base

Load VIC level-0 static inputs on the grid domain.

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

load_dem()

Load SRTM DEM values for the current grid shapefile.

Returns:

Dictionary with key "dem" whose value is the updated grid GeoDataFrame containing DEM-related columns.

Return type:

dict

load_soil()

Load CONUS soil attributes for each grid cell.

Returns:

Dictionary with key "soil" whose value is the updated grid GeoDataFrame containing soil columns.

Return type:

dict

__init__(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

aggregate_grid_to_basins()

Aggregate grid-level variables to basin-level summaries.

Notes

This method is intentionally left for subclasses.

clear_data_from_cache(save_names: List[str] | None = None, step_name: str | None = None)

Clear cached entries by key list or clear all entries.

Parameters:

• save_names (list of str, optional) – Keys to remove. If None, all cache entries are removed.

• step_name (str, optional) – Step name to discard from _executed_steps while clearing keys.

discard_step_name(step_name: str)

Mark one step as not executed.

Parameters:

step_name (str) – Step name to remove from _executed_steps.

get_data_from_cache(save_name: str, default: Any | None = None) → Any

Get cached data and its level by key.

Parameters:

• save_name (str) – Cache key to retrieve.

• default (Any, optional) – Value used when the key is not found.

Returns:

(data, data_level) if key exists; otherwise (default, None).

Return type:

tuple

list_cache() → List[str]

List available keys in cache.

Returns:

Current cache key names.

Return type:

list of str

load_basin_shp()

Load basin shapefile-like object from loaddata_kwargs.

Returns:

Dictionary containing key "basin_shp".

Return type:

dict

load_grid_shp()

Load grid shapefile-like object and grid resolution from inputs.

Returns:

Dictionary containing keys "grid_shp" and "grid_res".

Return type:

dict

load_state(load_path: str, reset_on_load_failure: bool = False, **kwargs) → dataProcess_base

Load processor state from a pickle file.

Parameters:

• load_path (str) – State file path.

• reset_on_load_failure (bool, optional) – Whether to reset to a clean state when loading fails.

• **kwargs (dict) – Reserved for compatibility.

Returns:

Current processor instance.

Return type:

dataProcess_base

Raises:

RuntimeError – Raised when loading fails and reset_on_load_failure is False.

loaddata_pipeline(save_path=None, loaddata_kwargs: Dict[str, Dict[str, Any]] | None = None)

Execute all registered steps in dependency order.

Parameters:

• save_path (str, optional) – Path for persisting state after each successful step.

• loaddata_kwargs (dict, optional) – Runtime input dictionary consumed by individual step methods.

merge_basin_data() → geopandas.GeoDataFrame

Merge all cached basin-level outputs into one GeoDataFrame.

Returns:

Basin GeoDataFrame containing original geometry plus all joinable basin-level products in cache.

Return type:

geopandas.GeoDataFrame

merge_grid_data() → geopandas.GeoDataFrame

Merge all cached grid-level outputs into one GeoDataFrame.

Returns:

Grid GeoDataFrame containing original grid fields and appended grid-level variables from cache.

Return type:

geopandas.GeoDataFrame

plot(fig=None, ax=None, grid_shp_kwargs={}, grid_shp_point_kwargs={}, basin_shp_kwargs={})

Plot cached basin and grid geometry.

Parameters:

• fig (matplotlib.figure.Figure, optional) – Existing figure object. A new one is created when omitted.

• ax (matplotlib.axes.Axes, optional) – Existing axes object. A new one is created when omitted.

• grid_shp_kwargs (dict, optional) – Keyword arguments passed to grid_shp.boundary.plot.

• grid_shp_point_kwargs (dict, optional) – Keyword arguments passed to grid_shp["point_geometry"].plot.

• basin_shp_kwargs (dict, optional) – Keyword arguments passed to basin_shp.plot.

Returns:

(fig, ax) with rendered basin/grid layout.

Return type:

tuple

register_processing_step(step_name: str, save_names: str | List[str], data_level: str, func: Callable, dependencies: List[str] | None = None)

Register one processing step in the execution graph.

Parameters:

• step_name (str) – Unique step identifier.

• save_names (str or list of str) – Cache key(s) expected to be produced by func.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• func (Callable) – Step callable with zero arguments. It must return a dict keyed by save_names.

• dependencies (list of str, optional) – Steps that must be executed before this step.

save_data_to_cache(save_name: str, data: Any, data_level: str, step_name: str | None = None) → None

Save data into cache and optionally reopen its step state.

Parameters:

• save_name (str) – Cache key for the data object.

• data (Any) – Data object to cache.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• step_name (str, optional) – Step name to discard from _executed_steps after updating cache.

save_state(save_path: str | None = None) → None

Serialize processor state to a pickle file.

Parameters:

save_path (str, optional) – Output state path. If omitted, self.load_path is used when set.

class easy_vic_build.tools.dpc_func.dpc_subclass.dataProcess_VIC_level1(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Bases: dataProcess_base

Load VIC level-1 eco-hydrological attributes on grids.

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

load_st()

Load ERA5 soil temperature for each grid.

Returns:

Dictionary with key "st" mapping to the updated grid data.

Return type:

dict

load_annual_P()

Load NLDAS annual precipitation for each grid.

Returns:

Dictionary with key "annual_P" mapping to updated grid data.

Return type:

dict

load_lulc()

Load UMD land-cover fractions and classes for each grid.

Returns:

Dictionary with key "lulc" mapping to updated grid data.

Return type:

dict

load_bsa()

Load MODIS black-sky albedo (BSA) climatology for each grid.

Returns:

Dictionary with key "bsa" mapping to updated grid data.

Return type:

dict

load_ndvi()

Load MODIS NDVI climatology for each grid.

Returns:

Dictionary with key "ndvi" mapping to updated grid data.

Return type:

dict

load_lai()

Load MODIS LAI climatology for each grid.

Returns:

Dictionary with key "lai" mapping to updated grid data.

Return type:

dict

__init__(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

aggregate_grid_to_basins()

Aggregate grid-level variables to basin-level summaries.

Notes

This method is intentionally left for subclasses.

clear_data_from_cache(save_names: List[str] | None = None, step_name: str | None = None)

Clear cached entries by key list or clear all entries.

Parameters:

• save_names (list of str, optional) – Keys to remove. If None, all cache entries are removed.

• step_name (str, optional) – Step name to discard from _executed_steps while clearing keys.

discard_step_name(step_name: str)

Mark one step as not executed.

Parameters:

step_name (str) – Step name to remove from _executed_steps.

get_data_from_cache(save_name: str, default: Any | None = None) → Any

Get cached data and its level by key.

Parameters:

• save_name (str) – Cache key to retrieve.

• default (Any, optional) – Value used when the key is not found.

Returns:

(data, data_level) if key exists; otherwise (default, None).

Return type:

tuple

list_cache() → List[str]

List available keys in cache.

Returns:

Current cache key names.

Return type:

list of str

load_basin_shp()

Load basin shapefile-like object from loaddata_kwargs.

Returns:

Dictionary containing key "basin_shp".

Return type:

dict

load_grid_shp()

Load grid shapefile-like object and grid resolution from inputs.

Returns:

Dictionary containing keys "grid_shp" and "grid_res".

Return type:

dict

load_state(load_path: str, reset_on_load_failure: bool = False, **kwargs) → dataProcess_base

Load processor state from a pickle file.

Parameters:

• load_path (str) – State file path.

• reset_on_load_failure (bool, optional) – Whether to reset to a clean state when loading fails.

• **kwargs (dict) – Reserved for compatibility.

Returns:

Current processor instance.

Return type:

dataProcess_base

Raises:

RuntimeError – Raised when loading fails and reset_on_load_failure is False.

loaddata_pipeline(save_path=None, loaddata_kwargs: Dict[str, Dict[str, Any]] | None = None)

Execute all registered steps in dependency order.

Parameters:

• save_path (str, optional) – Path for persisting state after each successful step.

• loaddata_kwargs (dict, optional) – Runtime input dictionary consumed by individual step methods.

merge_basin_data() → geopandas.GeoDataFrame

Merge all cached basin-level outputs into one GeoDataFrame.

Returns:

Basin GeoDataFrame containing original geometry plus all joinable basin-level products in cache.

Return type:

geopandas.GeoDataFrame

merge_grid_data() → geopandas.GeoDataFrame

Merge all cached grid-level outputs into one GeoDataFrame.

Returns:

Grid GeoDataFrame containing original grid fields and appended grid-level variables from cache.

Return type:

geopandas.GeoDataFrame

plot(fig=None, ax=None, grid_shp_kwargs={}, grid_shp_point_kwargs={}, basin_shp_kwargs={})

Plot cached basin and grid geometry.

Parameters:

• fig (matplotlib.figure.Figure, optional) – Existing figure object. A new one is created when omitted.

• ax (matplotlib.axes.Axes, optional) – Existing axes object. A new one is created when omitted.

• grid_shp_kwargs (dict, optional) – Keyword arguments passed to grid_shp.boundary.plot.

• grid_shp_point_kwargs (dict, optional) – Keyword arguments passed to grid_shp["point_geometry"].plot.

• basin_shp_kwargs (dict, optional) – Keyword arguments passed to basin_shp.plot.

Returns:

(fig, ax) with rendered basin/grid layout.

Return type:

tuple

register_processing_step(step_name: str, save_names: str | List[str], data_level: str, func: Callable, dependencies: List[str] | None = None)

Register one processing step in the execution graph.

Parameters:

• step_name (str) – Unique step identifier.

• save_names (str or list of str) – Cache key(s) expected to be produced by func.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• func (Callable) – Step callable with zero arguments. It must return a dict keyed by save_names.

• dependencies (list of str, optional) – Steps that must be executed before this step.

save_data_to_cache(save_name: str, data: Any, data_level: str, step_name: str | None = None) → None

Save data into cache and optionally reopen its step state.

Parameters:

• save_name (str) – Cache key for the data object.

• data (Any) – Data object to cache.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• step_name (str, optional) – Step name to discard from _executed_steps after updating cache.

save_state(save_path: str | None = None) → None

Serialize processor state to a pickle file.

Parameters:

save_path (str, optional) – Output state path. If omitted, self.load_path is used when set.

class easy_vic_build.tools.dpc_func.dpc_subclass.dataProcess_VIC_level2(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Bases: dataProcess_base

Load VIC level-2 meteorological forcing data.

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

load_forcing()

Load NLDAS forcing time series for each grid.

Returns:

Dictionary with key "NLDAS_forcing" mapping to updated grid data.

Return type:

dict

__init__(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

aggregate_grid_to_basins()

Aggregate grid-level variables to basin-level summaries.

Notes

This method is intentionally left for subclasses.

clear_data_from_cache(save_names: List[str] | None = None, step_name: str | None = None)

Clear cached entries by key list or clear all entries.

Parameters:

• save_names (list of str, optional) – Keys to remove. If None, all cache entries are removed.

• step_name (str, optional) – Step name to discard from _executed_steps while clearing keys.

discard_step_name(step_name: str)

Mark one step as not executed.

Parameters:

step_name (str) – Step name to remove from _executed_steps.

get_data_from_cache(save_name: str, default: Any | None = None) → Any

Get cached data and its level by key.

Parameters:

• save_name (str) – Cache key to retrieve.

• default (Any, optional) – Value used when the key is not found.

Returns:

(data, data_level) if key exists; otherwise (default, None).

Return type:

tuple

list_cache() → List[str]

List available keys in cache.

Returns:

Current cache key names.

Return type:

list of str

load_basin_shp()

Load basin shapefile-like object from loaddata_kwargs.

Returns:

Dictionary containing key "basin_shp".

Return type:

dict

load_grid_shp()

Load grid shapefile-like object and grid resolution from inputs.

Returns:

Dictionary containing keys "grid_shp" and "grid_res".

Return type:

dict

load_state(load_path: str, reset_on_load_failure: bool = False, **kwargs) → dataProcess_base

Load processor state from a pickle file.

Parameters:

• load_path (str) – State file path.

• reset_on_load_failure (bool, optional) – Whether to reset to a clean state when loading fails.

• **kwargs (dict) – Reserved for compatibility.

Returns:

Current processor instance.

Return type:

dataProcess_base

Raises:

RuntimeError – Raised when loading fails and reset_on_load_failure is False.

loaddata_pipeline(save_path=None, loaddata_kwargs: Dict[str, Dict[str, Any]] | None = None)

Execute all registered steps in dependency order.

Parameters:

• save_path (str, optional) – Path for persisting state after each successful step.

• loaddata_kwargs (dict, optional) – Runtime input dictionary consumed by individual step methods.

merge_basin_data() → geopandas.GeoDataFrame

Merge all cached basin-level outputs into one GeoDataFrame.

Returns:

Basin GeoDataFrame containing original geometry plus all joinable basin-level products in cache.

Return type:

geopandas.GeoDataFrame

merge_grid_data() → geopandas.GeoDataFrame

Merge all cached grid-level outputs into one GeoDataFrame.

Returns:

Grid GeoDataFrame containing original grid fields and appended grid-level variables from cache.

Return type:

geopandas.GeoDataFrame

plot(fig=None, ax=None, grid_shp_kwargs={}, grid_shp_point_kwargs={}, basin_shp_kwargs={})

Plot cached basin and grid geometry.

Parameters:

• fig (matplotlib.figure.Figure, optional) – Existing figure object. A new one is created when omitted.

• ax (matplotlib.axes.Axes, optional) – Existing axes object. A new one is created when omitted.

• grid_shp_kwargs (dict, optional) – Keyword arguments passed to grid_shp.boundary.plot.

• grid_shp_point_kwargs (dict, optional) – Keyword arguments passed to grid_shp["point_geometry"].plot.

• basin_shp_kwargs (dict, optional) – Keyword arguments passed to basin_shp.plot.

Returns:

(fig, ax) with rendered basin/grid layout.

Return type:

tuple

register_processing_step(step_name: str, save_names: str | List[str], data_level: str, func: Callable, dependencies: List[str] | None = None)

Register one processing step in the execution graph.

Parameters:

• step_name (str) – Unique step identifier.

• save_names (str or list of str) – Cache key(s) expected to be produced by func.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• func (Callable) – Step callable with zero arguments. It must return a dict keyed by save_names.

• dependencies (list of str, optional) – Steps that must be executed before this step.

save_data_to_cache(save_name: str, data: Any, data_level: str, step_name: str | None = None) → None

Save data into cache and optionally reopen its step state.

Parameters:

• save_name (str) – Cache key for the data object.

• data (Any) – Data object to cache.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• step_name (str, optional) – Step name to discard from _executed_steps after updating cache.

save_state(save_path: str | None = None) → None

Serialize processor state to a pickle file.

Parameters:

save_path (str, optional) – Output state path. If omitted, self.load_path is used when set.

class easy_vic_build.tools.dpc_func.dpc_subclass.dataProcess_VIC_level3(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Bases: dataProcess_base

Load VIC level-3 basin-level observations and attributes.

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

load_grid_shp()

Disable grid loading for level-3 workflows.

Level-3 processing operates at basin level only, so this method is a no-op override of the base implementation.

__init__(load_path: str | None = None, reset_on_load_failure=False, **kwargs)

Initialize the processing object and optionally restore saved state.

Parameters:

• load_path (str, optional) – Path to a serialized processor state (pickle file). If provided, the state will be loaded immediately.

• reset_on_load_failure (bool, optional) – If True, reset to a clean state when state loading fails. If False, loading errors raise RuntimeError.

• **kwargs (dict) – Extra keyword arguments forwarded to load_state().

aggregate_grid_to_basins()

Aggregate grid-level variables to basin-level summaries.

Notes

This method is intentionally left for subclasses.

clear_data_from_cache(save_names: List[str] | None = None, step_name: str | None = None)

Clear cached entries by key list or clear all entries.

Parameters:

• save_names (list of str, optional) – Keys to remove. If None, all cache entries are removed.

• step_name (str, optional) – Step name to discard from _executed_steps while clearing keys.

discard_step_name(step_name: str)

Mark one step as not executed.

Parameters:

step_name (str) – Step name to remove from _executed_steps.

get_data_from_cache(save_name: str, default: Any | None = None) → Any

Get cached data and its level by key.

Parameters:

• save_name (str) – Cache key to retrieve.

• default (Any, optional) – Value used when the key is not found.

Returns:

(data, data_level) if key exists; otherwise (default, None).

Return type:

tuple

list_cache() → List[str]

List available keys in cache.

Returns:

Current cache key names.

Return type:

list of str

load_basin_shp()

Load basin shapefile-like object from loaddata_kwargs.

Returns:

Dictionary containing key "basin_shp".

Return type:

dict

load_state(load_path: str, reset_on_load_failure: bool = False, **kwargs) → dataProcess_base

Load processor state from a pickle file.

Parameters:

• load_path (str) – State file path.

• reset_on_load_failure (bool, optional) – Whether to reset to a clean state when loading fails.

• **kwargs (dict) – Reserved for compatibility.

Returns:

Current processor instance.

Return type:

dataProcess_base

Raises:

RuntimeError – Raised when loading fails and reset_on_load_failure is False.

load_streamflow()

Load CAMELS streamflow for the configured basin and period.

Returns:

Dictionary with key "streamflow" mapping to basin-level data.

Return type:

dict

loaddata_pipeline(save_path=None, loaddata_kwargs: Dict[str, Dict[str, Any]] | None = None)

Execute all registered steps in dependency order.

Parameters:

• save_path (str, optional) – Path for persisting state after each successful step.

• loaddata_kwargs (dict, optional) – Runtime input dictionary consumed by individual step methods.

merge_basin_data() → geopandas.GeoDataFrame

Merge all cached basin-level outputs into one GeoDataFrame.

Returns:

Basin GeoDataFrame containing original geometry plus all joinable basin-level products in cache.

Return type:

geopandas.GeoDataFrame

merge_grid_data() → geopandas.GeoDataFrame

Merge all cached grid-level outputs into one GeoDataFrame.

Returns:

Grid GeoDataFrame containing original grid fields and appended grid-level variables from cache.

Return type:

geopandas.GeoDataFrame

plot(fig=None, ax=None, grid_shp_kwargs={}, grid_shp_point_kwargs={}, basin_shp_kwargs={})

Plot cached basin and grid geometry.

Parameters:

• fig (matplotlib.figure.Figure, optional) – Existing figure object. A new one is created when omitted.

• ax (matplotlib.axes.Axes, optional) – Existing axes object. A new one is created when omitted.

• grid_shp_kwargs (dict, optional) – Keyword arguments passed to grid_shp.boundary.plot.

• grid_shp_point_kwargs (dict, optional) – Keyword arguments passed to grid_shp["point_geometry"].plot.

• basin_shp_kwargs (dict, optional) – Keyword arguments passed to basin_shp.plot.

Returns:

(fig, ax) with rendered basin/grid layout.

Return type:

tuple

register_processing_step(step_name: str, save_names: str | List[str], data_level: str, func: Callable, dependencies: List[str] | None = None)

Register one processing step in the execution graph.

Parameters:

• step_name (str) – Unique step identifier.

• save_names (str or list of str) – Cache key(s) expected to be produced by func.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• func (Callable) – Step callable with zero arguments. It must return a dict keyed by save_names.

• dependencies (list of str, optional) – Steps that must be executed before this step.

save_data_to_cache(save_name: str, data: Any, data_level: str, step_name: str | None = None) → None

Save data into cache and optionally reopen its step state.

Parameters:

• save_name (str) – Cache key for the data object.

• data (Any) – Data object to cache.

• data_level (str) – Data scope label, usually "basin_level" or "grid_level".

• step_name (str, optional) – Step name to discard from _executed_steps after updating cache.

save_state(save_path: str | None = None) → None

Serialize processor state to a pickle file.

Parameters:

save_path (str, optional) – Output state path. If omitted, self.load_path is used when set.

load_basin_attribute()

Load CAMELS static attributes for the configured basin.

Returns:

Dictionary with key "basin_attribute" mapping to basin-level attribute data.

Return type:

dict