from logging import debug
import pandas as pd
import xarray as xr
import numpy as np
from .time_coord import convert_calendar, decode_datetimes
from .....utils.check.errclass import CifError, CifValueError
OFFSET = pd.offsets.Nano(1) # One nanosecond time offest
[docs]
def get_time_intervals(
ds, tracer, time_varname, date_i, date_f,
calendar,
time_midpoint=False,
time_endpoint=False
):
"""Compute the list of validity time intervals covered by a file.
If ``tracer.period_varname`` is set, the intervals are read directly
from that variable. Otherwise, they are derived from the time
coordinate by converting it to Pandas periods (using ``tracer.var_freq``,
or falling back to ``tracer.file_freq`` for single-timestep files), and
filtered down to those overlapping ``[date_i, date_f]``.
Args:
ds (xr.Dataset): Dataset holding the time coordinate (and, if
relevant, the period variable).
tracer: The flux tracer plugin, providing ``var_freq``, ``file_freq``
and, optionally, ``period_varname``.
time_varname (str): Name of the time coordinate.
date_i (datetime.datetime): Start of the simulation window.
date_f (datetime.datetime): End of the simulation window.
calendar (str): Calendar of the time coordinate (unused directly,
kept for interface consistency).
time_midpoint (bool): Unused directly, kept for interface
consistency.
time_endpoint (bool): Unused directly, kept for interface
consistency.
Returns:
list: A list of ``[start, end]`` `datetime.datetime` pairs
overlapping the simulation window.
Raises:
CifError: If ``var_freq`` ends with the ``'S'`` (start) anchor,
which `to_period` does not accept.
CifValueError: If the time coordinate cannot be parsed/converted to
periods, if a single-period file does not overlap the
simulation window, or if no period overlaps the simulation
window.
"""
var_freq = getattr(tracer, 'var_freq', None)
if hasattr(tracer, 'period_varname'):
# Read periods form the NetCDF file
periods = ds[tracer.period_varname].values
time_intervals = np.concatenate([
pd.DatetimeIndex(periods[:, [0]]).to_pydatetime(),
pd.DatetimeIndex(periods[:, [1]]).to_pydatetime()],
axis=1
)
else:
# Compute periods with the NetCDF file time coordinates and the
# optional 'var_freq' argument
if ds[time_varname].size == 1 and var_freq is None:
# Cannot infer period with only one time value, using 'file_freq'
# argument as 'var_freq'
var_freq = tracer.file_freq
# Converting 'Start' offset (not useable as period dtype) to 'End'
# offset (useable as period dtype)
if var_freq[-1] == 'S':
var_freq = var_freq[:-1]
debug(f"single time value in file '{ds.encoding['source']}'\n"
f"using argument 'file_freq' as 'var_freq' ('{var_freq}')")
if var_freq is not None:
if var_freq[-1] == 'S':
raise CifError(
f"The frequency {var_freq} is not accepted by 'to_period'. "
f"The 'S' anchor at the end is not accepted for periods. "
f"Please remove it."
)
try:
dates = pd.to_datetime(ds[time_varname]).values
# Infer periods if argument 'var_freq' is not provided
periods = pd.to_datetime(dates).to_period(freq=var_freq)
if var_freq is not None:
debug(f"Inferred frequency for file '{ds.encoding['source']}': "
f"'{periods.freqstr}'")
except ValueError as e:
if var_freq is None:
msg = ("could not infer frequency from 'time' coordinate "
f"values in file '{ds.encoding['source']}'. "
f"Please use the 'var_freq' argument.")
else:
msg = ("could not parse 'time' coordinate values in file "
f"'{ds.encoding['source']}' with var_freq={var_freq}.")
raise CifValueError(msg) from e
time_intervals = np.concatenate([
periods.start_time.to_pydatetime()[:, np.newaxis],
(periods.end_time + OFFSET).to_pydatetime()[:, np.newaxis]
], axis=1)
if len(time_intervals) == 1:
di, df = time_intervals[0]
# Checking if time interval and simulation window are overlaping
if df <= date_i or date_f <= di:
raise CifValueError(
f"Simulation window [{date_i.isoformat(), date_f.isoformat()}] "
f"and the single period [{di.isoformat(), df.isoformat()}] "
f"in file '{ds.encoding['source']}' are not overlaping"
)
else:
mask = ((time_intervals[:, 0] <= date_f)
& (time_intervals[:, 1] >= date_i))
time_intervals = time_intervals[mask]
if time_intervals.size == 0:
raise CifValueError(
"No period found whithin the simulation window "
f"[{date_i.isoformat(), date_f.isoformat()}] in file "
f"'{ds.encoding['source']}'"
)
return time_intervals.tolist()
[docs]
def get_year_offset(
ds: xr.Dataset,
time_varname: str,
ref_year: int,
is_climatology: bool = False
) -> int:
"""Compute the year shift needed to align a file's time coordinate with a target year.
For climatology data, the offset simply maps the file's (single) year
onto `ref_year`. Otherwise, the file is only shifted if `ref_year` falls
entirely outside the range of years present in the file (shifting to the
nearest edge); no shift is applied if `ref_year` is already covered.
Args:
ds (xr.Dataset): Dataset holding the time coordinate.
time_varname (str): Name of the time coordinate.
ref_year (int): Target year to align the file with.
is_climatology (bool): Whether the data is climatological.
Returns:
int: The number of years to add to the file's time coordinate.
"""
if is_climatology:
years = ref_year - ds[time_varname].dt.year.values[0]
else:
year_file = ds[time_varname].dt.year.values
if ref_year > year_file.max():
years = ref_year - year_file.max()
elif ref_year < year_file.min():
years = ref_year - year_file.min()
else:
years = 0
return years
