Merge pull request #21 from EcoExtreML/add_model_class

Add model class

Author: SarahAlidoost

PyStemmusScope/__init__.py

@@ -1,9 +1,6 @@
 """Documentation about PyStemmusScope"""
 import logging
-from . import forcing_io
-from . import iostreamer
-from .iostreamer import create_io_dir
-from .iostreamer import read_config
+from .stemmus_scope import StemmusScope
 
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())

PyStemmusScope/iostreamer.py PyStemmusScope/config_io.pyPyStemmusScope/iostreamer.py renamed to PyStemmusScope/config_io.py

@@ -7,7 +7,7 @@
 import os
 import shutil
 import time
-from pathlib import Path
+from . import utils
 
 
 logger = logging.getLogger(__name__)
@@ -42,27 +42,28 @@ def create_io_dir(forcing_filename, config):
     # get start time with the format Y-M-D-HM
     timestamp = time.strftime('%Y-%m-%d-%H%M')
     station_name = forcing_filename.split('_')[0]
+
     # create input directory
-    input_dir = Path(f"{config['WorkDir']}/input/{station_name}_{timestamp}")
-    Path(input_dir).mkdir(parents=True, exist_ok=True)
+    work_dir = utils.to_absolute_path(config['WorkDir'])
+    input_dir = work_dir / "input" / f"{station_name}_{timestamp}"
+    input_dir.mkdir(parents=True, exist_ok=True)
     message = f"Prepare work directory {input_dir} for the station: {station_name}"
     logger.info("%s", message)
+
     # copy model parameters to work directory
     _copy_data(input_dir, config)
-    input_dir = str(input_dir)
 
     # create output directory
-    output_dir = Path(f"{config['WorkDir']}/output/{station_name}_{timestamp}")
+    output_dir = work_dir / "output" / f"{station_name}_{timestamp}"
     output_dir.mkdir(parents=True, exist_ok=True)
     message = f"Prepare work directory {output_dir} for the station: {station_name}"
     logger.info("%s", message)
-    output_dir = str(output_dir)
 
     # update config file for ForcingFileName and InputPath
     config_file_path = _update_config_file(forcing_filename, input_dir, output_dir,
         config, station_name, timestamp)
 
-    return input_dir, output_dir, config_file_path
+    return str(input_dir), str(output_dir), config_file_path
 
 def _copy_data(input_dir, config):
     """Copy required data to the work directory.
@@ -98,7 +99,7 @@ def _update_config_file(nc_file, input_dir, output_dir, config, station_name, ti
     Returns:
         Path to updated config file.
     """
-    config_file_path = Path(input_dir, f"{station_name}_{timestamp}_config.txt")
+    config_file_path = input_dir / f"{station_name}_{timestamp}_config.txt"
     with open(config_file_path, 'w', encoding="utf8") as f:
         for key, value in config.items():
             if key == "ForcingFileName":

PyStemmusScope/forcing_io.py

@@ -156,20 +156,18 @@ def prepare_global_variables(data, input_path, config):
     hdf5storage.savemat(input_path / 'forcing_globals.mat', matfiledata, appendmat=False)
 
 
-def prepare_forcing(input_dir, forcing_file, config):
+def prepare_forcing(config):
     """Function to prepare the forcing files required by STEMMUS_SCOPE. The input
         directory should be taken from the model configuration file.
 
     Args:
-        input_dir (path or str): Path to the input directory that will be read by
-            STEMMUS_SCOPE.
-        forcing_file (path or str): Path to the netCDF forcing file that will be used
-            to generate the input data.
         config (dict): The PyStemmusScope configuration dictionary.
     """
-    input_path = Path(input_dir)
+
+    input_path = Path(config["InputPath"])
 
     # Read the required data from the forcing file into a dictionary
+    forcing_file = Path(config["ForcingPath"]) / config["ForcingFileName"]
     data = read_forcing_data(forcing_file)
 
     # Write the single-column ascii '.dat' files to the input directory

PyStemmusScope/soil_io.py

@@ -205,28 +205,25 @@ def _retrieve_latlon(file):
     return lat, lon
 
 
-def prepare_soil_data(soil_data_dir, matfile_path, run_config):
+def prepare_soil_data(config):
     """Function that prepares the soil input data for the STEMMUS_SCOPE model. It parses
     the data for the input location, and writes a file that can be easily read in by
     Matlab.
 
     Args:
-        soil_data_dir (Path): Path to the directory which contains the soil data.
-        mathfile_path (Path): Path to the directory where soil parameter file
-            should be written to.
-        run_config (dict): Dictionary containing the configuration for the current
-            STEMMUS_SCOPE run.
+        config (dict): The PyStemmusScope configuration dictionary.
     """
-    forcing_file = Path(run_config["ForcingPath"]) / run_config["ForcingFileName"]
+
+    forcing_file = Path(config["ForcingPath"]) / config["ForcingFileName"]
 
     # Data missing at ID-Pag site. See github.com/EcoExtreML/STEMMUS_SCOPE/issues/77
-    if run_config["ForcingFileName"].startswith("ID"):
+    if config["ForcingFileName"].startswith("ID"):
         lat, lon = -1., 112.
     else:
         lat, lon = _retrieve_latlon(forcing_file)
 
-    matfiledata = _collect_soil_data(Path(soil_data_dir), lat, lon)
+    matfiledata = _collect_soil_data(Path(config['SoilPropertyPath']), lat, lon)
 
     hdf5storage.savemat(
-        Path(matfile_path) / "soil_parameters.mat", mdict=matfiledata, appendmat=False,
+        Path(config["InputPath"]) / "soil_parameters.mat", mdict=matfiledata, appendmat=False,
     )

