Initial Hardening of ConfigOptions

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,

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