Source code for pycif.plugins.models.dummy.ini_mapper

import numpy as np
import datetime




def ini_mapper(model, general_mapper={}, backup_comps={},
               transforms_order=[], ref_transform="", **kwargs):
    """Build the data-flow mapper for the dummy Gaussian plume model.

    Declares:

    * **Inputs** — meteorological fields (``windspeed``, ``winddir``,
      ``stabclass``) and one surface flux field per active species.
    * **Outputs** — one sparse (observation-sampled) concentration field per
      active species.
    * **outputs2inputs** — links each concentration output to the flux input
      of the corresponding species so the adjoint pipeline knows which flux
      sensitivity to accumulate.

    Input date windows are hourly intervals derived from
    ``model.input_dates``.

    Args:
        model: dummy model plugin instance (carries ``input_dates``,
            ``domain``, and ``chemistry.acspecies``).
        general_mapper (dict): unused.
        backup_comps (dict): unused.
        transforms_order (list): unused.
        ref_transform (str): unused.
        **kwargs: unused.

    Returns:
        dict: mapper with ``inputs``, ``outputs``, and ``outputs2inputs``.
    """
    input_intervals = {
        ddi: np.append(
            model.input_dates[ddi][:-1, np.newaxis],
            model.input_dates[ddi][:-1, np.newaxis]
            + datetime.timedelta(hours=1),
            axis=1)
        for ddi in model.input_dates}

    default_dict = {
        "input_dates": input_intervals,
        "force_loadin": True,
        "force_dump": True,
        "sparse_data": False,
        "sampled": False,
    }
    dict_surface = dict(default_dict, **{"domain": model.domain})

    # Executable
    mapper = {
        "inputs": {("meteo", param): dict_surface
                   for param in ["windspeed", "winddir", "stabclass"]},
        "outputs": {("concs", s): {
            "force_loadout": True,
            "input_dates": input_intervals,
            "continuous_hdomain": True,
            "continuous_vdomain": True,
            "domain": model.domain,
            "sampled": True,
            "sparse_data": False
        }
            for s in model.chemistry.acspecies.attributes},
    }

    emis = {
        ("flux", s): dict_surface
        for s in model.chemistry.acspecies.attributes
    }
    mapper["inputs"].update(emis)

    # Save propagation of perturbations
    mapper["outputs2inputs"] = {
        ("concs", s): [("flux", s)]
        for s in model.chemistry.acspecies.attributes
    }

    return mapper