Source code for pycif.plugins.transforms.complex.conc2ratio.utils

import pandas as pd




def propagate_incompatible_input_dates(self, trans_mapper, trid_in, mode="backward"):
    """Synchronise the input-date grid with the union of all output-date grids.

    When ``conc2ratio`` inputs and outputs have incompatible sub-simulation
    schedules (e.g. outputs are split by observation window while inputs
    cover the full period), this function re-assigns ``input_dates`` of
    *trid_in* to the union of all output date windows and rebuilds the
    ``subsimus`` dict accordingly.

    Args:
        self (Plugin): conc2ratio plugin instance.
        trans_mapper (dict): the transform mapper; ``'inputs'`` and
            ``'outputs'`` are modified in-place.
        trid_in (tuple): ``(component, parameter)`` key of the input tracer
            whose date grid should be updated.
        mode (str): propagation direction (currently unused; kept for API
            consistency).
    """

    # Modify input dates as concatenation of output dates
    input_dates = {}
    list_ddi = []
    for trid_out in trans_mapper["outputs"]:
        input_dates[trid_out] = trans_mapper["outputs"][trid_out].get(
            "input_dates", {})
        list_ddi.extend(input_dates[trid_out].keys())

    # Stop here if all output dates are empty
    if list_ddi == []:
        return

    new_input_dates = {
        ddi: pd.concat(
            [input_dates[trid_out].get(ddi, pd.DataFrame())
             for trid_out in input_dates], axis=0)
        for ddi in set(list_ddi)
    }

    trans_mapper["inputs"][trid_in]["input_dates"] = new_input_dates

    # Now update subsimulations
    trans_mapper["subsimus"] = {
        ddi: {
            "inputs": {
                trid_in: {ddi: []}
                for trid_in in trans_mapper["inputs"]
            },
            "outputs": {
                trid_out: {
                    ddi: []
                }
                for trid_out in trans_mapper["outputs"]
                if ddi in trans_mapper["outputs"][trid_out].get("input_dates", {})
            }
        }
        for ddi in new_input_dates}