Source code for pycif.plugins.datastreams.fields.grib2_ecmwf.fetch

import os

import pandas as pd

from .....utils import path
from .utils import find_valid_file




def fetch(ref_dir, ref_file, input_interval, target_dir, tracer=None, **kwargs):
    """Link the closest valid ECMWF GRIB files and build the date/file maps.

    For each date spaced by ``tracer.file_freq`` across ``input_interval``,
    calls :func:`~pycif.plugins.datastreams.fields.grib2_ecmwf.utils.find_valid_file`
    to locate the closest available GRIB file (and, when ``tracer.decumul``
    is set, the surrounding file needed for decumulation), links the file
    into ``target_dir``, and records the corresponding sub-interval:
    ``[dd, dd + file_freq]`` when decumulating or when
    ``tracer.valid_interval == "left"``, otherwise a symmetric interval
    centered on ``dd``.

    Args:
        ref_dir: Date-format directory pattern where the original files are
            found.
        ref_file: Date-format pattern for the original file names.
        input_interval: List of two dates, the beginning and end of the
            period to fetch.
        target_dir: Directory where links to the original files are
            created.
        tracer: Tracer/component configuration (``file_freq``,
            ``cumul_length``, ``decumul``, ``valid_interval``,
            ``delta_tolerance``, etc.).
        **kwargs: Unused.

    Returns:
        A tuple ``(list_files, list_dates)`` of dictionaries keyed by file
        date, mapping to the list of files and covered sub-intervals.
    """
    list_period_dates = pd.date_range(
        input_interval[0], input_interval[1], freq=tracer.file_freq
    )
    time_freq = pd.to_timedelta(tracer.file_freq).to_pytimedelta()
    cumul_freq = pd.to_timedelta(f"{tracer.cumul_length}h").to_pytimedelta()

    list_dates = {}
    list_files = {}
    for dd in list_period_dates:
        dir_dd = dd.strftime(ref_dir)
        dir_dd_next = (dd + cumul_freq).strftime(ref_dir)
        dir_dd_previous = (dd - cumul_freq).strftime(ref_dir)
        files_3d, dates_3d = find_valid_file(
            ref_file,
            dd,
            time_freq,
            dir_dd,
            dir_dd_next,
            dir_dd_previous,
            cumul_length=tracer.cumul_length,
            cumul_variable=getattr(tracer, "decumul", False),
            delta_tolerance=tracer.delta_tolerance,
        )

        if os.path.isfile(files_3d[0]):
            if tracer.decumul or tracer.valid_interval == "left":
                list_dates[dd] = [[dd, dd + time_freq]]
                if list_files:
                    list_files[dd] = [files_3d]
                else:
                    list_files[dd] = [files_3d]

            else:
                list_dates[dd] = [[dd - time_freq / 2, dd + time_freq / 2]]
                list_files[dd] = [[files_3d[0]]]

            # Fetching
            # Renaming target files according to date in case
            # the file to fetch is a forecast
            local_files = []
            target_file = f"{target_dir}/{dd.strftime(ref_file)}"
            path.link(files_3d[0], target_file)
            local_files.append(target_file)

    return list_files, list_dates