Source code for hydroflows.methods.fiat.fiat_run
"""Method for running a FIAT model."""
import subprocess
from pathlib import Path
from typing import Literal, Optional
from pydantic import model_validator
from hydroflows._typing import FileDirPath
from hydroflows.workflow.method import Method
from hydroflows.workflow.method_parameters import Parameters
[docs]
class Output(Parameters):
"""Output parameters.
This class represents the output data
generated by the :py:class:`FIATRun` method.
"""
fiat_out_gpkg: FileDirPath
"""The resulting spatial file from the fiat calculations."""
fiat_out_csv: Path
"""The resulting csv file from the fiat calculations."""
[docs]
class Params(Parameters):
"""Parameters.
Instances of this class are used in the :py:class:`FIATRun`
method to define the required settings.
"""
run_method: Literal["exe", "python"] = "exe"
"""How to run the FIAT model. Options are 'exe' for running the executable directly (only on Windows),
'python' for running the model in a Python environment."""
fiat_exe: Optional[Path] = None
"""The path to the FIAT executable."""
threads: int = 1
"""The number of the threads to be used."""
@model_validator(mode="after")
def check_fiat_exe(self):
"""Check the FIAT binary path."""
if self.run_method == "exe" and self.fiat_exe is None:
raise ValueError("FIAT binary path is required when run_method is 'exe'.")
return self
[docs]
class FIATRun(Method):
"""Method for running a FIAT model.
Parameters
----------
fiat_cfg : Path
Path to the FIAT config file.
run_method : Literal["exe", "python"]
How to run the FIAT model. Options are 'exe' for running the executable directly (only on Windows),
'python' for running the model in a Python environment.
fiat_exe : Path
Path to the FIAT executable
**params
Additional parameters to pass to the FIATRun instance.
See :py:class:`fiat_run Params <hydroflows.methods.fiat.fiat_run.Params>`.
See Also
--------
:py:class:`fiat_run Input <hydroflows.methods.fiat.fiat_run.Input>`
:py:class:`fiat_run Output <hydroflows.methods.fiat.fiat_run.Output>`
:py:class:`fiat_run Params <hydroflows.methods.fiat.fiat_run.Params>`
"""
name: str = "fiat_run"
_test_kwargs = {
"fiat_cfg": Path("fiat.toml"),
"fiat_exe": Path("fiat.exe"),
}
def __init__(
self,
fiat_cfg: Path,
run_method: Literal["exe", "python"] = "exe",
fiat_exe: Optional[Path] = None,
**params,
):
self.params: Params = Params(fiat_exe=fiat_exe, run_method=run_method, **params)
self.input: Input = Input(fiat_cfg=fiat_cfg)
self.output: Output = Output(
fiat_out_gpkg=self.input.fiat_cfg.parent / "output" / "spatial.gpkg",
fiat_out_csv=self.input.fiat_cfg.parent / "output" / "output.csv",
)
# TODO check if cfg matches output
def _run(self):
"""Run the FIATRun method."""
# Get basic info
cwd = self.input.fiat_cfg.parent
fiat_cfg_path = self.input.fiat_cfg.name
entrypoint = (
"fiat"
if self.params.run_method == "python"
else self.params.fiat_exe.as_posix()
)
# Setup the cli command
command = [
entrypoint,
"run",
fiat_cfg_path,
"-t",
str(self.params.threads),
]
# Execute the rule
subprocess.run(command, check=True, cwd=cwd)
