"""HYSPLIT as a transport model."""
from __future__ import annotations
import datetime as dt
import hashlib
import tempfile
from collections.abc import Iterable
from importlib.resources import files as pkg_files
from pathlib import Path
from typing import TYPE_CHECKING
from stilt.exceptions import SimulationError
from stilt.transport import ModelRun
from .config import HysplitConfig
from .driver import (
LOG_FILE,
PARTICLE_STILT_FILE,
_bundled_data_dir,
check_met_stop,
check_release,
read_particle_dat,
run_hycs_std,
write_inputs,
)
from .failures import FailureReason
from .met import Met, MetConfig
if TYPE_CHECKING:
import pandas as pd
from arlmet import Grid
from stilt.receptors import Receptor
def hysplit_version(exe_dir: str | Path | None = None) -> str:
"""
Return the version of the HYSPLIT build in *exe_dir*, or of the bundled build.
A build directory names its version in a ``version`` file beside
``hycs_std``, as the bundled one does. Two builds with the same settings
can give different particles, so the version is part of a run's
identity.
Raises
------
FileNotFoundError
If *exe_dir* has no ``version`` file.
"""
if exe_dir is None:
path = Path(str(pkg_files("stilt.transport.hysplit") / "bin" / "version"))
else:
path = Path(exe_dir) / "version"
if not path.exists():
raise FileNotFoundError(
f"{exe_dir} has no 'version' file. A custom hycs_std build needs "
"one beside the binary, holding its version string (such as "
"v5.3.2+t0-rows), so its runs are told apart from other builds'."
)
return path.read_text().strip()
def _sha256(path: Path) -> str:
"""Return the SHA-256 hex digest of a file."""
return hashlib.sha256(path.read_bytes()).hexdigest()
[docs]
class HysplitModel:
"""
HYSPLIT as a transport model, the object ``get_model("hysplit")`` returns.
:meth:`run` picks the met files a receptor needs, writes HYSPLIT's input
files (:func:`~stilt.transport.hysplit.write_inputs`), runs ``hycs_std``,
and reads the particles it writes
(:func:`~stilt.transport.hysplit.read_particle_dat`). HYSPLIT reads the
met where it is (the cropped copies when the met is cropped), so nothing
is staged per run.
"""
name = "hysplit"
config_class = HysplitConfig
#: HYSPLIT reads ARL files, found, downloaded, or cropped as a met says.
met_config_class = MetConfig
#: HYSPLIT writes its inputs and outputs as files in a directory.
needs_workdir = True
#: One hycs_std process per receptor.
batched = False
[docs]
def version(self, config: HysplitConfig) -> str:
"""Return the version of the bundled build, or of the one in ``config.exe_dir``."""
return hysplit_version(config.exe_dir)
[docs]
def data_files(self, config: HysplitConfig) -> dict[str, str] | None:
"""
Return the SHA-256 of each table in ``config.data_dir`` that differs from the bundled one.
``None`` when there is no ``data_dir`` or every table in it matches
the bundled table of the same name.
"""
if config.data_dir is None:
return None
bundled = _bundled_data_dir()
changed = {}
for path in sorted(Path(config.data_dir).iterdir()):
if not path.is_file():
continue
digest = _sha256(path)
own = bundled / path.name
if not own.is_file() or _sha256(own) != digest:
changed[path.name] = digest
return changed or None
[docs]
def check_met(
self, met: MetConfig, receptors: pd.DataFrame, n_hours: int
) -> pd.DataFrame:
"""
Return what would stop HYSPLIT from running *receptors* with *met*, found before any run.
The checks of :meth:`Met.check_receptors`: the met's directory
holds ARL files, every file a run needs is there, and the receptors
are inside the crop. Nothing is run, opened past the first file's
header, or downloaded.
Parameters
----------
met : MetConfig
The met, with an absolute directory in a project.
receptors : pandas.DataFrame
One row per release point, with ``receptor``, ``time``,
``longitude``, and ``latitude`` columns.
n_hours : int
The length of the runs, which sets the files each needs.
Returns
-------
pandas.DataFrame
``check``, ``receptor`` (``NA`` for a problem of the met
itself), and ``message`` columns, one row for each problem.
"""
return Met("met", met).check_receptors(receptors, n_hours)
[docs]
def fetch_met(
self, met: MetConfig, windows: Iterable[tuple[dt.datetime, dt.datetime]]
) -> pd.DataFrame:
"""
Get the met files that cover *windows*, downloading the ones a download lacks.
The files :meth:`run` would ask for, one request to the archive for
windows that overlap (:meth:`Met.fetch`). A met without ``download``
is only checked for missing files.
Parameters
----------
met : MetConfig
The met, with an absolute directory in a project.
windows : iterable of tuple of datetime
The time each run covers (:func:`stilt.meteorology.run_window`).
Returns
-------
pandas.DataFrame
``file`` and ``message`` columns: a row for each file, then a
row for each window whose files could not be had.
"""
return Met("met", met).fetch(windows)
[docs]
def run(
self,
receptor: Receptor,
config: HysplitConfig,
met: MetConfig,
window: tuple[dt.datetime, dt.datetime],
workdir: Path | None = None,
timeout: int | None = None,
) -> ModelRun:
"""
Run HYSPLIT for one receptor in *workdir* and return its particles.
It finds the met files that cover *window*, up to the met step at
or after its end, which HYSPLIT interpolates, cropped when the met
is. Its output is ``stilt.log`` in *workdir*, returned as the run's
log. Without *workdir* it runs in a temporary directory. The release
heights and the near-field correction are added by the caller, as
for any model (:func:`stilt.transport.run_model`).
Raises
------
SimulationError
With HYSPLIT's log in ``log``, and ``reason`` ``TIMEOUT`` when
the run exceeded *timeout* seconds; the :class:`FailureReason`
of a known failure message in the log or ``WARNING`` file, such
as the met running out (``MET_COVERAGE``); ``EXIT_STATUS`` when
``hycs_std`` was killed by a signal or exited with a non-zero
status, whatever it had written; ``PARTICLE_LIMIT``
when ``maxpar`` stopped a release from several locations before
the last ones got their particles; or ``NO_PARTICLE_DATA`` when
HYSPLIT wrote no particle file. A run whose particles stopped
before ``n_hours`` well inside the met's domain fails as
``MET_COVERAGE`` too, when ``outdt`` is 0: more than 5 grid
cells from every side of the met grid.
Particles that all left the met's domain before the end of the
run are kept.
MeteorologyError
Before HYSPLIT runs, when a met file the run needs is missing or
damaged, or the files do not hold every time the run covers
(:meth:`Met.check`).
"""
if workdir is None:
with tempfile.TemporaryDirectory(prefix="stilt-") as tmp:
return self.run(receptor, config, met, window, Path(tmp), timeout)
files = Met("met", met)
source = files.files_for(window)
readable = files.readable(source)
files.check(readable, window)
write_inputs(workdir, receptor, config, readable)
try:
run_hycs_std(workdir, timeout)
check_release(workdir, receptor, config)
path = workdir / PARTICLE_STILT_FILE
if not path.exists():
raise SimulationError(
f"HYSPLIT wrote no {PARTICLE_STILT_FILE}.",
reason=FailureReason.NO_PARTICLE_DATA,
)
particles = read_particle_dat(path, config.varsiwant)
if config.outdt == 0: # the check needs a row every time step
check_met_stop(particles, config.n_hours, _met_grid(readable[0]))
except SimulationError as error:
error.log = _read_log(workdir)
raise
# The record names the source files; the crop settings are in the
# run's settings.
return ModelRun(particles=particles, log=_read_log(workdir), met_files=source)
def _met_grid(path: Path) -> Grid:
"""Return the horizontal grid of the ARL file at *path*."""
from arlmet import File
with File(path) as met:
return met.grid
def _read_log(workdir: Path) -> str:
"""Return HYSPLIT's log in *workdir*, or an empty string when it wrote none."""
path = workdir / LOG_FILE
return path.read_text(encoding="utf-8", errors="replace") if path.exists() else ""
__all__ = ["HysplitModel", "hysplit_version"]