Source code for pycif.plugins.models.lmdz_ico.io.native2inputs

from __future__ import annotations

import datetime
from os import PathLike
from typing import Any, Literal
from .....utils.check.errclass import CifValueError

from .inputs import (
    make_chemfields,
    make_endconcs,
    make_fluxes,
    make_inicond,
    make_kinetic,
    make_meteo,
)




def native2inputs(
    self,
    datastore: dict[tuple[str, str], dict[str | datetime.datetime, Any]] | None,
    input_type: str,
    datei: datetime.datetime,
    datef: datetime.datetime,
    runsubdir: str | PathLike,
    mode: Literal["fwd", "tl", "adj"] = "fwd",
    onlyinit: bool = False,
    do_simu: bool = True,
    check_transforms: bool = False,
    **kwargs,
) -> None:
    """Converts data at the model data resolution to model compatible input
    files.

    Args:
        self: the model Plugin
        input_type (str): one of 'fluxes', 'obs'
        datastore: data to convert
            if input_type == 'fluxes', a dictionary with flux maps
            if input_type == 'obs', a pandas dataframe with the observations
        datei, datef: date interval of the sub-simulation
        mode (str): running mode: one of 'fwd', 'adj' and 'tl'
        runsubdir (str): sub-directory for the current simulation
        workdir (str): the directory of the whole pycif simulation

    Notes:
        - LMDZ expects daily inputs; if the periods in the control vector are
        longer than one day, period values are uniformly de-aggregated to the
        daily scale; this is done with pandas function 'asfreq' and the option
        'ffill' as 'forward-filling'
        See Pandas page for details:
        https://pandas.pydata.org/pandas-docs/stable/generated/pandas
        .DataFrame.asfreq.html

    """
    if datastore is None:
        datastore = {}

    if not do_simu:
        return

    # Switching datei and datef if adjoint
    ddi = min(datei, datef)
    ddf = max(datei, datef)

    # If mode is "fwd" or "tl" but onlyinit is True, it means that we are
    # initializing an adjoint by running backward the adjoint pipeline
    if mode in ("fwd", "tl") and onlyinit:
        mode = "adj"

    # Deals with fluxes
    if input_type == "flux":
        make_fluxes(self, datastore, ddi, runsubdir, mode)

    # Deals with chemical fields
    # TODO: deposition may not work there (3D field, not 4D)
    elif input_type in ("prodloss3d", "prescrconcs", "deposition", "photorates"):
        make_chemfields(self, datastore, input_type, ddi, runsubdir, mode)

    # Deals with photolysis rates
    elif input_type == "kinetic":
        make_kinetic(self, datastore, ddi, runsubdir)

    # Deals with initial conditions
    elif input_type in ("inicond", "restart_inicond"):
        make_inicond(self, datastore, input_type, ddi, runsubdir, mode, onlyinit)

    # Initial conditions from previous simulations
    elif input_type == "endconcs":
        make_endconcs(self, datastore, ddi, ddf, runsubdir, mode, onlyinit)

    # Deals with meteo
    elif input_type == "meteo":
        make_meteo(self, datastore, ddi, runsubdir)

    else:
        raise CifValueError(f"Unknown input type '{input_type}'")