diff --git a/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/config.py b/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/config.py index c061e8c5..8704a9e4 100755 --- a/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/config.py +++ b/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/config.py @@ -4,6 +4,7 @@ import os import re from datetime import datetime, timedelta, timezone +import typing import numpy as np @@ -13,11 +14,15 @@ from NextGen_Forcings_Engine_BMI.NextGen_Forcings_Engine.core.time_handling import ( calculate_lookback_window, ) +from NextGen_Forcings_Engine_BMI.NextGen_Forcings_Engine.general_utils import ( + setter_hardener, +) from nextgen_forcings_ewts import MODULE_NAME LOG = logging.getLogger(MODULE_NAME) FORCE_COUNT = 27 +DATETIME_FORMAT = "%Y%m%d%H%M" class ConfigOptions: @@ -57,7 +62,9 @@ def __init__(self, config: dict, b_date=None, geogrid_arg=None): self.realtime_flag = None self.refcst_flag = None self.ana_flag = None - self.b_date_proc = b_date + self.b_date_proc: datetime | None = ( + None if b_date is None else datetime.strptime(b_date, DATETIME_FORMAT) + ) self.e_date_proc = None self.first_fcst_cycle = None self.current_fcst_cycle = None @@ -145,18 +152,138 @@ def __init__(self, config: dict, b_date=None, geogrid_arg=None): self.geogrid = geogrid_arg self.geopackage = None + ### BEGIN hardened properties + # b_date_proc + @property + def b_date_proc(self): + return self._b_date_proc + + @b_date_proc.setter + @setter_hardener + def b_date_proc(self, val: datetime | None): + if (val is not None) and (not isinstance(val, datetime)): + raise TypeError( + f"Expected type datetime or None for b_date_proc, but new val has type: {type(val)}" + ) + self._b_date_proc = val + + # fcst_freq + @property + def fcst_freq(self): + return self._fcst_freq + + @fcst_freq.setter + @setter_hardener + def fcst_freq(self, val: typing.Any): + self._fcst_freq = val + + # fcst_input_horizons + @property + def fcst_input_horizons(self): + return self._fcst_input_horizons + + @fcst_input_horizons.setter + @setter_hardener + def fcst_input_horizons(self, val: typing.Any): + self._fcst_input_horizons = val + + # aorc_conus_source + @property + def aorc_conus_source(self): + return self._aorc_conus_source + + @aorc_conus_source.setter + @setter_hardener + def aorc_conus_source(self, val: typing.Any): + self._aorc_conus_source = val + + # aorc_conus_year_url + @property + def aorc_conus_year_url(self): + return self._aorc_conus_year_url + + @aorc_conus_year_url.setter + @setter_hardener + def aorc_conus_year_url(self, val: typing.Any): + self._aorc_conus_year_url = val + + # aorc_alaska_source + @property + def aorc_alaska_source(self): + return self._aorc_alaska_source + + @aorc_alaska_source.setter + @setter_hardener + def aorc_alaska_source(self, val: typing.Any): + self._aorc_alaska_source = val + + # aorc_alaska_url + @property + def aorc_alaska_url(self): + return self._aorc_alaska_url + + @aorc_alaska_url.setter + @setter_hardener + def aorc_alaska_url(self, val: typing.Any): + self._aorc_alaska_url = val + + # nwm_source + @property + def nwm_source(self): + return self._nwm_source + + @nwm_source.setter + @setter_hardener + def nwm_source(self, val: typing.Any): + self._nwm_source = val + + # nwm_geogrid + @property + def nwm_geogrid(self): + return self._nwm_geogrid + + @nwm_geogrid.setter + @setter_hardener + def nwm_geogrid(self, val: typing.Any): + self._nwm_geogrid = val + + # geogrid + @property + def geogrid(self): + return self._geogrid + + @geogrid.setter + @setter_hardener + def geogrid(self, val: typing.Any): + self._geogrid = val + + # geopackage + @property + def geopackage(self): + return self._geopackage + + @geopackage.setter + @setter_hardener + def geopackage(self, val: typing.Any): + self._geopackage = val + + ### END hardened properties + def validate_config(self, cfg_bmi: dict) -> None: """Validate in options from the configuration file and check that proper options were provided.""" # Ensure b_date_proc is set; if not, read from the configuration file if self.b_date_proc is None: try: - self.b_date_proc = cfg_bmi.get( - "RefcstBDateProc", None - ) # Default to None if not found - if self.b_date_proc is None: + # Default to None if not found + b_date_proc_from_cfg_bmi = cfg_bmi.get("RefcstBDateProc", None) + if b_date_proc_from_cfg_bmi is None: err_out_screen( "Unable to locate RefcstBDateProc under Logistics section in configuration file." ) + self.b_date_proc = datetime.strptime( + b_date_proc_from_cfg_bmi, DATETIME_FORMAT + ) + except KeyError as e: err_out_screen( "Unable to locate RefcstBDateProc under Logistics section in configuration file.", @@ -606,8 +733,15 @@ def validate_config(self, cfg_bmi: dict) -> None: e, ) try: - self.b_date_proc = datetime.strptime(beg_date_tmp, "%Y%m%d%H%M") - except ValueError as e: + if isinstance(beg_date_tmp, datetime): + self.b_date_proc = beg_date_tmp + elif isinstance(beg_date_tmp, str): + self.b_date_proc = datetime.strptime(beg_date_tmp, DATETIME_FORMAT) + else: + raise TypeError( + f"Expected datetime.datetime or str for beg_date_tmp, but got: {type(self.b_date_proc)}" + ) + except Exception as e: err_out_screen( "Improper RefcstBDateProc value entered into the configuration file. Please check your entry.", e, diff --git a/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/general_utils.py b/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/general_utils.py new file mode 100644 index 00000000..8fb70774 --- /dev/null +++ b/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/general_utils.py @@ -0,0 +1,32 @@ +"""General utilities.""" + +import functools + + +def setter_hardener(func): + """ + Decorator for setters. Causes the setter to be hardened such that it + asserts that the new value is either replacing None, or that the new value + is equal to the existing value. + + NOTE: since `==` is used to determine if the new value is equal to the old value, + it is possible to change the value if the `==` check passes, for example setting 5.0 (float) to replace 5 (int) + would not cause an error to be raised. + """ + + @functools.wraps(func) + def wrapper(self, new_value): + # Private attr attr_private is typical pattern, with underscore prepending the public attr attr_public. + attr_public = func.__name__ + attr_private = f"_{attr_public}" + # Current value, or None if not yet set. + val_existing = getattr(self, attr_private, None) + + if val_existing is None or val_existing == new_value: + return func(self, new_value) + else: + raise ValueError( + f"Public attr {attr_public} (private attr {attr_private}) is hardened. It had already been set to non-None value {repr(val_existing)}, and proposed new value is {repr(new_value)}, which is not equal to the existing value." + ) + + return wrapper