import os
from logging import info
import pandas as pd
import xarray as xr
from ......utils import path
from ......utils.hdf5 import _hdf5_lock
from ...ini_mapper import meteo_components, static_components
[docs]
def make_meteo(self, datastore, runsubdir, mode, datei, datef):
"""Build D&B's ``input/dynforcing.nc`` from the ``meteo`` data-store.
See :func:`_make_forcing` for the shared logic (also used by
:func:`make_static`): for each of D&B's mandatory hourly meteo
components (see
:data:`pycif.plugins.models.dalecbethy.ini_mapper.meteo_components`),
either the raw forcing file is linked as-is, or the component is
(re)written via ``self.meteo.write``. The mandatory ``yyyymmddhh``
calendar variable -- not itself a per-component mapper input -- is then
copied as-is from the original raw file, for the hours covered by
``[datei, datef]``.
Args:
self: the dalecbethy model Plugin.
datastore (dict): the pyCIF data-store, keyed by
``("meteo", <component>)``.
runsubdir (str): sub-directory for the current simulation.
mode (str): running mode ('fwd', 'tl', 'adj'); only 'fwd' is
supported so far (D&B has no adjoint/tangent-linear wiring for
meteo yet).
datei, datef: date interval of the sub-simulation.
Raises:
NotImplementedError: if ``mode != "fwd"``.
"""
_make_forcing(
self, datastore, runsubdir, mode, datei, datef,
components=meteo_components,
trid_type="meteo",
subplug=self.meteo,
target_name="dynforcing.nc",
needs_calendar=True,
)
[docs]
def make_static(self, datastore, runsubdir, mode, datei, datef):
"""Build D&B's ``input/staticforcing.nc`` from the ``dalecbethy_static``
data-store.
See :func:`_make_forcing` for the shared logic (also used by
:func:`make_meteo`): for each of D&B's time-invariant land-surface/PFT
components (see
:data:`pycif.plugins.models.dalecbethy.ini_mapper.static_components`),
either the raw forcing file is linked as-is, or the component is
(re)written via ``self.static.write``. Unlike ``dynforcing.nc``, there
is no ``yyyymmddhh`` calendar variable to copy.
Args:
self: the dalecbethy model Plugin.
datastore (dict): the pyCIF data-store, keyed by
``("dalecbethy_static", <component>)``.
runsubdir (str): sub-directory for the current simulation.
mode (str): running mode ('fwd', 'tl', 'adj'); only 'fwd' is
supported so far (D&B has no adjoint/tangent-linear wiring for
static forcing yet).
datei, datef: date interval of the sub-simulation.
Raises:
NotImplementedError: if ``mode != "fwd"``.
"""
_make_forcing(
self, datastore, runsubdir, mode, datei, datef,
components=static_components,
trid_type="dalecbethy_static",
subplug=self.static,
target_name="staticforcing.nc",
needs_calendar=False,
)
def _make_forcing(self, datastore, runsubdir, mode, datei, datef, *,
components, trid_type, subplug, target_name,
needs_calendar):
"""Shared logic behind :func:`make_meteo` and :func:`make_static`.
Mirrors ``pycif.plugins.models.chimere.io.inputs.make_meteo.make_meteo``:
for each of ``components``, if it was not materialized in memory (no
``"spec"`` key in the data-store, i.e. ``force_loadin`` was not needed
-- see :func:`pycif.plugins.models.dalecbethy.ini_mapper.ini_mapper`),
the *whole* original raw forcing file is simply linked to
``runsubdir/input/<target_name>``. Unlike CHIMERE (one file per
species), a single D&B raw file already carries every component of a
given kind (dynamic or static) natively, so one link covers every
component -- no need to read/rewrite anything. Otherwise (a transform
perturbed/materialized the data), it is written via ``subplug.write``
(see :func:`pycif.plugins.datastreams.fluxes.dalecbethy.write.write`).
Args:
self: the dalecbethy model Plugin.
datastore (dict): the pyCIF data-store.
runsubdir (str): sub-directory for the current simulation.
mode (str): running mode ('fwd', 'tl', 'adj'); only 'fwd' is
supported so far.
datei, datef: date interval of the sub-simulation.
components (list[str]): component names to look for in the
data-store, keyed as ``(trid_type, component)``.
trid_type (str): first element of the data-store key, e.g.
``"meteo"`` or ``"dalecbethy_static"``.
subplug: the shared fluxes sub-plugin instance to write components
with (``self.meteo`` or ``self.static``).
target_name (str): name of the native file to build, within
``runsubdir/input`` (``"dynforcing.nc"`` or
``"staticforcing.nc"``).
needs_calendar (bool): whether the mandatory ``yyyymmddhh`` calendar
variable must be copied in, once any component was written
(dynamic forcing only).
Raises:
NotImplementedError: if ``mode != "fwd"``.
"""
if mode != "fwd":
raise NotImplementedError(
f"dalecbethy make_forcing ({trid_type}): mode={mode!r} not "
f"implemented yet"
)
ddi = min(datei, datef)
ddf = max(datei, datef)
input_dir = os.path.join(runsubdir, "input")
os.makedirs(input_dir, exist_ok=True)
target_file = os.path.join(input_dir, target_name)
wrote_any = False
linked = False
ref_file = None
for comp in components:
trid = (trid_type, comp)
if trid not in datastore:
continue
tracer = datastore[trid]
tracer_data = tracer["data"][ddi]
if ref_file is None:
# The original (un-materialized) raw file, kept around so that
# _write_yyyymmddhh can copy the calendar variable from it even
# when every component ends up (re)written below.
ref_file = ddi.strftime(
os.path.join(tracer["dirorig"], tracer["fileorig"]))
if "spec" not in tracer_data:
# Not perturbed/materialized: the raw file already carries
# every component natively, so a single link suffices.
if not os.path.isfile(target_file):
path.link(ref_file, target_file)
linked = True
else:
# Replace an existing link by a real copy before appending to
# it, so we never write through the link into the raw file.
if linked:
path.copyfromlink(target_file)
linked = False
xr_mode = "a" if os.path.isfile(target_file) else "w"
subplug.write(
comp, target_file, tracer_data["spec"], mode=xr_mode)
wrote_any = True
if wrote_any and needs_calendar:
_write_yyyymmddhh(ref_file, target_file, ddi, ddf)
if wrote_any or linked:
info(f"D&B forcing written to '{target_file}'")
def _write_yyyymmddhh(ref_file, dynforcing_file, ddi, ddf):
"""Copy the ``yyyymmddhh`` calendar variable from the raw forcing file.
D&B's own ``time`` records mark the *end* of each hourly period (see
``pycif.plugins.datastreams.fluxes.dalecbethy.fetch.fetch``), hence the
``(times > ddi) & (times <= ddf)`` mask below, matching that same
convention.
Args:
ref_file (str): the original, un-materialized dynamic forcing
file to copy the calendar variable from (same file any
unperturbed component in this sub-simulation would have been
linked from, see :func:`_make_forcing`).
dynforcing_file (str): the ``dynforcing.nc`` file to append to.
ddi, ddf: date interval of the sub-simulation.
"""
with _hdf5_lock:
with xr.open_dataset(ref_file) as ds:
times = pd.DatetimeIndex(ds["time"].values)
mask = (times > ddi) & (times <= ddf)
yyyymmddhh = ds["yyyymmddhh"].isel(time=mask).values
with _hdf5_lock:
xr.Dataset(
{"yyyymmddhh": (("ntc", "time"), yyyymmddhh)}
).to_netcdf(dynforcing_file, mode="a")
