"""The run orchestration.
This is where the five contracts meet. A run writes a keyword file, writes the tree list beside
it, drives BC through PyFVS's step API, reads the outputs back, and records its provenance. The
reference pages govern each part: :doc:`/reference/tree-list` the records,
:doc:`/reference/keywords` the set, :doc:`/reference/outputs` the read-back,
:doc:`/reference/provenance` the record, and :doc:`/reference/nemora-ingest` the adapter that feeds
it.
Three facts about the run are not choices, and each of them is a way to get a run quietly wrong
- **The tree-data file is named after the keyword file, in lowercase.** PyFVS computes its treelist
path as ``keywords_path.with_suffix('.tre')`` and refuses to start when that file is absent, so a
list written as ``.TRE`` passes FVS and fails PyFVS.
- **The output database has a fixed name.** ``dbsqlite/dbsblkd.f`` has ``DATA DSNOUT/"FVSOut.db"``,
so the database lands in the working directory under that name whatever the keyword file is
called. Two runs in one directory overwrite each other's database.
- **The tree-record format must be the variant's.** PyFVS injects its own ``TREEFMT`` when it is
given a keyword *set*, using a default format rather than BC's. A run here passes a keyword
*file*, so nothing is injected and BC's format is the only one a run can get.
Because the database's name is fixed, and because FVS keeps its state in the process, **a run
drives the projection in a child process** (ADR 0010). The child owns the working directory and the
Fortran state; this process shares nothing between runs, so two runs in one process are independent
and a run is thread-safe. The child writes the summary to ``summary.csv`` and the build's variant
list to ``variants.json``, and the parent reads them. The summary is still FVS's own frame rather
than a re-derivation, and it is inspectable on disk.
What the orchestration writes, and what it refuses
--------------------------------------------------
The caller writes the stand: ``STDIDENT``, ``STDINFO``, ``DESIGN``, ``NUMCYCLE``, ``SITECODE``, and
whatever else the case needs. The orchestration writes the tree handoff, which is ``TREEFMT`` with
BC's format and ``TREEDATA`` with a blank parameter, placed before the first ``PROCESS`` or ``STOP``
so that FVS reads the trees before it starts the projection. A set that already carries either
keyword is refused rather than quietly overridden, so nothing the caller wrote is replaced.
That insertion is a reading of the keyword contract's rule 4, which allows one exception, "the
tree-record format, which it must control". Without ``TREEDATA`` FVS never opens the tree file, so
the format and the instruction to use it are one handoff. The reading is recorded on
:doc:`/reference/keywords` rather than left to be inferred from this module.
The keyword hash covers what FVS read
-------------------------------------
Provenance is taken after the handoff is inserted, because a hash of the set the caller wrote would
identify a set no run ever saw.
"""
from __future__ import annotations
import importlib.util
import json
import math
import os
import subprocess
import sys
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
import pandas as pd
from fresh_fvsbc.bec import BecCode, BecCodeError
from fresh_fvsbc.bundle import BundleError, CalibrationBundle
from fresh_fvsbc.calibration import SPECIES_GROUPS
from fresh_fvsbc.keywords import Keyword, KeywordSet
from fresh_fvsbc.nemora import PlotFields, to_tree_list
from fresh_fvsbc.outputs import check_run, read_cases
from fresh_fvsbc.provenance import Provenance, built_variants, from_cases, treelist_hash
from fresh_fvsbc.treelist import BC_TREFMT, TreeListFields, write_tree_file
__all__ = [
"DATABASE_NAME",
"KEYWORD_FILE_NAME",
"PROVENANCE_FILE_NAME",
"SUMMARY_FILE_NAME",
"TREEFMT_LINES",
"VARIANTS_FILE_NAME",
"ExtensionUnavailableError",
"KeywordConflictError",
"RunError",
"RunResult",
"database_path",
"run_case",
"run_population",
"tree_file_name",
"treelist_format_lines",
"with_tree_handoff",
]
#: The keyword file's name inside a run's directory. FVS and PyFVS both derive other names from it.
KEYWORD_FILE_NAME: str = "run.key"
#: The database's name, which ``dbsblkd.f`` fixes and no caller can change.
DATABASE_NAME: str = "FVSOut.db"
#: The provenance record's name, which the provenance contract puts beside the outputs.
PROVENANCE_FILE_NAME: str = "provenance.json"
#: What the child writes for the parent: the summary FVS produced, and the build's variant list.
SUMMARY_FILE_NAME: str = "summary.csv"
VARIANTS_FILE_NAME: str = "variants.json"
#: The exit code the child uses when PyFVS is not importable, so the parent re-raises the same type.
EXTENSION_UNAVAILABLE_CODE: int = 3
#: The child's module, run with ``python -m``.
_CHILD_MODULE: str = "fresh_fvsbc._child"
#: This package's parent directory, which the child needs on its path when it is not installed.
_PACKAGE_ROOT: Path = Path(__file__).resolve().parents[1]
#: The two keywords the orchestration writes, in the order FVS needs them.
TREE_FORMAT_KEYWORD: str = "TREEFMT"
TREE_DATA_KEYWORD: str = "TREEDATA"
_TREE_HANDOFF: tuple[str, ...] = (TREE_FORMAT_KEYWORD, TREE_DATA_KEYWORD)
#: The keywords that start and end a projection. The handoff goes before the first of them.
_PROJECTION_KEYWORDS: tuple[str, ...] = ("PROCESS", "STOP")
#: The point in BC's format where the format is split across the two lines FVS reads it from. It is
#: the split the variant's own keyword files use, and a split inside a group is safe because FVS
#: joins the two lines without padding.
_SPLIT_AFTER: str = "2I3,"
[docs]
class RunError(Exception):
"""A run that could not be set up, driven, or read."""
[docs]
class ExtensionUnavailableError(RunError):
"""PyFVS is not importable, or the variant it was asked for did not load."""
[docs]
class KeywordConflictError(RunError):
"""A keyword set that carries a keyword the orchestration writes itself."""
#: BC's format, split into the two lines a ``TREEFMT`` keyword carries.
TREEFMT_LINES: tuple[str, str] = treelist_format_lines()
[docs]
def database_path(directory: str | Path) -> Path:
"""Return the path of the database a run in ``directory`` writes.
Parameters
----------
directory : str or Path
The run's directory.
Returns
-------
Path
``directory/FVSOut.db``, the **default** ``DSNOUT`` that ``dbsblkd.f`` initialises. A
keyword set carrying a ``DATABASE`` block with a ``DSNOUT`` option names another, and this
function does not read it; the route this module runs carries no such block.
"""
return Path(directory) / DATABASE_NAME
[docs]
def tree_file_name(key_path: str | Path) -> Path:
"""Return the tree-data path PyFVS expects beside a keyword file.
Parameters
----------
key_path : str or Path
The keyword file's path.
Returns
-------
Path
The same path with a lowercase ``.tre`` suffix. FVS would also accept ``.TRE``, but PyFVS
checks for ``.tre`` and refuses to start without it, so the lowercase name is the only one
that works through this route.
"""
return Path(key_path).with_suffix(".tre")
[docs]
def with_tree_handoff(keywords: Sequence[Keyword]) -> tuple[Keyword, ...]:
"""Return a keyword set with the tree handoff inserted, in the caller's order.
The handoff goes before the first ``PROCESS`` or ``STOP``, because FVS reads the trees when it
reaches ``TREEDATA`` and starts the projection at ``PROCESS``. A set that carries neither ends
with the handoff.
Parameters
----------
keywords : Sequence of Keyword
The caller's set, which describes the stand and the run.
Returns
-------
tuple of Keyword
The caller's keywords in order, with ``TREEFMT`` and ``TREEDATA`` inserted.
Raises
------
KeywordConflictError
If the set already carries either keyword. Overriding it would replace something the caller
wrote, which the keyword contract forbids.
"""
given = tuple(keywords)
carried = [keyword.name for keyword in given if keyword.name in _TREE_HANDOFF]
if carried:
raise KeywordConflictError(
f"the keyword set carries {', '.join(carried)}, which the run writes itself"
)
handoff = (Keyword(TREE_FORMAT_KEYWORD, TREEFMT_LINES), Keyword(TREE_DATA_KEYWORD))
for index, keyword in enumerate(given):
if keyword.name in _PROJECTION_KEYWORDS:
return (*given[:index], *handoff, *given[index:])
return (*given, *handoff)
def _seed_of(keywords: Sequence[Keyword]) -> int | None:
"""Return the seed the set carries, or ``None`` when it carries none."""
for keyword in keywords:
if keyword.name != "RANNSEED":
continue
if len(keyword.parameters) != 1:
raise RunError(f"RANNSEED takes one parameter, not {len(keyword.parameters)}")
try:
return int(keyword.parameters[0])
except ValueError as error:
raise RunError(f"RANNSEED is not a whole number: {keyword.parameters[0]!r}") from error
return None
@dataclass(frozen=True, slots=True)
class _Drive:
"""What driving a projection yields: the summary, and the build's variant list."""
summary: pd.DataFrame
variants: tuple[str, ...]
def _drive_here(
key_name: str, *, variant: str, directory: Path
) -> tuple[pd.DataFrame, tuple[str, ...]]:
"""Drive one projection in **this** process, from inside ``directory``.
This is the child process's work (ADR 0010). It is not what :func:`run_case` calls, and it is
public to this module only so the child can reach it.
Parameters
----------
key_name : str
The keyword file's name, relative to ``directory``.
variant : str
The variant to run, which for this project is ``bc``.
directory : Path
The run's directory, which this function changes into and always leaves.
Returns
-------
tuple
The summary the run produced in memory, and the build's variant list.
Raises
------
ExtensionUnavailableError
If PyFVS is not importable, or the variant did not load.
RunError
If the projection ends with a non-zero return code.
"""
if importlib.util.find_spec("pyfvs") is None:
raise ExtensionUnavailableError(
"pyfvs is not importable, so no variant can run; the build recipe is in ADR 0004 and a "
"worked run is in notes/reference/fvsbc-smoke-run.md"
)
import pyfvs
from pyfvs import fvs
try:
simulator = fvs.FVS(variant)
except Exception as error: # PyFVS raises its own types when a variant is absent
raise ExtensionUnavailableError(f"the {variant} variant did not load: {error}") from error
previous = Path.cwd()
os.chdir(directory)
try:
simulator.init_projection(key_name)
for _ in range(int(simulator.globals.ncyc)):
simulator.grow()
code = int(simulator.end_projection())
finally:
os.chdir(previous)
if code != 0:
raise RunError(f"the projection ended with return code {code}")
return simulator.summary, built_variants(pyfvs.list_variants())
def _child_environment() -> dict[str, str]:
"""Return the environment the child needs to import this package.
The parent may have been started with ``src`` on ``sys.path`` rather than with the package
installed, so the child is given the package's own location explicitly.
Returns
-------
dict
The current environment with ``PYTHONPATH`` naming this package's parent directory.
"""
existing = os.environ.get("PYTHONPATH", "")
parts = [str(_PACKAGE_ROOT), *([existing] if existing else [])]
return {**os.environ, "PYTHONPATH": os.pathsep.join(parts)}
def _drive(key_name: str, *, variant: str, directory: Path) -> _Drive:
"""Drive one projection in a child process, and read what it wrote.
FVS keeps its state in the process, so a run after another projection produces different numbers
(ADR 0010). The child owns the projection and the working directory; this process shares nothing
with the next run.
Parameters
----------
key_name : str
The keyword file's name, relative to ``directory``.
variant : str
The variant to run, which for this project is ``bc``.
directory : Path
The run's directory, which the child changes into.
Returns
-------
_Drive
The summary the child wrote, and the build's variant list.
Raises
------
ExtensionUnavailableError
If the child reports that PyFVS is not importable, or the variant did not load.
RunError
If the child fails, or does not write what the parent needs.
"""
finished = subprocess.run(
[sys.executable, "-m", _CHILD_MODULE, str(directory), key_name, variant],
capture_output=True,
text=True,
check=False,
env=_child_environment(),
)
if finished.returncode == EXTENSION_UNAVAILABLE_CODE:
raise ExtensionUnavailableError(
finished.stderr.strip() or "the child reported that the extension is unavailable"
)
if finished.returncode != 0:
detail = (finished.stderr or finished.stdout).strip()[-2000:]
raise RunError(f"the child process that drives the projection failed: {detail}")
summary_path = directory / SUMMARY_FILE_NAME
variants_path = directory / VARIANTS_FILE_NAME
if not summary_path.is_file() or not variants_path.is_file():
detail = (finished.stderr or finished.stdout).strip()[-2000:]
raise RunError(
"the child process exited cleanly but wrote no "
f"{SUMMARY_FILE_NAME} or {VARIANTS_FILE_NAME}: {detail}"
)
return _Drive(
summary=pd.read_csv(summary_path),
variants=tuple(json.loads(variants_path.read_text(encoding="utf-8"))),
)
def _first_case(cases: pd.DataFrame) -> dict[str, object]:
"""Return the first ``FVS_Cases`` row as a mapping, refusing an empty table."""
if cases.empty:
raise RunError("the run wrote no FVS_Cases row, so it cannot record its provenance")
return {str(name): value for name, value in cases.iloc[0].items()}
[docs]
@dataclass(frozen=True, slots=True)
class RunResult:
"""What a run returns: its tables, its provenance, and where it left them.
Parameters
----------
provenance : Provenance
The record, as the provenance contract defines it.
cases : pandas.DataFrame
``FVS_Cases``, which identifies the run.
warnings : pandas.DataFrame
The rows of ``FVS_Error`` whose severity is ``WARNING``. A row of severity ``ERROR`` raises
instead, so a result never carries one.
summary : pandas.DataFrame
The summary the run produced in memory, which the output contract names as the route to
that table through this API.
directory : Path
The run's directory, which holds the keyword file, the tree list, the database, the reports,
the provenance record, and the two files the child process wrote: ``summary.csv`` and
``variants.json``.
keyword_file : Path
The keyword file written.
database : Path
The database read.
"""
provenance: Provenance
cases: pd.DataFrame
warnings: pd.DataFrame
summary: pd.DataFrame
directory: Path
keyword_file: Path
database: Path
def _species_groups(records: Sequence[TreeListFields]) -> tuple[str, ...]:
"""Return the calibration groups present in a tree list."""
return tuple(
sorted(
{SPECIES_GROUPS.get(str(record["species"]).upper(), "UNKNOWN") for record in records}
)
)
def _ba(records: Sequence[TreeListFields]) -> float:
"""Return the expansion-weighted basal area of a tree list, in m2/ha."""
return sum(
float(record["tree_count"]) * math.pi / 4.0 * (float(record["diam"]) / 100.0) ** 2
for record in records
)
def _qmd(records: Sequence[TreeListFields]) -> float | None:
"""Return the expansion-weighted QMD of a tree list, or ``None`` when empty.
Parameters
----------
records : Sequence of TreeListFields
The trees handed to FVS.
Returns
-------
float or None
The initial quadratic mean diameter in centimetres, or ``None`` when
the tree list has no expansion weight.
"""
total = sum(float(record["tree_count"]) for record in records)
if total <= 0:
return None
squares = sum(float(record["diam"]) ** 2 * float(record["tree_count"]) for record in records)
return math.sqrt(squares / total)
def _management(keywords: Sequence[Keyword]) -> str:
"""Return the management regime a run declares.
``MANAGED`` is present in the baseline set with a zero first field; that is
an untreated run, not a treatment. A non-zero field is managed.
"""
for keyword in keywords:
if keyword.name != "MANAGED":
continue
value = keyword.parameters[0] if keyword.parameters else "0"
try:
return "untreated" if float(value) == 0.0 else "managed"
except ValueError:
return "managed"
return "untreated"
def _bec_of(keywords: Sequence[Keyword]) -> BecCode:
"""Return the BEC a run carries, or refuse a set that does not name one.
Parameters
----------
keywords : Sequence of Keyword
The caller's set, before the tree handoff is inserted.
Returns
-------
BecCode
The validated BEC from the first two ``STDINFO`` fields.
Raises
------
BecCodeError
If the set carries no ``STDINFO``, or its first two fields do not parse.
A BC run without one leaves FVS on its documented default, ``ICHmw2/01``,
which is not the stand the caller described.
"""
for keyword in keywords:
if keyword.name == "STDINFO":
if len(keyword.parameters) < 2:
raise BecCodeError(
"STDINFO needs a BEC in its first two fields, as in "
"'STDINFO=CAR,IDFdk3/01,...'; without one FVS silently uses ICHmw2/01"
)
return BecCode.from_fields(keyword.parameters[0], keyword.parameters[1])
raise BecCodeError(
"a BC run needs STDINFO with a BEC in its first two fields, as in "
"'STDINFO=CAR,IDFdk3/01,...'; without one FVS silently uses ICHmw2/01"
)
def _echoed_bec(directory: Path) -> str | None:
"""Return the BEC in the run's ``STDINFO`` echo, or ``None`` when there is no line.
Parameters
----------
directory : Path
The run's directory, which holds ``run.out``.
Returns
-------
str or None
The code after ``BEC=`` on the first ``STDINFO`` line, without the trailing
semicolon. ``None`` when the report or the line is absent.
"""
report = directory / "run.out"
if not report.is_file():
return None
for line in report.read_text(encoding="latin-1").splitlines():
if line.lstrip().startswith("STDINFO") and "BEC=" in line:
return line.split("BEC=", 1)[1].split()[0].rstrip(";")
return None
def _refuse_unresolved_bec(directory: Path, bec: BecCode) -> None:
"""Refuse a run whose build resolved a different BEC from the requested one.
The wrapper's token list and the variant's ``HABTYP`` list must agree. If the
installed extension predates a token, FVS silently falls back to
``ICHmw2/01``, so the run's own echo is the oracle that the code was
represented.
Parameters
----------
directory : Path
The run's directory.
bec : BecCode
The BEC the caller requested.
Raises
------
BecCodeError
If the run wrote no ``STDINFO`` echo, or it names a different code.
"""
echoed = _echoed_bec(directory)
if echoed is None:
raise BecCodeError(
f"the run wrote no STDINFO BEC= echo, so {bec.pretty} cannot be confirmed; "
"the installed BC build may not support it"
)
if echoed.upper() != bec.pretty.upper():
raise BecCodeError(
f"FVS resolved the requested BEC {bec.pretty} as {echoed}; the installed BC "
"build does not carry this code. Build the variant with the W2.3 patch or use "
"a build that resolves it"
)
[docs]
def run_case(
records: Sequence[TreeListFields],
keywords: Sequence[Keyword],
*,
directory: str | Path,
variant: str = "bc",
key_name: str = KEYWORD_FILE_NAME,
calibration: CalibrationBundle | None = None,
tsa: str | None = None,
) -> RunResult:
"""Run one case: write the inputs, drive BC, read the outputs, and record the provenance.
The caller describes the stand and the run in ``keywords`` and supplies the trees in
``records``. The orchestration writes the tree handoff, the keyword file, and the tree list,
drives one projection in ``directory``, and reads the database back. A run that reports an error
row raises rather than returning a result that holds one.
Parameters
----------
records : Sequence of TreeListFields
The trees, in the order to write them. :func:`fresh_fvsbc.nemora.to_tree_list` builds these
from a synthetic population.
keywords : Sequence of Keyword
The keyword set, in the order FVS will read it. It must not carry ``TREEFMT`` or
``TREEDATA``.
directory : str or Path
Where the run writes. It is created if it does not exist, and the run owns it: the
database's name is fixed, so two runs in one directory would overwrite each other.
variant : str, optional
The variant to run. Defaults to ``bc``, which is the only variant this project wraps.
key_name : str, optional
The keyword file's name inside ``directory``. The tree file takes the same base name.
calibration : CalibrationBundle, optional
A bundle the run uses. The caller renders the bundle into ``keywords``;
the run checks the domain and refuses a set that does not carry the
bundle's keywords. The bundle id is recorded in provenance.
tsa : str, optional
The TSA name, when a bundle domain declares one. It is recorded in
provenance and checked against the bundle.
Returns
-------
RunResult
The tables, the provenance, and the paths.
Raises
------
BecCodeError
If the set carries no ``STDINFO``, its first two fields are not a BEC the
variant resolves, or the run echoes a different BEC because the installed
build does not carry the requested code.
BundleError
If the calibration bundle's domain does not cover the run, or the set does
not carry the keywords the bundle renders.
KeywordConflictError
If the keyword set carries a keyword the orchestration writes.
UnknownKeywordError
If the set carries a keyword FVS's table does not hold.
ExtensionUnavailableError
If PyFVS is not importable, or the variant did not load.
RunFailedError
If the run holds a row of severity ``ERROR``.
MissingOutputError
If the run wrote no database.
RunError
If the projection ends badly, or the run recorded no case.
"""
bec = _bec_of(keywords)
qmd = _qmd(records)
ba = _ba(records)
if calibration is not None:
calibration.check_domain(
bec,
_species_groups(records),
_management(keywords),
dbh_values=tuple(float(record["diam"]) for record in records),
qmd=qmd,
ba=ba,
tsa=tsa,
)
missing = [keyword for keyword in calibration.render() if keyword not in keywords]
if missing:
names = ", ".join(keyword.name for keyword in missing)
raise BundleError(
f"{calibration.bundle_id} renders {names}, and the keyword set does not "
"carry them; render the bundle into the set before the run"
)
handoff = with_tree_handoff(keywords)
keyword_set = KeywordSet(handoff)
target = Path(directory).expanduser().resolve()
target.mkdir(parents=True, exist_ok=True)
keyword_file = target / key_name
keyword_file.write_text(keyword_set.render())
tree_file_name(keyword_file).write_text(write_tree_file(records))
driven = _drive(keyword_file.name, variant=variant, directory=target)
_refuse_unresolved_bec(target, bec)
database = database_path(target)
cases = read_cases(database)
warnings = check_run(database)
case_id, run_datetime = from_cases(_first_case(cases))
provenance = Provenance(
variant=variant,
keyword_set_hash=keyword_set.digest,
treelist_hash=treelist_hash(records),
case_id=case_id,
run_datetime=run_datetime,
bec=str(bec),
qmd=qmd,
qmd_min=calibration.domain.qmd_min if calibration is not None else None,
ba=ba,
ba_min=calibration.domain.ba_min if calibration is not None else None,
tsa=tsa,
calibration_id=calibration.bundle_id if calibration is not None else "",
variant_list=driven.variants,
random_seed=_seed_of(handoff),
)
provenance.write(target / PROVENANCE_FILE_NAME)
return RunResult(
provenance=provenance,
cases=cases,
warnings=warnings,
summary=driven.summary,
directory=target,
keyword_file=keyword_file,
database=database,
)
[docs]
def run_population(
population: Sequence[Mapping[str, object]],
*,
directory: str | Path,
species: str | None = None,
tree_age: float | None = None,
keywords: Sequence[Keyword] = (),
variant: str = "bc",
plot_fields: PlotFields | None = None,
key_name: str = KEYWORD_FILE_NAME,
) -> RunResult:
"""Run one case built from a synthetic stem population.
This is :func:`fresh_fvsbc.nemora.to_tree_list` followed by :func:`run_case`, and it exists
because that pair is the project's own path from a population to a run. It adds no behaviour of
its own, and the adapter's refusals are its refusals.
Parameters
----------
population : Sequence of Mapping
The stems, shaped like `nemora`'s output.
directory : str or Path
Where the run writes.
species : str, optional
The species code. Required, because `nemora` emits none.
tree_age : float, optional
The age at breast height, in years. Required, for the same reason.
keywords : Sequence of Keyword, optional
The keyword set. The orchestration writes the tree handoff.
variant : str, optional
The variant to run. Defaults to ``bc``.
plot_fields : PlotFields, optional
The plot-level fields, which belong to a run.
key_name : str, optional
The keyword file's name inside ``directory``.
Returns
-------
RunResult
The tables, the provenance, and the paths.
Raises
------
MissingAssignmentError
If ``species`` or ``tree_age`` is not supplied.
PopulationError
If a record is missing a field the adapter reads, or carries a non-numeric one.
CrownRatioError
If a crown ratio cannot be represented by the tree-list field.
SpeciesError
If the species code resolves to no BC species, or to more than one.
RunError
If the run cannot be set up, driven, or read.
"""
records = to_tree_list(population, species=species, tree_age=tree_age, plot_fields=plot_fields)
return run_case(records, keywords, directory=directory, variant=variant, key_name=key_name)