hydrocnhs package

Subpackages

Submodules

hydrocnhs.abm module

class hydrocnhs.abm.Base(**kwargs)

Bases: object

Agent_type class’s available items: * name: agent’s name. * config: agent’s configuration dictionary the model file (.yaml). * start_date: datetime object. * data_length: length of the simulation. * data_collector: a container to store simulated data. * rn_gen: random number generator to ensure reproducibility (e.g., * self.rn_gen.random()). Note that do NOT set a global random seed in * this module! All type of random number should be created by “rn_gen.” * dm: decision making object if assigned in the model file (.yaml).

Decision-making class’s available items:

  • start_date: datetime object.

  • data_length: length of the simulation.

  • abm: the ABM configuration dictionary from the model file (.yaml).

  • data_collector: a container to store simulated data.

  • rn_gen: random number generator to ensure reproducibility (e.g.,

  • self.rn_gen.random()). Note that do NOT set a global random seed in

  • this module! All type of random number should be created by “rn_gen.

class hydrocnhs.abm.agent

Bases: object

hydrocnhs.abm.read_factor(ag_config, outlet)

Read factor from agent’s ag_config at a given outlet.

The function will automatically parse the list-like value assignment. e.g., [ReturnFactor, 0, Plus].

Parameters:

  • ag_config (dict) – Agent’s configuration.

  • outlet (str) – Outlet name.

hydrocnhs.abm_script module

hydrocnhs.abm_script.add_agt_class(agt_type)
hydrocnhs.abm_script.add_dm_class(dm_type, is_institution=False)

hydrocnhs.data_collector module

class hydrocnhs.data_collector.Data_collector

Bases: object

add_field(field, data_type={}, desc=None, unit=None, check_exist=True)

Add a field to the data collector.

A field can be a dictionary or a list. A Data_collector object cannot have duplicated field name.

Parameters:

  • field (str) – Field name. Cannot have space in a field name.

  • data_type (dict, optional) – Data type of the field (e.g., {} and []), by default {}. User can also populate the field by directly assigning data here.

  • desc (str) – Field description.

  • unit – Unit of the field.

  • check_exist (bool, optional) – If Ture, check the given field name is not existed before adding, by default True.

del_field(field)

Delete a field from the data collector.

Parameters:

field (str) – Field name.

get_dict(copy=False)

Get data collector in dictionary format.

Note that if copy=False, any modification on a variable assigned with the returned dictionary will also affect the data stored in the data collector object.

Parameters:

copy (str) – If true, a copy of dictionary will be returned, else a pointer will be returned. Default False.

Returns:

A dictionary contains all fields.

Return type:

dict

get_field(field, copy=False)

Get a field.

This function create a shortcut to access a field. Namely, changes of a local variable of get_field() will be accumulated back to the original data_collector. copy=Ture if a copy of a field is needed.

Parameters:

  • field (str) – Field name.

  • copy (bool) – If true, create a copy of a field, which has seperate storage

  • Otherwise (pointer than the original data_collector.)

  • shortcut (return a)

  • data_collector. (of a field to the original)

Return type:

Assigned field type.

list_fields()

Print available fields in the data collector.

hydrocnhs.hydrocnhs module

class hydrocnhs.hydrocnhs.Model(model, name=None, rn_gen=None, checked=False, parsed=False, log_settings={'level': 'WARNING', 'log_filename': None}, paral_setting={'cores_formUH': 1, 'cores_pet': 1, 'cores_runoff': 1, 'verbose': 0})

Bases: object

get_model_object()

Get the model object in a dictionary form.

Returns:

model object dictionary.

Return type:

dict

load_weather_data(temp, prec, pet=None, outlets=[])

Load temperature, precipitation, and potential evapotranspiration data.

Parameters:

  • temp (dict) – [degC] Daily mean temperature time series data (value) for each subbasin named by its outlet. E.g., {“subbasin1”:[…], “subbasin2”:[…]}

  • prec (dict) – [cm] Daily precipitation time series data (value) for each subbasin named by its outlet. E.g., {“subbasin1”:[…], “subbasin2”:[…]}

  • pet (dict, optional) – [cm] Daily potential evapotranspiration time series data (value) for each subbasin named by its outlet, by default None. E.g., {“subbasin1”:[…], “subbasin2”:[…]}

  • outlets (list, optional) – Outlets that need weather data for rainfall-runoff simulation.

run(temp, prec, pet=None, assigned_Q={}, assigned_UH={}, disable=False)

Run HydroCNHS simulation.

