API reference

Here you find the code reference for the two main components users of PyStemmusScope might need. First is the StemmusScope model class. Second is the save module, with which you can save the STEMMUS_SCOPE model output to a netCDF file.

PyStemmusScope:

PyStemmusScope wrapper around Stemmus_Scope model.

For a detailed model description, look at this publication.

Configures the model and prepares forcing and soil data for the model run.

Parameters:

Name	Type	Description	Default
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.	required
model_src_path	str	Path to Stemmus_Scope executable file or to a directory containing model source codes.	required
interpreter	str	Use Matlab or Octave. Only required if model_src_path is a path to model source codes.	None

Example

See notebooks/run_model_in_notebook.ipynb at the STEMMUS_SCOPE_Processing repository

Source code in PyStemmusScope/stemmus_scope.py

def __init__(self, config_file: str, model_src_path: str, interpreter: str = None):
    # make sure paths are abolute and path objects
    config_file = utils.to_absolute_path(config_file)
    model_src_path = utils.to_absolute_path(model_src_path)

    # check the path to model source
    self.exe_file = None
    if _is_model_src_exe(model_src_path):
        self.exe_file = model_src_path
    else:
        _check_interpreter(interpreter)

    self.model_src = model_src_path
    self.interpreter = interpreter

    # read config template
    self._config = config_io.read_config(config_file)

config: Dict property

Return the configurations for this model.

run()

Run model using executable.

Returns:

Type	Description
str	The model log.

Source code in PyStemmusScope/stemmus_scope.py

def run(self) -> str:
    """Run model using executable.

    Args:

    Returns:
        The model log.
    """
    if self.exe_file:
        # run using MCR
        args = [f"{self.exe_file} {self.cfg_file}"]
        # set matlab log dir
        os.environ['MATLAB_LOG_DIR'] = str(self._config["InputPath"])
        result = _run_sub_process(args, None)
    if self.interpreter=="Matlab":
        # set Matlab arguments
        path_to_config = f"'{self.cfg_file}'"
        eval_code= f'STEMMUS_SCOPE_exe({path_to_config});exit;'
        args = ["matlab", "-r", eval_code, "-nodisplay", "-nosplash", "-nodesktop"]
        # seperate args dont work on linux!
        if utils.os_name() !="nt":
            args = shlex.join(args)
        result = _run_sub_process(args, self.model_src)
    if self.interpreter=="Octave":
        # set Octave arguments
        # use subprocess instead of oct2py,
        # see issue STEMMUS_SCOPE_Processing/issues/46
        path_to_config = f"'{self.cfg_file}'"
        # fix for windows
        path_to_config = path_to_config.replace("\\", "/")
        eval_code = f'STEMMUS_SCOPE_exe({path_to_config});exit;'
        args = ["octave", "--eval", eval_code, "--no-gui", "--silent"]
        # seperate args dont work on linux!
        if utils.os_name() !="nt":
            args = shlex.join(args)
        result = _run_sub_process(args, self.model_src)
    return result

setup(WorkDir=None, Location=None, StartTime=None, EndTime=None)

Configure model run.

1. Creates config file and input/output directories based on the config template.
2. Prepare forcing and soil data

Parameters:

Name	Type	Description	Default
WorkDir	str	path to a directory where input/output directories should be created.	None
ForcingFileName		forcing file name. Forcing file should be in netcdf format.	required
StartTime	str	Start time of the model run. It must be in ISO format (e.g. 2007-01-01T00:00).	None
EndTime	str	End time of the model run. It must be in ISO format (e.g. 2007-01-01T00:00).	None

Returns:

Type	Description
Tuple[str, str]	Path to the config file

Source code in PyStemmusScope/stemmus_scope.py

def setup(
    self,
    WorkDir: str = None,
    Location: str = None,
    StartTime: str = None,
    EndTime: str = None,
) -> Tuple[str, str]:
    """Configure model run.

    1. Creates config file and input/output directories based on the config template.
    2. Prepare forcing and soil data

    Arguments:
        WorkDir: path to a directory where input/output directories should be created.
        ForcingFileName: forcing file name. Forcing file should be in netcdf format.
        StartTime: Start time of the model run. It must be in
            ISO format (e.g. 2007-01-01T00:00).
        EndTime: End time of the model run. It must be in ISO format
            (e.g. 2007-01-01T00:00).

    Returns:
        Path to the config file
    """
    # update config template if needed
    if WorkDir:
        self._config["WorkDir"] = WorkDir

    if Location:
        self._config["Location"] = Location

    if StartTime:
        self._config["StartTime"] = StartTime

    if EndTime:
        self._config["EndTime"] = EndTime

    # validate config *before* directory creation
    config_io.validate_config(self._config)

    # create customized config file and input/output directories for model run
    _, _, self.cfg_file = config_io.create_io_dir(self._config)

    self._config = config_io.read_config(self.cfg_file)

    forcing_io.prepare_forcing(self._config)
    soil_io.prepare_soil_data(self._config)
    soil_io.prepare_soil_init(self._config)

    return str(self.cfg_file)

