HydroCNHS package

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_configdict

Agent’s configuration.

outletstr

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

fieldstr

Field name. Cannot have space in a field name.

data_typedict, optional

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

descstr

Field description.

unit

Unit of the field.

check_existbool, 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

fieldstr

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

copystr

If true, a copy of dictionary will be returned, else a pointer will be returned. Default False.

Returns

dict

A dictionary contains all fields.

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

fieldstr

Field name.

copybool

If true, create a copy of a field, which has seperate storage

pointer than the original data_collector. Otherwise, return a shortcut of a field to the original data_collector. Returns ——- 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_filename=None)

Bases: object

get_model_object()

Get the model object in a dictionary form.

Returns

dict

model object dictionary.

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

Load temperature, precipitation, and otential evapotranpiration data.

Parameters

tempdict

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

precdict

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

petdict, optional

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

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

Run HydroCNHS simulation.

Parameters

tempdict

[degC] Daily mean temperature.

precdict

[cm] Daily precipitation.

petdict, optional

[cm] Potential evapotranspiration, by default None. If none, pet is calculted by Hamon’s method.

assigned_Qdict, optional

[cms] If user want to manually assign Q for certain outlets {“outlet”: array}, by default {}.

assigned_UHdict, optional

If user want to manually assign UH (Lohmann) for certain outlet {“outlet”: array}, by default {}.

disablebool, optional

Disable display progress bar, by default False.

Returns

dict

A dictionary of flow time series.

HydroCNHS.indicators module

class HydroCNHS.indicators.Indicator

Bases: object

static CP(x_obv, y_sim, r_na=False)

Correlation of persistence.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Correlation of persistence.

static KGE(x_obv, y_sim, r_na=True)

Kling–Gupta efficiency.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Kling–Gupta efficiency.

static NSE(x_obv, y_sim, r_na=False)

Nash–Sutcliffe efficiency.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Nash–Sutcliffe efficiency.

static RSR(x_obv, y_sim, r_na=False)

RMSE-observations standard deviation ratio.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

RMSE-observations standard deviation ratio.

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

static iKGE(x_obv, y_sim, r_na=True)

Inverse Kling–Gupta efficiency.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Inverse Kling–Gupta efficiency.

static iNSE(x_obv, y_sim, r_na=False)

Inverse Nash–Sutcliffe efficiency.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Inverse Nash–Sutcliffe efficiency.

static r(x_obv, y_sim, r_na=True)

Correlation.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

r coefficient.

static r2(x_obv, y_sim, r_na=True)

Coefficient of determination.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

r2 coefficient.

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_obvarray

Observation data.

y_simarray

Simulation data.

Returns

tuple

Updated (x_obv, y_sim)

static rmse(x_obv, y_sim, r_na=False)

Root mean square error.

Parameters

x_obvarray

Observation data.

y_simarray

Simulation data.

r_nabool, optional

Remove nan, by default True

Returns

float

Root mean square error.

HydroCNHS.model_builder module

class HydroCNHS.model_builder.ModelBuilder(wd)

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_classstr

Assigned agent type class.

agt_namestr

Agent name.

apistr

The API to integrate the agent to the HydroCNHS. e.g., mb.Dam.

priorityint, optional

Priority of the agent if conflicts occur, by default 1.

link_dictdict, optional

Linkage dictionary, by default {}.

dm_classstr, optional

Assigned decision-making class, by default None

par_dictdict, optional

Parameter dictionary, by default {}

attr_dictdict, optional

Attribution dictionary, by default {}

add_institution(institution, instit_dm_class, agent_list)

Add a institution.

Parameters

institutionstr

Institution name.

instit_dm_classstr

Assigned institutional decision-making class.

agent_listlist

Agent member list of the institute.

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

indentorstr, optional

Indentor, by default ” “.

levelint, 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_pathstr, optional

Folder directory of ABM modules. It it is not given, working directory will be assigned, by default None.

abm_module_namestr, optional

The ABM module name, by default “ABM_module.py”

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

Set up RainfallRunoff.

Parameters

outlet_listlist

A list of subbasin outlet names.

area_listlist, optional

Area [ha] list corresponding to outlet_list, by default None.

lat_liststr, optional

Latitude [deg] list corresponding to outlet_list, by default None.

runoff_modelstr, 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')

Set up a routing outlet.

Parameters

routing_outletstr

Name of routing outlet. routing_outlet should be one of outlets in RainfallRunoff.

upstream_outlet_listlist

A list of outlets or dam agents that contribute to the streamflow at routing_outlet.

instream_objectslist, optional

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

flow_length_listlist, optional

A list of flow lengths. The order has to consist to the upstream_outlet_list.

routing_modellist, optional

Routing model, by default “Lohmann”.

set_water_system(start_date, end_date)

Set up WaterSystem.

Parameters

start_datestr

“yyyy/mm/dd”

end_datestr

“yyyy/mm/dd”

write_model_to_yaml(filename)

Output model configuration file (.yaml)

Parameters

filenamestr

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

inputsdict

Inputs dictionary containing FlowLength [m] Travel distence of flow between two outlets [float] and InstreamControl [bool].

routing_parsdict

Four parameters for routing: GShape, GScale, Velo, Diff [float]

force_ingrid_offbool, 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)

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

Parameters

routing_outletstr

routing outlet.

routingdict

Routing setting dictionary from model.yaml file.

UH_Lohmanndict

