import numpy as np
[docs]
def forward(
transform,
inout_datastore,
controlvect,
obsvect,
mapper,
di,
df,
mode,
runsubdir,
workdir,
onlyinit=False,
save_debug=True,
**kwargs
):
"""Add background concentrations to regional model simulations.
For each observation station, reindexes the background field
(forward-filled) to the same date/station index as the regional
concentrations, then computes:
.. math::
y_{out} = y_{concs} + y_{background}
In TL mode the same addition is applied to the ``incr`` field.
The output replaces the ``('concs', spec)`` entry in the datastore.
Args:
transform (Plugin): background instance (carries ``spec``).
inout_datastore (dict): mutable datastore; ``'inputs'`` must contain
both ``('concs', spec)`` and ``('background', spec)`` entries.
controlvect: unused.
obsvect: unused.
mapper (dict): transform mapper.
di (datetime): sub-simulation start date.
df (datetime): sub-simulation end date.
mode (str): ``'fwd'`` or ``'tl'``.
runsubdir (str): unused.
workdir (str): unused.
onlyinit (bool): unused.
save_debug (bool): if ``True``, propagate extra diagnostic columns
from inputs to the output datastore.
**kwargs: unused.
"""
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
