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

import datetime
import os

import netCDF4 as nc
import numpy as np

from .....utils import path
from .....utils.check.errclass import CifValueError

# Copied from chimere_icbc, adapted to WRF




def fetch(
    ref_dir, ref_file, input_interval, target_dir, tracer=None, component=None
):
    """Fetch the WRF-Chem initial or lateral boundary condition file.

    For ``component.orig_name == "inicond"``, links the single
    ``wrfinput``-style file for the start date. For
    ``component.orig_name == "latcond"``, links the ``wrfbdy`` file and
    parses its ``Times`` variable to build the sequence of boundary
    intervals, extrapolating one extra timestep past the last one found
    in the file (using the last inter-timestep spacing).

    Args:
        ref_dir: directory where the original file is found.
        ref_file: (template) name of the original file.
        input_interval: list of two dates; only the start date is used.
        target_dir: directory where the link to the original file is
            created.
        tracer: unused, accepted for interface compatibility.
        component: the component Plugin; ``component.orig_name`` selects
            between ``"inicond"`` and ``"latcond"`` handling.

    Returns:
        list_files: single-entry dict mapping the start date to a
            one-element list with the file path.
        list_dates: single-entry dict mapping the start date to the list
            of boundary intervals (for ``latcond``) or
            ``[[start date, start date]]`` (for ``inicond``).

    Raises:
        CifValueError: if ``component.orig_name`` is neither
            ``"inicond"`` nor ``"latcond"``, or if the ``wrfbdy`` file
            has fewer than two timestamps.
    """

    # Two cases: initial conditions or LBC
    orig_name = getattr(component, "orig_name", "")
    if 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)

    elif orig_name == "latcond":
        datei = input_interval[0]
        filei = datei.strftime(f"{ref_dir}/{ref_file}")
        list_files = {datei: [filei]}
        # Get time intervals from file as datetime.datetime
        ncf = nc.Dataset(filei)
        times_char = ncf["Times"][:].astype(str)
        ncf.close()
        times_str = np.apply_along_axis(lambda x: "".join(x), 1, times_char)
        times = [datetime.datetime.strptime(t, "%Y-%m-%d_%H:%M:%S") for t in times_str]
        # Add one more timestep to cover the period of the last input
        # time. In the flux plugin, I added flux_freq to the config to
        # do this, but here, it's all in the same file and comes from
        # real.exe. So the intervals are all the same, and I use them
        # to compute the end time.
        if len(times) < 2:
            raise CifValueError("Can't compute end date of wrfbdy_d01"
                             "because there is only one input time!")
        times.append(times[-1] + (times[-1] - times[-2]))
        # list_dates = {datei: [[times[n-1], times[n]] for n in range(1, len(times))]}
        list_dates = {datei: [[times[n], times[n]] for n in range(len(times))]}

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

    else:
        raise CifValueError("Unknown orig_name: '" + orig_name + "'")

    return list_files, list_dates