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

from __future__ import annotations

from datetime import datetime
from pathlib import Path

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


# pylint: disable=unused-argument


def fetch(
    ref_dir: str,
    ref_file: str,
    input_interval: dict[datetime, list[datetime]],
    target_dir: str,
    tracer: object | None = None,
    **kwargs,
):
    """Locate LMDZ monthly mass-flux files and link them to the working directory.

    For each period listed in ``tracer.model.meteo_input_dates``, builds
    the expected ``fluxstoke``/``phystoke_csr`` file names (plus
    ``fluxstokev`` when ``tracer.model.grid == "regular"``) following the
    ``<prefix>.<YYYY>.m<MM><suffix>.nc`` convention, links each into
    ``target_dir``, and records the sub-date-interval pairs covered by
    each period.

    Args:
        ref_dir: directory holding the LMDZ mass-flux files.
        ref_file: unused (kept for interface consistency; this plugin
            derives its own file names from ``tracer.model``).
        input_interval: unused (kept for interface consistency; dates
            come from ``tracer.model.meteo_input_dates``).
        target_dir: directory where matching files are symlinked.
        tracer: the meteo datastream Plugin, used for
            ``tracer.model.grid`` and ``tracer.model.meteo_input_dates``.

    Returns:
        tuple: ``(list_files, list_dates)``, dicts keyed by each period's
        start date. ``list_files`` maps each key to a placeholder list
        (one ``"dummy"`` entry per sub-period; the actual linked file
        names are not tracked here). ``list_dates`` maps each key to the
        list of consecutive ``(start, end)`` date pairs within the
        period.

    Raises:
        CifValueError: if ``tracer`` is ``None``.
        CifFileNotFoundError: if an expected mass-flux file is missing
            from ``ref_dir``.
    """
    if tracer is None:
        raise CifValueError("tracer is None")

    filetype_list = [("fluxstoke", ""), ("phystoke", "_csr")]

    if tracer.model.grid == "regular":
        filetype_list.append(("fluxstokev", ""))

    input_dates = tracer.model.meteo_input_dates  # type: ignore

    list_dates: dict[datetime, list[tuple[datetime, datetime]]] = {}
    list_files: dict[datetime, list[str]] = {}

    # Link meteo files
    for di, dates in input_dates.items():
        # Set dates
        list_dates[di] = [(di, df) for di, df in zip(dates[:-1], dates[1:])]
        list_files[di] = len(list_dates[di]) * ["dummy"]

        # Link meteo files
        for prefix, suffix in filetype_list:
            filename = f"{prefix}.{di:an%Y.m%m}{suffix}.nc"
            source = Path(ref_dir, filename)

            if not source.is_file():
                raise CifFileNotFoundError(f"Meteo file '{source}' for {di} not found")

            target = Path(target_dir, source.name)
            path.link(source, target)

    return list_files, list_dates