diff --git a/README.md b/README.md
index bc0243f3..f67651c5 100644
--- a/README.md
+++ b/README.md
@@ -112,6 +112,17 @@ RUBEM depends on [PCRaster](https://pcraster.geo.uu.nl/) and [GDAL](https://gdal
rubem -h
```
+- Python usage example
+
+ ```python
+ from rubem.api import Model
+
+ result = Model.from_file("config.json").run()
+ print(result.output_directory)
+ ```
+
+ `rubem.api` is the supported programmatic surface: see the [Python API page](https://rubem.readthedocs.io/en/latest/api.html) for the result it returns, the stability policy and how to run repeated simulations in separate processes.
+
_For more examples, please refer to the [user guide](https://rubem.readthedocs.io/en/latest/userguide.html) and [tutorials](https://rubem.readthedocs.io/en/latest/tutorials.html)._
diff --git a/doc/source/api.rst b/doc/source/api.rst
index 1882d9aa..0821512b 100644
--- a/doc/source/api.rst
+++ b/doc/source/api.rst
@@ -1,11 +1,236 @@
-Python API Reference
-====================
+Python API
+==========
+
+``rubem.api`` runs the model from Python: it loads a configuration, runs the
+simulation and reports what the run wrote.
+
+Quickstart
+----------
+
+Run a configuration file, the same file the ``rubem run`` command takes:
+
+.. code-block:: python
+
+ from rubem.api import Model
+
+ model = Model.from_file("config.json")
+ result = model.run()
+ print(result.output_directory)
+
+Relative paths of the file are anchored on its directory, unless ``base_dir``
+says otherwise. A configuration held in memory is passed as a dictionary
+instead, in the legacy format or in format 1.0; a dictionary has no directory
+of its own, so ``base_dir`` anchors its relative paths:
+
+.. code-block:: python
+
+ from rubem.api import ConfigurationError, Model
+
+ try:
+ model = Model.from_config(document, base_dir="/data/basin")
+ except ConfigurationError as error:
+ for problem in error.problems:
+ print(problem)
+ raise
+
+The dictionary is copied when it is loaded, so it can be edited and passed
+again to configure another model: each model keeps the document it was built
+from, and runs it in either mode.
+
+Both loaders validate the input files and their content unless
+``validate_input=False`` is passed, and both raise ``ConfigurationError`` when
+the validation finds blocking problems, a missing member of an input raster
+series among them. Not every failed load is one of those: a file that is not
+JSON raises ``json.JSONDecodeError``, a document whose sections do not match
+the schema a pydantic ``ValidationError``, and a single raster or table that is
+not there a ``FileNotFoundError``. The clone and the DEM are read while the
+configuration is built, so they raise ``FileNotFoundError`` whether or not the
+input is validated; with the validation off the other missing inputs surface
+only when the run reaches them.
+
+An already loaded ``ModelConfiguration`` may be passed to ``Model.from_config``
+as well, and is used as it is, neither validated again nor re-anchored.
+
+``Model.run_isolated`` runs the same simulation in a fresh subprocess:
+
+.. code-block:: python
+
+ result = model.run_isolated()
+
+Either form returns a ``RunResult``, which describes the run from its
+configuration rather than by listing the output directory, and whose paths are
+absolute:
+
+.. code-block:: python
+
+ result.output_directory # Path of the directory the run wrote to
+ result.rasters # {variable id: (raster member, ...)}, per enabled format
+ result.time_series # {variable id: (table, ...)}, the CSV and/or TSS tables
+ result.metadata # Path of metadata.json (format 1.0), or None
+ result.first_step # first simulated time step
+ result.last_step # last simulated time step
+ result.elapsed_seconds # wall-clock duration of the simulation itself
+
+Stability
+---------
+
+``rubem.api`` is the public surface of the package: ``Model``, ``RunResult``
+and ``ConfigurationError``. Every other module is internal and may change
+without notice, including the classes the internal modules expose today
+(``ModelConfiguration``, ``DynamicFrameworkWrapper`` and the rest).
+
+While the version is below 1.0, a breaking change to ``rubem.api`` bumps the
+minor version and is listed in the changelog.
+
+``Model`` hands out the loaded ``ModelConfiguration``: the constructor takes
+one, ``Model.from_config`` accepts one and the ``Model.configuration`` property
+returns one, so that a caller that already holds a configuration can use it.
+That object is internal all the same: it is reachable through the public
+surface, but its own shape carries no stability guarantee.
+
+Process model and limitations
+-----------------------------
+
+PCRaster keeps its state process-wide: ``setclone`` is global to the process
+and the raster memory of a run is not reclaimed. Two consequences follow for
+``Model.run``, which runs in the calling process:
+
+* successive runs in one interpreter grow the resident memory;
+* two runs must not happen at the same time in one process. Every run sets the
+ clone for itself, so runs on different grids may follow one another; runs
+ started from several threads, though, share that state while they are under
+ way. On different grids one of them fails; on the same grid they may all
+ finish and write wrong results, without any error. Parallel runs need one
+ process each.
+
+``Model.run_isolated`` avoids both: it runs the simulation in a subprocess
+started with the ``spawn`` method, used for that one run and shut down before
+the call returns. The configuration crosses as its document, its base
+directory and the validation flag, and is rebuilt on the other side; the result
+crosses as plain data. A ``ConfigurationError`` raised while the subprocess
+rebuilds the configuration reaches the caller as the same exception, with its
+problems; any other exception of the run propagates as itself, and a subprocess
+that dies before the run finishes is reported as a ``RuntimeError``. Every call
+pays a full interpreter start-up, which is the price of keeping the state of
+PCRaster out of the caller.
+
+A caller that only uses ``run_isolated`` never loads PCRaster itself: the
+library is imported in the subprocess alone. GDAL is a different matter:
+loading a configuration imports it, whether or not the inputs are validated.
+
+The ``spawn`` method imports the main module of the caller in the subprocess,
+as it does for any :mod:`multiprocessing` worker, so a script that calls
+``run_isolated`` must guard its entry point:
+
+.. code-block:: python
+
+ if __name__ == "__main__":
+ result = Model.from_file("config.json").run_isolated()
+
+For the same reason the script must be a file. One read from standard input
+(``python - < script.py``) has no file for the subprocess to import, and on
+Python 3.13 and 3.14 the subprocess fails at start-up; a command passed with
+``-c`` is fine. Both mistakes, the missing guard and the script on standard
+input, kill the subprocess before the run starts and reach the caller as the
+``RuntimeError`` above, whose message names them; the error of the subprocess
+itself is on standard error.
+
+``run_isolated`` waits for its subprocess. An interrupt that reaches the
+subprocess as well, as Ctrl-C in a terminal does, stops the run and raises
+``KeyboardInterrupt`` at once. An interrupt delivered to the calling process
+alone (``kill -INT`` on its pid, or a scheduler that signals only the process it
+started) is raised only after the subprocess has finished the run.
+
+Importing ``rubem.api`` does not require PCRaster or GDAL; running the model
+does. Without them, loading a configuration or running raises ``ImportError``
+with the same installation guidance the command line prints.
+
+Parallel runs
+-------------
+
+Parallel runs need one process each. The form to use is a process pool of the
+caller, with ``Model.run`` called inside every worker:
+
+.. code-block:: python
+
+ import multiprocessing
+ from concurrent.futures import ProcessPoolExecutor
+
+ from rubem.api import Model
+
+
+ def simulate(path):
+ return Model.from_file(path).run()
+
+
+ if __name__ == "__main__":
+ paths = ["basin_a/config.json", "basin_b/config.json", "basin_c/config.json"]
+ context = multiprocessing.get_context("spawn")
+ with ProcessPoolExecutor(max_workers=2, mp_context=context) as pool:
+ for result in pool.map(simulate, paths):
+ print(result.output_directory)
+
+The model is built inside the worker, so what crosses the boundary is the path
+(or the configuration dictionary) on the way in and the ``RunResult`` on the way
+back; a ``ConfigurationError`` of a worker reaches the caller as itself, with
+its problems. Give every run its own output directory.
+
+A worker is reused from one run to the next, on the same grid or on another, so
+the interpreter start-up is paid once per worker and not once per run, which is
+what makes the pool cheaper than calling ``run_isolated`` from several threads.
+The memory of a reused worker grows with its runs, as it does for any
+in-process run; ``max_tasks_per_child`` bounds it. ``spawn`` is the start method
+``run_isolated`` uses and the one available on every platform, and the entry
+point guard above applies to it.
+
+A thread pool is not a substitute: threads that call ``Model.run`` share the
+state of PCRaster, with the consequences described above.
+
+Logging
+-------
+
+The package reports its progress through the ``rubem`` logger and writes
+nothing to standard output of its own. Configure that logger to follow a run:
+
+.. code-block:: python
+
+ import logging
+
+ logging.basicConfig(level=logging.INFO)
+ logging.getLogger("rubem").setLevel(logging.INFO)
+
+This configures the calling process, and an isolated run happens in another
+one: the records of the simulation are emitted in the subprocess, which starts
+from the default logging configuration and does not inherit the handlers of the
+caller. What the calling process does reaches the configured handlers, loading
+the configuration included, since the loaders run there; the run itself does
+not.
+
+Two things on standard output are not this package's to suppress. The PCRaster
+framework registers an ``atexit`` hook that writes a single newline when the
+interpreter exits, in the caller for an in-process run and in the subprocess
+for an isolated one; the subprocess shares the standard output of the caller
+and exits with every call, so a batch of isolated runs writes one newline per
+run. And at level ``DEBUG`` an in-process run asks the framework for its own
+progress output, which it writes to standard output as one dot per time step;
+at ``INFO`` it is silent, and an isolated run is silent at any level, its
+subprocess starting from the default configuration.
+
+Module reference
+----------------
+
+.. autoexception:: rubem.api.ConfigurationError
+ :members:
+
+The reference below covers the internal modules as well, because the model is
+documented from them; only ``rubem.api`` is stable.
.. autosummary::
:toctree: generated
:template: custom-module-template.rst
:recursive:
+ rubem.api
rubem.cli
rubem.configuration
rubem.core
diff --git a/doc/source/changelog.rst b/doc/source/changelog.rst
index ca4918d5..f3a56ff2 100644
--- a/doc/source/changelog.rst
+++ b/doc/source/changelog.rst
@@ -13,6 +13,17 @@ The format follows `Keep a Changelog `__.
Added
`````
+- Added ``rubem.api``, the public Python surface: ``Model.from_file`` and
+ ``Model.from_config`` load a configuration file or document, ``Model.run()``
+ runs the simulation in the current process and ``Model.run_isolated()`` in a
+ fresh spawned subprocess, both returning a ``RunResult`` that lists the
+ rasters, time series and metadata the run wrote, enumerated from the
+ configuration. Importing the module needs neither PCRaster nor GDAL; running
+ without them raises ``ImportError`` with the installation guidance.
+ ``ConfigurationError`` now survives pickling with its problems
+ (`#341 `__). The Python API
+ page documents the stability policy, the process model and how to run in
+ parallel.
- Added ``rubem preprocess kp``, which builds the Class A pan coefficient series from wind
speed and relative humidity raster series and a fetch distance with the formula of the
published model (supplement S29), through the shared raster I/O of the preprocessing tools;
diff --git a/rubem/_deps.py b/rubem/_deps.py
index b9693a32..a27c0548 100644
--- a/rubem/_deps.py
+++ b/rubem/_deps.py
@@ -6,20 +6,51 @@
_ENVIRONMENT_YML_URL = "https://github.com/LabSid-USP/RUBEM/blob/main/environment.yml"
+def missing_runtime_deps() -> list[str]:
+ """Return the conda-only runtime dependencies that are not importable.
+
+ A dependency whose lookup itself fails (a broken installation whose parent
+ package raises on import) counts as missing: the run would not get past it
+ either.
+
+ :return: The missing dependency names, in the order they are checked.
+ """
+ missing = []
+ for name in _CONDA_ONLY_DEPENDENCIES:
+ try:
+ found = importlib.util.find_spec(name) is not None
+ except (ImportError, ValueError):
+ found = False
+ if not found:
+ missing.append(name)
+ return missing
+
+
+def runtime_deps_message(missing: list[str]) -> str:
+ """Return the installation guidance for the missing runtime dependencies.
+
+ The command line raises it as ``SystemExit`` and the Python API as
+ ``ImportError``, so both front ends give the same instructions.
+
+ :param missing: The missing dependency names, see :func:`missing_runtime_deps`.
+ """
+ return (
+ "RUBEM cannot run because the following conda-only dependencies are "
+ f"not installed: {', '.join(missing)}. Install them from conda-forge, "
+ "for example with 'conda install -c conda-forge pcraster gdal' (or the "
+ "micromamba equivalent), and run RUBEM from that environment. The pinned "
+ f"specification is environment.yml: {_ENVIRONMENT_YML_URL}"
+ )
+
+
def require_runtime_deps() -> None:
"""Fail fast with guidance when pcraster or GDAL are not importable.
:raises SystemExit: If any conda-only dependency is missing.
"""
- missing = [name for name in _CONDA_ONLY_DEPENDENCIES if importlib.util.find_spec(name) is None]
+ missing = missing_runtime_deps()
if missing:
- raise SystemExit(
- "RUBEM cannot run because the following conda-only dependencies are "
- f"not installed: {', '.join(missing)}. Install them from conda-forge, "
- "for example with 'conda install -c conda-forge pcraster gdal' (or the "
- "micromamba equivalent), and run RUBEM from that environment. The pinned "
- f"specification is environment.yml: {_ENVIRONMENT_YML_URL}"
- )
+ raise SystemExit(runtime_deps_message(missing))
_PREPROCESSING_DEPENDENCIES = ("pykrige", "skgstat")
diff --git a/rubem/api.py b/rubem/api.py
new file mode 100644
index 00000000..a26112d8
--- /dev/null
+++ b/rubem/api.py
@@ -0,0 +1,425 @@
+"""The public Python API of RUBEM.
+
+Three names are public: :class:`Model`, the entry point that loads a
+configuration and runs the model; :class:`RunResult`, the description of what a
+finished run wrote; and :class:`ConfigurationError`, re-exported here, raised
+when a configuration carries blocking problems.
+
+Stability
+ ``rubem.api`` is the supported programmatic surface. Every other module of
+ the package is internal and may change without notice. While the version is
+ below 1.0, a breaking change to ``rubem.api`` bumps the minor version and is
+ listed in the changelog.
+
+Process limitation
+ PCRaster keeps the clone and the raster memory process-wide (``setclone``
+ is global to the process and the memory of a run is not reclaimed), so
+ repeated :meth:`Model.run` calls in one interpreter grow the resident
+ memory and must not overlap: every run sets the clone for itself, which
+ lets runs on different grids follow one another, but runs under way at the
+ same time, in several threads, share that state and fail or, on the same
+ grid, may write wrong results without any error.
+ :meth:`Model.run_isolated` runs the simulation in a fresh spawned
+ subprocess instead, at the cost of an interpreter start-up per call, and
+ keeps that state out of the caller. Parallel runs need one process each:
+ the form to use is a process pool of the caller
+ (:class:`concurrent.futures.ProcessPoolExecutor`) whose workers call
+ :meth:`Model.run`, not a thread pool.
+
+Importing this module does not require PCRaster or GDAL; running the model
+does, and raises :class:`ImportError` with the installation guidance when they
+are missing.
+
+Logging
+ The package reports its progress through the ``rubem`` logger and writes
+ nothing to standard output of its own; configure that logger to follow a
+ run. The records of the simulation of an isolated run are emitted in its
+ subprocess, which starts from the default logging configuration and does
+ not inherit the handlers of the caller; the loading of the configuration
+ happens in the calling process and is logged there.
+"""
+
+import logging
+import multiprocessing
+import time
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures.process import BrokenProcessPool
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from ._deps import missing_runtime_deps, runtime_deps_message
+from ._paths import PathInput
+from .configuration._problems import ConfigurationError
+from .configuration.output_format import OutputFileFormat, TimeSeriesFileFormat
+from .file._naming import get_raster_series_filepath, output_raster_filename
+
+if TYPE_CHECKING:
+ from .configuration.model_configuration import ModelConfiguration
+
+__all__ = ["Model", "RunResult", "ConfigurationError"]
+
+logger = logging.getLogger(__name__)
+
+_METADATA_FILENAME = "metadata.json"
+
+
+@dataclass(frozen=True)
+class RunResult:
+ """What a finished run wrote, enumerated from the configuration.
+
+ The paths are derived from the configuration and the naming rules of the
+ model, never from a listing of the output directory, so a result describes
+ the run that produced it even when the directory holds the outputs of
+ earlier runs as well. They are absolute, so a result of an isolated run
+ names the same files whatever the caller does with its working directory
+ afterwards.
+
+ :param output_directory: Directory the run wrote to.
+ :param rasters: Raster members per variable id, the PCRaster series in step
+ order followed by the GeoTIFF series in step order, restricted to the
+ enabled formats.
+ :param time_series: Time series tables per variable id, the CSV table
+ followed by the PCRaster TSS table, restricted to the enabled formats.
+ Empty when the run writes no time series. The ``zones_mapping.csv`` a
+ run aggregated over zones writes is not listed: it belongs to the
+ aggregation, not to a variable.
+ :param metadata: The ``metadata.json`` of a format 1.0 run, ``None`` for a
+ legacy configuration, which has no metadata section.
+ :param first_step: First simulated time step.
+ :param last_step: Last simulated time step.
+ :param elapsed_seconds: Wall-clock duration of the run itself. An isolated
+ run measures the simulation inside its subprocess, not the interpreter
+ start-up the call also pays.
+ """
+
+ output_directory: Path
+ rasters: dict[str, tuple[Path, ...]]
+ time_series: dict[str, tuple[Path, ...]]
+ metadata: Path | None
+ first_step: int
+ last_step: int
+ elapsed_seconds: float
+
+
+class Model:
+ """A configured model, ready to run.
+
+ Build one with :meth:`from_file` or :meth:`from_config` rather than by
+ calling the constructor, unless a :class:`ModelConfiguration` has already
+ been loaded elsewhere.
+
+ :param configuration: The loaded configuration.
+ :param validate_input: Whether an isolated run revalidates the input files
+ and their content when it rebuilds the configuration. It has no effect
+ on ``configuration``, which is already loaded; pass the value the
+ configuration was loaded with.
+ """
+
+ def __init__(self, configuration: "ModelConfiguration", *, validate_input: bool = True) -> None:
+ self._configuration = configuration
+ self._validate_input = bool(validate_input)
+
+ @classmethod
+ def from_file(
+ cls,
+ path: PathInput,
+ *,
+ validate_input: bool = True,
+ base_dir: PathInput | None = None,
+ ) -> "Model":
+ """Load a configuration file and return the model it configures.
+
+ :param path: The JSON configuration file, legacy or format 1.0.
+ :param validate_input: Whether to validate the input files and their content.
+ :param base_dir: Directory the relative paths of the configuration are
+ anchored on. Defaults to the directory of the file.
+ :raises ImportError: If PCRaster or GDAL are not installed.
+ :raises FileNotFoundError: If ``path``, or a raster or table it names,
+ is not there.
+ :raises json.JSONDecodeError: If the file is not JSON.
+ :raises pydantic.ValidationError: If the document does not match the schema.
+ :raises ConfigurationError: If the inputs carry blocking problems.
+ """
+ _require_runtime_deps()
+ from .configuration.model_configuration import ModelConfiguration
+
+ configuration = ModelConfiguration(path, validate_input, base_dir)
+ return cls(configuration, validate_input=validate_input)
+
+ @classmethod
+ def from_config(
+ cls,
+ config: "dict | ModelConfiguration",
+ *,
+ validate_input: bool = True,
+ base_dir: PathInput | None = None,
+ ) -> "Model":
+ """Return the model a configuration document, or a loaded configuration, configures.
+
+ :param config: The configuration document (legacy or format 1.0) as a
+ dictionary, or an already loaded :class:`ModelConfiguration`, which
+ is used as it is. A dictionary is copied, so it can be edited and
+ passed again for another model without changing this one.
+ :param validate_input: Whether to validate the input files and their
+ content. An already loaded configuration is not validated again;
+ the flag then only says whether an isolated run revalidates when it
+ rebuilds the configuration.
+ :param base_dir: Directory the relative paths of the document are
+ anchored on. A dictionary has no anchor unless this is passed.
+ Ignored when ``config`` is already loaded.
+ :raises ImportError: If PCRaster or GDAL are not installed.
+ :raises FileNotFoundError: If a raster or table the document names is
+ not there.
+ :raises pydantic.ValidationError: If the document does not match the schema.
+ :raises ConfigurationError: If the inputs carry blocking problems.
+ """
+ _require_runtime_deps()
+ from .configuration.model_configuration import ModelConfiguration
+
+ if isinstance(config, ModelConfiguration):
+ return cls(config, validate_input=validate_input)
+ return cls(
+ ModelConfiguration(config, validate_input, base_dir), validate_input=validate_input
+ )
+
+ @property
+ def configuration(self) -> "ModelConfiguration":
+ """The loaded configuration this model runs."""
+ return self._configuration
+
+ def run(self) -> RunResult:
+ """Run the simulation in the current process.
+
+ A fresh model framework is built for every call, and sets the clone of
+ its own grid. PCRaster state is process-wide all the same, so successive
+ calls in one interpreter grow the resident memory, and calls must not
+ run at the same time in several threads: they fail or, on the same grid,
+ may write wrong results without any error. See :meth:`run_isolated`.
+
+ :return: What the run wrote.
+ :raises ImportError: If PCRaster or GDAL are not installed.
+ :raises Exception: Whatever the simulation itself raises, unchanged.
+ """
+ _require_runtime_deps()
+ from .core import DynamicFrameworkWrapper
+
+ logger.info("Starting an in-process run.")
+ started = time.perf_counter()
+ DynamicFrameworkWrapper(self._configuration).run()
+ elapsed = time.perf_counter() - started
+ return _describe_run(self._configuration, elapsed)
+
+ def run_isolated(self) -> RunResult:
+ """Run the simulation in a fresh spawned subprocess.
+
+ The configuration crosses as its document (the copy taken when it was
+ loaded) plus its base directory and the validation flag, and is rebuilt
+ on the other side; the result crosses as plain data. The subprocess is
+ started for this call only and the executor is shut down before
+ returning, so the PCRaster state of the run leaves nothing behind. Every
+ call therefore pays a full interpreter start-up.
+
+ The spawn start method imports the main module of the caller in the
+ subprocess, so a script that calls this method must guard its entry
+ point with ``if __name__ == "__main__":`` to avoid re-running itself,
+ and must be a file: a script read from standard input cannot be
+ imported again. Either mistake kills the subprocess at start-up and is
+ reported as the ``RuntimeError`` below.
+
+ The call waits for the subprocess. An interrupt that reaches the
+ subprocess as well (Ctrl-C in a terminal) stops the run; one delivered
+ to the calling process alone is raised only after the run has finished.
+
+ A configuration problem found while the subprocess rebuilds the
+ configuration reaches the caller as the same
+ :class:`ConfigurationError`, with its problems; any other exception of
+ the run propagates as itself.
+
+ The log records of the simulation are emitted in the subprocess, which
+ starts from the default logging configuration: the handlers of the
+ caller do not see them.
+
+ :return: What the run wrote.
+ :raises ImportError: If PCRaster or GDAL are not installed.
+ :raises RuntimeError: If the subprocess dies before the run finishes.
+ """
+ _require_runtime_deps()
+ logger.info("Starting an isolated run in a spawned subprocess.")
+ executor = ProcessPoolExecutor(
+ max_workers=1,
+ mp_context=multiprocessing.get_context("spawn"),
+ max_tasks_per_child=1,
+ )
+ try:
+ future = executor.submit(
+ _run_document,
+ self._configuration.config,
+ self._configuration.base_dir,
+ self._validate_input,
+ )
+ document = future.result()
+ except BrokenProcessPool as error:
+ raise RuntimeError(
+ "The subprocess of Model.run_isolated() died before the run finished. When "
+ "it dies at start-up the cause is usually the calling script: the spawn start "
+ "method imports it again, so the call must sit under an "
+ '`if __name__ == "__main__":` guard, and a script read from standard input '
+ "cannot be imported again at all; the error of the subprocess is on standard "
+ "error. Otherwise the subprocess was killed from outside or the native "
+ "libraries crashed it."
+ ) from error
+ finally:
+ executor.shutdown(wait=True)
+ return _result_from_json(document)
+
+
+def _run_document(document: dict, base_dir: str | None, validate_input: bool) -> dict[str, Any]:
+ """Rebuild a configuration, run it, and return the result as plain data.
+
+ The entry point of the subprocess of :meth:`Model.run_isolated`. It lives at
+ module level because the spawn start method imports it by name in the child;
+ it is not part of the public surface.
+
+ :param document: The configuration document, legacy or format 1.0.
+ :param base_dir: Directory the relative paths of the document are anchored on.
+ :param validate_input: Whether to validate the input files and their content.
+ :return: The :class:`RunResult` of the run, as a JSON-able dictionary.
+ """
+ from .configuration.model_configuration import ModelConfiguration
+ from .core import DynamicFrameworkWrapper
+
+ configuration = ModelConfiguration(document, validate_input, base_dir)
+ started = time.perf_counter()
+ DynamicFrameworkWrapper(configuration).run()
+ elapsed = time.perf_counter() - started
+ return _result_to_json(_describe_run(configuration, elapsed))
+
+
+def _require_runtime_deps() -> None:
+ """Raise with the installation guidance when PCRaster or GDAL are missing.
+
+ :raises ImportError: If any conda-only runtime dependency is missing.
+ """
+ missing = missing_runtime_deps()
+ if missing:
+ raise ImportError(runtime_deps_message(missing))
+
+
+def _describe_run(configuration: "ModelConfiguration", elapsed_seconds: float) -> RunResult:
+ """Enumerate the outputs of a finished run from its configuration.
+
+ The output directory is made absolute first: a configuration may name it
+ relatively, and a result whose members were half absolute (the PCRaster
+ naming helper returns absolute paths) and half relative would be read
+ against two different directories.
+ """
+ directory = Path(configuration.output_directory.path).absolute()
+ period = configuration.simulation_period
+ metadata = directory / _METADATA_FILENAME if configuration.file_v1 is not None else None
+ return RunResult(
+ output_directory=directory,
+ rasters=_raster_members(configuration, directory),
+ time_series=_time_series_tables(configuration, directory),
+ metadata=metadata,
+ first_step=period.first_step,
+ last_step=period.last_step,
+ elapsed_seconds=elapsed_seconds,
+ )
+
+
+def _raster_members(
+ configuration: "ModelConfiguration", directory: Path
+) -> dict[str, tuple[Path, ...]]:
+ """The raster members of every enabled variable, per enabled format."""
+ formats = configuration.output_variables.file_formats
+ period = configuration.simulation_period
+ steps = range(period.first_step, period.last_step + 1)
+ members = {}
+ for variable in configuration.output_variables.get_enabled_raster_series():
+ paths = []
+ if OutputFileFormat.PCRASTER in formats:
+ paths.extend(
+ Path(get_raster_series_filepath(directory, variable.raster_filename_prefix, step))
+ for step in steps
+ )
+ if OutputFileFormat.GEOTIFF in formats:
+ paths.extend(
+ directory / output_raster_filename(variable.raster_filename_prefix, step, "tif")
+ for step in steps
+ )
+ members[variable.id] = tuple(paths)
+ return members
+
+
+def _time_series_tables(
+ configuration: "ModelConfiguration", directory: Path
+) -> dict[str, tuple[Path, ...]]:
+ """The time series tables of every enabled variable, per enabled format."""
+ if not _writes_time_series(configuration):
+ return {}
+ formats = configuration.output_variables.time_series_formats
+ tables = {}
+ for variable in configuration.output_variables.get_enabled_time_series():
+ paths = []
+ if TimeSeriesFileFormat.CSV in formats:
+ paths.append(directory / f"{variable.table_filename_prefix}.csv")
+ # The run writes a TSS file for every enabled time series and converts
+ # it to CSV afterwards, removing the source unless the TSS format is
+ # enabled as well; with the CSV format disabled nothing is converted.
+ if TimeSeriesFileFormat.CSV not in formats or TimeSeriesFileFormat.PCRASTER_TSS in formats:
+ paths.append(directory / f"{variable.table_filename_prefix}.tss")
+ tables[variable.id] = tuple(paths)
+ return tables
+
+
+def _writes_time_series(configuration: "ModelConfiguration") -> bool:
+ """Whether the run writes time series at all.
+
+ The same condition the model itself applies: the time series must be
+ enabled and the raster their aggregation reads must be configured.
+ """
+ variables = configuration.output_variables
+ if not variables.tss:
+ return False
+ if variables.aggregation == "zones":
+ return bool(configuration.raster_files.zones)
+ return bool(configuration.raster_files.sample_locations)
+
+
+def _result_to_json(result: RunResult) -> dict[str, Any]:
+ """Return ``result`` as a JSON-able dictionary, with the paths as strings."""
+ return {
+ "output_directory": str(result.output_directory),
+ "rasters": {
+ variable: [str(path) for path in paths] for variable, paths in result.rasters.items()
+ },
+ "time_series": {
+ variable: [str(path) for path in paths]
+ for variable, paths in result.time_series.items()
+ },
+ "metadata": None if result.metadata is None else str(result.metadata),
+ "first_step": result.first_step,
+ "last_step": result.last_step,
+ "elapsed_seconds": result.elapsed_seconds,
+ }
+
+
+def _result_from_json(document: dict[str, Any]) -> RunResult:
+ """Rebuild a :class:`RunResult` from the dictionary of :func:`_result_to_json`."""
+ metadata = document["metadata"]
+ return RunResult(
+ output_directory=Path(document["output_directory"]),
+ rasters={
+ variable: tuple(Path(path) for path in paths)
+ for variable, paths in document["rasters"].items()
+ },
+ time_series={
+ variable: tuple(Path(path) for path in paths)
+ for variable, paths in document["time_series"].items()
+ },
+ metadata=None if metadata is None else Path(metadata),
+ first_step=int(document["first_step"]),
+ last_step=int(document["last_step"]),
+ elapsed_seconds=float(document["elapsed_seconds"]),
+ )
diff --git a/rubem/configuration/_problems.py b/rubem/configuration/_problems.py
index e3c3ae0e..758cd2a3 100644
--- a/rubem/configuration/_problems.py
+++ b/rubem/configuration/_problems.py
@@ -34,6 +34,8 @@ class ConfigurationError(ValueError):
"""Raised when a configuration carries blocking problems.
:param problems: The problems found; the blocking ones are listed in the message.
+ :ivar problems: The problems the constructor was given, blocking or not.
+ :vartype problems: list[Problem]
"""
def __init__(self, problems: list[Problem]) -> None:
@@ -42,3 +44,15 @@ def __init__(self, problems: list[Problem]) -> None:
lines = [f"The configuration has {len(blocking)} blocking problem(s):"]
lines.extend(f"- {problem}" for problem in blocking)
super().__init__("\n".join(lines))
+
+ def __reduce__(self) -> tuple:
+ """Rebuild the exception from its problems, not from its message.
+
+ ``BaseException`` pickles itself through ``args``, which here is the
+ formatted message: unpickling would call the constructor with a string
+ and fail. Rebuilding from :attr:`problems` (each a pydantic model,
+ itself picklable) keeps the exception crossing a process boundary as
+ itself, with its problems intact. The instance dictionary travels as
+ the pickle state, so notes added to the exception survive as well.
+ """
+ return (type(self), (self.problems,), self.__dict__)
diff --git a/rubem/configuration/model_configuration.py b/rubem/configuration/model_configuration.py
index e15d42a3..2cda9ddb 100644
--- a/rubem/configuration/model_configuration.py
+++ b/rubem/configuration/model_configuration.py
@@ -1,3 +1,4 @@
+import copy
import json
import logging
import os
@@ -45,7 +46,8 @@ class ModelConfiguration:
required for running the model. It supports loading configuration from either a dictionary or a JSON file.
:param config_input: The configuration input: a dictionary with the legacy sections, or the
- path of a legacy JSON file.
+ path of a legacy JSON file. A dictionary is copied, so changing it after the load does not
+ change the loaded configuration.
:param validate_input: Whether to validate the input files and their content. Defaults to `True`.
:type validate_input: bool, optional
:param base_dir: Directory the relative paths of the configuration are anchored on. Defaults to
@@ -75,8 +77,11 @@ def __init__(
self.file_v1 = None
if isinstance(config_input, dict):
self.logger.debug("Reading configuration from dictionary")
- self.config = config_input
- self.__parse(config_input, duplicates=[])
+ # The document outlives the load (an isolated run rebuilds the
+ # configuration from it), so it must not follow what the caller
+ # does to the dictionary afterwards.
+ self.config = copy.deepcopy(config_input)
+ self.__parse(self.config, duplicates=[])
elif isinstance(config_input, (str, bytes, os.PathLike)):
config_input_path = as_path(config_input)
config_input_str = str(config_input_path)
diff --git a/tests/unit/api/__init__.py b/tests/unit/api/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/unit/api/_child.py b/tests/unit/api/_child.py
new file mode 100644
index 00000000..72f40be1
--- /dev/null
+++ b/tests/unit/api/_child.py
@@ -0,0 +1,60 @@
+"""Entry points a spawned subprocess imports by name.
+
+Those of ``Model.run_isolated``, and the worker of the process pool a caller
+builds for its parallel runs.
+
+A subprocess entry point is pickled by reference, so the child imports this
+module to find the function it has to call. Nothing here imports the model at
+module level: a probe must be able to look at the fresh interpreter before the
+native libraries are loaded, which is what tells a spawned child from a forked
+one.
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+
+MARKER_FILENAME = "child.json"
+
+
+def record_and_run(document, base_dir, validate_input):
+ """Record who runs the simulation, then delegate to the real entry point.
+
+ Installed over ``rubem.api._run_document`` in the parent; the child
+ unpickles this function by name and finds its own, unpatched
+ ``rubem.api._run_document``, so the delegation below is not a loop.
+
+ :param document: The configuration document.
+ :param base_dir: Directory the relative paths of the document are anchored on.
+ :param validate_input: Whether to validate the input files and their content.
+ :return: The result of the run, as the plain data the boundary carries.
+ """
+ preloaded = sorted(name for name in ("pcraster", "osgeo") if name in sys.modules)
+
+ from rubem import api
+
+ result = api._run_document(document, base_dir, validate_input)
+ marker = Path(document["DIRECTORIES"]["output"]) / MARKER_FILENAME
+ marker.write_text(json.dumps({"pid": os.getpid(), "preloaded": preloaded}), encoding="utf8")
+ return result
+
+
+def simulate(config):
+ """Build the model in this worker and run it here, as the documented pool does.
+
+ :param config: The configuration document.
+ :return: The pid of the worker and the result of the run.
+ """
+ from rubem.api import Model
+
+ return os.getpid(), Model.from_config(config).run()
+
+
+def die(document, base_dir, validate_input):
+ """Kill the subprocess the way a crashing native library does.
+
+ ``os._exit`` leaves no exception for the executor to send back, which is
+ the situation ``Model.run_isolated`` has to report on its own.
+ """
+ os._exit(3)
diff --git a/tests/unit/api/test_model.py b/tests/unit/api/test_model.py
new file mode 100644
index 00000000..4a10a1cc
--- /dev/null
+++ b/tests/unit/api/test_model.py
@@ -0,0 +1,675 @@
+"""The public Python API: loading a configuration, running it, and what it reports."""
+
+import json
+import math
+import multiprocessing
+import os
+import pickle
+import shutil
+import subprocess
+import sys
+import textwrap
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures.process import BrokenProcessPool
+from inspect import signature
+
+import numpy as np
+import pydantic
+import pytest
+
+from rubem import _deps, api
+from rubem.api import ConfigurationError, Model, RunResult
+from rubem.configuration._problems import Problem
+from rubem.configuration.model_configuration import ModelConfiguration
+from rubem.configuration.model_configuration_file import ModelConfigurationFile
+from rubem.configuration.model_configuration_file_v1 import ModelConfigurationFileV1
+from tests.helpers.config import REPO_ROOT, base_model_config
+from tests.helpers.synthetic import MISSING, series_name, write_synthetic_dataset
+from tests.unit.api import _child
+from tests.unit.core.test_aggregation import write_zones
+from tests.unit.core.test_core import expected_outputs
+
+VARIABLES = ("itp", "bfw", "srn", "eta", "lfw", "rec", "smc", "rnf", "arn")
+
+
+def reported_paths(result):
+ """Every path a result reports, rasters first, then time series tables."""
+ paths = [path for members in result.rasters.values() for path in members]
+ paths.extend(path for tables in result.time_series.values() for path in tables)
+ return paths
+
+
+def reported_names(result):
+ return sorted(path.name for path in reported_paths(result))
+
+
+def missing_files(result):
+ return [str(path) for path in reported_paths(result) if not path.is_file()]
+
+
+def written_bytes(result):
+ """The content of every file a result reports, by file name."""
+ return {path.name: path.read_bytes() for path in reported_paths(result)}
+
+
+def break_kp(config):
+ """Make the first pan coefficient raster non-positive, a blocking problem."""
+ import pcraster as pcr
+
+ pcr.setclone(config["RASTERS"]["clone"])
+ path = os.path.join(config["DIRECTORIES"]["kp"], series_name("kp", 1))
+ pcr.report(pcr.numpy2pcr(pcr.Scalar, np.zeros((3, 3), dtype=np.float32), -9999.0), path)
+
+
+def v1_document(legacy_config, **metadata):
+ """The format 1.0 equivalent of a legacy configuration document."""
+ legacy = ModelConfigurationFile.model_validate(legacy_config)
+ return ModelConfigurationFileV1.from_legacy(legacy, metadata or None).to_dict()
+
+
+def relative_to(config, base_dir):
+ """The same configuration with every path relative to ``base_dir``."""
+ relative = json.loads(json.dumps(config))
+ for section in ("DIRECTORIES", "RASTERS", "TABLES"):
+ relative[section] = {
+ key: (os.path.relpath(value, base_dir) if value else value)
+ for key, value in config[section].items()
+ }
+ return relative
+
+
+def v1_document_with_time_series_formats(legacy_config, formats):
+ """A format 1.0 document whose time series ask for ``formats``."""
+ document = v1_document(legacy_config)
+ document["model_simulation_output"]["time_series_samples"]["formats"] = list(formats)
+ return document
+
+
+class TestInProcessRuns:
+ @pytest.mark.unit
+ def test_a_run_from_a_dictionary_reports_every_file_it_wrote(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+
+ result = Model.from_config(config).run()
+
+ assert isinstance(result, RunResult)
+ assert result.output_directory == tmp_path / "out"
+ assert (result.first_step, result.last_step) == (1, 2)
+ assert result.metadata is None, "a legacy configuration has no metadata section"
+ assert math.isfinite(result.elapsed_seconds) and result.elapsed_seconds >= 0
+ assert reported_names(result) == sorted(expected_outputs())
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_raster_members_are_grouped_by_format_in_step_order(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+
+ result = Model.from_config(config).run()
+
+ assert sorted(result.rasters) == sorted(VARIABLES)
+ assert [path.name for path in result.rasters["itp"]] == [
+ series_name("itp", 1),
+ series_name("itp", 2),
+ "itp0000001.tif",
+ "itp0000002.tif",
+ ]
+ assert [path.name for path in result.time_series["itp"]] == ["tss_itp.csv"]
+
+ @pytest.mark.unit
+ def test_a_run_from_a_file_anchors_and_runs(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+ config_file = tmp_path / "config.json"
+ config_file.write_text(json.dumps(config), encoding="utf8")
+
+ result = Model.from_file(config_file).run()
+
+ assert result.output_directory == tmp_path / "out"
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_a_loaded_configuration_is_used_as_it_is(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+ loaded = ModelConfiguration(config, validate_input=False)
+
+ model = Model.from_config(loaded)
+
+ assert model.configuration is loaded
+ assert not missing_files(model.run())
+
+ @pytest.mark.unit
+ def test_a_format_1_0_run_reports_its_metadata(self, tmp_path):
+ document = v1_document(write_synthetic_dataset(str(tmp_path)), title="api")
+
+ result = Model.from_config(document).run()
+
+ assert result.metadata == tmp_path / "out" / "metadata.json"
+ assert not missing_files(result)
+ written = json.loads(result.metadata.read_text(encoding="utf8"))
+ assert written["version"] == "1.0" and written["title"] == "api"
+
+ @pytest.mark.unit
+ def test_the_reported_files_are_exactly_the_files_in_the_directory(self, tmp_path):
+ """The description of the run must match the listing it refuses to read."""
+ result = Model.from_config(write_synthetic_dataset(str(tmp_path))).run()
+
+ assert reported_names(result) == sorted(
+ path.name for path in result.output_directory.iterdir()
+ )
+
+ @pytest.mark.unit
+ def test_every_reported_path_is_absolute(self, tmp_path, monkeypatch):
+ """A relative output directory must not split the result in two anchors."""
+ config = write_synthetic_dataset(str(tmp_path))
+ config["DIRECTORIES"]["output"] = "out"
+ monkeypatch.chdir(tmp_path)
+
+ result = Model.from_config(config).run()
+
+ assert result.output_directory == tmp_path / "out"
+ assert all(path.is_absolute() for path in reported_paths(result))
+ assert {path.parent for path in reported_paths(result)} == {result.output_directory}
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_time_series_kept_as_pcraster_tss_are_the_reported_ones(self, tmp_path):
+ document = v1_document_with_time_series_formats(
+ write_synthetic_dataset(str(tmp_path)), ["PCRasterTSS"]
+ )
+
+ result = Model.from_config(document).run()
+
+ assert [path.name for path in result.time_series["itp"]] == ["tss_itp.tss"]
+ assert not missing_files(result)
+ assert not list(result.output_directory.glob("*.csv"))
+
+ @pytest.mark.unit
+ def test_both_time_series_formats_are_reported_csv_first(self, tmp_path):
+ document = v1_document_with_time_series_formats(
+ write_synthetic_dataset(str(tmp_path)), ["CSV", "PCRasterTSS"]
+ )
+
+ result = Model.from_config(document).run()
+
+ assert [path.name for path in result.time_series["itp"]] == ["tss_itp.csv", "tss_itp.tss"]
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_an_aggregated_run_reports_the_tables_under_their_aggregated_names(self, tmp_path):
+ """``zones`` renames the tables and writes a mapping that belongs to no variable."""
+ document = v1_document(write_synthetic_dataset(str(tmp_path)))
+ document["rasters"]["zones"] = write_zones(
+ document, [20, 20, 20, 7, 7, 7, MISSING, MISSING, MISSING]
+ )
+ document["model_simulation_output"]["time_series_samples"]["aggregation"] = "zones"
+ del document["rasters"]["samples"]
+
+ result = Model.from_config(document).run()
+
+ assert [path.name for path in result.time_series["itp"]] == ["tss_itp_zones.csv"]
+ assert not missing_files(result)
+ assert (result.output_directory / "zones_mapping.csv").is_file()
+ assert "zones_mapping.csv" not in reported_names(result)
+
+ @pytest.mark.unit
+ def test_runs_on_different_grids_follow_one_another_in_one_process(self, tmp_path):
+ """Every run sets the clone for itself, so the grid of the last one is no constraint."""
+ small = Model.from_config(write_synthetic_dataset(str(tmp_path / "small")))
+ large = Model.from_config(base_model_config(str(tmp_path / "large")))
+ small_alone = written_bytes(small.run())
+ large_alone = written_bytes(large.run())
+
+ small_after_large = written_bytes(small.run())
+ large_after_small = written_bytes(large.run())
+
+ member = series_name("itp", 1)
+ assert len(small_alone[member]) != len(large_alone[member]), "the grids must differ"
+ assert small_after_large == small_alone
+ assert large_after_small == large_alone
+
+ @pytest.mark.unit
+ def test_a_run_writes_nothing_to_stdout(self, tmp_path, capsys):
+ """The library only logs; a front end is what prints."""
+ Model.from_config(write_synthetic_dataset(str(tmp_path))).run()
+
+ assert capsys.readouterr().out == ""
+
+
+class TestIsolatedRuns:
+ @pytest.mark.unit
+ def test_two_sequential_isolated_runs_write_two_directories(self, tmp_path):
+ results = []
+ for name in ("first", "second"):
+ base = tmp_path / name
+ base.mkdir()
+ results.append(Model.from_config(write_synthetic_dataset(str(base))).run_isolated())
+
+ first, second = results
+ assert first.output_directory == tmp_path / "first" / "out"
+ assert second.output_directory == tmp_path / "second" / "out"
+ for result in results:
+ assert reported_names(result) == sorted(expected_outputs())
+ assert not missing_files(result)
+ assert (result.first_step, result.last_step) == (1, 2)
+ assert math.isfinite(result.elapsed_seconds) and result.elapsed_seconds >= 0
+
+ @pytest.mark.unit
+ def test_reusing_the_dictionary_for_another_model_leaves_the_first_run_alone(self, tmp_path):
+ """A batch that edits one dictionary per experiment must not redirect an earlier model."""
+ config = write_synthetic_dataset(str(tmp_path))
+ first = Model.from_config(config)
+ other_output = tmp_path / "other"
+ config["DIRECTORIES"]["output"] = str(other_output)
+ Model.from_config(config)
+
+ result = first.run_isolated()
+
+ assert result.output_directory == tmp_path / "out"
+ assert not missing_files(result)
+ assert list(other_output.iterdir()) == [], "the first run wrote to the second directory"
+
+ @pytest.mark.unit
+ def test_each_model_submits_the_document_it_was_built_from(self, tmp_path, mocker):
+ """Both execution modes must run the same parameters, those of the load."""
+ config = write_synthetic_dataset(str(tmp_path))
+ first = Model.from_config(config, validate_input=False)
+ config["CALIBRATION"]["alpha"] = 9.0
+ second = Model.from_config(config, validate_input=False)
+ executor = mocker.MagicMock()
+ executor.submit.return_value.result.return_value = api._result_to_json(
+ api._describe_run(first.configuration, 0.0)
+ )
+ mocker.patch.object(api, "ProcessPoolExecutor", return_value=executor)
+
+ first.run_isolated()
+ second.run_isolated()
+
+ submitted = [
+ call.args[1]["CALIBRATION"]["alpha"] for call in executor.submit.call_args_list
+ ]
+ assert submitted == [4.5, 9.0]
+ assert first.configuration.calibration_parameters.alpha == 4.5
+ assert second.configuration.calibration_parameters.alpha == 9.0
+
+ @pytest.mark.unit
+ def test_the_anchor_of_a_file_crosses_the_boundary(self, tmp_path, monkeypatch):
+ """The child anchors on the directory of the file, not on its own cwd."""
+ config_file = tmp_path / "config.json"
+ config_file.write_text(
+ json.dumps(relative_to(write_synthetic_dataset(str(tmp_path)), tmp_path)),
+ encoding="utf8",
+ )
+ elsewhere = tmp_path / "elsewhere"
+ elsewhere.mkdir()
+ monkeypatch.chdir(elsewhere)
+
+ result = Model.from_file(config_file).run_isolated()
+
+ assert result.output_directory == tmp_path / "out"
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_an_explicit_base_dir_crosses_the_boundary(self, tmp_path, monkeypatch):
+ """A document has no anchor of its own; the one given must travel with it."""
+ config = relative_to(write_synthetic_dataset(str(tmp_path)), tmp_path)
+ elsewhere = tmp_path / "elsewhere"
+ elsewhere.mkdir()
+ monkeypatch.chdir(elsewhere)
+
+ result = Model.from_config(config, base_dir=tmp_path).run_isolated()
+
+ assert result.output_directory == tmp_path / "out"
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_the_run_happens_in_a_spawned_child_that_does_not_outlive_the_call(
+ self, tmp_path, monkeypatch
+ ):
+ """The child is a fresh interpreter, and the executor takes it with it."""
+ config = write_synthetic_dataset(str(tmp_path))
+ monkeypatch.setattr(api, "_run_document", _child.record_and_run)
+
+ result = Model.from_config(config).run_isolated()
+
+ assert not missing_files(result)
+ marker = json.loads(
+ (result.output_directory / _child.MARKER_FILENAME).read_text(encoding="utf8")
+ )
+ assert marker["pid"] != os.getpid(), "the run must not happen in the calling process"
+ assert "pcraster" in sys.modules, "the parent of this test has the model loaded"
+ assert marker["preloaded"] == [], "a spawned child starts without the parent's modules"
+ # The executor's worker is a multiprocessing child of this process, so an
+ # empty active_children() proves it was joined; a probe with os.kill would
+ # not be portable (Windows has no signal 0) and could hit a reused pid.
+ assert multiprocessing.active_children() == []
+
+ @pytest.mark.unit
+ def test_a_child_that_dies_is_reported_against_run_isolated(self, tmp_path, monkeypatch):
+ """A child killed mid-run, without an exception to send back."""
+ model = Model.from_config(write_synthetic_dataset(str(tmp_path)), validate_input=False)
+ monkeypatch.setattr(api, "_run_document", _child.die)
+
+ with pytest.raises(RuntimeError, match="Model.run_isolated") as error:
+ model.run_isolated()
+
+ assert isinstance(error.value.__cause__, BrokenProcessPool)
+ assert multiprocessing.active_children() == []
+
+ @pytest.mark.unit
+ def test_another_exception_of_the_child_crosses_the_boundary_as_itself(self, tmp_path):
+ """Not every failure is a configuration problem; the type must survive."""
+ config = write_synthetic_dataset(str(tmp_path))
+ model = Model.from_config(config, validate_input=False)
+ output = tmp_path / "out"
+ shutil.rmtree(output)
+ output.write_text("not a directory", encoding="utf8")
+
+ with pytest.raises(NotADirectoryError, match="out"):
+ model.run_isolated()
+
+ @pytest.mark.unit
+ def test_a_configuration_problem_crosses_the_boundary_as_itself(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+ model = Model.from_config(config)
+ break_kp(config)
+
+ with pytest.raises(ConfigurationError) as error:
+ model.run_isolated()
+
+ assert any("not positive" in str(problem) for problem in error.value.problems)
+ assert any(problem.blocking for problem in error.value.problems)
+
+ @pytest.mark.unit
+ def test_skipping_the_validation_crosses_the_boundary_too(self, tmp_path):
+ config = write_synthetic_dataset(str(tmp_path))
+ break_kp(config)
+ with pytest.raises(ConfigurationError):
+ Model.from_config(config)
+
+ result = Model.from_config(config, validate_input=False).run_isolated()
+
+ assert not missing_files(result)
+
+ @pytest.mark.unit
+ def test_a_dead_subprocess_is_reported_against_run_isolated(self, tmp_path, mocker):
+ model = Model.from_config(write_synthetic_dataset(str(tmp_path)), validate_input=False)
+ executor = mocker.MagicMock()
+ executor.submit.return_value.result.side_effect = BrokenProcessPool("killed")
+ mocker.patch.object(api, "ProcessPoolExecutor", return_value=executor)
+
+ with pytest.raises(RuntimeError, match="Model.run_isolated") as error:
+ model.run_isolated()
+
+ executor.shutdown.assert_called_once_with(wait=True)
+ # A death at start-up is the caller's script far more often than a crash.
+ assert '__name__ == "__main__"' in str(error.value)
+ assert "standard input" in str(error.value)
+
+ @pytest.mark.unit
+ def test_a_script_without_the_entry_point_guard_is_told_so(self, tmp_path):
+ """The child dies re-importing the script; the caller must learn why, and not hang."""
+ config_file = tmp_path / "config.json"
+ config_file.write_text(json.dumps(write_synthetic_dataset(str(tmp_path))), encoding="utf8")
+ script = tmp_path / "unguarded.py"
+ script.write_text(_UNGUARDED_SCRIPT, encoding="utf8")
+
+ completed = subprocess.run(
+ [sys.executable, str(script), str(config_file)],
+ cwd=REPO_ROOT,
+ env={**os.environ, "PYTHONPATH": REPO_ROOT},
+ capture_output=True,
+ text=True,
+ check=False,
+ timeout=300,
+ )
+
+ assert completed.returncode != 0
+ assert "finished" not in completed.stdout
+ raised = [line for line in completed.stderr.splitlines() if "Model.run_isolated()" in line]
+ assert raised, completed.stderr
+ assert '__name__ == "__main__"' in raised[-1]
+
+
+class TestParallelRuns:
+ @pytest.mark.unit
+ def test_a_process_pool_of_the_caller_runs_in_parallel(self, tmp_path):
+ """One process per run at a time; reused workers change grid between their runs."""
+ references = {
+ "small": written_bytes(
+ Model.from_config(write_synthetic_dataset(str(tmp_path / "small"))).run()
+ ),
+ "large": written_bytes(
+ Model.from_config(base_model_config(str(tmp_path / "large"))).run()
+ ),
+ }
+ grids, jobs = [], []
+ for number in range(6):
+ grid = "small" if number % 2 else "large"
+ base = tmp_path / f"job{number}"
+ base.mkdir()
+ config = (
+ write_synthetic_dataset(str(base))
+ if grid == "small"
+ else base_model_config(str(base / "out"))
+ )
+ grids.append(grid)
+ jobs.append(config)
+
+ context = multiprocessing.get_context("spawn")
+ with ProcessPoolExecutor(max_workers=2, mp_context=context) as pool:
+ outcomes = list(pool.map(_child.simulate, jobs))
+
+ workers = {pid for pid, _ in outcomes}
+ assert os.getpid() not in workers
+ assert len(workers) <= 2, "the workers are reused from one run to the next"
+ for grid, (_, result) in zip(grids, outcomes, strict=True):
+ assert written_bytes(result) == references[grid]
+
+
+class TestDocumentedLoaderFailures:
+ """The loaders fail before validation too, with the documented types."""
+
+ @pytest.mark.unit
+ def test_a_configuration_file_that_is_not_there_is_reported_as_such(self, tmp_path):
+ with pytest.raises(FileNotFoundError):
+ Model.from_file(tmp_path / "absent.json")
+
+ @pytest.mark.unit
+ def test_a_configuration_file_that_is_not_json_is_reported_as_such(self, tmp_path):
+ path = tmp_path / "config.json"
+ path.write_text("{not json", encoding="utf8")
+
+ with pytest.raises(json.JSONDecodeError):
+ Model.from_file(path)
+
+ @pytest.mark.unit
+ def test_a_document_that_does_not_match_the_schema_is_reported_as_such(self, tmp_path):
+ with pytest.raises(pydantic.ValidationError):
+ Model.from_config({"bogus": 1}, base_dir=str(tmp_path))
+
+
+class TestConfigurationErrorPickling:
+ @pytest.mark.unit
+ def test_a_round_trip_preserves_the_problems_and_the_message(self):
+ # The payload is the exception built right above, not foreign data:
+ # the round trip is what an isolated run does across its own pipe.
+ error = ConfigurationError(
+ [
+ Problem(description="kp", reason="is not positive", blocking=True),
+ Problem(description="ndvi", reason="has gaps"),
+ ]
+ )
+
+ error.add_note("while rebuilding the configuration")
+
+ restored = pickle.loads(pickle.dumps(error))
+
+ assert type(restored) is ConfigurationError
+ assert restored.problems == error.problems
+ assert str(restored) == str(error)
+ assert restored.__notes__ == ["while rebuilding the configuration"]
+
+
+class TestResultSerialization:
+ @pytest.mark.unit
+ def test_a_result_survives_the_json_round_trip_the_process_boundary_uses(self, tmp_path):
+ result = Model.from_config(write_synthetic_dataset(str(tmp_path))).run()
+
+ restored = api._result_from_json(json.loads(json.dumps(api._result_to_json(result))))
+
+ assert restored == result
+
+
+_WITHOUT_NATIVE_DEPENDENCIES = textwrap.dedent(
+ """
+ import sys
+
+
+ class BlockNativeDependencies:
+ def find_spec(self, name, path=None, target=None):
+ if name.partition(".")[0] in ("pcraster", "osgeo"):
+ raise ModuleNotFoundError(f"blocked: {name}", name=name)
+ return None
+
+
+ sys.meta_path.insert(0, BlockNativeDependencies())
+
+ from rubem import api
+
+ assert "pcraster" not in sys.modules, "importing rubem.api must not import pcraster"
+ assert "osgeo" not in sys.modules, "importing rubem.api must not import osgeo"
+
+ calls = {
+ "run": lambda: api.Model.__new__(api.Model).run(),
+ "run_isolated": lambda: api.Model.__new__(api.Model).run_isolated(),
+ "from_config": lambda: api.Model.from_config({}),
+ "from_file": lambda: api.Model.from_file("config.json"),
+ }
+ for name, call in calls.items():
+ try:
+ call()
+ except ImportError as error:
+ print(name, error)
+ else:
+ raise AssertionError(f"{name} without the native dependencies must raise ImportError")
+ """
+)
+
+_ISOLATION_OF_THE_CALLER = textwrap.dedent(
+ """
+ import json
+ import sys
+
+ from rubem.api import Model
+
+ model = Model.from_file(sys.argv[1], validate_input=False)
+ result = model.run_isolated()
+ assert result.rasters, "the isolated run must report what it wrote"
+ print(json.dumps({name: name in sys.modules for name in ("pcraster", "osgeo")}))
+ """
+)
+
+
+_UNGUARDED_SCRIPT = textwrap.dedent(
+ """
+ import sys
+
+ from rubem.api import Model
+
+ Model.from_file(sys.argv[1], validate_input=False).run_isolated()
+ print("finished")
+ """
+)
+
+
+class TestWithoutTheNativeDependencies:
+ @pytest.mark.unit
+ def test_the_module_imports_and_a_run_explains_what_is_missing(self):
+ completed = subprocess.run(
+ [sys.executable, "-c", _WITHOUT_NATIVE_DEPENDENCIES],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+
+ assert completed.returncode == 0, completed.stderr
+ for name in ("run", "run_isolated", "from_config", "from_file"):
+ assert f"{name} RUBEM cannot run" in completed.stdout
+ assert "pcraster, osgeo" in completed.stdout
+ assert "conda-forge" in completed.stdout
+ assert "environment.yml" in completed.stdout
+
+ @pytest.mark.unit
+ def test_the_command_line_still_exits_where_the_api_raises(self, monkeypatch):
+ """The two front ends share the text, not the exception."""
+ assert _deps.missing_runtime_deps() == [], "this environment has the native dependencies"
+ # The API imported the function by name, so both bindings are replaced.
+ monkeypatch.setattr(_deps, "missing_runtime_deps", lambda: ["pcraster"])
+ monkeypatch.setattr(api, "missing_runtime_deps", lambda: ["pcraster"])
+
+ with pytest.raises(SystemExit) as exit_:
+ _deps.require_runtime_deps()
+ with pytest.raises(ImportError) as imported:
+ Model.from_config({})
+
+ assert str(exit_.value) == _deps.runtime_deps_message(["pcraster"])
+ assert str(imported.value) == str(exit_.value)
+
+
+class TestTheCallerStaysFreeOfTheNativeState:
+ @pytest.mark.unit
+ def test_an_isolated_run_never_loads_pcraster_in_the_caller(self, tmp_path):
+ """Only the child touches PCRaster: that is what run_isolated buys."""
+ config_file = tmp_path / "config.json"
+ config_file.write_text(json.dumps(write_synthetic_dataset(str(tmp_path))), encoding="utf8")
+
+ completed = subprocess.run(
+ [sys.executable, "-c", _ISOLATION_OF_THE_CALLER, str(config_file)],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+
+ assert completed.returncode == 0, completed.stderr
+ loaded = json.loads(completed.stdout)
+ assert loaded["pcraster"] is False
+ # Loading a configuration imports GDAL in the caller even with the
+ # validation off; only the PCRaster state stays in the child.
+ assert loaded["osgeo"] is True
+
+
+class TestPublicSurface:
+ @pytest.mark.unit
+ def test_the_exported_names_are_the_documented_ones(self):
+ assert sorted(api.__all__) == ["ConfigurationError", "Model", "RunResult"]
+
+ @pytest.mark.unit
+ def test_the_signatures_are_the_documented_ones(self):
+ assert str(signature(Model.__init__)) == (
+ "(self, configuration: 'ModelConfiguration', *, validate_input: bool = True) -> None"
+ )
+ assert str(signature(Model.from_file)) == (
+ "(path: str | os.PathLike[str] | bytes, *, validate_input: bool = True,"
+ " base_dir: str | os.PathLike[str] | bytes | None = None) -> 'Model'"
+ )
+ assert str(signature(Model.from_config)) == (
+ "(config: 'dict | ModelConfiguration', *, validate_input: bool = True,"
+ " base_dir: str | os.PathLike[str] | bytes | None = None) -> 'Model'"
+ )
+ assert str(signature(Model.run)) == "(self) -> rubem.api.RunResult"
+ assert str(signature(Model.run_isolated)) == "(self) -> rubem.api.RunResult"
+ assert isinstance(Model.configuration, property)
+
+ @pytest.mark.unit
+ def test_the_result_fields_are_the_documented_ones(self):
+ assert list(RunResult.__dataclass_fields__) == [
+ "output_directory",
+ "rasters",
+ "time_series",
+ "metadata",
+ "first_step",
+ "last_step",
+ "elapsed_seconds",
+ ]
+ assert RunResult.__dataclass_params__.frozen
diff --git a/tests/unit/configuration/test_model_configuration.py b/tests/unit/configuration/test_model_configuration.py
index dc2e7e3b..6f4a53de 100644
--- a/tests/unit/configuration/test_model_configuration.py
+++ b/tests/unit/configuration/test_model_configuration.py
@@ -8,6 +8,7 @@
from rubem.configuration.input_raster_series import InputRasterSeries
from rubem.configuration.model_configuration import ModelConfiguration
from rubem.configuration.raster_map import RasterBand
+from tests.helpers.synthetic import write_synthetic_dataset
class TestModelConfiguration:
@@ -299,3 +300,18 @@ def test_init_with_incomplete_json_file(self, fs):
)
with pytest.raises(Exception):
_ = ModelConfiguration("/test_path/config.json")
+
+
+class TestTheLoadedDocumentIsOwned:
+ @pytest.mark.unit
+ def test_changing_the_dictionary_afterwards_leaves_the_loaded_document_alone(self, tmp_path):
+ """The document feeds the rebuild of an isolated run, long after the load."""
+ document = write_synthetic_dataset(str(tmp_path))
+
+ loaded = ModelConfiguration(document, validate_input=False)
+ document["CALIBRATION"]["alpha"] = 9.0
+ document["DIRECTORIES"]["output"] = str(tmp_path / "elsewhere")
+
+ assert loaded.config is not document
+ assert loaded.config["CALIBRATION"]["alpha"] == 4.5
+ assert loaded.config["DIRECTORIES"]["output"] == str(tmp_path / "out")