Parameters:

  • temp (dict) – [degC] Daily mean temperature.

  • prec (dict) – [cm] Daily precipitation.

  • pet (dict, optional) – [cm] Potential evapotranspiration, by default None. If None, pet is calculated by Hamon’s method.

  • assigned_Q (dict, optional) – [cms] If the user wants to manually assign Q for certain outlets {“outlet”: array}, by default {}.

  • assigned_UH (dict, optional) – If the user wants to manually assign UH (Lohmann) for certain outlets {“outlet”: array}, by default {}.

  • disable (bool, optional) – Disable display progress bar, by default False.

Returns:

A dictionary of flow time series.

Return type:

dict

hydrocnhs.indicators module

class hydrocnhs.indicators.Indicator

Bases: object

A class for calculating indicators.

static cal_indicator_df(x_obv, y_sim, index_name='value', indicators_list=None, r_na=True)

Calculate indicators and return as a DataFrame.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • index_name (str, optional) – Index name, by default “value”.

  • indicators_list (list, optional) – List of indicators, by default None.

  • r_na (bool, optional) – Remove nan, by default True.

Returns:

A DataFrame of indicators.

Return type:

DataFrame

static get_cp(x_obv, y_sim, r_na=False)

Correlation of persistence.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Correlation of persistence.

Return type:

float

static get_df_indicator(x_obv_df, y_sim_df, indicators_list=None, r_na=True)

Calculate indicators and return as a DataFrame.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • indicators_list (list, optional) – List of indicators, by default None.

  • r_na (bool, optional) – Remove nan, by default True.

Returns:

A DataFrame of indicators.

Return type:

DataFrame

static get_ikge(x_obv, y_sim, r_na=True)

Inverse Kling-Gupta efficiency.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Inverse Kling-Gupta efficiency.

Return type:

float

static get_inse(x_obv, y_sim, r_na=False)

Inverse Nash-Sutcliffe efficiency.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Inverse Nash-Sutcliffe efficiency.

Return type:

float

static get_kge(x_obv, y_sim, r_na=True)

Kling-Gupta efficiency.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Kling-Gupta efficiency.

Return type:

float

static get_nse(x_obv, y_sim, r_na=False)

Nash-Sutcliffe efficiency.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Nash-Sutcliffe efficiency.

Return type:

float

static get_r(x_obv, y_sim, r_na=True)

Correlation.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

r coefficient.

Return type:

float

static get_r2(x_obv, y_sim, r_na=True)

Coefficient of determination.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

r2 coefficient.

Return type:

float

static get_rmse(x_obv, y_sim, r_na=False)

Root mean square error.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

Root mean square error.

Return type:

float

static get_rsr(x_obv, y_sim, r_na=False)

RMSE-observations standard deviation ratio.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • r_na (bool, optional) – Remove nan, by default True

Returns:

RMSE-observations standard deviation ratio.

Return type:

float

static remove_na(x_obv, y_sim)

Remove nan in x_obv and y_sim.

This function makes sure there is no nan involves in the indicator calculation. If nan is detected, data points will be remove from x_obv and y_sim simultaneously.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

Returns:

Updated (x_obv, y_sim)

Return type:

tuple

hydrocnhs.model_builder module

class hydrocnhs.model_builder.ModelBuilder(wd, model=None)

Bases: object

add_agent(agt_type_class, agt_name, api, priority=1, link_dict={}, dm_class=None, par_dict={}, attr_dict={})

Add agent.

Parameters:

  • agt_type_class (str) – Assigned agent type class.

  • agt_name (str) – Agent name.

  • api (str) – The API to integrate the agent to the HydroCNHS. e.g., mb.Dam.

  • priority (int, optional) – Priority of the agent if conflicts occur, by default 1.

  • link_dict (dict, optional) – Linkage dictionary, by default {}.

  • dm_class (str, optional) – Assigned decision-making class, by default None

  • par_dict (dict, optional) – Parameter dictionary, by default {}

  • attr_dict (dict, optional) – Attribution dictionary, by default {}

add_institution(institution, instit_dm_class, agent_list)

Add a institution.

Parameters:

  • institution (str) – Institution name.

  • instit_dm_class (str) – Assigned institutional decision-making class.

  • agent_list (list) – Agent member list of the institute.

add_sediment(subbasin, area_list, cool_months, K_list, Ac, Aw, LS_list, DR, id_list=None, CP=-99, Sa=-99, Sb=-99, Sq=-99)
gen_ABM_module_template()

Generate ABM module template based on the ABM setting.

help()
print_model(indentor='  ', level=1)

Print model to the console

Parameters:

  • indentor (str, optional) – Indentor, by default “ “.

  • level (int, optional) – Print out level of a nested dictionary, by default 1.

