import os
import pathlib
import numpy as np
import pandas as pd
import xarray as xr
from ......utils import path
from ......utils.datastores.dump import dump_datastore
from ......utils.check.errclass import CifTypeError
def _dataset_meta(ds):
"""Return a human-readable metadata string for an :class:`xr.Dataset`.
Reports, for the ``spec`` variable and — when present — the ``incr``
variable:
* dimension names and sizes,
* min and max values (NaN-safe),
* whether any NaN values are present.
Args:
ds (xr.Dataset): dataset to summarise.
Returns:
str: multi-line metadata string, one block per variable.
"""
lines = []
for varname in ("spec", "incr"):
if varname not in ds:
continue
da = ds[varname]
lines.append(f"[{varname}]")
dims = dict(zip(da.dims, da.shape))
lines.append(f" dims : {dims}")
try:
vals = da.values.astype(float)
lines.append(f" min : {float(np.nanmin(vals)):.6g}")
lines.append(f" max : {float(np.nanmax(vals)):.6g}")
lines.append(f" has_nan: {bool(np.isnan(vals).any())}")
except Exception as exc:
lines.append(f" stats : (could not compute: {exc})")
return "\n".join(lines)
def _dataframe_meta(df):
"""Return a human-readable metadata string for a :class:`pd.DataFrame`.
Reports the number of rows and, for each of the ``maindata``, ``spec``,
and ``incr`` columns that exist in the dataframe:
* min and max values,
* whether any NaN values are present.
Args:
df (pd.DataFrame): dataframe to summarise.
Returns:
str: multi-line metadata string, one block per column.
"""
lines = [f"length : {len(df)}"]
for colname in ("spec", "incr"):
if colname not in df.columns:
continue
col = df[("maindata", colname)]
lines.append(f"[{colname}]")
try:
lines.append(f" min : {float(col.min()):.6g}")
lines.append(f" max : {float(col.max()):.6g}")
except (TypeError, ValueError):
lines.append(f" min : {col.min()}")
lines.append(f" max : {col.max()}")
lines.append(f" has_nan: {bool(col.isna().any())}")
return "\n".join(lines)
[docs]
def dump_debug(
transform, transf_mapper, tmp_datastore, runsubdir, ddi,
entry="outputs", transform_onlyinit=False,
dump_metadata_only=False):
"""Dump inputs or outputs of a transform step for post-run inspection.
For each tracer ID (``trid``) in ``tmp_datastore[entry]``, writes debug
files under::
<runsubdir>/../transform_debug/<transform>/<ddi>/<component>/<parameter>/
Two modes are supported:
**Full dump** (``dump_metadata_only=False``)
Writes the complete datastore to disk as NetCDF files
(:class:`xr.Dataset`) or pyCIF datastore files (:class:`pd.DataFrame`).
One file is produced per tracer ID and date. This mode is accurate
but slow and disk-intensive.
**Metadata-only dump** (``dump_metadata_only=True``)
Writes lightweight plain-text files instead of full data files.
Dramatically reduces wall-time overhead and disk usage while still
capturing enough information to trace data flow and detect anomalies.
* For :class:`xr.Dataset` values — records the dimension names and
sizes, min/max values, and NaN presence for the ``spec`` variable
(and ``incr`` when it exists in the dataset).
* For :class:`pd.DataFrame` values — records the row count and, for
each of the ``maindata``, ``spec``, and ``incr`` columns that are
present, the min/max values and NaN presence.
Args:
transform (str): name of the transform being debugged.
transf_mapper (dict): mapper entry for the transform (not directly
used here; kept for API consistency).
tmp_datastore (dict): the datastore for the current transform step,
keyed by ``"inputs"`` and ``"outputs"``.
runsubdir (str): the per-period run sub-directory. Debug files are
written to ``<runsubdir>/../transform_debug/…``.
ddi (datetime.datetime): the current simulation date (used both for
directory naming and for per-date filename formatting).
entry (str, optional): which side of the datastore to dump —
``"inputs"`` or ``"outputs"``. Defaults to ``"outputs"``.
transform_onlyinit (bool, optional): when ``True``, the transform ran
in dry-run / init-only mode. Currently unused inside this
function but kept for API consistency with the call sites in
:func:`~.do_transforms.do_transforms`. Defaults to ``False``.
dump_metadata_only (bool, optional): when ``True``, write lightweight
metadata text files instead of full NetCDF/datastore files.
Defaults to ``False``.
Raises:
TypeError: if a datastore entry is neither an :class:`xr.Dataset`
nor a :class:`dict` of :class:`xr.Dataset` / :class:`pd.DataFrame`.
"""
for trid in tmp_datastore[entry]:
debug_dir = (
f"{runsubdir}/../transform_debug/{transform}"
f"/{ddi.strftime('%Y-%m-%d_%H:%M')}/{trid[0]}/{trid[1]}"
)
path.init_dir(debug_dir)
# ------------------------------------------------------------------ #
# Branch 1: the datastore entry is itself an xr.Dataset
# ------------------------------------------------------------------ #
if isinstance(tmp_datastore[entry][trid], xr.Dataset):
if dump_metadata_only:
meta_file = f"{debug_dir}/dataarray_meta_{entry}.txt"
with open(meta_file, "w") as fh:
fh.write(_dataset_meta(tmp_datastore[entry][trid]))
else:
debug_file = f"{debug_dir}/dataarray_debug_{entry}.nc"
if os.path.isfile(debug_file):
pathlib.Path(debug_file).unlink()
xr.Dataset(tmp_datastore[entry][trid]).to_netcdf(
debug_file, mode="w")
continue
# ------------------------------------------------------------------ #
# Branch 2: the datastore entry is a dict of per-date values
# ------------------------------------------------------------------ #
if not isinstance(tmp_datastore[entry][trid], dict):
raise CifTypeError(
f"dump_debug got an unexpected type for tmp_datastore: "
f"{type(tmp_datastore[entry][trid])}"
)
for d in tmp_datastore[entry][trid]:
value = tmp_datastore[entry][trid][d]
if isinstance(value, pd.DataFrame):
if dump_metadata_only:
meta_file = d.strftime(
f"{debug_dir}/monitor_meta_%Y%m%d%H%M_{entry}.txt")
with open(meta_file, "w") as fh:
fh.write(_dataframe_meta(value))
else:
debug_file = d.strftime(
f"{debug_dir}/monitor_debug_%Y%m%d%H%M_{entry}.nc")
if os.path.isfile(debug_file):
pathlib.Path(debug_file).unlink()
dump_datastore(
value, debug_file,
col2dump=value.columns,
dump_default=False,
mode='w'
)
elif isinstance(value, xr.Dataset):
if dump_metadata_only:
meta_file = d.strftime(
f"{debug_dir}/dataarray_meta_%Y%m%d%H%M_{entry}.txt")
with open(meta_file, "w") as fh:
fh.write(_dataset_meta(value))
else:
debug_file = d.strftime(
f"{debug_dir}/dataarray_debug_%Y%m%d%H%M_{entry}.nc")
if os.path.isfile(debug_file):
pathlib.Path(debug_file).unlink()
xr.Dataset(value).to_netcdf(debug_file, mode="w")
