Skip to content

API reference

Here you find the code reference for the main components of PyStemmusScope.

Run the model:

PyStemmusScope wrapper around Stemmus_Scope model.

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 Union[str, Path]

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 Union[str, Path]

Path to Stemmus_Scope executable file or to a directory containing model source codes.

required
interpreter optional

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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __init__(
    self,
    config_file: Union[str, Path],
    model_src_path: Union[str, Path],
    interpreter: Optional[str] = None,
):
    """PyStemmusScope wrapper around Stemmus_Scope model.

    For a detailed model description, look at
    [this publication](https://gmd.copernicus.org/articles/14/1379/2021/).

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

    Arguments:
        config_file: 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).
        model_src_path: Path to Stemmus_Scope executable file or to a
            directory containing model source codes.
        interpreter (optional): Use `Matlab` or `Octave`. Only required if
            `model_src_path` is a path to model source codes.

    Example:
        See notebooks/run_model_in_notebook.ipynb at the [STEMMUS_SCOPE_Processing
        repository](https://github.com/EcoExtreML/STEMMUS_SCOPE_Processing)
    """
    # make sure paths are abolute and path objects
    config_path = utils.to_absolute_path(config_file)
    model_src = utils.to_absolute_path(model_src_path)

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

    self.model_src = model_src
    self.interpreter = interpreter

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

config property

Return the configurations for this model.

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

Configure the 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 Optional[str]

path to a directory where input/output directories should be created.

None
Location Optional[str]

Location of the model run. Can be a site ("FI-Hyy") or lat/lon, e.g., "(52.0, 4.05)".

None
ForcingFileName

forcing file name. Forcing file should be in netcdf format.

required
StartTime Optional[str]

Start time of the model run. It must be in ISO format (e.g. 2007-01-01T00:00).

None
EndTime Optional[str]

End time of the model run. It must be in ISO format (e.g. 2007-01-01T00:00).

None

Returns:

Type Description
str

Path to the config file