PyStemmusScope/stemmus_scope.py

@@ -0,0 +1,116 @@
+"""PyStemmusScope wrapper around Stemmus_Scope."""
+
+import logging
+import os
+import subprocess
+from typing import Dict
+from . import config_io
+from . import forcing_io
+from . import soil_io
+from . import utils
+
+
+logger = logging.getLogger(__name__)
+
+
+class StemmusScope():
+    """PyStemmusScope wrapper around Stemmus_Scope model.
+    see https://gmd.copernicus.org/articles/14/1379/2021/
+
+    It sets the model with a configuration file and executable file.
+    It also prepares forcing and soil data for model run.
+
+    Args:
+        config_file(str): path to Stemmus_Scope configuration file. An example
+        config_file can be found in tests/test_data in `STEMMUS_SCOPE_Processing
+        repository <https://github.com/EcoExtreML/STEMMUS_SCOPE_Processing>`_
+        exe_file(str): path to Stemmus_Scope executable file.
+
+    Example:
+        See notebooks/run_model_in_notebook.ipynb in `STEMMUS_SCOPE_Processing
+        repository <https://github.com/EcoExtreML/STEMMUS_SCOPE_Processing>`_
+    """
+
+    def __init__(self, config_file: str, exe_file: str):
+        # make sure paths are abolute and path objects
+        config_file = utils.to_absolute_path(config_file)
+        self.exe_file = utils.to_absolute_path(exe_file)
+
+        # read config template
+        self._configs = config_io.read_config(config_file)
+
+    def setup(
+        self,
+        WorkDir: str = None,
+        ForcingFileName: str = None,
+        NumberOfTimeSteps: str = None,
+    ) -> str:
+        """Configure model run.
+
+        1. Creates config file and input/output directories based on the config template.
+        2. Prepare forcing and soil data
+
+        Args:
+            WorkDir: path to a directory where input/output directories should be created.
+            ForcingFileName: forcing file name. Forcing file should be in netcdf format.
+            NumberOfTimeSteps: total number of time steps in which model runs. It can be
+                `NA` or a number. Example `10` runs the model for 10 time steps.
+
+        Returns:
+            Paths to config file and input/output directories
+        """
+        # update config template if needed
+        if WorkDir:
+            self._configs["WorkDir"] = WorkDir
+
+        if ForcingFileName:
+            self._configs["ForcingFileName"] = ForcingFileName
+
+        if NumberOfTimeSteps:
+            self._configs["NumberOfTimeSteps"] = NumberOfTimeSteps
+
+        # create customized config file and input/output directories for model run
+        _, _, self.cfg_file = config_io.create_io_dir(
+            self._configs["ForcingFileName"], self._configs
+            )
+
+        # read the run config file
+        self._configs = config_io.read_config(self.cfg_file)
+
+        # prepare forcing data
+        forcing_io.prepare_forcing(self._configs)
+
+        # prepare soil data
+        soil_io.prepare_soil_data(self._configs)
+
+        # set matlab log dir
+        os.environ['MATLAB_LOG_DIR'] = str(self._configs["InputPath"])
+
+        return str(self.cfg_file)
+
+    def run(self) -> str:
+        """Run model using executable.
+
+        Args:
+
+        Returns:
+            Tuple with stdout and stderr
+        """
+
+        # run the model
+        args = [f"{self.exe_file} {self.cfg_file}"]
+        result = subprocess.run(
+            args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, check=True,
+        )
+        stdout = result.stdout
+
+        # TODO return log info line by line!
+        logger.info("%s", stdout)
+
+        return stdout
+
+
+    @property
+    def config(self) -> Dict:
+        """Return the configurations for this model."""
+        return self._configs

PyStemmusScope/utils.py

@@ -1,3 +1,5 @@
+import os
+from pathlib import Path
 import numpy as np
 
 
@@ -22,3 +24,46 @@ def convert_to_lsm_coordinates(lat, lon):
         lon += 720
 
     return np.round(lat).astype(int), np.round(lon).astype(int)
+
+
+def os_name():
+    return os.name
+
+
+def to_absolute_path(
+    input_path: str,
+    parent: Path = None,
+    must_be_in_parent=True,
+) -> Path:
+    """Parse input string as :py:class:`pathlib.Path` object.
+
+    Args:
+        input_path: Input string path that can be a relative or absolute path.
+        parent: Optional parent path of the input path
+        must_exist: Optional argument to check if the input path exists.
+        must_be_in_parent: Optional argument to check if the input path is
+            subpath of parent path
+
+    Returns:
+        The input path that is an absolute path and a :py:class:`pathlib.Path` object.
+    """
+
+    must_exist = False
+    pathlike = Path(input_path)
+    if parent:
+        if not parent.is_absolute():
+            # care for windows, see issue 22
+            must_exist = os_name() == 'nt'
+        pathlike = parent.joinpath(pathlike)
+        if must_be_in_parent:
+            try:
+                pathlike.relative_to(parent)
+            except ValueError as exc:
+                raise ValueError(
+                    f"Input path {input_path} is not a subpath of parent {parent}"
+                ) from exc
+    else:
+        # care for windows, see issue 22
+        must_exist = os_name() == 'nt'
+
+    return pathlike.expanduser().resolve(strict=must_exist)