set_ABM(abm_module_folder_path=None, abm_module_name='ABM_module.py')

Set up ABM

Parameters:

  • abm_module_folder_path (str, optional) – Folder directory of ABM modules. It it is not given, working directory will be assigned, by default None.

  • abm_module_name (str, optional) – The ABM module name, by default “ABM_module.py”

set_rainfall_runoff(outlet_list, area_list=None, lat_list=None, runoff_model='GWLF', activate=True)

Set up RainfallRunoff.

Parameters:

  • outlet_list (list) – A list of subbasin outlet names.

  • area_list (list, optional) – Area [ha] list corresponding to outlet_list, by default None.

  • lat_list (str, optional) – Latitude [deg] list corresponding to outlet_list, by default None.

  • runoff_model (str, optional) – “GWLF” or “ABCD” or “Other”, by default None.

  • Note – If “Other” is selected for runoff_model, users must provide precalculated runoffs for each subbasin as an input to HydroCNHS.

set_routing_outlet(routing_outlet, upstream_outlet_list, instream_objects=[], flow_length_list=None, routing_model='Lohmann', activate=True)

Set up a routing outlet.

Parameters:

  • routing_outlet (str) – Name of routing outlet. routing_outlet should be one of outlets in RainfallRunoff.

  • upstream_outlet_list (list) – A list of outlets or dam agents that contribute to the streamflow at routing_outlet.

  • instream_objects (list, optional) – A list of instream objects’ names (i.e., dam agents), by default [].

  • flow_length_list (list, optional) – A list of flow lengths. The order has to consist to the upstream_outlet_list.

  • routing_model (list, optional) – Routing model, by default “Lohmann”.

set_sediment(start_month)
set_water_system(start_date, end_date)

Set up WaterSystem.

Parameters:

  • start_date (str) – “yyyy/mm/dd”

  • end_date (str) – “yyyy/mm/dd”

write_model_to_yaml(filename)

Output model configuration file (.yaml)

Parameters:

filename (str) – Filename

hydrocnhs.routing module

hydrocnhs.routing.form_UH_Lohmann(inputs, routing_pars, force_ingrid_off=False)

Derive HRU’s UH at the (watershed) outlet.

Parameters:

  • inputs (dict) – Inputs dictionary containing FlowLength [m] Travel distence of flow between two outlets [float] and InstreamControl [bool].

  • routing_pars (dict) – Four parameters for routing: GShape, GScale, Velo, Diff [float]

  • force_ingrid_off (bool, optional) – If True, then within subbasin routing will be forced to turn off, by default False.

hydrocnhs.routing.run_step_Lohmann(routing_outlet, routing, UH_Lohmann, Q, Q_runoff, t, *args)

Calculate a single time step routing for a given routing_outlet at time t.

Parameters:

  • routing_outlet (str) – routing outlet.

  • routing (dict) – Routing setting dictionary from model.yaml file.

  • UH_Lohmann (dict) – A dictionary containing pre-formed UHs.

  • Q (dict) – A dictionary containing newest routed flows.

  • Q_runoff (dict) – A dictionary containing newest unrouted flows without.

  • t (int) – Index of current time step [day].

Returns:

Routed flow of routing_outlet at time t.

Return type:

float

hydrocnhs.routing.run_step_Lohmann_convey(routing_outlet, routing, UH_Lohmann_convey, Q_convey, t, *args)

Calculate a single time step conveying water routing for a given routing_outlet at time t.

Parameters:

  • routing_outlet (str) – routing outlet.

  • routing (dict) – Routing setting dictionary from model.yaml file.

  • UH_Lohmann_convey (dict) – A dictionary containing pre-formed conveying UHs (i.e., no within subbasin routing).

  • Q_convey (dict) – A dictionary containing conveying water.

  • t (int) – Index of current time step [day].

Returns:

Routed conveying flow of routing_outlet at time t.

Return type:

float

hydrocnhs.routing.run_step_Lohmann_convey_sed(routing_outlet, routing, UH_Lohmann_convey, Q_convey, t, Q_frac)

Calculate a single time step conveying water routing for a given routing_outlet at time t.

Parameters:

  • routing_outlet (str) – routing outlet.

  • routing (dict) – Routing setting dictionary from model.yaml file.

  • UH_Lohmann_convey (dict) – A dictionary containing pre-formed conveying UHs (i.e., no within subbasin routing).

  • Q_convey (dict) – A dictionary containing conveying water.

  • t (int) – Index of current time step [day].

Returns:

Routed conveying flow of routing_outlet at time t.

