From f2c2aa9eeaf66fa4321a01664120abfb808360a3 Mon Sep 17 00:00:00 2001 From: Debasish Pal <48341250+debpal@users.noreply.github.com> Date: Fri, 19 Sep 2025 13:53:06 +0300 Subject: [PATCH 1/5] remove typing_extensions dependency because pydantic contains it --- pyproject.toml | 1 - requirements.txt | 1 - 2 files changed, 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 7907c6a..47b08b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,7 +34,6 @@ classifiers = [ ] dependencies = [ "SALib", - "typing-extensions", "pydantic" ] diff --git a/requirements.txt b/requirements.txt index 974e692..9e7cf3a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,4 @@ pytest pytest-cov SALib -typing-extensions pydantic \ No newline at end of file From fb3ef6291f18efb8eaa966d717454a86feb49009 Mon Sep 17 00:00:00 2001 From: Debasish Pal <48341250+debpal@users.noreply.github.com> Date: Fri, 19 Sep 2025 13:54:30 +0300 Subject: [PATCH 2/5] rename files to understand better --- ...ityanalyzer.md => sensitivity_analyzer.md} | 0 .../{txtinoutreader.md => txtinout_reader.md} | 2 +- mkdocs.yml | 4 +- pySWATPlus/__init__.py | 4 +- ...nalyzer.py => cal_sensitivity_analyzer.py} | 2 +- .../{txtinoutreader.py => txtinout_reader.py} | 1520 ++++++++--------- ...metrics.py => test_performance_metrics.py} | 0 ...alyzer.py => test_sensitivity_analyzer.py} | 0 ...inoutreader.py => test_txtinout_reader.py} | 0 9 files changed, 766 insertions(+), 766 deletions(-) rename docs/api/{sensitivityanalyzer.md => sensitivity_analyzer.md} (100%) rename docs/api/{txtinoutreader.md => txtinout_reader.md} (96%) rename pySWATPlus/{calsensitivityanalyzer.py => cal_sensitivity_analyzer.py} (99%) rename pySWATPlus/{txtinoutreader.py => txtinout_reader.py} (97%) rename tests/{test_performancemetrics.py => test_performance_metrics.py} (100%) rename tests/{test_sensitivityanalyzer.py => test_sensitivity_analyzer.py} (100%) rename tests/{test_txtinoutreader.py => test_txtinout_reader.py} (100%) diff --git a/docs/api/sensitivityanalyzer.md b/docs/api/sensitivity_analyzer.md similarity index 100% rename from docs/api/sensitivityanalyzer.md rename to docs/api/sensitivity_analyzer.md diff --git a/docs/api/txtinoutreader.md b/docs/api/txtinout_reader.md similarity index 96% rename from docs/api/txtinoutreader.md rename to docs/api/txtinout_reader.md index eea59fa..4137a28 100644 --- a/docs/api/txtinoutreader.md +++ b/docs/api/txtinout_reader.md @@ -1 +1 @@ -::: pySWATPlus.TxtinoutReader +::: pySWATPlus.TxtinoutReader diff --git a/mkdocs.yml b/mkdocs.yml index 8bf318b..9480f2e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -41,9 +41,9 @@ nav: - Read Output: userguide/read_output.md - API Reference: - - TxtinoutReader: api/txtinoutreader.md + - TxtinoutReader: api/txtinout_reader.md - FileReader: api/filereader.md - - SensitivityAnalyzer: api/sensitivityanalyzer.md + - SensitivityAnalyzer: api/sensitivity_analyzer.md - PerformanceMetrics: api/performance_metrics.md - pySWATPlus.types: api/types.md diff --git a/pySWATPlus/__init__.py b/pySWATPlus/__init__.py index 6ffcc2f..717e8d9 100644 --- a/pySWATPlus/__init__.py +++ b/pySWATPlus/__init__.py @@ -1,6 +1,6 @@ -from .txtinoutreader import TxtinoutReader +from .txtinout_reader import TxtinoutReader from .filereader import FileReader -from .calsensitivityanalyzer import SensitivityAnalyzer +from .cal_sensitivity_analyzer import SensitivityAnalyzer from .performance_metrics import PerformanceMetrics from importlib.metadata import version, PackageNotFoundError diff --git a/pySWATPlus/calsensitivityanalyzer.py b/pySWATPlus/cal_sensitivity_analyzer.py similarity index 99% rename from pySWATPlus/calsensitivityanalyzer.py rename to pySWATPlus/cal_sensitivity_analyzer.py index 9847f86..17d0877 100644 --- a/pySWATPlus/calsensitivityanalyzer.py +++ b/pySWATPlus/cal_sensitivity_analyzer.py @@ -8,7 +8,7 @@ import pathlib from . import utils from . import validators -from .txtinoutreader import TxtinoutReader +from .txtinout_reader import TxtinoutReader class SensitivityAnalyzer(BaseSensitivityAnalyzer): diff --git a/pySWATPlus/txtinoutreader.py b/pySWATPlus/txtinout_reader.py similarity index 97% rename from pySWATPlus/txtinoutreader.py rename to pySWATPlus/txtinout_reader.py index 356ae0b..8c7bd78 100644 --- a/pySWATPlus/txtinoutreader.py +++ b/pySWATPlus/txtinout_reader.py @@ -1,760 +1,760 @@ -import subprocess -import shutil -import pathlib -import typing -import logging -from .filereader import FileReader -from .types import ParametersType, ParameterModel -from . import utils -from . import validators -from datetime import date - -logger = logging.getLogger(__name__) - - -class TxtinoutReader: - ''' - Provide functionality for seamless reading, editing, and writing of - SWAT+ model files located in the `TxtInOut` folder. - ''' - - IGNORED_FILE_PATTERNS: typing.Final[tuple[str, ...]] = tuple( - f'_{suffix}.{ext}' - for suffix in ('day', 'mon', 'yr', 'aa') - for ext in ('txt', 'csv') - ) - - def __init__( - self, - path: str | pathlib.Path - ) -> None: - ''' - Create a TxtinoutReader instance for accessing SWAT+ model files. - - Args: - path (str or Path): Path to the `TxtInOut` folder, which must contain - exactly one SWAT+ executable `.exe` file. - - Raises: - TypeError: If the path is not a valid string or Path, or if the folder contains - zero or multiple `.exe` files. - ''' - - path = utils._ensure_path(path) - - # check if folder exists - if not path.is_dir(): - raise FileNotFoundError('Folder does not exist') - - # check .exe files in the directory - exe_list = [file for file in path.iterdir() if file.suffix == ".exe"] - - # raise error on .exe file - if len(exe_list) != 1: - raise TypeError( - 'Expected exactly one .exe file in the parent folder, but found none or multiple.' - ) - - # find parent directory - self.root_folder = path - self.swat_exe_path = path / exe_list[0] - - def enable_object_in_print_prt( - self, - obj: typing.Optional[str], - daily: bool, - monthly: bool, - yearly: bool, - avann: bool, - allow_unavailable_object: bool = False - ) -> None: - ''' - Update or add an object in the `print.prt` file with specified time frequency flags. - - This method modifies the `print.prt` file in a SWAT+ project to enable or disable output - for a specific object (or all objects if `obj` is None) at specified time frequencies - (daily, monthly, yearly, or average annual). If the object does not exist in the file - and `obj` is not None, it is appended to the end of the file. - - Note: - This input does not provide complete control over `print.prt` outputs. - Some files are internally linked in the SWAT+ model and may still be - generated even when disabled. - - Args: - obj (str or None): The name of the object to update (e.g., 'channel_sd', 'reservoir'). - If `None`, all objects in the `print.prt` file are updated with the specified time frequency settings. - daily (bool): If `True`, enable daily frequency output. - monthly (bool): If `True`, enable monthly frequency output. - yearly (bool): If `True`, enable yearly frequency output. - avann (bool): If `True`, enable average annual frequency output. - allow_unavailable_object (bool, optional): If True, allows adding an object not in - the standard SWAT+ output object list. If False and `obj` is not in the standard list, - a ValueError is raised. Defaults to False. - ''' - - obj_dict = { - 'model_components': ['channel_sd', 'channel_sdmorph', 'aquifer', 'reservoir', 'recall', 'ru', 'hyd', 'water_allo'], - 'basin_model_components': ['basin_sd_cha', 'basin_sd_chamorph', 'basin_aqu', 'basin_res', 'basin_psc'], - 'nutrient_balance': ['basin_nb', 'lsunit_nb', 'hru-lte_nb'], - 'water_balance': ['basin_wb', 'lsunit_wb', 'hru_wb', 'hru-lte_wb'], - 'plant_weather': ['basin_pw', 'lsunit_pw', 'hru_pw', 'hru-lte_pw'], - 'losses': ['basin_ls', 'lsunit_ls', 'hru_ls', 'hru-lte_ls'], - 'salts': ['basin_salt', 'hru_salt', 'ru_salt', 'aqu_salt', 'channel_salt', 'res_salt', 'wetland_salt'], - 'constituents': ['basin_cs', 'hru_cs', 'ru_cs', 'aqu_cs', 'channel_cs', 'res_cs', 'wetland_cs'] - } - - obj_list = [i for v in obj_dict.values() for i in v] - - # Check 'obj' is either string or NoneType - if not (isinstance(obj, str) or obj is None): - raise TypeError(f'Input "obj" to be string type or None, got {type(obj).__name__}') - - # Check 'obj' is valid - if obj and obj not in obj_list and not allow_unavailable_object: - raise ValueError( - f'Object "{obj}" not found in print.prt file. Use allow_unavailable_object=True to proceed.' - ) - - # Time frequency dictionary - time_dict = { - 'daily': daily, - 'monthly': monthly, - 'yearly': yearly, - 'avann': avann - } - - for key, val in time_dict.items(): - if not isinstance(val, bool): - raise TypeError(f'Variable "{key}" for "{obj}" must be a bool value') - - # read all print_prt file, line by line - print_prt_path = self.root_folder / 'print.prt' - new_print_prt = "" - found = False - - # Check if file exists - if not print_prt_path.exists(): - raise FileNotFoundError("print.prt file does not exist") - - with open(print_prt_path, 'r', newline='') as file: - for i, line in enumerate(file, start=1): - if i <= 10: - # Always keep first 10 lines as-is - new_print_prt += line - continue - - stripped = line.strip() - if not stripped: - # Keep blank lines unchanged - new_print_prt += line - continue - - parts = stripped.split() - line_obj = parts[0] - - if obj is None: - # Update all objects - new_print_prt += utils._build_line_to_add(line_obj, daily, monthly, yearly, avann) - elif line_obj == obj: - # obj already exist, replace it in same position - new_print_prt += utils._build_line_to_add(line_obj, daily, monthly, yearly, avann) - found = True - else: - new_print_prt += line - - if not found and obj is not None: - new_print_prt += utils._build_line_to_add(obj, daily, monthly, yearly, avann) - - # store new print_prt - with open(print_prt_path, 'w', newline='') as file: - file.write(new_print_prt) - - def set_begin_and_end_date( - self, - begin_date: date, - end_date: date, - step: typing.Optional[int] = 0 - ) -> None: - ''' - Modify the simulation period by updating - the begin and end dates in the `time.sim` file. - - Args: - begin_date (date): Beginning date of the simulation. - end_date (date): Ending date of the simulation. - step (int, optional): Timestep of the simulation. - 0 = daily - 1 = increment (12 hrs) - 24 = hourly - 96 = 15 mins - 1440 = minute - ''' - if not isinstance(begin_date, date) or not isinstance(end_date, date): - raise TypeError("begin_date and end_date must be datetime.date objects") - - if begin_date >= end_date: - raise ValueError("begin_date must be earlier than end_date") - - if not isinstance(step, int): - raise TypeError("step must be an integer") - valid_steps = [0, 1, 24, 96, 1440] - if step not in valid_steps: - raise ValueError(f"Invalid step: {step}. Must be one of {valid_steps}") - - # Extract years and Julian days - begin_day = begin_date.timetuple().tm_yday - begin_year = begin_date.year - end_day = end_date.timetuple().tm_yday - end_year = end_date.year - - nth_line = 3 # line in time.sim file to modify - - time_sim_path = self.root_folder / 'time.sim' - - # Check if file exists - if not time_sim_path.exists(): - raise FileNotFoundError("time.sim file does not exist") - - # Open the file in read mode and read its contents - with open(time_sim_path, 'r') as file: - lines = file.readlines() - - # Split existing line - elements = lines[2].split() - - # Update values - elements[0] = str(begin_day) - elements[1] = str(begin_year) - elements[2] = str(end_day) - elements[3] = str(end_year) - elements[4] = str(step) - - # Reconstruct the result string while maintaining spaces - result_string = '{: >8} {: >10} {: >10} {: >10} {: >10} \n'.format(*elements) - - lines[nth_line - 1] = result_string - - with open(time_sim_path, 'w') as file: - file.writelines(lines) - - def set_warmup_year( - self, - warmup: int - ) -> None: - ''' - Modify the warm-up years in the `print.prt` file. - - Args: - warmup (int): A positive integer representing the number of years - the simulation will use for warm-up (e.g., 1). - - Raises: - ValueError: If the warmup year is less than or equal to 0. - ''' - - if not isinstance(warmup, int): - raise TypeError('warmup must be an integer value') - if warmup <= 0: - raise ValueError('warmup must be a positive integer') - - print_prt_path = self.root_folder / 'print.prt' - - # Check if file exists - if not print_prt_path.exists(): - raise FileNotFoundError("print.prt file does not exist") - - # Open the file in read mode and read its contents - with open(print_prt_path, 'r') as file: - lines = file.readlines() - - nth_line = 3 - year_line = lines[nth_line - 1] - - # Split the input string by spaces - elements = year_line.split() - - # Modify warmup year - elements[0] = str(warmup) - - # Reconstruct the result string while maintaining spaces - result_string = '{: <12} {: <11} {: <11} {: <10} {: <10} {: <10} \n'.format(*elements) - - lines[nth_line - 1] = result_string - - with open(print_prt_path, 'w') as file: - file.writelines(lines) - - def _enable_disable_csv_print( - self, - enable: bool = True - ) -> None: - ''' - Enable or disable print in the `print.prt` file. - ''' - - # read - nth_line = 7 - - print_prt_path = self.root_folder / 'print.prt' - - # Check if file exists - if not print_prt_path.exists(): - raise FileNotFoundError("print.prt file does not exist") - - # Open the file in read mode and read its contents - with open(print_prt_path, 'r') as file: - lines = file.readlines() - - if enable: - lines[nth_line - 1] = 'y' + lines[nth_line - 1][1:] - else: - lines[nth_line - 1] = 'n' + lines[nth_line - 1][1:] - - with open(print_prt_path, 'w') as file: - file.writelines(lines) - - def enable_csv_print( - self - ) -> None: - ''' - Enable print in the `print.prt` file. - ''' - - self._enable_disable_csv_print(enable=True) - - def disable_csv_print( - self - ) -> None: - ''' - Disable print in the `print.prt` file. - ''' - - self._enable_disable_csv_print(enable=False) - - def register_file( - self, - filename: str, - has_units: bool, - ) -> FileReader: - ''' - Register a file to work with in the SWAT+ model. - - Args: - filename (str): Path to the file to register, located in the `TxtInOut` folder. - has_units (bool): If True, the second row of the file contains units. - - Returns: - A FileReader instance for the registered file. - ''' - - file_path = self.root_folder / filename - - return FileReader(file_path, has_units) - - def copy_required_files( - self, - target_dir: str | pathlib.Path, - ) -> pathlib.Path: - ''' - Copy the required file from the input folder associated with the - `TxtinoutReader` instance to the specified directory for SWAT+ simulation. - - Args: - target_dir (str or Path): Path to the directory where the required files will be copied. - - Returns: - The path to the target directory containing the copied files. - ''' - - target_dir = utils._ensure_path(target_dir) - - # Enforce that it's a directory, not a file path - if target_dir.suffix: - raise ValueError(f"`target_dir` must be a directory, not a file path: {target_dir}") - - # Create the directory if it does not exist and copy necessary files - target_dir.mkdir(parents=True, exist_ok=True) - - dest_path = pathlib.Path(target_dir) - - if any(dest_path.iterdir()): - raise FileExistsError(f"Target directory {dest_path} is not empty.") - - # Copy files from source folder - for file in self.root_folder.iterdir(): - if file.is_dir() or file.name.endswith(self.IGNORED_FILE_PATTERNS): - continue - shutil.copy2(file, dest_path / file.name) - - return dest_path - - def _write_calibration_file( - self, - parameters: list[ParameterModel] - ) -> None: - ''' - Writes `calibration.cal` file with parameter changes. - ''' - - outfile = self.root_folder / "calibration.cal" - - # If calibration.cal exists, remove it (always recreate) - if outfile.exists(): - outfile.unlink() - - # make sure calibration.cal is enabled in file.cio - self._add_or_remove_calibration_cal_to_file_cio(add=True) - - # Number of parameters (number of rows in the DataFrame) - num_parameters = len(parameters) - - # Column widths for right-alignment - col_widths = { - "NAME": 12, # left-aligned - "CHG_TYPE": 8, - "VAL": 16, - "CONDS": 16, - "LYR1": 8, - "LYR2": 8, - "YEAR1": 8, - "YEAR2": 8, - "DAY1": 8, - "DAY2": 8, - "OBJ_TOT": 8 - } - - calibration_cal_rows = [] - for change in parameters: - units = change.units - - # Convert to compact representation - compacted_units = utils._compact_units(units) if units else [] - - # get conditions - parsed_conditions = utils._parse_conditions(change) - - calibration_cal_rows.append({ - "NAME": change.name, - "CHG_TYPE": change.change_type, - "VAL": change.value, - "CONDS": len(parsed_conditions), - "LYR1": 0, - "LYR2": 0, - "YEAR1": 0, - "YEAR2": 0, - "DAY1": 0, - "DAY2": 0, - "OBJ_TOT": len(compacted_units), - "OBJ_LIST": compacted_units, # Store the compacted units - "PARSED_CONDITIONS": parsed_conditions - }) - - with open(outfile, "w") as f: - # Write header - f.write(f"Number of parameters:\n{num_parameters}\n") - headers = ( - f"{'NAME':<12}{'CHG_TYPE':<21}{'VAL':<14}{'CONDS':<9}" - f"{'LYR1':<8}{'LYR2':<7}{'YEAR1':<8}{'YEAR2':<9}" - f"{'DAY1':<8}{'DAY2':<5}{'OBJ_TOT':>7}" - ) - f.write(f"{headers}\n") - - # Write rows - for row in calibration_cal_rows: - line = "" - for col in ["NAME", "CHG_TYPE", "VAL", "CONDS", "LYR1", "LYR2", - "YEAR1", "YEAR2", "DAY1", "DAY2", "OBJ_TOT"]: - if col == "NAME": - line += f"{row[col]:<{col_widths[col]}}" # left-align - elif col == "VAL" and isinstance(row[col], float): - line += utils._format_val_field(typing.cast(float, row[col])) # special VAL formatting - else: - line += f"{row[col]:>{col_widths[col]}}" # right-align numeric columns - - # Append compacted units at the end (space-separated) - if row["OBJ_LIST"]: - line += " " + " ".join(str(u) for u in typing.cast(list[str], row["OBJ_LIST"])) - - if row["PARSED_CONDITIONS"]: - parsed_conditions = typing.cast(list[str], row["PARSED_CONDITIONS"]) - line += "\n" + "\n".join(parsed_conditions) - - f.write(line + "\n") - - def _add_or_remove_calibration_cal_to_file_cio( - self, - add: bool - ) -> None: - ''' - Adds or removes the calibration line to 'file.cio' - ''' - file_path = self.root_folder / "file.cio" - if not file_path.exists(): - raise FileNotFoundError("file.cio file does not exist in the TxtInOut folder") - - fmt = ( - f"{'{:<18}'}" # chg - f"{'{:<18}'}" # cal_parms.cal / null - f"{'{:<18}'}" # calibration.cal - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<18}'}" # null - f"{'{:<4}'}" # null - ) - - # Prepare the values for the line - cal_line_values = [ - "chg", - "cal_parms.cal" if add else "null", - "calibration.cal", - ] + ["null"] * 9 # Fill remaining columns with null - - line_to_add = fmt.format(*cal_line_values) - - line_index = 21 - - # Read all lines - with file_path.open("r") as f: - lines = f.readlines() - - # Safety check: ensure the file has enough lines - if line_index >= len(lines): - raise IndexError(f"The file only has {len(lines)} lines, cannot replace line {line_index+1}.") - - # Replace the line, ensure it ends with a newline - lines[line_index] = line_to_add.rstrip() + "\n" - - # Write back - with file_path.open("w") as f: - f.writelines(lines) - - def _apply_swat_configuration( - self, - begin_and_end_date: typing.Optional[dict[str, typing.Any]] = None, - warmup: typing.Optional[int] = None, - print_prt_control: typing.Optional[dict[str, dict[str, bool]]] = None - ) -> None: - ''' - Sets begin and end year for the simulation, the warm-up period, and toggles the elements in print.prt file - ''' - # Set simulation range time - if begin_and_end_date is not None: - if not isinstance(begin_and_end_date, dict): - raise TypeError('begin_and_end_date must be a dictionary') - - self.set_begin_and_end_date(**begin_and_end_date) - - # Set warmup period - if warmup is not None: - self.set_warmup_year( - warmup=warmup - ) - - # Update print.prt file to write output - if print_prt_control is not None: - if not isinstance(print_prt_control, dict): - raise TypeError('print_prt_control must be a dictionary') - if len(print_prt_control) == 0: - raise ValueError('print_prt_control cannot be an empty dictionary') - default_dict = { - 'daily': True, - 'monthly': True, - 'yearly': True, - 'avann': True - } - for key, val in print_prt_control.items(): - if not isinstance(val, dict): - raise ValueError(f'Value of key "{key}" must be a dictionary') - if len(val) == 0: - raise ValueError(f'Value of key "{key}" cannot be an empty dictionary') - key_dict = default_dict.copy() - for sub_key, sub_val in val.items(): - if sub_key not in key_dict: - raise ValueError(f'Sub-key "{sub_key}" for key "{key}" is not valid') - key_dict[sub_key] = sub_val - self.enable_object_in_print_prt( - obj=key, - daily=key_dict['daily'], - monthly=key_dict['monthly'], - yearly=key_dict['yearly'], - avann=key_dict['avann'] - ) - - def _run_swat( - self, - ) -> None: - ''' - Run the SWAT+ simulation. - ''' - - # Run simulation - try: - process = subprocess.Popen( - [str(self.swat_exe_path.resolve())], - cwd=str(self.root_folder.resolve()), # Sets working dir just for this subprocess - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - bufsize=1, # Line buffered - text=True # Handles text output - ) - - # Real-time output handling - if process.stdout: - for line in process.stdout: - clean_line = line.strip() - if clean_line: - logger.info(clean_line) - - # Wait for process and check for errors - return_code = process.wait() - if return_code != 0: - stderr = process.stderr.read() if process.stderr else None - raise subprocess.CalledProcessError( - return_code, - process.args, - stderr=stderr - ) - - except Exception as e: - logger.error(f"Failed to run SWAT: {str(e)}") - raise - - def run_swat( - self, - target_dir: typing.Optional[str | pathlib.Path], - parameters: typing.Optional[ParametersType] = None, - begin_and_end_date: typing.Optional[dict[str, typing.Any]] = None, - warmup: typing.Optional[int] = None, - print_prt_control: typing.Optional[dict[str, dict[str, bool]]] = None, - skip_validation: bool = False - ) -> pathlib.Path: - ''' - Run the SWAT+ simulation with optional parameter changes. - - Args: - - target_dir (str or Path, optional): Path to the directory where the simulation will be done. - If None, the simulation runs directly in the current folder. - - parameters (ParametersType, optional): Nested dictionary specifying parameter changes. - - The `parameters` dictionary should follow this structure: - - ```python - parameters = [ - { - "name": str, # Name of the parameter to which the changes will be applied - "value": float # The value to apply to the parameter - "change_type": str, # One of: 'absval', 'abschg', 'pctchg' - "units": Iterable[int], # (Optional) An optional list of unit IDs to constrain the parameter change. - **Unit IDs should be 1-based**, i.e., the first object has ID 1. - "conditions": dict[str: list[str]], # (Optional) A dictionary of conditions to apply to the parameter change. - }, - ... - ] - ``` - - begin_and_end_date (dict, optional): Dictionary defining the simulation period (and optionally timestep). - Must contain the following keys: - - - `begin_date` (date): Start date of the simulation. - - `end_date` (date): End date of the simulation. - - `step` (int, optional): Timestep of the simulation. Defaults to `0` (daily) if not provided. - Allowed values: - - `0` = daily - - `1` = 12-hour increments - - `24` = hourly - - `96` = 15-minute increments - - `1440` = minute - - warmup (int): A positive integer representing the number of warm-up years (e.g., 1). - - print_prt_control (dict[str, dict[str, bool]], optional): A dictionary to control output printing in the `print.prt` file. - Each outer key is an object name from `print.prt` (e.g., 'channel_sd', 'basin_wb'). - Each value is a dictionary with keys `daily`, `monthly`, `yearly`, or `avann`, mapped to boolean values. - Set to `False` to disable printing for that time step; defaults to `True` if not specified. - An error is raised if an outer key has an empty dictionary. - The time step keys represent: - - - `daily`: Output for each day of the simulation. - - `monthly`: Output aggregated for each month. - - `yearly`: Output aggregated for each year. - - `avann`: Average annual output over the entire simulation period. - - skip_validation (bool): If `True`, skip validation of units and conditions in parameter changes. - - - Returns: - Path where the SWAT+ simulation was executed. - - Example: - ```python - simulation = pySWATPlus.TxtinoutReader.run_swat( - target_dir="C:\\\\Users\\\\Username\\\\simulation_folder", - parameters = [ - { - "name": 'perco', - "change_type": "absval", - "value": 0.5, - "conditions": {"hsg": ["A"]} - }, - { - 'name': 'bf_max', - "change_type": "absval", - "value": 0.3, - "units": range(1, 194) - } - ] - begin_and_end_date={ - "begin_date": date(2012, 1, 1), - "end_date": date(2016, 12, 31) - # step defaults to 0 (daily) - }, - warmup=1, - print_prt_control = { - 'channel_sd': {'daily': False}, - 'channel_sdmorph': {'monthly': False} - } - ) - ``` - ''' - - if target_dir: - _target_dir = utils._ensure_path(target_dir) - - # Resolve to absolute paths - if self.root_folder.resolve() == _target_dir.resolve(): - raise ValueError( - "`target_dir` parameter must be different from the existing TxtInOut path!" - ) - run_path = self.copy_required_files(target_dir=target_dir) - - # Initialize new TxtinoutReader class - reader = TxtinoutReader(run_path) - else: - reader = self - run_path = self.root_folder - - # Apply SWAT+ configuration changes - reader._apply_swat_configuration(begin_and_end_date, warmup, print_prt_control) - - if parameters: - _params = [ParameterModel(**param) for param in parameters] - - validators._validate_cal_parameters(reader.root_folder, _params) - - if not skip_validation: - validators._validate_conditions_and_units(_params, reader.root_folder) - - reader._write_calibration_file(_params) - - # Run simulation - reader._run_swat() - - return run_path +import subprocess +import shutil +import pathlib +import typing +import logging +from .filereader import FileReader +from .types import ParametersType, ParameterModel +from . import utils +from . import validators +from datetime import date + +logger = logging.getLogger(__name__) + + +class TxtinoutReader: + ''' + Provide functionality for seamless reading, editing, and writing of + SWAT+ model files located in the `TxtInOut` folder. + ''' + + IGNORED_FILE_PATTERNS: typing.Final[tuple[str, ...]] = tuple( + f'_{suffix}.{ext}' + for suffix in ('day', 'mon', 'yr', 'aa') + for ext in ('txt', 'csv') + ) + + def __init__( + self, + path: str | pathlib.Path + ) -> None: + ''' + Create a TxtinoutReader instance for accessing SWAT+ model files. + + Args: + path (str or Path): Path to the `TxtInOut` folder, which must contain + exactly one SWAT+ executable `.exe` file. + + Raises: + TypeError: If the path is not a valid string or Path, or if the folder contains + zero or multiple `.exe` files. + ''' + + path = utils._ensure_path(path) + + # check if folder exists + if not path.is_dir(): + raise FileNotFoundError('Folder does not exist') + + # check .exe files in the directory + exe_list = [file for file in path.iterdir() if file.suffix == ".exe"] + + # raise error on .exe file + if len(exe_list) != 1: + raise TypeError( + 'Expected exactly one .exe file in the parent folder, but found none or multiple.' + ) + + # find parent directory + self.root_folder = path + self.swat_exe_path = path / exe_list[0] + + def enable_object_in_print_prt( + self, + obj: typing.Optional[str], + daily: bool, + monthly: bool, + yearly: bool, + avann: bool, + allow_unavailable_object: bool = False + ) -> None: + ''' + Update or add an object in the `print.prt` file with specified time frequency flags. + + This method modifies the `print.prt` file in a SWAT+ project to enable or disable output + for a specific object (or all objects if `obj` is None) at specified time frequencies + (daily, monthly, yearly, or average annual). If the object does not exist in the file + and `obj` is not None, it is appended to the end of the file. + + Note: + This input does not provide complete control over `print.prt` outputs. + Some files are internally linked in the SWAT+ model and may still be + generated even when disabled. + + Args: + obj (str or None): The name of the object to update (e.g., 'channel_sd', 'reservoir'). + If `None`, all objects in the `print.prt` file are updated with the specified time frequency settings. + daily (bool): If `True`, enable daily frequency output. + monthly (bool): If `True`, enable monthly frequency output. + yearly (bool): If `True`, enable yearly frequency output. + avann (bool): If `True`, enable average annual frequency output. + allow_unavailable_object (bool, optional): If True, allows adding an object not in + the standard SWAT+ output object list. If False and `obj` is not in the standard list, + a ValueError is raised. Defaults to False. + ''' + + obj_dict = { + 'model_components': ['channel_sd', 'channel_sdmorph', 'aquifer', 'reservoir', 'recall', 'ru', 'hyd', 'water_allo'], + 'basin_model_components': ['basin_sd_cha', 'basin_sd_chamorph', 'basin_aqu', 'basin_res', 'basin_psc'], + 'nutrient_balance': ['basin_nb', 'lsunit_nb', 'hru-lte_nb'], + 'water_balance': ['basin_wb', 'lsunit_wb', 'hru_wb', 'hru-lte_wb'], + 'plant_weather': ['basin_pw', 'lsunit_pw', 'hru_pw', 'hru-lte_pw'], + 'losses': ['basin_ls', 'lsunit_ls', 'hru_ls', 'hru-lte_ls'], + 'salts': ['basin_salt', 'hru_salt', 'ru_salt', 'aqu_salt', 'channel_salt', 'res_salt', 'wetland_salt'], + 'constituents': ['basin_cs', 'hru_cs', 'ru_cs', 'aqu_cs', 'channel_cs', 'res_cs', 'wetland_cs'] + } + + obj_list = [i for v in obj_dict.values() for i in v] + + # Check 'obj' is either string or NoneType + if not (isinstance(obj, str) or obj is None): + raise TypeError(f'Input "obj" to be string type or None, got {type(obj).__name__}') + + # Check 'obj' is valid + if obj and obj not in obj_list and not allow_unavailable_object: + raise ValueError( + f'Object "{obj}" not found in print.prt file. Use allow_unavailable_object=True to proceed.' + ) + + # Time frequency dictionary + time_dict = { + 'daily': daily, + 'monthly': monthly, + 'yearly': yearly, + 'avann': avann + } + + for key, val in time_dict.items(): + if not isinstance(val, bool): + raise TypeError(f'Variable "{key}" for "{obj}" must be a bool value') + + # read all print_prt file, line by line + print_prt_path = self.root_folder / 'print.prt' + new_print_prt = "" + found = False + + # Check if file exists + if not print_prt_path.exists(): + raise FileNotFoundError("print.prt file does not exist") + + with open(print_prt_path, 'r', newline='') as file: + for i, line in enumerate(file, start=1): + if i <= 10: + # Always keep first 10 lines as-is + new_print_prt += line + continue + + stripped = line.strip() + if not stripped: + # Keep blank lines unchanged + new_print_prt += line + continue + + parts = stripped.split() + line_obj = parts[0] + + if obj is None: + # Update all objects + new_print_prt += utils._build_line_to_add(line_obj, daily, monthly, yearly, avann) + elif line_obj == obj: + # obj already exist, replace it in same position + new_print_prt += utils._build_line_to_add(line_obj, daily, monthly, yearly, avann) + found = True + else: + new_print_prt += line + + if not found and obj is not None: + new_print_prt += utils._build_line_to_add(obj, daily, monthly, yearly, avann) + + # store new print_prt + with open(print_prt_path, 'w', newline='') as file: + file.write(new_print_prt) + + def set_begin_and_end_date( + self, + begin_date: date, + end_date: date, + step: typing.Optional[int] = 0 + ) -> None: + ''' + Modify the simulation period by updating + the begin and end dates in the `time.sim` file. + + Args: + begin_date (date): Beginning date of the simulation. + end_date (date): Ending date of the simulation. + step (int, optional): Timestep of the simulation. + 0 = daily + 1 = increment (12 hrs) + 24 = hourly + 96 = 15 mins + 1440 = minute + ''' + if not isinstance(begin_date, date) or not isinstance(end_date, date): + raise TypeError("begin_date and end_date must be datetime.date objects") + + if begin_date >= end_date: + raise ValueError("begin_date must be earlier than end_date") + + if not isinstance(step, int): + raise TypeError("step must be an integer") + valid_steps = [0, 1, 24, 96, 1440] + if step not in valid_steps: + raise ValueError(f"Invalid step: {step}. Must be one of {valid_steps}") + + # Extract years and Julian days + begin_day = begin_date.timetuple().tm_yday + begin_year = begin_date.year + end_day = end_date.timetuple().tm_yday + end_year = end_date.year + + nth_line = 3 # line in time.sim file to modify + + time_sim_path = self.root_folder / 'time.sim' + + # Check if file exists + if not time_sim_path.exists(): + raise FileNotFoundError("time.sim file does not exist") + + # Open the file in read mode and read its contents + with open(time_sim_path, 'r') as file: + lines = file.readlines() + + # Split existing line + elements = lines[2].split() + + # Update values + elements[0] = str(begin_day) + elements[1] = str(begin_year) + elements[2] = str(end_day) + elements[3] = str(end_year) + elements[4] = str(step) + + # Reconstruct the result string while maintaining spaces + result_string = '{: >8} {: >10} {: >10} {: >10} {: >10} \n'.format(*elements) + + lines[nth_line - 1] = result_string + + with open(time_sim_path, 'w') as file: + file.writelines(lines) + + def set_warmup_year( + self, + warmup: int + ) -> None: + ''' + Modify the warm-up years in the `print.prt` file. + + Args: + warmup (int): A positive integer representing the number of years + the simulation will use for warm-up (e.g., 1). + + Raises: + ValueError: If the warmup year is less than or equal to 0. + ''' + + if not isinstance(warmup, int): + raise TypeError('warmup must be an integer value') + if warmup <= 0: + raise ValueError('warmup must be a positive integer') + + print_prt_path = self.root_folder / 'print.prt' + + # Check if file exists + if not print_prt_path.exists(): + raise FileNotFoundError("print.prt file does not exist") + + # Open the file in read mode and read its contents + with open(print_prt_path, 'r') as file: + lines = file.readlines() + + nth_line = 3 + year_line = lines[nth_line - 1] + + # Split the input string by spaces + elements = year_line.split() + + # Modify warmup year + elements[0] = str(warmup) + + # Reconstruct the result string while maintaining spaces + result_string = '{: <12} {: <11} {: <11} {: <10} {: <10} {: <10} \n'.format(*elements) + + lines[nth_line - 1] = result_string + + with open(print_prt_path, 'w') as file: + file.writelines(lines) + + def _enable_disable_csv_print( + self, + enable: bool = True + ) -> None: + ''' + Enable or disable print in the `print.prt` file. + ''' + + # read + nth_line = 7 + + print_prt_path = self.root_folder / 'print.prt' + + # Check if file exists + if not print_prt_path.exists(): + raise FileNotFoundError("print.prt file does not exist") + + # Open the file in read mode and read its contents + with open(print_prt_path, 'r') as file: + lines = file.readlines() + + if enable: + lines[nth_line - 1] = 'y' + lines[nth_line - 1][1:] + else: + lines[nth_line - 1] = 'n' + lines[nth_line - 1][1:] + + with open(print_prt_path, 'w') as file: + file.writelines(lines) + + def enable_csv_print( + self + ) -> None: + ''' + Enable print in the `print.prt` file. + ''' + + self._enable_disable_csv_print(enable=True) + + def disable_csv_print( + self + ) -> None: + ''' + Disable print in the `print.prt` file. + ''' + + self._enable_disable_csv_print(enable=False) + + def register_file( + self, + filename: str, + has_units: bool, + ) -> FileReader: + ''' + Register a file to work with in the SWAT+ model. + + Args: + filename (str): Path to the file to register, located in the `TxtInOut` folder. + has_units (bool): If True, the second row of the file contains units. + + Returns: + A FileReader instance for the registered file. + ''' + + file_path = self.root_folder / filename + + return FileReader(file_path, has_units) + + def copy_required_files( + self, + target_dir: str | pathlib.Path, + ) -> pathlib.Path: + ''' + Copy the required file from the input folder associated with the + `TxtinoutReader` instance to the specified directory for SWAT+ simulation. + + Args: + target_dir (str or Path): Path to the directory where the required files will be copied. + + Returns: + The path to the target directory containing the copied files. + ''' + + target_dir = utils._ensure_path(target_dir) + + # Enforce that it's a directory, not a file path + if target_dir.suffix: + raise ValueError(f"`target_dir` must be a directory, not a file path: {target_dir}") + + # Create the directory if it does not exist and copy necessary files + target_dir.mkdir(parents=True, exist_ok=True) + + dest_path = pathlib.Path(target_dir) + + if any(dest_path.iterdir()): + raise FileExistsError(f"Target directory {dest_path} is not empty.") + + # Copy files from source folder + for file in self.root_folder.iterdir(): + if file.is_dir() or file.name.endswith(self.IGNORED_FILE_PATTERNS): + continue + shutil.copy2(file, dest_path / file.name) + + return dest_path + + def _write_calibration_file( + self, + parameters: list[ParameterModel] + ) -> None: + ''' + Writes `calibration.cal` file with parameter changes. + ''' + + outfile = self.root_folder / "calibration.cal" + + # If calibration.cal exists, remove it (always recreate) + if outfile.exists(): + outfile.unlink() + + # make sure calibration.cal is enabled in file.cio + self._add_or_remove_calibration_cal_to_file_cio(add=True) + + # Number of parameters (number of rows in the DataFrame) + num_parameters = len(parameters) + + # Column widths for right-alignment + col_widths = { + "NAME": 12, # left-aligned + "CHG_TYPE": 8, + "VAL": 16, + "CONDS": 16, + "LYR1": 8, + "LYR2": 8, + "YEAR1": 8, + "YEAR2": 8, + "DAY1": 8, + "DAY2": 8, + "OBJ_TOT": 8 + } + + calibration_cal_rows = [] + for change in parameters: + units = change.units + + # Convert to compact representation + compacted_units = utils._compact_units(units) if units else [] + + # get conditions + parsed_conditions = utils._parse_conditions(change) + + calibration_cal_rows.append({ + "NAME": change.name, + "CHG_TYPE": change.change_type, + "VAL": change.value, + "CONDS": len(parsed_conditions), + "LYR1": 0, + "LYR2": 0, + "YEAR1": 0, + "YEAR2": 0, + "DAY1": 0, + "DAY2": 0, + "OBJ_TOT": len(compacted_units), + "OBJ_LIST": compacted_units, # Store the compacted units + "PARSED_CONDITIONS": parsed_conditions + }) + + with open(outfile, "w") as f: + # Write header + f.write(f"Number of parameters:\n{num_parameters}\n") + headers = ( + f"{'NAME':<12}{'CHG_TYPE':<21}{'VAL':<14}{'CONDS':<9}" + f"{'LYR1':<8}{'LYR2':<7}{'YEAR1':<8}{'YEAR2':<9}" + f"{'DAY1':<8}{'DAY2':<5}{'OBJ_TOT':>7}" + ) + f.write(f"{headers}\n") + + # Write rows + for row in calibration_cal_rows: + line = "" + for col in ["NAME", "CHG_TYPE", "VAL", "CONDS", "LYR1", "LYR2", + "YEAR1", "YEAR2", "DAY1", "DAY2", "OBJ_TOT"]: + if col == "NAME": + line += f"{row[col]:<{col_widths[col]}}" # left-align + elif col == "VAL" and isinstance(row[col], float): + line += utils._format_val_field(typing.cast(float, row[col])) # special VAL formatting + else: + line += f"{row[col]:>{col_widths[col]}}" # right-align numeric columns + + # Append compacted units at the end (space-separated) + if row["OBJ_LIST"]: + line += " " + " ".join(str(u) for u in typing.cast(list[str], row["OBJ_LIST"])) + + if row["PARSED_CONDITIONS"]: + parsed_conditions = typing.cast(list[str], row["PARSED_CONDITIONS"]) + line += "\n" + "\n".join(parsed_conditions) + + f.write(line + "\n") + + def _add_or_remove_calibration_cal_to_file_cio( + self, + add: bool + ) -> None: + ''' + Adds or removes the calibration line to 'file.cio' + ''' + file_path = self.root_folder / "file.cio" + if not file_path.exists(): + raise FileNotFoundError("file.cio file does not exist in the TxtInOut folder") + + fmt = ( + f"{'{:<18}'}" # chg + f"{'{:<18}'}" # cal_parms.cal / null + f"{'{:<18}'}" # calibration.cal + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<18}'}" # null + f"{'{:<4}'}" # null + ) + + # Prepare the values for the line + cal_line_values = [ + "chg", + "cal_parms.cal" if add else "null", + "calibration.cal", + ] + ["null"] * 9 # Fill remaining columns with null + + line_to_add = fmt.format(*cal_line_values) + + line_index = 21 + + # Read all lines + with file_path.open("r") as f: + lines = f.readlines() + + # Safety check: ensure the file has enough lines + if line_index >= len(lines): + raise IndexError(f"The file only has {len(lines)} lines, cannot replace line {line_index+1}.") + + # Replace the line, ensure it ends with a newline + lines[line_index] = line_to_add.rstrip() + "\n" + + # Write back + with file_path.open("w") as f: + f.writelines(lines) + + def _apply_swat_configuration( + self, + begin_and_end_date: typing.Optional[dict[str, typing.Any]] = None, + warmup: typing.Optional[int] = None, + print_prt_control: typing.Optional[dict[str, dict[str, bool]]] = None + ) -> None: + ''' + Sets begin and end year for the simulation, the warm-up period, and toggles the elements in print.prt file + ''' + # Set simulation range time + if begin_and_end_date is not None: + if not isinstance(begin_and_end_date, dict): + raise TypeError('begin_and_end_date must be a dictionary') + + self.set_begin_and_end_date(**begin_and_end_date) + + # Set warmup period + if warmup is not None: + self.set_warmup_year( + warmup=warmup + ) + + # Update print.prt file to write output + if print_prt_control is not None: + if not isinstance(print_prt_control, dict): + raise TypeError('print_prt_control must be a dictionary') + if len(print_prt_control) == 0: + raise ValueError('print_prt_control cannot be an empty dictionary') + default_dict = { + 'daily': True, + 'monthly': True, + 'yearly': True, + 'avann': True + } + for key, val in print_prt_control.items(): + if not isinstance(val, dict): + raise ValueError(f'Value of key "{key}" must be a dictionary') + if len(val) == 0: + raise ValueError(f'Value of key "{key}" cannot be an empty dictionary') + key_dict = default_dict.copy() + for sub_key, sub_val in val.items(): + if sub_key not in key_dict: + raise ValueError(f'Sub-key "{sub_key}" for key "{key}" is not valid') + key_dict[sub_key] = sub_val + self.enable_object_in_print_prt( + obj=key, + daily=key_dict['daily'], + monthly=key_dict['monthly'], + yearly=key_dict['yearly'], + avann=key_dict['avann'] + ) + + def _run_swat( + self, + ) -> None: + ''' + Run the SWAT+ simulation. + ''' + + # Run simulation + try: + process = subprocess.Popen( + [str(self.swat_exe_path.resolve())], + cwd=str(self.root_folder.resolve()), # Sets working dir just for this subprocess + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=1, # Line buffered + text=True # Handles text output + ) + + # Real-time output handling + if process.stdout: + for line in process.stdout: + clean_line = line.strip() + if clean_line: + logger.info(clean_line) + + # Wait for process and check for errors + return_code = process.wait() + if return_code != 0: + stderr = process.stderr.read() if process.stderr else None + raise subprocess.CalledProcessError( + return_code, + process.args, + stderr=stderr + ) + + except Exception as e: + logger.error(f"Failed to run SWAT: {str(e)}") + raise + + def run_swat( + self, + target_dir: typing.Optional[str | pathlib.Path], + parameters: typing.Optional[ParametersType] = None, + begin_and_end_date: typing.Optional[dict[str, typing.Any]] = None, + warmup: typing.Optional[int] = None, + print_prt_control: typing.Optional[dict[str, dict[str, bool]]] = None, + skip_validation: bool = False + ) -> pathlib.Path: + ''' + Run the SWAT+ simulation with optional parameter changes. + + Args: + + target_dir (str or Path, optional): Path to the directory where the simulation will be done. + If None, the simulation runs directly in the current folder. + + parameters (ParametersType, optional): Nested dictionary specifying parameter changes. + + The `parameters` dictionary should follow this structure: + + ```python + parameters = [ + { + "name": str, # Name of the parameter to which the changes will be applied + "value": float # The value to apply to the parameter + "change_type": str, # One of: 'absval', 'abschg', 'pctchg' + "units": Iterable[int], # (Optional) An optional list of unit IDs to constrain the parameter change. + **Unit IDs should be 1-based**, i.e., the first object has ID 1. + "conditions": dict[str: list[str]], # (Optional) A dictionary of conditions to apply to the parameter change. + }, + ... + ] + ``` + + begin_and_end_date (dict, optional): Dictionary defining the simulation period (and optionally timestep). + Must contain the following keys: + + - `begin_date` (date): Start date of the simulation. + - `end_date` (date): End date of the simulation. + - `step` (int, optional): Timestep of the simulation. Defaults to `0` (daily) if not provided. + Allowed values: + - `0` = daily + - `1` = 12-hour increments + - `24` = hourly + - `96` = 15-minute increments + - `1440` = minute + + warmup (int): A positive integer representing the number of warm-up years (e.g., 1). + + print_prt_control (dict[str, dict[str, bool]], optional): A dictionary to control output printing in the `print.prt` file. + Each outer key is an object name from `print.prt` (e.g., 'channel_sd', 'basin_wb'). + Each value is a dictionary with keys `daily`, `monthly`, `yearly`, or `avann`, mapped to boolean values. + Set to `False` to disable printing for that time step; defaults to `True` if not specified. + An error is raised if an outer key has an empty dictionary. + The time step keys represent: + + - `daily`: Output for each day of the simulation. + - `monthly`: Output aggregated for each month. + - `yearly`: Output aggregated for each year. + - `avann`: Average annual output over the entire simulation period. + + skip_validation (bool): If `True`, skip validation of units and conditions in parameter changes. + + + Returns: + Path where the SWAT+ simulation was executed. + + Example: + ```python + simulation = pySWATPlus.TxtinoutReader.run_swat( + target_dir="C:\\\\Users\\\\Username\\\\simulation_folder", + parameters = [ + { + "name": 'perco', + "change_type": "absval", + "value": 0.5, + "conditions": {"hsg": ["A"]} + }, + { + 'name': 'bf_max', + "change_type": "absval", + "value": 0.3, + "units": range(1, 194) + } + ] + begin_and_end_date={ + "begin_date": date(2012, 1, 1), + "end_date": date(2016, 12, 31) + # step defaults to 0 (daily) + }, + warmup=1, + print_prt_control = { + 'channel_sd': {'daily': False}, + 'channel_sdmorph': {'monthly': False} + } + ) + ``` + ''' + + if target_dir: + _target_dir = utils._ensure_path(target_dir) + + # Resolve to absolute paths + if self.root_folder.resolve() == _target_dir.resolve(): + raise ValueError( + "`target_dir` parameter must be different from the existing TxtInOut path!" + ) + run_path = self.copy_required_files(target_dir=target_dir) + + # Initialize new TxtinoutReader class + reader = TxtinoutReader(run_path) + else: + reader = self + run_path = self.root_folder + + # Apply SWAT+ configuration changes + reader._apply_swat_configuration(begin_and_end_date, warmup, print_prt_control) + + if parameters: + _params = [ParameterModel(**param) for param in parameters] + + validators._validate_cal_parameters(reader.root_folder, _params) + + if not skip_validation: + validators._validate_conditions_and_units(_params, reader.root_folder) + + reader._write_calibration_file(_params) + + # Run simulation + reader._run_swat() + + return run_path diff --git a/tests/test_performancemetrics.py b/tests/test_performance_metrics.py similarity index 100% rename from tests/test_performancemetrics.py rename to tests/test_performance_metrics.py diff --git a/tests/test_sensitivityanalyzer.py b/tests/test_sensitivity_analyzer.py similarity index 100% rename from tests/test_sensitivityanalyzer.py rename to tests/test_sensitivity_analyzer.py diff --git a/tests/test_txtinoutreader.py b/tests/test_txtinout_reader.py similarity index 100% rename from tests/test_txtinoutreader.py rename to tests/test_txtinout_reader.py From 4b94f821ecbf7d4c86ab15768b0436985e8bc32c Mon Sep 17 00:00:00 2001 From: Debasish Pal <48341250+debpal@users.noreply.github.com> Date: Fri, 19 Sep 2025 14:09:11 +0300 Subject: [PATCH 3/5] rename build-check.yml to building.yml for consistency --- .github/workflows/{build-check.yml => building.yml} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename .github/workflows/{build-check.yml => building.yml} (97%) diff --git a/.github/workflows/build-check.yml b/.github/workflows/building.yml similarity index 97% rename from .github/workflows/build-check.yml rename to .github/workflows/building.yml index 7b0a40e..a277abd 100644 --- a/.github/workflows/build-check.yml +++ b/.github/workflows/building.yml @@ -1,4 +1,4 @@ -name: Build and Validate +name: build on: push: From fc69efba9a5a57fbfa1ed81c1b41f73b63e45578 Mon Sep 17 00:00:00 2001 From: Debasish Pal <48341250+debpal@users.noreply.github.com> Date: Fri, 19 Sep 2025 14:13:26 +0300 Subject: [PATCH 4/5] add badge for build status --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 5f1c882..2027573 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ [![flake8](https://github.com/swat-model/pySWATPlus/actions/workflows/linting.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/linting.yml) [![mypy](https://github.com/swat-model/pySWATPlus/actions/workflows/typing.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/typing.yml) [![pytest](https://github.com/swat-model/pySWATPlus/actions/workflows/testing.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/testing.yml) +[![pytest](https://github.com/debpal/pySWATPlus/actions/workflows/building.yml/badge.svg)](https://github.com/debpal/pySWATPlus/actions/workflows/building.yml) ![Codecov](https://img.shields.io/codecov/c/github/debpal/pySWATPlus) ![GitHub Repo stars](https://img.shields.io/github/stars/swat-model/pySWATPlus) From db40c6de18489795a902d0da233ec9a6d409052e Mon Sep 17 00:00:00 2001 From: Debasish Pal <48341250+debpal@users.noreply.github.com> Date: Fri, 19 Sep 2025 14:14:26 +0300 Subject: [PATCH 5/5] fix build badge path --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 2027573..af2dbc8 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ [![flake8](https://github.com/swat-model/pySWATPlus/actions/workflows/linting.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/linting.yml) [![mypy](https://github.com/swat-model/pySWATPlus/actions/workflows/typing.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/typing.yml) [![pytest](https://github.com/swat-model/pySWATPlus/actions/workflows/testing.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/testing.yml) -[![pytest](https://github.com/debpal/pySWATPlus/actions/workflows/building.yml/badge.svg)](https://github.com/debpal/pySWATPlus/actions/workflows/building.yml) +[![pytest](https://github.com/swat-model/pySWATPlus/actions/workflows/building.yml/badge.svg)](https://github.com/swat-model/pySWATPlus/actions/workflows/building.yml) ![Codecov](https://img.shields.io/codecov/c/github/debpal/pySWATPlus) ![GitHub Repo stars](https://img.shields.io/github/stars/swat-model/pySWATPlus)