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

import datetime
import os

import pandas as pd
import xarray as xr

from .....utils import path
from .....utils.check.errclass import CifError
from .....utils.hdf5 import _hdf5_lock




def fetch(ref_dir, ref_file, input_interval, target_dir, tracer=None, **kwargs):
    """Fetch a D&B forcing file and list the period(s) it covers.

    D&B writes its whole forcing to a single NetCDF file: it is linked once
    into ``target_dir``, and, if the file has a ``time`` coordinate (D&B's
    hourly *dynamic* forcing, e.g. ``dynforcing.nc``), it is read to list
    the hourly sub-periods overlapping ``input_interval``. Time-invariant
    *static* forcing files (e.g. ``staticforcing.nc``, see
    ``src/ncread_forcing.f90::ncread_static_forcing``) have no ``time``
    coordinate: they are instead listed as a single period covering the
    whole of ``input_interval``.

    Args:
        ref_dir (str): directory where the forcing file is found.
        ref_file (str): name of the forcing file.
        input_interval (list): simulation interval, as a list of the two
            bounding dates.
        target_dir (str): directory where a link to the forcing file is
            created.
        tracer: the tracer Plugin, giving access to ``name``.
        **kwargs: unused, kept for interface compatibility.

    Returns:
        (dict, dict): ``list_files`` and ``list_dates``, both with a single
        key (the first covered hour, or ``input_interval``'s start for
        static forcing), following the same ``{key: [...per-period...]}``
        convention as other fluxes plugins (see e.g.
        ``pycif.plugins.datastreams.fluxes.chimere.fetch.fetch``).

    Raises:
        CifError: if the forcing file is not found, or (dynamic forcing
            only) has no ``time`` record overlapping ``input_interval``.
    """
    file = os.path.join(ref_dir, ref_file)

    if not os.path.isfile(file):
        raise CifError(
            f"Could not find the D&B forcing file for tracer "
            f"{tracer.name}:\n   - ref_dir: {ref_dir}\n   - ref_file: {ref_file}"
        )

    with _hdf5_lock:
        with xr.open_dataset(file) as ds:
            has_time = "time" in ds

    date_i, date_f = input_interval

    if has_time:
        with _hdf5_lock:
            with xr.open_dataset(file) as ds:
                times = pd.DatetimeIndex(ds["time"].values)

        hours = times[(times > date_i) & (times <= date_f)]

        if hours.empty:
            raise CifError(
                f"D&B dynamic forcing file '{file}' has no 'time' record "
                f"within the requested interval {input_interval} for tracer "
                f"{tracer.name}."
            )

        ddi = hours[0].to_pydatetime() - datetime.timedelta(hours=1)
        list_dates = {
            ddi: [
                [hh.to_pydatetime() - datetime.timedelta(hours=1), hh.to_pydatetime()]
                for hh in hours
            ]
        }
        list_files = {ddi: len(hours) * [file]}
    else:
        # Static forcing: time-invariant, so a single period covering the
        # whole requested interval is enough.
        list_dates = {date_i: [[date_i, date_f]]}
        list_files = {date_i: [file]}

    # Fetching
    target_file = os.path.join(target_dir, os.path.basename(file))
    path.link(file, target_file)

    return list_files, list_dates