Return type:

float

hydrocnhs.routing.run_step_Lohmann_sed(routing_outlet, routing, UH_Lohmann, Q, Q_runoff, t, Q_frac)

Calculate a single time step routing for a given routing_outlet at time t.

Parameters:

  • routing_outlet (str) – routing outlet.

  • routing (dict) – Routing setting dictionary from model.yaml file.

  • UH_Lohmann (dict) – A dictionary containing pre-formed UHs.

  • Q (dict) – A dictionary containing newest routed flows.

  • Q_runoff (dict) – A dictionary containing newest unrouted flows without.

  • t (int) – Index of current time step [day].

Returns:

Routed flow of routing_outlet at time t.

Return type:

float

hydrocnhs.sediment module

hydrocnhs.sediment.cal_LS(SL, PS)

Calculate topographic factor, LS.

Parameters:

  • SL (1darray) – Slope length.

  • PS (1darray) – Percent slope.

Returns:

LS

Return type:

1darray

hydrocnhs.sediment.cal_SX(DR, X, pd_date_index)

Calculate total sediment supply, SX, of subbasins.

Parameters:

  • DR (1darray) – Delivery ratio.

  • X (2darray) – Sediment supply.

  • pd_date_index (pd.Datetime) – Pandas datetime index.

Returns:

SX

Return type:

1darray

hydrocnhs.sediment.cal_TR_B(Q_frac, pd_date_index, fi, ti)

Calculate monthly allocation propotions based on transportation capacity.

Parameters:

  • Q_frac (dict) – A dictionary of streamflow portion of each subbasion that contributes their corresponding routing outlets.

  • pd_date_index (pd.Datetime) – Pandas datetime index.

Returns:

A dictionary of monthly allocation ratios.

Return type:

dict

hydrocnhs.sediment.cal_usle(RE, K, CP, LS, Area)

Calculate the erosion using USLE.

Parameters:

  • RE (1darray) – Rainfall erosivity.

  • K (1darray) – Soil erodibility factor.

  • CP (1darray) – Vegetation management factor = cover and management factor * support practice factor.

  • LS (1darray) – Topographic factor.

  • Area (1darray) – [ha] Area

Returns:

[Mg] Sediment supply

Return type:

2darray

hydrocnhs.sediment.run_TSS(RE_dict, sediment_setting, routing_setting, dam_agts, pd_date_index, Q_frac, dc_TSS, yi, fi, ti)

hydrocnhs.util module

hydrocnhs.util.check_RainfallRunoff(model_dict)
hydrocnhs.util.check_WS(model_dict)
hydrocnhs.util.check_agent_in_routing(model_dict)
hydrocnhs.util.check_model(model_dict)

Check the model dictionary

Parameters:

model_dict (dict) – Model dictionary

Returns:

True if pass the check.

Return type:

bool

hydrocnhs.util.create_rn_gen(seed)

Create random number generator based on numpy module.

Parameters:

seed (int) – Random seed.

Returns:

Random number generator.

Return type:

object

hydrocnhs.util.dict_to_string(dictionary, indentor='  ', level=1)

Ture a dictionary into a printable string.

Parameters:

  • dictionary (dict) – A dictionary.

  • indentor (str, optional) – Indentor, by default “ “

Returns:

String.

Return type:

str

hydrocnhs.util.form_sim_seq(node_list, back_tracking_dict)
hydrocnhs.util.gen_default_bounds(model_dict, key_option=['Pars'])

Generate default parameter bounds in the format of a list of DataFrames.

Parameters:

  • model_dict (dict) – Model dictionary.

  • key_option (list, optional) – Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

Returns:

A list of parameter bound DataFrames, A list of DataFrame names.

Return type:

tuple

hydrocnhs.util.list_callable_public_object(obj)
hydrocnhs.util.load_customized_module_to_class(Class, module_name, path)

Load classes and functions in a user defined module (.py) into a given Class.

Parameters:

  • Class (class) – A class to collect classes and functions in a given module.

  • module_name (str) – filename.py or filename.

  • path (str) – Directory of filename.py.

hydrocnhs.util.load_df_to_model_dict(model_dict, df, section, key)

Load dataframe to model dictionary.

Note that the column and index names have to be identical as outputted from write_model_to_df() function.

Parameters:

  • model_dict (dict) – Model dictionary.

  • df (DataFrame) – Dataframe.

  • section (str) – RainfallRunoff or Routing or ABM.

  • key (str) – nputs or Pars or Attributes.

Returns:

Updated model_dict.

Return type:

dict