A dictionary containing pre-formed UHs.

Qdict

A dictionary containing newest routed flows.

Q_runoffdict

A dictionary containing newest unrouted flows without.

tint

Index of current time step [day].

Returns

float

Routed flow of routing_outlet at time t.

HydroCNHS.routing.run_step_Lohmann_convey(routing_outlet, routing, UH_Lohmann_convey, Q_convey, t)

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

Parameters

routing_outletstr

routing outlet.

routingdict

Routing setting dictionary from model.yaml file.

UH_Lohmann_conveydict

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

Q_conveydict

A dictionary containing conveying water.

tint

Index of current time step [day].

Returns

float

Routed conveying flow of routing_outlet at time t.

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_dictdict

Model dictionary

Returns

bool

True if pass the check.

HydroCNHS.util.create_rn_gen(seed)

Create random number generator based on numpy module.

Parameters

seedint

Random seed.

Returns

object

Random number generator.

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

Ture a dictionary into a printable string.

Parameters

dictionarydict

A dictionary.

indentorstr, optional

Indentor, by default ” “

Returns

str

String.

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_dictdict

Model dictionary.

key_optionlist, optional

Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

Returns

tuple

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

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

Classclass

A class to collect classes and functions in a given module.

module_namestr

filename.py or filename.

pathstr

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_dictdict

Model dictionary.

dfDataFrame

Dataframe.

sectionstr

RainfallRunoff or Routing or ABM.

keystr

nputs or Pars or Attributes.

Returns

dict

Updated model_dict.

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

Load model.yaml or model dictionary.

Parameters

modelstr/dict

Filename (e.g., model.yaml) or model dictionary.

checkedbool, optional

If True, no pre-check will be implemented, by default False.

parsedbool, optional

If True, model will not be re-parses, by default False.

print_summarybool, optional

Print the summary of loaded model, by default True.

Returns

dict

Model dictionary.

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

Parse model dictionary. Populate SystemParsedData.

Args:

model_dict (dict): Load from model.yaml.

Returns:

dict: model_dict

HydroCNHS.util.parse_sim_seq(model_dict, print_summary=True)

HydroCNHS.util.set_logging_config(log_filename=None)

Set up logging config.

Parameters

log_filenamestr, 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_dictdict

Model dictionary.

filenamestr

Model configuration filename (e.g. model.yaml).

org_modelstr, 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_pathstr

Output folder directory.

model_dictdict

Model dictionary.

key_optionlist, optional

Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

prefixstr, optional

Prefix for the filenames, by default “”.

Returns

list

A list of filenames.

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_dictdict

Model dictionary.

key_optionlist, optional

Output sections (e.g., Pars, Inputs, Attributes), by default [“Pars”].

prefixstr, 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_obvarray

Observation data.

y_simarray

Simulation data.

titlestr, optional

Title, by default None.

xy_laballist, optional

List of x and y labels, by default None.

same_xy_limitbool, optional

If True same limit will be applied to x and y axis, by default True.

return_reg_parbool, optional

If True, slope and interception will be return, by default False.

save_fig_pathstr, optional

If given, plot will be save as .png, by default None.

showbool, optional

If True, the plot will be shown in the console, by default True.

Returns

ax or list

axis object or [slope, intercept].

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

Plot timeseries.

Parameters

dfDataFrame

Dataframe.

titlestr, optional

Title, by default None.

xy_laballist, optional

List of x and y labels, by default None.

data_dotsbool, optional

If Ture, show data marker, by default True.

save_fig_pathstr, optional

If given, plot will be save as .png, by default None.

Returns

object

axis 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_obvarray/DataFrame

Observation data.

y_simarray/DataFrame

Simulation data.

xtickslist, optional

Ticks for x-axis, by default None.

titlestr, optional

Title, by default None.

xy_laballist, optional

List of x and y labels, by default None.

save_fig_pathstr, optional

If given, plot will be save as .png, by default None.

legendbool, optional

If True, plot legend, by default None.

showbool, optional

If True, the plot will be shown in the console, by default True.

kwargsoptional

Other keywords for matplotlib.

Returns

object

axis object.

Module contents

Subpackages

• HydroCNHS.calibration package
  • Submodules
  • HydroCNHS.calibration.convertor module
    • Convertor
      • Convertor.gen_cali_inputs()
      • Convertor.gen_formatter()
      • Convertor.to_1D_array()
      • Convertor.to_df_list()
  • HydroCNHS.calibration.ga_deap module
    • GA_DEAP
      • GA_DEAP.auto_save()
      • GA_DEAP.find_best_and_record()
      • GA_DEAP.gen_rn_gens()
      • GA_DEAP.load()
      • GA_DEAP.run()
      • GA_DEAP.run_individual()
      • GA_DEAP.set()
    • cxUniform()
    • descale()
    • gen_init_pop()
    • logger
    • mut_middle()
    • mut_uniform()
    • sample_by_LHC()
    • sample_by_MC()
    • scale()
    • selRoulette()
  • Module contents
    • get_config_template()
    • get_inputs_template()
    • helper()
• HydroCNHS.rainfall_runoff_model package
  • Submodules
  • HydroCNHS.rainfall_runoff_model.abcd module
    • run_ABCD()
  • HydroCNHS.rainfall_runoff_model.gwlf module
    • run_GWLF()
  • HydroCNHS.rainfall_runoff_model.pet_hamon module
    • cal_pet_Hamon()
  • Module contents