Initial unit test set up by pytest

.github/workflows/linting.yml

@@ -35,4 +35,4 @@ jobs:
 
     - name: Lint with flake8
       run: |
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        flake8 . --count --exit-zero --max-complexity=10 --statistics

.github/workflows/testing.yml

@@ -0,0 +1,40 @@
+name: pytest
+
+on:
+  push:
+    branches:
+      - main  # Your branch name
+    paths:
+      - '**/*.py'  # Trigger for changes in Python files
+  pull_request:
+    branches:
+      - main  # Your branch name
+    paths:
+      - '**/*.py'  # Trigger for changes in Python files
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -r requirements.txt  # Install dependencies
+
+    - name: Run tests with pytest
+      run: |
+        export PYTHONPATH=$(pwd)
+        pytest -rA -Wignore::DeprecationWarning --cov=GeoAnalyze --cov-report=xml  # Run tests and generate coverage report in XML format
+

pySWATPlus/FileReader.py

@@ -17,6 +17,7 @@ def read_csv(
         engine: Literal['c', 'python'],
         mode: Literal['dask', 'pandas'] = 'dask'
 ) -> Union[pd.DataFrame, dd.DataFrame]:
+
     '''
     Read a CSV file using either Dask or Pandas and filter the data based on criteria.
 
@@ -38,7 +39,16 @@ def read_csv(
         - When `mode` is 'pandas', a Pandas DataFrame is returned.
 
         Example:
-        read_csv('plants.plt', skip_rows=[0], usecols=['name', 'plnt_typ', 'gro_trig'], filter_by={'plnt_typ': 'perennial'}, separator=r"[ ]{2,}", encoding="utf-8", engine='python', mode='dask')
+
+        read_csv(
+            'plants.plt',
+            skip_rows=[0],
+            usecols=['name', 'plnt_typ', 'gro_trig'],
+            filter_by={'plnt_typ': 'perennial'}, separator=r"[ ]{2,}",
+            encoding="utf-8",
+            engine='python',
+            mode='dask'
+        )
     '''
 
     if mode == 'dask':
@@ -93,6 +103,7 @@ def __init__(
         usecols: List[str] = None,
         filter_by: Dict[str, Union[Any, List[Any], re.Pattern]] = {}
     ):
+
         '''
         Initialize a FileReader instance to read data from a file.
 
@@ -101,7 +112,7 @@ def __init__(
         has_units (bool): Indicates if the file has units (default is False).
         index (str, optional): The name of the index column (default is None).
         usecols (List[str], optional): A list of column names to read (default is None).
-        filter_by (Dict[str, Union[Any, List[Any], re.Pattern]): A dictionary of column names and values to filter by (default is an empty dictionary).
+        filter_by (Dict[str, Union[Any, List[Any], re.Pattern], optional): A dictionary of column names and values to filter.
 
         Raises:
         FileNotFoundError: If the specified file does not exist.
@@ -118,8 +129,8 @@ def __init__(
 
         Example:
         FileReader('plants.plt', has_units = False, index = 'name', usecols=['name', 'plnt_typ', 'gro_trig'], filter_by={'plnt_typ': 'perennial'})
-
         '''
+
         if not isinstance(path, (str, os.PathLike)):
             raise TypeError("path must be a string or os.PathLike object")
 