PyStemmusScope.save:

PyStemmusScope.save

PyStemmusScope save module.

Module designed to create a netcdf file following the ALMA convention from csv files following the SCOPE format in the output directory.

The file required_netcf_variables.csv lists required variable names and their attributes based on the ALMA+CF convention table.

Note

See notebooks/run_model_in_notebook.ipynb in the STEMMUS_SCOPE_Processing repository.

to_netcdf(config_file, cf_filename)

Save csv files generated by Stemmus_Scope model to a netcdf file using information provided by ALMA conventions.

Parameters:

Name	Type	Description	Default
config_file	str	Path to the config file.	required
cf_filename	str	Path to a csv file for ALMA conventions.	required

Returns:

Type	Description
str	Path to a csv file under the output directory.

Source code in PyStemmusScope/save.py

def to_netcdf(config_file: str, cf_filename: str) -> str:
    """Save csv files generated by Stemmus_Scope model to a netcdf file using
        information provided by ALMA conventions.

    Args:
        config_file: Path to the config file.
        cf_filename: Path to a csv file for ALMA conventions.

    Returns:
        Path to a csv file under the output directory.
    """
    config = config_io.read_config(config_file)
    loc, fmt = utils.check_location_fmt(config["Location"])

    # list of required forcing variables, Alma_short_name: forcing_io_name, # model_name
    var_names = {
        "RH": "rh", # RH
        "SWdown_ec": "sw_down", # Rin
        "LWdown_ec": "lw_down", # Rli
        "Qair": "Qair",
        "Tair": "t_air_celcius", # Ta
        "Psurf": "psurf_hpa", # P
        "Wind": "wind_speed", # u
        "Precip": "precip_conv", # Pre
    }

    if fmt == "site":
        # read forcing file into a dict
        forcing_dict = forcing_io.read_forcing_data_plumber2(
            utils.get_forcing_file(config),
            config["StartTime"],
            config["EndTime"],
        )
    elif fmt == "latlon":
        forcing_dict = forcing_io.read_forcing_data_global(
            Path(config["ForcingPath"]),
            lat=loc[0],
            lon=loc[1],
            start_time=config["StartTime"],
            end_time=config["EndTime"],
        )

    # get time info
    time = forcing_dict["time"].values

    # read convention file
    conventions = pd.read_csv(cf_filename)

    alma_short_names = conventions["short_name_alma"]
    data_list = []
    for alma_name in alma_short_names:
        df = conventions.loc[alma_short_names == alma_name].iloc[0]
        file_name = Path(config["OutputPath"]) / df["file_name_STEMMUS-SCOPE"]

        if alma_name in var_names:
            # select data
            data_array = _select_forcing_variables(forcing_dict, var_names[alma_name], alma_name)

        # create data array
        elif alma_name in {"SoilTemp", "SoilMoist"}:
            data_array = _prepare_soil_data(file_name, alma_name, time)
        else:
            data_array = _prepare_simulated_data(
                file_name, df["short_name_STEMMUS-SCOPE"], alma_name, time
                )

        # update attributes of array
        data_array.attrs = {
            "units": df["unit"],
            "long_name": df["long_name"],
            "standard_name": df["standard_name"],
            "STEMMUS-SCOPE_name": df["short_name_STEMMUS-SCOPE"],
            "definition": df["definition"],
        }

        # add to list
        data_list.append(data_array)

    # merge to a dataset
    dataset = xr.merge(data_list)

    # update dimensions
    dataset = _update_dataset_attrs_dims(dataset, forcing_dict)

    # # save to nc file
    nc_filename = Path(config["OutputPath"]) / f"{Path(config['OutputPath']).stem}_STEMMUS_SCOPE.nc"
    dataset.to_netcdf(path= nc_filename)

    return str(nc_filename)