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")