config.py — Konfigurationsmodul¶
Quelldatei
Lädt die YAML-Konfigurationsdatei und stellt alle Parameter als Attribute eines Config-Objekts bereit. Ersetzt das frühere hartcodierte Paths_and_params.py-System.
Projektverzeichnis automatisch bestimmen Da dieses Modul in src/ liegt, ist das Projektverzeichnis ein Ordner hoeher. Alle relativen Pfade in config.yaml werden relativ zu diesem Verzeichnis aufgeloest.
PROJECT_ROOT: Path = Path(__file__).resolve().parent.parent
def load_config(config_path: Path | str | None = None) -> "Config":
Laedt die YAML-Konfigurationsdatei und gibt ein Config-Objekt zurueck.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
config_path |
Path, str oder None, optional |
Pfad zur YAML-Datei. Falls None, wird config.yaml im Projektverzeichnis verwendet. Standard ist None. |
Rueckgabe
Config
Config-Objekt mit allen Parametern als Attribute.
Fehler
FileNotFoundError
Wenn die Konfigurationsdatei nicht gefunden wird.
ValueError
Wenn erforderliche Parameter fehlen.
if config_path is None:
config_path = PROJECT_ROOT / "config.yaml"
else:
config_path = Path(config_path)
if not config_path.exists():
raise FileNotFoundError(
f"Konfigurationsdatei nicht gefunden: {config_path}\n"
f"Kopiere config.yaml.example nach config.yaml und passe die Werte an."
)
with open(config_path, "r", encoding="utf-8") as f:
raw = yaml.safe_load(f)
return Config(raw)
class Config:
Haelt alle Konfigurationsparameter fuer einen WorldQual Lite Modelllauf.
Die Parameter werden aus einem Dictionary (YAML) gelesen und als Attribute bereitgestellt. Pfade werden automatisch relativ zum Projektverzeichnis aufgeloest.
Initialisiert das Config-Objekt aus einem YAML-Dictionary.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
raw |
dict |
Das Konfigurationswoerterbuch aus der YAML-Datei. |
self._raw = raw
self._load_run_config(raw.get("run", {}))
self._load_time_config(raw.get("time", {}))
self._load_spatial_config(raw.get("spatial", {}))
self._load_calibration(raw.get("calibration", {}))
self._load_data_structure(raw.get("data_structure", {}))
self._load_paths(raw.get("paths", {}))
self._load_continents(raw.get("continents", {}))
self._derive_database_names()
self._validate()
def _load_run_config(self, run: dict) -> None:
Laedt Laufkonfigurationsparameter (run).
Bestimmt ob ein historischer oder projizierter Lauf durchgefuehrt wird, welches Klimaszenario (SSP/RCP/GCM) verwendet wird und ob die Daten aus der MySQL-Datenbank oder aus CSV-Dateien geladen werden.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
run |
dict |
Woerterbuch mit Laufkonfigurationsparametern aus der YAML-Datei. |
self.run_type = run.get("type", "Historical")
self.data_source = run.get("data_source", "DB")
self.Scenario = run.get("scenario", "SSP2")
self.rcp = run.get("rcp", "rcp6p0")
self.GCM = run.get("gcm", "MIROC5")
def _load_time_config(self, time: dict) -> None:
Laedt zeitliche Konfigurationsparameter (time).
Definiert den Simulationszeitraum: Anfangs- und Endjahr, Zeitschritt (fuer Zukunftsszenarien) und welche Monate berechnet werden sollen (Standard: alle 12).
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
time |
dict |
Woerterbuch mit zeitlichen Parametern aus der YAML-Datei. |
self.initial_year = time.get("initial_year", 2000)
self.final_year_included = time.get("final_year", 2000)
self.time_step = time.get("time_step", 10)
months_list = time.get("months", list(range(1, 13)))
self.months = range(months_list[0], months_list[-1] + 1)
def _load_spatial_config(self, spatial: dict) -> None:
Laedt raeumliche Konfigurationsparameter (spatial).
Legt das Untersuchungsgebiet fest: Laenderkennung (ISO 3166), Kontinent-Index (0=Europa), FAO-Region, Phosphor-Parameter-ID (60), Szenario-IDs fuer die Datenbankabfrage und die Auslass-Zellen des Einzugsgebiets fuer das Routing in BasinDelineation.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
spatial |
dict |
Woerterbuch mit raeumlichen Parametern aus der YAML-Datei. |
self.country_id = spatial.get("country_id", 276)
self.continent_index = spatial.get("continent_index", 0)
self.IDReg = spatial.get("id_reg", 1)
self.parameter_id = spatial.get("parameter_id", 60)
self.IDScen = spatial.get("id_scen", 27)
self.Future_IDScen = spatial.get("future_id_scen", 27)
self.Point_load_corr = spatial.get("point_load_corr", 1)
self.downstream_cells = spatial.get("downstream_cells", [])
def _load_calibration(self, cal: dict) -> None:
Laedt Kalibrierungsparameter (calibration).
Zwei Modellkomponenten werden kalibriert: 1. Erosionsmodell nach Fink et al.: Lmax, a, b, c bestimmen den Anteil diffuser Frachten, der das Gewaesser erreicht. sc_corr und bg_corr sind Korrekturfaktoren. 2. Retentionsmodell nach Vollenweider: a_ret und b_ret beschreiben die Phosphorretention in stehenden Gewaessern als Funktion der hydraulischen Belastung (HL).
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
cal |
dict |
Woerterbuch mit Kalibrierungsparametern aus der YAML-Datei. |
Erosionsmodell (Fink et al.)
self.Lmax_calib = cal.get("lmax", 6.34e-02)
self.a_calib = cal.get("a", 900)
self.b_calib = cal.get("b", -2)
self.c_calib = cal.get("c", 1e-12)
self.sc_corr_calib = cal.get("sc_corr", 1)
self.bg_corr_calib = cal.get("bg_corr", 1)
Retentionsmodell (Vollenweider)
self.a_ret = cal.get("a_ret", 13.2)
self.b_ret = cal.get("b_ret", -0.93)
def _load_data_structure(self, ds: dict) -> None:
Laedt Datenstruktur-Konstanten (data_structure).
Feste Strukturparameter, die sich aus dem WaterGAP-Datenmodell ergeben: 21 Kulturtypen (GLCC-Klassifikation), 12 Viehkategorien (FAO), und die GCRC-Datenbankversion fuer die Zell-ID-Konvertierung.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
ds |
dict |
Woerterbuch mit Datenstrukturparametern aus der YAML-Datei. |
self.crop_type_count = ds.get("crop_type_count", 21)
self.livestock_type_count = ds.get("livestock_type_count", 12)
self.db_gcrc_version = ds.get("db_gcrc_version", 3)
def _load_paths(self, paths: dict) -> None:
Laedt Pfade zu Eingabedateien (paths).
Alle Pfade koennen absolut oder relativ angegeben werden. Relative Pfade werden automatisch relativ zum Projektverzeichnis aufgeloest. Die UNF-Ordner enthalten die binaeren Rasterdateien fuer Abfluss, Viehbestand, Duenger, Erosionsfaktoren etc.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
paths |
dict |
Woerterbuch mit Dateipfaden aus der YAML-Datei. |
self.Basin_cells_list_csv_path = self._resolve_path(
paths.get("basin_cells_csv", "data/example/basin_cells.csv")
)
self.data_path = self._resolve_path(
paths.get("data_path", "data/input")
)
self.cell_input_folder = str(self._resolve_path(
paths.get("cell_input_folder", "data/Europe_Cell_Input_Files")
))
self.output_folder = str(self._resolve_path(
paths.get("output_folder", "data/output")
))
UNF-Ordner (historisch)
self.Surface_Runoff_folder = str(self._resolve_path(
paths.get("surface_runoff_folder", "data/Europe_Input_UNF_Files/G_SURFACE_RUNOFF")
))
self.Urban_Runoff_folder = str(self._resolve_path(
paths.get("urban_runoff_folder", "data/Europe_Input_UNF_Files/G_URBAN_RUNOFF")
))
self.Livestock_Density_folder = str(self._resolve_path(
paths.get("livestock_density_folder", "data/Europe_Input_UNF_Files/G_LIVESTOCK_NR")
))
self.Correction_Factor_folder = str(self._resolve_path(
paths.get("correction_factor_folder", "data/Europe_Input_UNF_Files/G_CORR_FACT_RTF")
))
self.P_Rate_ton_folder = str(self._resolve_path(
paths.get("p_rate_ton_folder", "data/Europe_Input_UNF_Files/P_RATE_TON_KM2")
))
self.CropLand_Corrected_folder = str(self._resolve_path(
paths.get("cropland_corrected_folder", "data/Europe_Input_UNF_Files/CROPLAND_CORR_KM2")
))
self.Other_UNF_files_folder = str(self._resolve_path(
paths.get("other_unf_files_folder", "data/Europe_Input_UNF_Files/OTHER_UNF_FILES")
))
UNF-Ordner (Zukunft)
self.Future_UNF_files = str(self._resolve_path(
paths.get("future_unf_files", "data/Future_UNF_files")
))
BasinDelineation-Pfade
self.outflow_file = str(self._resolve_path(
paths.get("outflow_file", "data/Europe_Input_UNF_Files/OTHER_UNF_FILES/G_OUTFLC.UNF4")
))
self.mother_grid_shapefile = str(self._resolve_path(
paths.get("mother_grid_shapefile", "data/WG_mothers/mother_eu.shp")
))
def _load_continents(self, cont: dict) -> None:
Laedt Kontinentdaten (continents).
WaterGAP unterteilt die Welt in 10 Kontinentbloecke. Jeder Block hat eine eigene Zellenzahl (ng), Zeilenanzahl (nrow) und Spaltenanzahl (ncol) im 5-Bogenminuten-Raster. Der continent_index waehlt aus, welcher Block fuer den aktuellen Modelllauf verwendet wird.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
cont |
dict |
Woerterbuch mit Kontinentparametern aus der YAML-Datei. |
self.name = cont.get("names",
["eu", "af", "as", "au", "na", "sa", "wg2", "wa", "clm", "clm025"]
)
self.ng = cont.get("cell_counts",
[180721, 371410, 841703, 109084, 461694, 226852, 66896, 67420, 70412, 281648]
)
self.nrow = cont.get("nrow",
[641, 1090, 1258, 740, 915, 824, 360, 360, 360, 720]
)
self.ncol = cont.get("ncol",
[1000, 1237, 4320, 4309, 1519, 1356, 720, 720, 720, 1440]
)
def _derive_database_names(self) -> None:
Leitet Datenbanknamen aus dem Kontinent-Index ab.
Die Datenbanknamen werden basierend auf dem Kontinentindex automatisch aus den Kontinentnamen generiert.
self.dbname_cell = "globewq_wq_load_" + self.name[self.continent_index]
self.dbname1 = "globewq_wq_load"
def _resolve_path(self, path_str: str) -> Path:
Loest einen Pfad relativ zum Projektverzeichnis auf.
Absolute Pfade bleiben absolut, relative Pfade werden relativ zum Projektverzeichnis aufgeloest.
Parameter
| Parameter | Typ | Beschreibung |
|---|---|---|
path_str |
str |
Der zu aufzuloesende Pfad (absolut oder relativ). |
Rueckgabe
Path
Das aufgeloeste Path-Objekt.
p = Path(path_str)
if p.is_absolute():
return p
return PROJECT_ROOT / p
def _validate(self) -> None:
Prueft ob die wichtigsten Konfigurationsparameter plausibel sind.
Validiert die Konfigurationsparameter auf Konsistenz und Plausiblitaet. Wirft ValueError, wenn kritische Parameter ungueltig sind.
Fehler
ValueError
Wenn die Konfiguration Fehler oder Ungueltigkeiten enthaelt.
errors = []
if self.run_type not in ("Historical", "Future"):
errors.append(
f"run.type muss 'Historical' oder 'Future' sein, ist aber '{self.run_type}'"
)
if self.data_source not in ("DB", "Excel"):
errors.append(
f"run.data_source muss 'DB' oder 'Excel' sein, ist aber '{self.data_source}'"
)
if self.continent_index < 0 or self.continent_index >= len(self.name):
errors.append(
f"spatial.continent_index={self.continent_index} ist ungueltig "
f"(muss 0-{len(self.name)-1} sein)"
)
if self.initial_year > self.final_year_included:
errors.append(
f"initial_year ({self.initial_year}) liegt nach "
f"final_year ({self.final_year_included})"
)
if errors:
msg = "Konfigurationsfehler:\n" + "\n".join(f" - {e}" for e in errors)
raise ValueError(msg)
def __repr__(self) -> str:
Gibt eine Stringrepraesentation des Config-Objekts zurueck.¶
return (
f"Config(run_type={self.run_type!r}, scenario={self.Scenario!r}, "
f"country_id={self.country_id}, years={self.initial_year}-{self.final_year_included})"
)
Globale Config-Instanz
Wird automatisch beim ersten Import geladen. Alle anderen Module
greifen ueber from config import cfg auf die Parameter zu.
Fuer Tests oder alternative Konfigurationen kann load_config()
mit einem anderen Pfad aufgerufen werden.