Source code for pycif.plugins.transforms.basic.background.forward

import numpy as np




def forward(
        transform,
        inout_datastore,
        controlvect,
        obsvect,
        mapper,
        di,
        df,
        mode,
        runsubdir,
        workdir,
        onlyinit=False,
        save_debug=True,
        **kwargs
):

    xmod_in = inout_datastore["inputs"]
    xmod_out = inout_datastore["outputs"]

    # Skip if empty output
    if type(xmod_out[("concs", transform.spec)][di]) == dict \
            or type(xmod_in[("concs", transform.spec)][di]) == dict:
        return

    concs = xmod_in[("concs", transform.spec)][di].set_index(
        [("metadata", "date"), ("metadata", "station")])
    bkg = xmod_in[("background", transform.spec)][di].set_index(
        [("metadata", "date"), ("metadata", "station")])

    # Drop duplicate stations
    bkg = bkg[~bkg.index.duplicated(keep='first')]

    # Reindex background to concentration resolution
    for stat in concs.index.get_level_values(1).unique():
        mask_stat_conc = concs.index.get_level_values(1) == stat
        mask_stat_bkg = bkg.index.get_level_values(1) == stat

        xmod_out[("concs", transform.spec)][di].loc[
            mask_stat_conc, ("maindata", "spec")] = \
            bkg.loc[mask_stat_bkg].reindex(
            concs.loc[mask_stat_conc].index, method="ffill"
        )[("maindata", "spec")].values \
            + concs.loc[mask_stat_conc, ("maindata", "spec")].values

        if mode == "tl":
            xmod_out[("concs", transform.spec)][di].loc[
                mask_stat_conc, ("maindata", "incr")] = \
                bkg.loc[mask_stat_bkg].reindex(
                    concs.loc[mask_stat_conc].index, method="ffill"
            )[("maindata", "incr")].values \
                + concs.loc[mask_stat_conc, ("maindata", "incr")].values

    # Propagate info from previous transforms if saving debug info
    if save_debug:
        outputs = xmod_out[("concs", transform.spec)][di]
        for input_type in ["concs", "background"]:
            data_in = xmod_in[(input_type, transform.spec)][di]

            for c in data_in.columns:
                if c[0] not in ["metadata", "maindata"]:
                    outputs[c] = np.nan
                    outputs.loc[:, c] = data_in[c].values