Source code for pycif.plugins.models.chimere.io.inputs.utils
import numpy as np
import xarray as xr
import datetime
from netCDF4 import Dataset
import pandas as pd
from logging import warning
from ......utils import path
from ......utils.hdf5 import _hdf5_lock
from ......utils.check.errclass import CifError
[docs]
def replace_dates(file_emisout, list_dates, ignore_dates):
"""Check (or ignore) that a CHIMERE NetCDF file's ``Times`` axis matches expected dates.
CHIMERE expects every input NetCDF to carry a ``Times`` variable encoded
as ``%Y-%m-%d_%H:00:00`` strings. If the dates in the file differ from
*list_dates*:
* ``ignore_dates=True``: emits a warning and continues (useful when
reusing files from a different time window).
* ``ignore_dates=False``: raises an ``Exception`` describing the
mismatch and advising the user to set ``ignore_input_dates: true``
in the YAML.
Args:
file_emisout (str): path to the NetCDF file to check.
list_dates (list[datetime]): expected sequence of dates.
ignore_dates (bool): if ``True``, suppress the mismatch error.
Raises:
Exception: if dates mismatch and ``ignore_dates`` is ``False``.
"""
with _hdf5_lock:
with xr.open_dataset(file_emisout) as ds:
times = np.array([
datetime.datetime.strptime(d.decode(), "%Y-%m-%d_%H:00:00")
for k, d in enumerate(ds.Times.values)
])
if not np.all(times == list_dates):
if ignore_dates:
warning(f"CHIMERE will not check the dates provided in {file_emisout}; it will be used as such")
# path.copyfromlink(file_emisout)
# with Dataset(file_emisout, "a") as ds:
# ds.variables["Times"][:] = [
# list(pd.to_datetime(d).strftime("%Y-%m-%d_%H:00:00"))
# for d in list_dates]
else:
raise CifError(f"Dates in {file_emisout} are not what CHIMERE expects: \nExpected dates: {list_dates}\nActual dates: {times}\nYou can set 'ignore_input_dates' to True in the yml paragraph for CHIMERE to ignore this message and replace the dates from the original file")