@@ -199,7 +210,10 @@ def __init__(
         self.df = df
         self.path = path
 
-    def _store_text(self) -> None:
+    def _store_text(
+        self
+    ) -> None:
+
         '''
         Store the DataFrame as a formatted text file.
 
@@ -211,6 +225,7 @@ def _store_text(self) -> None:
         Returns:
         None
         '''
+
         data_str = self.df.to_string(index=False, justify='left', col_space=15)
 
         # Find the length of the longest string in each column
@@ -227,7 +242,10 @@ def _store_text(self) -> None:
             file.write(self.header_file)
             file.write(data_str)
 
-    def _store_csv(self) -> None:
+    def _store_csv(
+        self
+    ) -> None:
+
         '''
         Store the DataFrame as a CSV file.
 
@@ -236,9 +254,13 @@ def _store_csv(self) -> None:
         Returns:
         None
         '''
+
         raise TypeError("Not implemented yet")
 
-    def overwrite_file(self) -> None:
+    def overwrite_file(
+        self
+    ) -> None:
+
         '''
         Overwrite the original file with the DataFrame.
 
@@ -247,6 +269,7 @@ def overwrite_file(self) -> None:
         Returns:
         None
         '''
+
         if self.path.suffix == '.csv':
             self._store_csv()
         else:

pySWATPlus/PymooBestSolution.py

@@ -11,33 +11,54 @@ class SolutionManager:
     Class to manage the best solution found during optimization.
     """
 
-    def __init__(self):
+    def __init__(
+        self
+    ):
+
         self.X = None
         self.path = None
         self.error = None
         self.lock = multiprocessing.Lock()
 
-    def add_solution(self, X: np.ndarray, path: Dict[str, str], error: float) -> None:
+    def add_solution(
+        self,
+        X: np.ndarray,
+        path: Dict[str, str],
+        error: float
+    ) -> None:
+
         """
         Add a solution if it is better than the current best solution.
         """
+
         with self.lock:
             if self.error is None or error < self.error:
                 self.X = X
                 self.path = path
                 self.error = error
 
-    def get_solution(self) -> Tuple[np.ndarray, Dict[str, str], float]:
+    def get_solution(
+        self
+    ) -> Tuple[np.ndarray, Dict[str, str], float]:
+
         """
         Retrieve the best solution.
         """
+
         with self.lock:
             return self.X, self.path, self.error
 
-    def add_solutions(self, X_array: np.ndarray, paths_array: List[Dict[str, str]], errors_array: np.ndarray) -> None:
+    def add_solutions(
+        self,
+        X_array: np.ndarray,
+        paths_array: List[Dict[str, str]],
+        errors_array: np.ndarray
+    ) -> None:
+
         """
         Update the best solution based on provided paths and errors. Only the best solution is kept; others are deleted.
         """
+
         if len(errors_array) == 0:
             return
 

pySWATPlus/SWATProblem.py

@@ -4,15 +4,17 @@
 
 class SWATProblem(SWATProblemMultimodel):
 
-    def __init__(self,
-                 params: Dict[str, Tuple[str, List[Tuple[str, str, float, float]]]],
-                 function_to_evaluate: Callable,
-                 param_arg_name: str,
-                 n_workers: int = 1,
-                 parallelization: str = 'threads',
-                 debug: bool = False,
-                 **kwargs: Dict[str, Any]
-                 ) -> None:
+    def __init__(
+        self,
+        params: Dict[str, Tuple[str, List[Tuple[str, str, float, float]]]],
+        function_to_evaluate: Callable,
+        param_arg_name: str,
+        n_workers: int = 1,
+        parallelization: str = 'threads',
+        debug: bool = False,
+        **kwargs: Dict[str, Any]
+    ) -> None:
+
         """
         This feature inicializes a SWATProblem instance, which is used to perform optimization of the desired SWAT+ parameters by using the pymoo library.
 
@@ -34,5 +36,4 @@ def __init__(self,
         None
         """
 
-        # SWATProblemMultimodel.__init__(self, params, function_to_evaluate, param_arg_name, n_workers, parallelization, None, None, None, None, None, debug, **kwargs)
         super().__init__(params, function_to_evaluate, param_arg_name, n_workers, parallelization, None, None, None, None, None, debug, **kwargs)

pySWATPlus/SWATProblemMultimodel.py

@@ -17,6 +17,7 @@ def minimize_pymoo(
         verbose: bool = False,
         callback: Optional[Callable] = None
 ) -> Tuple[Optional[np.ndarray], Optional[str], Optional[float]]:
+
     """
     Perform optimization using the pymoo library.
 
@@ -29,7 +30,8 @@ def minimize_pymoo(
     - callback (Optional[Callable], optional): A callback function that is called after each generation (default is None).
 
     Returns:
-    - Tuple[np.ndarray, Dict[str, str], float]: The best solution found during the optimization process, in the form of a tuple containing the decision variables, the path to the output files with the identifier, and the error.
+    - Tuple[np.ndarray, Dict[str, str], float]: The best solution found during the optimization process, in the form of a tuple containing the decision variables,
+    the path to the output files with the identifier, and the error.
     """
 
     problem.solution_manager = SolutionManager()
@@ -77,7 +79,9 @@ def __init__(
     ) -> None:
 
         """
-        This class serves the same purpose as SWATProblem, with the added capability of running another model before executing SWAT+. This enables running a prior model in the same calibration process, wherein the parameters are calibrated simultaneously. For example, the prior model can modify an input file of SWAT+ before initiating SWAT+ (according to the parameters of the calibration).
+        This class serves the same purpose as SWATProblem, with the added capability of running another model before executing SWAT+.
+        This enables running a prior model in the same calibration process, wherein the parameters are calibrated simultaneously.
+        For example, the prior model can modify an input file of SWAT+ before initiating SWAT+ (according to the parameters of the calibration).
 
         Parameters:
         - params Dict[str, Tuple[str, List[Tuple[str, str, int, int]]]]): A dictionary containing the range of values to optimize.
@@ -136,11 +140,14 @@ def __init__(
         self.solution_manager = None  # it is initialized in the minize_pymoo function
         super().__init__(n_var=n_vars, n_obj=1, n_constr=0, xl=lb, xu=ub, elementwise_evaluation=False)
 
-    def _evaluate(self,
-                  X: np.ndarray,
-                  out: Dict[str, Any],
-                  *args: Any,
-                  **kwargs: Any):
+    def _evaluate(
+        self,
+        X: np.ndarray,
+        out: Dict[str, Any],
+        *args: Any,
+        **kwargs: Any
+    ):
+
         """
         Evaluate the objective function for a given set of input parameters.
 
@@ -153,6 +160,7 @@ def _evaluate(self,
         Returns:
         None
         """
+
         if self.debug:
             print('starting _evaluate')