-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmodel_configuration.py
206 lines (192 loc) · 10.9 KB
/
model_configuration.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
import configparser
from datetime import datetime
import json
import logging
import os
import textwrap
from typing import Union
from rubem.configuration.calibration_parameters import CalibrationParameters
from rubem.configuration.initial_soil_conditions import InitialSoilConditions
from rubem.configuration.input_raster_files import InputRasterFiles
from rubem.configuration.input_raster_series import InputRasterSeries
from rubem.configuration.input_table_files import InputTableFiles
from rubem.configuration.model_constants import ModelConstants
from rubem.configuration.output_data_directory import OutputDataDirectory
from rubem.configuration.output_format import OutputFileFormat
from rubem.configuration.output_variables import OutputVariables
from rubem.configuration.raster_grid_area import RasterGrid
from rubem.configuration.simulation_period import SimulationPeriod
class ModelConfiguration:
"""Represents the configuration settings for the model.
The `ModelConfiguration` class is responsible for loading and storing the configuration settings
required for running the model. It supports loading configuration from either a dictionary,
an INI file, or a JSON file.
:param config_input: The configuration input. It can be a dictionary containing the configuration
settings, a file path to an INI or JSON file, or a file-like object.
:param validate_input: Whether to validate the input. Defaults to `True`.
:type validate_input: bool, optional
:raises FileNotFoundError: If the specified config file is not found.
:raises ValueError: If the config file type is not supported.
:raises configparser.Error: If the INI file is not valid.
:raises json.JSONDecodeError: If the JSON file is not valid.
:raises KeyError: If a required setting is missing.
:raises ValueError: If a setting value is invalid.
"""
def __init__(
self, config_input: Union[dict, str, bytes, os.PathLike], validate_input: bool = True
):
self.logger = logging.getLogger(__name__)
print(f"Loading configuration{' and validating inputs' if validate_input else ''}...")
try:
if isinstance(config_input, dict):
self.logger.debug("Reading configuration from dictionary")
self.config = config_input
elif isinstance(config_input, (str, bytes, os.PathLike)):
config_input_str = str(config_input)
if not os.path.exists(config_input_str) or not os.path.isfile(config_input_str):
self.logger.error("Config file not found: %s", config_input_str)
raise FileNotFoundError(f"Config file not found: {config_input_str}")
if config_input_str.endswith(".ini"):
self.config = self.__read_ini(config_input_str)
elif config_input_str.endswith(".json"):
self.config = self.__read_json(config_input_str)
else:
self.logger.error("Unsupported file type: %s", config_input_str)
raise ValueError("Unsupported file type")
self.logger.debug("Loading configuration...")
self.simulation_period = SimulationPeriod(
datetime.strptime(self.__get_setting("SIM_TIME", "start"), "%d/%m/%Y").date(),
datetime.strptime(self.__get_setting("SIM_TIME", "end"), "%d/%m/%Y").date(),
)
self.grid = RasterGrid(float(self.__get_setting("GRID", "grid")))
self.calibration_parameters = CalibrationParameters(
alpha=float(self.__get_setting("CALIBRATION", "alpha")),
beta=float(self.__get_setting("CALIBRATION", "b")),
w_1=float(self.__get_setting("CALIBRATION", "w_1")),
w_2=float(self.__get_setting("CALIBRATION", "w_2")),
w_3=float(self.__get_setting("CALIBRATION", "w_3")),
rcd=float(self.__get_setting("CALIBRATION", "rcd")),
f=float(self.__get_setting("CALIBRATION", "f")),
alpha_gw=float(self.__get_setting("CALIBRATION", "alpha_gw")),
x=float(self.__get_setting("CALIBRATION", "x")),
)
self.initial_soil_conditions = InitialSoilConditions(
initial_soil_moisture_content=float(
self.__get_setting("INITIAL_SOIL_CONDITIONS", "t_ini")
),
initial_baseflow=float(self.__get_setting("INITIAL_SOIL_CONDITIONS", "bfw_ini")),
baseflow_limit=float(self.__get_setting("INITIAL_SOIL_CONDITIONS", "bfw_lim")),
initial_saturated_zone_storage=float(
self.__get_setting("INITIAL_SOIL_CONDITIONS", "s_sat_ini")
),
)
self.constants = ModelConstants(
fraction_photo_active_radiation_max=float(
self.__get_setting("CONSTANTS", "fpar_max")
),
fraction_photo_active_radiation_min=float(
self.__get_setting("CONSTANTS", "fpar_min")
),
leaf_area_interception_max=float(self.__get_setting("CONSTANTS", "lai_max")),
impervious_area_interception=float(self.__get_setting("CONSTANTS", "i_imp")),
)
self.output_directory = OutputDataDirectory(self.__get_setting("DIRECTORIES", "output"))
self.output_variables = OutputVariables(
itp=bool(self.__get_setting("GENERATE_FILE", "itp")),
bfw=bool(self.__get_setting("GENERATE_FILE", "bfw")),
srn=bool(self.__get_setting("GENERATE_FILE", "srn")),
eta=bool(self.__get_setting("GENERATE_FILE", "eta")),
lfw=bool(self.__get_setting("GENERATE_FILE", "lfw")),
rec=bool(self.__get_setting("GENERATE_FILE", "rec")),
smc=bool(self.__get_setting("GENERATE_FILE", "smc")),
rnf=bool(self.__get_setting("GENERATE_FILE", "rnf")),
tss=bool(self.__get_setting("GENERATE_FILE", "tss")),
output_format=(
OutputFileFormat.PCRASTER
if bool(self.__get_setting("RASTER_FILE_FORMAT", "map_raster_series"))
else OutputFileFormat.GEOTIFF
),
)
self.raster_series = InputRasterSeries(
etp=self.__get_setting("DIRECTORIES", "etp"),
etp_filename_prefix=self.__get_setting("FILENAME_PREFIXES", "etp_prefix"),
precipitation=self.__get_setting("DIRECTORIES", "prec"),
precipitation_filename_prefix=self.__get_setting(
"FILENAME_PREFIXES", "prec_prefix"
),
ndvi=self.__get_setting("DIRECTORIES", "ndvi"),
ndvi_filename_prefix=self.__get_setting("FILENAME_PREFIXES", "ndvi_prefix"),
kp=self.__get_setting("DIRECTORIES", "kp"),
kp_filename_prefix=self.__get_setting("FILENAME_PREFIXES", "kp_prefix"),
landuse=self.__get_setting("DIRECTORIES", "landuse"),
landuse_filename_prefix=self.__get_setting("FILENAME_PREFIXES", "landuse_prefix"),
validate_input=validate_input,
)
self.raster_files = InputRasterFiles(
dem=self.__get_setting("RASTERS", "dem"),
demtif=self.__get_setting("RASTERS", "demtif"),
clone=self.__get_setting("RASTERS", "clone"),
ndvi_max=self.__get_setting("RASTERS", "ndvi_max"),
ndvi_min=self.__get_setting("RASTERS", "ndvi_min"),
soil=self.__get_setting("RASTERS", "soil"),
sample_locations=self.__get_setting("RASTERS", "samples"),
validate_input=validate_input,
)
self.lookuptable_files = InputTableFiles(
rainy_days=self.__get_setting("TABLES", "rainydays"),
a_i=self.__get_setting("TABLES", "a_i"),
a_o=self.__get_setting("TABLES", "a_o"),
a_s=self.__get_setting("TABLES", "a_s"),
a_v=self.__get_setting("TABLES", "a_v"),
manning=self.__get_setting("TABLES", "manning"),
bulk_density=self.__get_setting("TABLES", "bulk_density"),
k_sat=self.__get_setting("TABLES", "k_sat"),
t_fcap=self.__get_setting("TABLES", "t_fcap"),
t_sat=self.__get_setting("TABLES", "t_sat"),
t_wp=self.__get_setting("TABLES", "t_wp"),
rootzone_depth=self.__get_setting("TABLES", "rootzone_depth"),
kc_min=self.__get_setting("TABLES", "k_c_min"),
kc_max=self.__get_setting("TABLES", "k_c_max"),
validate_input=validate_input,
)
except Exception as e:
self.logger.error("Failed to load configuration: %s", e)
raise
def __read_ini(self, file_path: Union[str, bytes, os.PathLike]):
self.logger.debug("Reading INI file: %s", file_path)
try:
parser = configparser.ConfigParser()
parser.read(file_path, encoding="utf-8")
return {section: dict(parser.items(section)) for section in parser.sections()}
except configparser.Error as e:
self.logger.error("Error parsing INI file: %s", e)
raise
def __read_json(self, file_path: Union[str, bytes, os.PathLike]):
self.logger.debug("Reading JSON file: %s", file_path)
try:
with open(file=file_path, mode="r", encoding="utf-8") as f:
return json.load(f)
except json.JSONDecodeError as e:
self.logger.error("Error parsing JSON file: %s", e)
raise
def __get_setting(self, section, setting):
try:
return self.config[section][setting]
except KeyError as e:
self.logger.error("Missing setting: %s in section: %s", setting, section)
raise ValueError(f"Missing setting: {setting} in section: {section}") from e
def __str__(self):
# Workaround for "Escape sequence (backslash) not allowed in expression portion of f-string prior to Python 3.12"
tab = "\t"
return (
f"Simulation period: {self.simulation_period}\n"
f"Grid area: {self.grid}\n"
f"Raster Series:\n{textwrap.indent(str(self.raster_series), tab)}\n"
f"Raster files:\n{textwrap.indent(str(self.raster_files), tab)}\n"
f"Lookuptable files:\n{textwrap.indent(str(self.lookuptable_files), tab)}\n"
f"Calibration parameters:\n{textwrap.indent(str(self.calibration_parameters), tab)}\n"
f"Initial soil conditions:\n{textwrap.indent(str(self.initial_soil_conditions), tab)}\n"
f"Constants:\n{textwrap.indent(str(self.constants), tab)}\n"
f"Output directory: {self.output_directory}\n"
f"Output Raster Series:\n{textwrap.indent(str(self.output_variables), tab)}"
)