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
):
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