Source code in PyStemmusScope/stemmus_scope.py
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def setup(
    self,
    WorkDir: Optional[str] = None,
    Location: Optional[str] = None,
    StartTime: Optional[str] = None,
    EndTime: Optional[str] = None,
) -> str:
    """Configure the 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.
        Location: Location of the model run. Can be a site ("FI-Hyy") or lat/lon,
            e.g., "(52.0, 4.05)".
        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)

run()

Run model using executable.

Returns:

Type Description
str

The model log.

Source code in PyStemmusScope/stemmus_scope.py
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def run(self) -> str:
    """Run model using executable.

    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!
        result = _run_sub_process(
            args if utils.os_name() == "nt" else shlex.join(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!
        result = _run_sub_process(
            args if utils.os_name() == "nt" else shlex.join(args), self.model_src
        )
    return result

Post processing:

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 to a ALMA compliant netCDF file.

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
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
def to_netcdf(config_file: str, cf_filename: str) -> str:
    """Save csv files generated by STEMMUS_SCOPE to a ALMA compliant netCDF file.

    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(Path(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],  # type: ignore
            lon=loc[1],  # type: ignore
            start_time=np.datetime64(config["StartTime"]),
            end_time=np.datetime64(config["EndTime"]),
        )

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

    # 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)

    # for writing to netcdf, time attrs should be added
    # time attrs should be the same as plumber 2 forcing data
    # otherwise it cannot be uploaded to modelevaluation portal
    start_time = time.dt.strftime("%Y-%m-%d").values[0]
    time_encode = {
        "time": {"units": f"seconds since {start_time}", "calendar": "standard"}
    }
    # save to nc file
    nc_filename = (
        Path(config["OutputPath"])
        / f"{Path(config['OutputPath']).stem}_STEMMUS_SCOPE.nc"
    )
    dataset.to_netcdf(path=nc_filename, encoding=time_encode)

    return str(nc_filename)

BMI interface:

PyStemmusScope.bmi.StemmusScopeBmi

Bases: InapplicableBmiMethods, Bmi

STEMMUS_SCOPE Basic Model Interface.

initialize(config_file)

Perform startup tasks for the model.

Parameters:

Name Type Description Default
config_file str

Path to the configuration file.

required
Source code in PyStemmusScope/bmi/implementation.py
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def initialize(self, config_file: str) -> None:
    """Perform startup tasks for the model.

    Args:
        config_file: Path to the configuration file.
    """
    self.config_file = config_file
    self.config = read_config(config_file)

    Path(self.config["OutputPath"]).mkdir(parents=True, exist_ok=True)
    self.state_file = Path(self.config["OutputPath"]) / "STEMMUS_SCOPE_state.mat"
    if self.state_file.exists():
        check_writable(self.state_file)
    else:
        self.state_file.touch()  # Prevent docker messing up file permission.

    self._run_mode = get_run_mode(self.config)

    self._process = start_process(self._run_mode, config_file)
    self._process.initialize()

update()

Advance the model state by one time step.

Source code in PyStemmusScope/bmi/implementation.py
243
244
245
246
247
248
249
250
251
252
253
254
def update(self) -> None:
    """Advance the model state by one time step."""
    if self.state is not None:
        self.state = self.state.close()  # Close file to allow matlab to write

    if self._process is not None:
        self._process.update()
    else:
        msg = "The STEMMUS_SCOPE process is not running/connected. Can't update!"
        raise ValueError(msg)

    self.state = load_state(self.config)

update_until(time)

Advance model state until the given time.

Parameters:

Name Type Description Default
time float

A model time later than the current model time.

required
Source code in PyStemmusScope/bmi/implementation.py
256
257
258
259
260
261
262
263
def update_until(self, time: float) -> None:
    """Advance model state until the given time.

    Args:
        time: A model time later than the current model time.
    """
    while time > self.get_current_time():
        self.update()

finalize()

Finalize the STEMMUS_SCOPE model.

Source code in PyStemmusScope/bmi/implementation.py
265
266
267
268
269
270
271
def finalize(self) -> None:
    """Finalize the STEMMUS_SCOPE model."""
    if self._process is not None:
        self._process.finalize()
    else:
        msg = "The STEMMUS_SCOPE process is not running/connected. Can't finalize!"
        raise ValueError(msg)

get_component_name()

Name of the component.

Returns:

Type Description
str

Name of the component (STEMMUS_SCOPE).

Source code in PyStemmusScope/bmi/implementation.py
273
274
275
276
277
278
279
def get_component_name(self) -> str:
    """Name of the component.

    Returns:
        Name of the component (STEMMUS_SCOPE).
    """
    return "STEMMUS_SCOPE"

get_input_item_count()

Get the number of model input variables.

Returns:

Type Description
int

The number of input variables.

Source code in PyStemmusScope/bmi/implementation.py
282
283
284
285
286
287
288
def get_input_item_count(self) -> int:
    """Get the number of model input variables.

    Returns:
        The number of input variables.
    """
    return len(MODEL_INPUT_VARNAMES)

get_output_item_count()

Get the number of model output variables.

Returns:

Type Description
int

The number of output variables.

Source code in PyStemmusScope/bmi/implementation.py
290
291
292
293
294
295
296
def get_output_item_count(self) -> int:
    """Get the number of model output variables.

    Returns:
        The number of output variables.
    """
    return len(MODEL_OUTPUT_VARNAMES)

get_input_var_names()

List of the model's input variables (as CSDMS Standard Names).

Source code in PyStemmusScope/bmi/implementation.py
300
301
302
def get_input_var_names(self) -> tuple[str, ...]:  # type: ignore
    """List of the model's input variables (as CSDMS Standard Names)."""
    return MODEL_INPUT_VARNAMES

get_output_var_names()

List of the model's output variables (as CSDMS Standard Names).

Source code in PyStemmusScope/bmi/implementation.py
304
305
306
def get_output_var_names(self) -> tuple[str, ...]:  # type: ignore
    """List of the model's output variables (as CSDMS Standard Names)."""
    return MODEL_OUTPUT_VARNAMES

get_var_grid(name)

Get grid identifier for the given variable.

Source code in PyStemmusScope/bmi/implementation.py
308
309
310
def get_var_grid(self, name: str) -> int:
    """Get grid identifier for the given variable."""
    return VARNAME_GRID[name]

get_var_type(name)

Get data type of the given variable.

Source code in PyStemmusScope/bmi/implementation.py
312
313
314
def get_var_type(self, name: str) -> str:
    """Get data type of the given variable."""
    return VARNAME_DTYPE[name]

get_var_units(name)

Get units of the given variable.

Source code in PyStemmusScope/bmi/implementation.py
316
317
318
def get_var_units(self, name: str) -> str:
    """Get units of the given variable."""
    return VARNAME_UNITS[name]

get_var_itemsize(name)

Get memory use for each array element in bytes.

Source code in PyStemmusScope/bmi/implementation.py
320
321
322
def get_var_itemsize(self, name: str) -> int:
    """Get memory use for each array element in bytes."""
    return np.array([], dtype=VARNAME_DTYPE[name]).itemsize

get_var_nbytes(name)

Get size, in bytes, of the given variable.

Source code in PyStemmusScope/bmi/implementation.py
324
325
326
def get_var_nbytes(self, name: str) -> int:
    """Get size, in bytes, of the given variable."""
    return self.get_grid_size(self.get_var_grid(name)) * self.get_var_itemsize(name)

get_current_time()

Get the current time of the model.

Source code in PyStemmusScope/bmi/implementation.py
329
330
331
332
333
334
def get_current_time(self) -> float:
    """Get the current time of the model."""
    if self.state is None:
        raise ValueError(NO_STATE_MSG)

    return self.get_start_time() + np.sum(self.state["TimeStep"][0])

get_start_time()

Start time of the model.

Source code in PyStemmusScope/bmi/implementation.py
336
337
338
339
340
341
342
343
344
345
def get_start_time(self) -> float:
    """Start time of the model."""
    if len(self.config) == 0:
        raise ValueError(NO_CONFIG_MSG)

    return (
        np.datetime64(self.config["StartTime"])
        .astype("datetime64[s]")
        .astype("float")
    )

get_end_time()

End time of the model.

Source code in PyStemmusScope/bmi/implementation.py
347
348
349
350
351
352
353
354
355
356
def get_end_time(self) -> float:
    """End time of the model."""
    if len(self.config) == 0:
        raise ValueError(NO_CONFIG_MSG)

    return (
        np.datetime64(self.config["EndTime"])
        .astype("datetime64[s]")
        .astype("float")
    )

get_time_units()

Time units of the model.

Source code in PyStemmusScope/bmi/implementation.py
358
359
360
def get_time_units(self) -> str:
    """Time units of the model."""
    return "seconds since 1970-01-01 00:00:00.0 +0000"

get_time_step()

Return the current time step of the model.

Source code in PyStemmusScope/bmi/implementation.py
362
363
364
365
366
def get_time_step(self) -> float:
    """Return the current time step of the model."""
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    return float(self.state["TimeStep"][0][0])

get_value(name, dest)

Get a copy of values of the given variable.

Parameters:

Name Type Description Default
name str

input or output variable name, a CSDMS Standard Name.

required
dest ndarray

numpy array into which to place the values.

required

Returns:

Type Description
ndarray

The same numpy array that was passed as an input buffer.

Source code in PyStemmusScope/bmi/implementation.py
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def get_value(self, name: str, dest: np.ndarray) -> np.ndarray:
    """Get a copy of values of the given variable.

    Args:
        name: input or output variable name, a CSDMS Standard Name.
        dest: numpy array into which to place the values.

    Returns:
        The same numpy array that was passed as an input buffer.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    dest[:] = get_variable(self.state, name)
    return dest

get_value_ptr(name)

Get a reference to values of the given variable.

Note: not possible due to the Matlab<>Python coupling.

Source code in PyStemmusScope/bmi/implementation.py
384
385
386
387
388
389
def get_value_ptr(self, name: str) -> np.ndarray:
    """Get a reference to values of the given variable.

    Note: not possible due to the Matlab<>Python coupling.
    """
    raise NotImplementedError()

get_value_at_indices(name, dest, inds)

Get values at particular indices.

Parameters:

Name Type Description Default
name str

Input or output variable name, a CSDMS Standard Name.

required
dest ndarray

numpy array into which to place the values.

required
inds ndarray

The indices into the variable array.

required

Returns:

Type Description
ndarray

Value of the model variable at the given location.

Source code in PyStemmusScope/bmi/implementation.py
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def get_value_at_indices(
    self, name: str, dest: np.ndarray, inds: np.ndarray
) -> np.ndarray:
    """Get values at particular indices.

    Args:
        name: Input or output variable name, a CSDMS Standard Name.
        dest: numpy array into which to place the values.
        inds: The indices into the variable array.

    Returns:
        Value of the model variable at the given location.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    dest[:] = get_variable(self.state, name)[inds]
    return dest

set_value(name, src)

Specify a new value for a model variable.

Parameters:

Name Type Description Default
name str

Input or output variable name, a CSDMS Standard Name.

required
src ndarray

The new value for the specified variable.

required
Source code in PyStemmusScope/bmi/implementation.py
409
410
411
412
413
414
415
416
417
418
419
420
421
def set_value(self, name: str, src: np.ndarray) -> None:
    """Specify a new value for a model variable.

    Args:
        name: Input or output variable name, a CSDMS Standard Name.
        src: The new value for the specified variable.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    if src.size != self.get_grid_size(self.get_var_grid(name)):
        msg = f"Size of `src` and variable '{name}' grid size are not equal!"
        raise ValueError(msg)
    self.state = set_variable(self.state, name, src)

set_value_at_indices(name, inds, src)

Specify a new value for a model variable at particular indices.

Parameters

name : str An input or output variable name, a CSDMS Standard Name. inds : array_like The indices into the variable array. src : array_like The new value for the specified variable.

Source code in PyStemmusScope/bmi/implementation.py
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
def set_value_at_indices(
    self, name: str, inds: np.ndarray, src: np.ndarray
) -> None:
    """Specify a new value for a model variable at particular indices.

    Parameters
    ----------
    name : str
        An input or output variable name, a CSDMS Standard Name.
    inds : array_like
        The indices into the variable array.
    src : array_like
        The new value for the specified variable.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    if inds.size != src.size:
        msg = "Sizes of `inds` and `src` are not equal!"
        raise ValueError(msg)
    self.state = set_variable(self.state, name, src, inds)

get_grid_rank(grid)

Get number of dimensions of the computational grid.

Parameters:

Name Type Description Default
grid int

A grid identifier.

required

Returns:

Type Description
int

Rank of the grid.

Source code in PyStemmusScope/bmi/implementation.py
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def get_grid_rank(self, grid: int) -> int:
    """Get number of dimensions of the computational grid.

    Args:
        grid: A grid identifier.

    Returns:
        Rank of the grid.
    """
    if grid == 0:
        return 2
    if grid == 1:
        return 3
    msg = f"Invalid grid identifier '{grid}'"
    raise ValueError(msg)

get_grid_size(grid)

Get the total number of elements in the computational grid.

Parameters:

Name Type Description Default
grid int

A grid identifier.

required

Returns:

Type Description
int

Size of the grid.

Source code in PyStemmusScope/bmi/implementation.py
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
def get_grid_size(self, grid: int) -> int:
    """Get the total number of elements in the computational grid.

    Args:
        grid: A grid identifier.

    Returns:
        Size of the grid.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)

    if grid == 0:
        return 1
    if grid == 1:
        return int(self.state["ModelSettings"]["mN"][0]) - 1

    msg = f"Invalid grid identifier '{grid}'"
    raise ValueError(msg)

get_grid_type(grid)

Get the grid type as a string.

Parameters:

Name Type Description Default
grid int

A grid identifier.

required

Returns:

Type Description
str

Type of grid as a string.

Source code in PyStemmusScope/bmi/implementation.py
481
482
483
484
485
486
487
488
489
490
def get_grid_type(self, grid: int) -> str:
    """Get the grid type as a string.

    Args:
        grid: A grid identifier.

    Returns:
        Type of grid as a string.
    """
    return "rectilinear"

get_grid_x(grid, x)

Get coordinates of grid nodes in the x direction.

Parameters:

Name Type Description Default
grid int

grid identifier.

required
x ndarray

numpy array to hold the x-coordinates of the grid node columns.

required

Returns:

Type Description
ndarray

The input numpy array that holds the grid's column x-coordinates.

Source code in PyStemmusScope/bmi/implementation.py
492
493
494
495
496
497
498
499
500
501
502
503
504
505
def get_grid_x(self, grid: int, x: np.ndarray) -> np.ndarray:
    """Get coordinates of grid nodes in the x direction.

    Args:
        grid: grid identifier.
        x: numpy array to hold the x-coordinates of the grid node columns.

    Returns:
        The input numpy array that holds the grid's column x-coordinates.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    x[:] = self.state["SiteProperties"]["longitude"][0]
    return x

get_grid_y(grid, y)

Get coordinates of grid nodes in the y direction.

Parameters:

Name Type Description Default
grid int

grid identifier.

required
y ndarray

numpy array to hold the y-coordinates of the grid node columns.

required

Returns:

Type Description
ndarray

The input numpy array that holds the grid's column y-coordinates.

Source code in PyStemmusScope/bmi/implementation.py
507
508
509
510
511
512
513
514
515
516
517
518
519
520
def get_grid_y(self, grid: int, y: np.ndarray) -> np.ndarray:
    """Get coordinates of grid nodes in the y direction.

    Args:
        grid: grid identifier.
        y: numpy array to hold the y-coordinates of the grid node columns.

    Returns:
        The input numpy array that holds the grid's column y-coordinates.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    y[:] = self.state["SiteProperties"]["latitude"][0]
    return y

get_grid_z(grid, z)

Get coordinates of grid nodes in the z direction.

Parameters:

Name Type Description Default
grid int

grid identifier.

required
z ndarray

numpy array to hold the z-coordinates of the grid node columns.

required

Returns:

Type Description
ndarray

The input numpy array that holds the grid's column z-coordinates.

Source code in PyStemmusScope/bmi/implementation.py
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
def get_grid_z(self, grid: int, z: np.ndarray) -> np.ndarray:
    """Get coordinates of grid nodes in the z direction.

    Args:
        grid: grid identifier.
        z: numpy array to hold the z-coordinates of the grid node columns.

    Returns:
        The input numpy array that holds the grid's column z-coordinates.
    """
    if self.state is None:
        raise ValueError(NO_STATE_MSG)
    if grid == 1:
        z[:] = (
            -np.hstack(
                (
                    self.state["ModelSettings"]["DeltZ_R"][:, 0].cumsum()[::-1],
                    np.array([0.0]),
                )
            )
            / 100
        )
        return z
    else:
        raise ValueError(f"Grid {grid} has no dimension `z`.")

get_grid_shape(grid, shape)

Get dimensions of the computational grid.

Source code in PyStemmusScope/bmi/implementation.py
548
549
550
551
552
553
554
555
556
557
558
def get_grid_shape(self, grid: int, shape: np.ndarray) -> np.ndarray:
    """Get dimensions of the computational grid."""
    if grid not in [0, 1]:
        msg = f"Unknown grid identifier '{grid}'"
        raise ValueError(msg)

    shape[-1] = 1  # Last element is x
    shape[-2] = 1  # Semi-last element is y
    if grid == 1:
        shape[-3] = self.get_grid_size(grid)  # First element is z
    return shape