import numpy as np
from logging import debug
from . import add_default
[docs]
def init_reindex(
self,
trid,
tmp_dict, trid_dict,
precursor_id, transform,
param,
all_transforms,
mapper,
backup_comps,
precursors,
do_pipe_entry=False
):
"""Insert a time-interpolation transform if the date indices differ.
Compares the ``input_dates`` of ``tmp_dict`` (a precursor's output)
against ``trid_dict`` (the target input), per sub-simulation. If they
already match exactly for every sub-simulation, nothing is inserted.
Otherwise, a ``time_interpolation`` transform is created via
:func:`add_default.add_default` and spliced between ``precursor_id``
and ``transform``.
Args:
self: The parent object (obs operator) holding the transform pipe.
trid: ``(component, parameter)`` tuple identifying the tracer.
tmp_dict: Attribute dictionary of the source side (precursor's
output), holding ``"input_dates"`` and ``"sparse_data"``.
trid_dict: Attribute dictionary of the target side (this
transform's input), holding ``"input_dates"`` to compare against.
precursor_id: Id of the precursor transform providing the data.
transform: Id of the transform whose input is being checked.
param: Object holding an optional ``time_interpolation``
configuration block used to parametrize the inserted transform.
all_transforms: Namespace holding all registered transform instances.
mapper: Dictionary mapping each transform id to its inputs/outputs
metadata, updated when a new transform is inserted.
backup_comps: Backup of components used to restore/compare state.
precursors: Precursors metadata, kept for interface consistency
with related init functions.
do_pipe_entry: Whether to run the pipe-entry initialization step
for the newly inserted transform.
Returns:
str: The id of the transform now directly feeding ``transform`` for
this trid — either the newly inserted ``time_interpolation``
transform, or the original ``precursor_id`` if indices already match.
"""
cmp, prm = trid
# Differentiate sparse data versus matrix data
if trid_dict.get("sparse_data", False):
same_index = [
np.all(
tmp_dict["input_dates"][ddi] == trid_dict["input_dates"].get(ddi, [
])
) if len(tmp_dict["input_dates"][ddi])
== len(trid_dict["input_dates"].get(ddi, []))
else False
for ddi in tmp_dict["input_dates"]
]
else:
# For matrix data, check if the index in the outputs (tmp_dict)
# is included in the input (trid_dict)
# If there are more dates available in inputs, don't do the
# interpolation
prec_dates = tmp_dict.get("input_dates", [])
ref_dates = trid_dict.get("input_dates", [])
same_index = [
np.all(ref_dates[ddi]
== prec_dates.get(ddi, []))
if len(ref_dates[ddi]) == len(prec_dates.get(ddi, []))
else False
for ddi in ref_dates
]
if sum(same_index) == len(same_index):
return precursor_id
debug(f' Temporal re-indexing if any: {prm} / {param}')
tinterp = getattr(param, "time_interpolation", None)
yml_dict = {
"plugin": {
"name": "time_interpolation",
"version": "std",
"type": "transform",
},
"method": getattr(tinterp, "method", "bilinear"),
"component": [cmp],
"parameter": [prm],
"successor": transform,
"precursor": precursor_id,
**{attr: getattr(tinterp, attr)
for attr in getattr(tinterp, "attributes", []) if attr != "plugin"}
}
ref_precursor = {(cmp, prm): precursor_id}
ref_successor = {(cmp, prm): transform}
new_transf, new_id = add_default.add_default(
self,
all_transforms,
yml_dict,
position="index",
index=all_transforms.attributes.index(transform),
mapper=mapper,
init=True,
backup_comps=backup_comps,
successor=ref_successor,
precursor=ref_precursor,
do_pipe_entry=do_pipe_entry
)
return new_id
