Source code for pycif.plugins.datastreams.fluxes.iconart.fetch

import os
import datetime as dt
import pandas as pd
import xarray as xr
import itertools
from .....utils import path
from .....utils.classes.setup import Setup




def fetch(ref_dir, ref_file, input_interval, target_dir, tracer=None, **kwargs):
    """Fetch ICON-ART flux files and build the hourly date intervals they cover.

    Splits the simulation interval into sub-periods using ``tracer.model.periods``
    (default ``'10D'``), links the file matching the start of each sub-period
    (if it exists) into `target_dir`, and expands each sub-period into hourly
    ``[start, end]`` intervals at ``tracer.model.input_resolution``.

    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 (list[datetime.datetime]): ``[date_i, date_f]``
            simulation interval to cover.
        target_dir (str): Directory where the resolved files are linked.
        tracer: The flux tracer plugin, providing access to ``model``
            (for ``periods`` and ``input_resolution``).

    Returns:
        tuple[dict, dict]: ``(list_files, list_dates)``, each keyed by the
        start date of a sub-period, mapping to the list of file paths and
        the list of hourly ``[start, end]`` date-interval pairs within that
        sub-period.
    """

    freq_subperiods = tracer.model.periods \
        if getattr(tracer.model, "periods", False) \
        else '10D'

    list_period_dates = \
        pd.date_range(input_interval[0], input_interval[1], freq=freq_subperiods)
    list_dates = {}
    list_files = {}

    for di, df in zip(list_period_dates[:-1], list_period_dates[1:]):

        file = di.strftime(f"{ref_dir}/{ref_file}")
        if os.path.isfile(file):
            target_file = f"{target_dir}/{os.path.basename(file)}"
            path.link(file, target_file)

        list_hours = pd.date_range(di, df, freq=tracer.model.input_resolution)
        list_files[di] = (len(list_hours) * [file])
        list_dates[di] = [[hi, hf] for hi, hf in zip(list_hours[:-1], list_hours[1:])]

    return list_files, list_dates