simplify name of validators and utils functions

Author: debpal

pySWATPlus/types.py

@@ -8,11 +8,11 @@ class ParameterBase(BaseModel):
     units: typing.Optional[list[int]] = None
     conditions: typing.Optional[dict[str, list[str]]] = None
 
-    @field_validator("units")
+    @field_validator('units')
     @classmethod
     def validate_units(cls, v: typing.Optional[typing.Iterable[int]]) -> typing.Optional[typing.Iterable[int]]:
         if v is not None and any(num <= 0 for num in v):
-            raise ValueError(f"All unit IDs must be > 0, got {list(v)}")
+            raise ValueError(f'All unit IDs must be > 0, got {list(v)}')
         return list(v) if v is not None else None
 
 

pySWATPlus/utils.py

@@ -7,6 +7,7 @@
 from pydantic import BaseModel
 import hashlib
 import json
+import datetime
 
 
 def _build_line_to_add(
@@ -35,6 +36,24 @@ def _build_line_to_add(
     return arg_to_add.rstrip() + '\n'
 
 
+def _date_str_to_object(
+    date_str: str
+) -> datetime.date:
+    '''
+    Converts a date string in 'YYYY-MM-DD' format to a `datetime.date` object
+    '''
+
+    date_fmt = '%d-%b-%Y'
+    try:
+        get_date = datetime.datetime.strptime(date_str, date_fmt).date()
+    except ValueError:
+        raise ValueError(
+            f'Invalid date format: "{date_str}"; expected format is DD-Mon-YYYY (e.g., 15-Mar-2010)'
+        )
+
+    return get_date
+
+
 def _clean(
     df: pandas.DataFrame
 ) -> pandas.DataFrame:
@@ -85,34 +104,39 @@ def _ensure_path(p: str | pathlib.Path) -> pathlib.Path:
     '''
     Validate and convert a path-like argument to a resolved pathlib.Path object.
     '''
+
     if not isinstance(p, (str, pathlib.Path)):
         raise TypeError(f"Argument must be a string or Path object, got {type(p).__name__}")
     return pathlib.Path(p).resolve()
 
 
-def _format_val_field(value: float) -> str:
-    """
+def _format_val_field(
+    value: float
+) -> str:
+    '''
     Format a number for the VAL column:
     - 16 characters total: 1 leading space + 15-character numeric field
     - Right-aligned
     - Fixed-point if integer part fits; scientific if too large
-    """
+    '''
 
     # Convert to string without formatting
     s = str(value)
 
     if len(s) > 15:
         # Use scientific notation
-        formatted = f"{value:.6e}"
+        formatted = f'{value:.6e}'
     else:
         # If it fits, just use normal string
         formatted = s
 
     # Right-align to 16 characters
-    return f"{formatted:>16}"
+    return f'{formatted:>16}'
 
 
-def _compact_units(unit_list: Iterable[int]) -> list[int]:
+def _compact_units(
+    unit_list: Iterable[int]
+) -> list[int]:
     '''
     Compact a 1-based list of unit IDs into SWAT units syntax.
 
@@ -122,6 +146,7 @@ def _compact_units(unit_list: Iterable[int]) -> list[int]:
 
     All IDs must be 1-based (Fortran-style).
     '''
+
     if not unit_list:
         return []
 
@@ -158,23 +183,28 @@ def _parse_conditions(
     '''
     Parse the conditions that must be added to that parameter in calibration.cal file
     '''
+
     conditions = parameters.conditions
     if not conditions:
         return []
 
     conditions_parsed = []
     for parameter, condition_keys in conditions.items():
         for key in condition_keys:
-            conditions_parsed.append(f"{parameter:<19}{'=':<15} {0:<16}{key}")
+            conditions_parsed.append(f'{parameter:<19}{'=':<15} {0:<16}{key}')
 
     return conditions_parsed
 
 
-def _make_unique_param_name(param_key: str, model: BaseModel) -> str:
-    """
+def _make_unique_param_name(
+    param_key: str,
+    model: BaseModel
+) -> str:
+    '''
     Generate a unique, deterministic parameter identifier by combining the parameter key
     with a hash of the Pydantic model's serialized contents.
-    """
+    '''
+
     payload = json.dumps(model.model_dump(), sort_keys=True)
     uid = hashlib.md5(payload.encode()).hexdigest()
-    return f"{param_key}|{uid}"
+    return f'{param_key}|{uid}'

pySWATPlus/validators.py

@@ -2,7 +2,7 @@
 import pathlib
 import typing
 import types
-from datetime import datetime
+import datetime
 from .types import ParameterModel, ParameterBoundedModel
 
 
@@ -11,7 +11,7 @@ def _variable_origin_static_type(
     vars_values: dict[str, typing.Any]
 ) -> None:
     '''
-    Validates input variables against their expected types.
+    Checks that input variables match their expected origin types.
     '''
 
     # iterate name and type of method variables
@@ -55,7 +55,7 @@ def _path_directory(
     path: pathlib.Path
 ) -> None:
     '''
-    Validates path of the direcotry.
+    Ensures the input path refers to a valid directory.
     '''
 
     if not path.is_dir():
@@ -66,20 +66,40 @@ def _path_directory(
     return None
 
 
+def _date_begin_earlier_end(
+    begin_date: datetime.date,
+    end_date: datetime.date
+) -> None:
+    '''
+    Checks that begin date is earlier than end date.
+    '''
+
+    date_fmt = '%d-%b-%Y'
+
+    if begin_date >= end_date:
+        raise ValueError(
+            f'begin_date {begin_date.strftime(date_fmt)} must be earlier than end_date {end_date.strftime(date_fmt)}'
+        )
+
+
 def _validate_date_str(
     date_str: str
 ) -> None:
     '''
     Validates a date string in 'YYYY-MM-DD' format.
     Raises ValueError if invalid.
     '''
+
     try:
-        datetime.strptime(date_str, '%Y-%m-%d').date()
+        datetime.datetime.strptime(date_str, '%Y-%m-%d').date()
     except ValueError:
         raise ValueError(f'Invalid date format: "{date_str}". Expected YYYY-MM-DD.')
 
 
-def _validate_units(param_change: ParameterModel, txtinout_path: pathlib.Path) -> None:
+def _calibration_units(
+    txtinout_path: pathlib.Path,
+    param_change: ParameterModel
+) -> None:
     '''
     Validate units for a given parameter change against calibration parameters.
     '''
@@ -98,26 +118,19 @@ def _validate_units(param_change: ParameterModel, txtinout_path: pathlib.Path) -
 
     # Get the object type for the parameter
     row = cal_parms_df.loc[cal_parms_df['name'] == name, 'obj_typ']
-    if row.empty:
-        raise ValueError(f"Parameter '{name}' not found in file 'cal_parms.cal'.")
-
     obj_type = row.iloc[0]
-    if not obj_type:
-        raise ValueError("Missing 'obj_typ' column in 'cal_parms.cal' file.")
 
-    # Supported mapping of obj_type → file
+    # Supported mapping of obj_type
     obj_type_files = {
         'hru': 'hru-data.hru',
         'sol': 'hru-data.hru',
         'res': 'reservoir.res',
-        'aqu': 'aquifer.aqu',
+        'aqu': 'aquifer.aqu'
     }
-
     if obj_type not in obj_type_files:
-        supported = ", ".join(obj_type_files.keys())
         raise ValueError(
-            f"Parameter '{name}' does not support units. "
-            f"Only parameters of type [{supported}] support units."
+            f'Parameter "{name}" with obj_type "{obj_type}" in "cal_parms.cal" does not support "units" key. '
+            f'Supported obj_type: [{', '.join(obj_type_files.keys())}].'
         )
 
     file = obj_type_files[obj_type]
@@ -135,13 +148,16 @@ def _validate_units(param_change: ParameterModel, txtinout_path: pathlib.Path) -
     max_unit = max(units)
     if len(df) < max_unit:
         raise ValueError(
-            f"Invalid units for parameter '{name}'. "
-            f"Some ids exceed the maximum available in {file} "
-            f"(requested up to {max_unit}, available {len(df)})."
+            f'Invalid units for parameter "{name}". '
+            f'Some ids exceed the maximum available in {file} '
+            f'(requested up to {max_unit}, available {len(df)}).'
         )
 
 
-def _validate_conditions(param_change: ParameterModel, txtinout_path: pathlib.Path) -> None:
+def _calibration_conditions(
+    txtinout_path: pathlib.Path,
+    param_change: ParameterModel
+) -> None:
     '''
     Validate conditions for a given parameter change against calibration parameters.
     '''
@@ -172,59 +188,69 @@ def _validate_conditions(param_change: ParameterModel, txtinout_path: pathlib.Pa
     for cond_name, cond_values in conditions.items():
         if cond_name not in supported_conditions:
             raise ValueError(
-                f"Condition '{cond_name}' for parameter '{name}' is not supported. "
-                f"Available conditions are: {', '.join(sorted(supported_conditions))}."
+                f'Condition "{cond_name}" for parameter "{name}" is not supported. '
+                f'Available conditions are: {', '.join(sorted(supported_conditions))}.'
             )
 
         valid_values = validators.get(cond_name, set())
         for val in cond_values:
             if val not in valid_values:
                 raise ValueError(
-                    f"Condition '{cond_name}' for parameter '{name}' "
-                    f"has invalid value '{val}'. "
-                    f"Valid values are: {sorted(valid_values)}."
+                    f'Condition "{cond_name}" for parameter "{name}" has invalid value "{val}". '
+                    f'Expected are: {sorted(valid_values)}.'
                 )
 
 
-def _validate_conditions_and_units(parameters: list[ParameterModel], txtinout_path: pathlib.Path) -> None:
+def _calibration_conditions_and_units(
+    txtinout_path: pathlib.Path,
+    parameters: list[ParameterModel]
+) -> None:
     '''
     This function checks:
     - That the parameter exists in the calibration parameters.
     - That the parameter supports units/conditions (based on 'obj_typ').
     - That specified units correspond to valid IDs in the relevant SWAT+ input files.
     - That conditions (if applicable) exist and are valid.
     '''
+
     for param_change in parameters:
         try:
-            _validate_conditions(param_change, txtinout_path)
-            _validate_units(param_change, txtinout_path)
+            _calibration_conditions(
+                txtinout_path=txtinout_path,
+                param_change=param_change
+            )
+            _calibration_units(
+                txtinout_path=txtinout_path,
+                param_change=param_change
+            )
         except ValueError as e:
             raise ValueError(
-                f"{e}\n\n"
-                f"If you want to ignore the validation, set "
-                f"'skip_units_and_conditions_validation=True'"
+                f'{e}\n\n'
+                f'If you want to ignore the validation, set "skip_validation=True"'
             ) from e
 
 
-def _validate_cal_parameters(
-    txtinout_folder: pathlib.Path,
+def _calibration_parameters(
+    txtinout_path: pathlib.Path,
     parameters: list[ParameterBoundedModel] | list[ParameterModel]
 ) -> None:
     '''
     Check if parameters exists in cal_parms.cal
     '''
 
-    file_path = txtinout_folder / "cal_parms.cal"
-
-    if not file_path.exists():
-        raise FileNotFoundError("cal_parms.cal file does not exist in the TxtInOut folder")
+    # Path of cal_parms.cal
+    file_path = txtinout_path / 'cal_parms.cal'
 
-    cal_parms_df = pandas.read_csv(
+    # DataFrame from cal_parms.cal file
+    parms_df = pandas.read_csv(
         filepath_or_buffer=file_path,
         skiprows=2,
         sep=r'\s+'
     )
 
+    # Check validity of input calibration parameter name
     for param in parameters:
-        if param.name not in cal_parms_df['name'].values:
-            raise ValueError(f"The parameter '{param.name}' is not in cal_parms.cal")
+        if param.name not in parms_df['name'].values:
+            raise ValueError(
+                f'Calibration parameter "{param.name}" not found in cal_parms.cal file'
+            )