from __future__ import annotations
from datetime import datetime
from logging import debug
from pathlib import Path
from typing import Any
import pandas as pd
import xarray as xr
from xarray import Dataset
from .....utils import path
OFFSET = pd.offsets.Nano(1)
[docs]
def get_period_times(ds: Dataset, var_freq: str | None = None):
"""Convert a dataset's ``time`` coordinate to validity period start/end times.
Converts the ``time`` coordinate values to Pandas periods (inferring the
frequency automatically, or using `var_freq` if given), then returns the
start and end timestamp of each period.
Args:
ds (Dataset): Dataset holding a ``time`` coordinate.
var_freq (str, optional): Time frequency of the data (a Pandas
offset alias, without a ``'S'``/start anchor), passed to
`pandas.DatetimeIndex.to_period`. Inferred automatically if not
given.
Returns:
tuple[numpy.ndarray, numpy.ndarray]: ``(period_start, period_end)``,
arrays of `datetime.datetime` marking the start and end of each
period.
Raises:
ValueError: If `var_freq` ends with the ``'S'`` anchor, or if the
time coordinate cannot be parsed/converted to periods.
"""
if var_freq is not None and var_freq[-1] == "S":
raise ValueError(
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:
# Infer periods if argument 'var_freq' is not provided
periods = pd.to_datetime(ds.time).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 ValueError(msg) from e
period_start = periods.start_time.to_pydatetime()
period_end = (periods.end_time + OFFSET).to_pydatetime()
return period_start, period_end
[docs]
def fetch(
ref_dir: str,
ref_file: str,
input_interval: tuple[datetime, datetime],
target_dir: str,
tracer: object | None = None,
**kwargs: Any,
) -> tuple[
dict[datetime, list[str]],
dict[datetime, list[tuple[datetime, datetime]]],
]:
"""Fetch LMDZ DYNAMICO NetCDF flux files and derive the validity time intervals they cover.
Builds the list of candidate file dates from `tracer.file_freq`, links
each matching file into `target_dir`, and computes the period
start/end times of every time record in the file via `get_period_times`
(falling back to `file_freq` as the period frequency for files with a
single timestamp).
Args:
ref_dir (str): Directory containing the reference input files.
ref_file (str): Filename pattern of the input files (a ``strftime``
format string).
input_interval (tuple[datetime, datetime]): ``(date_i, date_f)``
simulation sub-period to cover.
target_dir (str): Directory where the resolved files are linked.
tracer: The flux tracer plugin, providing ``file_freq`` and,
optionally, ``var_freq``.
Returns:
tuple[dict, dict]: ``(list_files, list_dates)``, each keyed by the
requested sub-period date, mapping to the list of resolved file
paths and the list of ``(start, end)`` date-interval pairs found in
those files. Both are empty if neither `ref_dir` nor `ref_file` is
given.
Raises:
FileNotFoundError: If a resolved file does not exist on disk.
"""
if not ref_dir and not ref_file:
return {}, {}
date_i, date_f = input_interval
file_freq = tracer.file_freq # type: ignore
var_freq = getattr(tracer, "var_freq", None)
file_dates = pd.date_range(date_i, date_f, freq=file_freq, inclusive="left")
if file_dates.empty:
file_dates = pd.to_datetime([date_i])
if file_dates[0] > date_i:
file_dates = pd.to_datetime([date_i] + file_dates.to_list())
# Getting files paths
file_paths = [Path(ref_dir, date.strftime(ref_file)) for date in file_dates]
list_dates = {}
list_files = {}
for date, source_path in zip(file_dates, file_paths):
if not source_path.is_file():
raise FileNotFoundError(f"file '{source_path}' not found")
# Fetching
target_path = Path(target_dir, source_path.name)
path.link(source_path, target_path)
with xr.open_dataset(source_path) as ds:
_var_freq = var_freq
if ds.sizes["time"] == 1 and var_freq is None:
_var_freq = file_freq if file_freq[-1] != 'S' else file_freq[:-1]
period_start, period_end = get_period_times(ds, _var_freq)
list_dates[date] = list(zip(period_start, period_end))
list_files[date] = len(period_start) * [str(target_path)]
return list_files, list_dates
