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

import datetime
import os

import numpy as np
import pandas as pd

from .....utils import path




def fetch(ref_dir, ref_file, input_interval, target_dir, tracer=None, component=None):
    """Link CHIMERE INI_CONCS/BOUN_CONCS files and build the date/file maps.

    For ``component.orig_name == "inicond"``, links the single initial
    condition file matching the start of ``input_interval``. Otherwise
    (lateral/top boundary conditions), iterates over dates spaced by
    ``tracer.file_freq`` across ``input_interval``, links each existing
    ``BOUN_CONCS``-style file found, and builds hourly sub-intervals
    covering the period spanned by each file.

    Args:
        ref_dir: Directory 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; ``tracer.file_freq`` gives
            the frequency at which BOUN_CONCS files are available.
        component: Component being fetched; ``component.orig_name`` is used
            to distinguish initial conditions from boundary conditions.

    Returns:
        A tuple ``(list_files, list_dates)``. For initial conditions, both
        dictionaries have a single key (the start date). For boundary
        conditions, each key is a file date mapping to the list of hourly
        files/sub-intervals covered by that file.
    """

    # Two cases: initial conditions or LBC
    if getattr(component, "orig_name", "") == "inicond":
        datei = input_interval[0]
        filei = datei.strftime(f"{ref_dir}/{ref_file}")
        list_files = {datei: [filei]}
        list_dates = {datei: [[datei, datei]]}

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

    else:
        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