Source code for pycif.plugins.datastreams.meteos.chimere_meteo.fetch

import datetime
import os

import pandas as pd

from .....utils import path




def fetch(ref_dir, ref_file, input_interval, target_dir, tracer=None, **kwargs):
    """Locate CHIMERE METEO.nc files and link them to the working directory.

    Iterates over period-start dates covering ``input_interval`` at
    ``tracer.file_freq``, formats ``ref_dir/ref_file`` for each period, and
    for each file that exists on disk records the hourly sub-periods it
    covers and symlinks it into ``target_dir``. Periods whose file does not
    exist are silently skipped (absent from the returned dicts).

    Args:
        ref_dir: directory holding the CHIMERE METEO.nc files.
        ref_file: strftime-style file name pattern (relative to
            ``ref_dir``), formatted once per period.
        input_interval: 2-element sequence ``(datei, datef)`` giving the
            requested date range.
        target_dir: directory where matching files are symlinked.
        tracer: the meteo datastream Plugin, used for ``tracer.file_freq``.

    Returns:
        tuple: ``(list_files, list_dates)``, dicts keyed by each period's
        start date. ``list_files`` maps each key to the (repeated) linked
        file path, one entry per hour in the period; ``list_dates`` maps
        each key to the corresponding list of ``[hour, hour + 1h]`` date
        pairs.
    """
    list_period_dates = pd.date_range(
        input_interval[0], input_interval[1], freq=tracer.file_freq
    )
    list_dates = {}
    list_files = {}
    for dd in list_period_dates:
        file = dd.strftime(f"{ref_dir}/{ref_file}")
        if os.path.isfile(file):
            file_hours = pd.date_range(
                dd, dd + pd.to_timedelta(tracer.file_freq), freq="1h"
            )
            list_dates[dd] = [
                [hh, hh + datetime.timedelta(hours=1)] for hh in file_hours
            ]
            list_files[dd] = len(file_hours) * [file]

            # Fetching
            target_file = f"{target_dir}/{os.path.basename(file)}"
            path.link(file, target_file)

    return list_files, list_dates