hydrocnhs.util.load_model(model, checked=False, parsed=False, print_summary=True)

Load model.yaml or model dictionary.

Parameters:

  • model (str/dict) – Filename (e.g., model.yaml) or model dictionary.

  • checked (bool, optional) – If True, no pre-check will be implemented, by default False.

  • parsed (bool, optional) – If True, model will not be re-parses, by default False.

  • print_summary (bool, optional) – Print the summary of loaded model, by default True.

Returns:

Model dictionary.

Return type:

dict

hydrocnhs.util.parse_model(model_dict, print_summary=True)

Parse model dictionary. Populate SystemParsedData.

Parameters:

model_dict (dict) – Load from model.yaml.

Returns:

model_dict

Return type:

dict

hydrocnhs.util.parse_sim_seq(model_dict, print_summary=True)
hydrocnhs.util.set_logging_config(log_filename=None, log_level='WARNING')

Set up logging config.

Parameters:

log_filename (str, optional) – Log filename, by default None

hydrocnhs.util.update_sim_seq_with_group(sim_seq, group, back_tracking_dict)
hydrocnhs.util.write_model(model_dict, filename, org_model=None)

Write model dictionary to .yaml file

If org_model is given, comments in the original file will be kept in the output model file.

Parameters:

  • model_dict (dict) – Model dictionary.

  • filename (str) – Model configuration filename (e.g. model.yaml).

  • org_model (str, optional) – Original model name (e.g. org_model.yaml), by default None.

Note

Common error :

  1. value type is numpy.float64 => convert it to float.

hydrocnhs.util.write_model_to_csv(folder_path, model_dict, key_option=['Pars'], prefix='')

Write model (dictionary) to csv files.

Parameters:

  • folder_path (str) – Output folder directory.

  • model_dict (dict) – Model dictionary.

  • key_option (list, optional) – Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

  • prefix (str, optional) – Prefix for the filenames, by default “”.

Returns:

A list of filenames.

Return type:

list

hydrocnhs.util.write_model_to_df(model_dict, key_option=['Pars'], prefix='')

Write model dictionary to dataframes given key options.

This function will convert model dictionary to dataframes according to user’s specified sections (key_option).

Parameters:

  • model_dict (dict) – Model dictionary.

  • key_option (list, optional) – Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

  • prefix (str, optional) – Prefix for the filenames, by default “”.

hydrocnhs.visual module

class hydrocnhs.visual.Visual

Bases: object

Collection of some plotting functions.

static plot_reg(x_obv, y_sim, title=None, xy_labal=None, same_xy_limit=True, return_reg_par=False, save_fig_path=None, show=True)

Plot regression.

Parameters:

  • x_obv (array) – Observation data.

  • y_sim (array) – Simulation data.

  • title (str, optional) – Title, by default None.

  • xy_labal (list, optional) – List of x and y labels, by default None.

  • same_xy_limit (bool, optional) – If True same limit will be applied to x and y axis, by default True.

  • return_reg_par (bool, optional) – If True, slope and interception will be return, by default False.

  • save_fig_path (str, optional) – If given, plot will be save as .png, by default None.

  • show (bool, optional) – If True, the plot will be shown in the console, by default True.

Returns:

axis object or [slope, intercept].

Return type:

ax or list

static plot_simple_ts(df, title=None, xy_labal=None, data_dots=True, save_fig_path=None, **kwargs)

Plot timeseries.

Parameters:

  • df (DataFrame) – Dataframe.

  • title (str, optional) – Title, by default None.

  • xy_labal (list, optional) – List of x and y labels, by default None.

  • data_dots (bool, optional) – If Ture, show data marker, by default True.

  • save_fig_path (str, optional) – If given, plot will be save as .png, by default None.

Returns:

axis object.

Return type:

object

static plot_timeseries(x_obv, y_sim, xticks=None, title=None, xy_labal=None, save_fig_path=None, legend=True, show=True, **kwargs)

Plot timeseries.

This function can plot two DataFrames with same column names.

Parameters:

  • x_obv (array/DataFrame) – Observation data.

  • y_sim (array/DataFrame) – Simulation data.

  • xticks (list, optional) – Ticks for x-axis, by default None.

  • title (str, optional) – Title, by default None.

  • xy_labal (list, optional) – List of x and y labels, by default None.

  • save_fig_path (str, optional) – If given, plot will be save as .png, by default None.

  • legend (bool, optional) – If True, plot legend, by default None.

  • show (bool, optional) – If True, the plot will be shown in the console, by default True.

  • kwargs (optional) – Other keywords for matplotlib.

Returns:

axis object.

Return type:

object

Module contents