Source code for pycif.plugins.transforms.basic.time_interpolation.utils.indexes
import numpy as np
import pandas as pd
import copy
import datetime
from logging import warning, debug
from ......utils.classes.transforms import Transform
from ......utils.check.errclass import CifError
[docs]
def calc_indexes(mapper, trid_out, all_transforms, general_mapper):
"""Calculate indexes correspondance between input dates and output dates.
The calculation relies on averages between inputs and outputs,
weighted by the relative duration of targets compared to available inputs.
Args:
mapper (dict[str]): the mapper for the present transform
trid_out (tuple[str]):
all_transforms (Transform): the object storing information on all transforms
general_mapper (dict[str]): the overall mapper for all transforms
"""
# Alias to variables
in_dict = mapper["inputs"][trid_out]
out_dict = mapper["outputs"][trid_out]
in_sparse = in_dict.get("sparse_data", False)
out_sparse = out_dict.get("sparse_data", False)
in_dates = in_dict["input_dates"]
out_dates = out_dict["input_dates"]
# Check if another transform with same settings was initialized
list_tinterp = [
transf for transf in all_transforms.attributes
if getattr(all_transforms, transf).plugin.name == "time_interpolation"]
for transf in list_tinterp:
inputs = general_mapper[transf]["inputs"]
in_trid = []
for trid in inputs:
target_dates = inputs[trid].get("input_dates", {})
if in_dates.keys() == target_dates.keys():
all_equals = np.all([
False if len(in_dates[dd]) != len(target_dates[dd])
else np.all(in_dates[dd] == target_dates[dd])
for dd in in_dates])
if all_equals:
in_trid.append(trid)
outputs = general_mapper[transf]["outputs"]
for trid in outputs:
if trid in in_trid:
target_dates = outputs[trid].get("input_dates", {})
if len(out_dates) != len(target_dates):
continue
if out_dates.keys() != target_dates.keys():
continue
same_dates = [
False if len(out_dates[ddi]) != len(target_dates[ddi])
else out_dates[ddi].equals(target_dates[ddi])
for ddi in target_dates
]
if ~np.all(same_dates):
continue
mapper["interpol_indexes"] = \
general_mapper[transf]["interpol_indexes"]
mapper["do_interpolation"] = \
general_mapper[transf]["do_interpolation"]
mapper["reorder_periods"] = \
general_mapper[transf]["reorder_periods"]
in_dict["force_loadin"] = \
inputs[trid]["force_loadin"]
debug("Fetching temporal interpolation indexes "
"from another transform")
return
# Otherwise, do the job
do_interpolation = {datei: False for datei in out_dates}
reorder_periods = {datei: False for datei in out_dates}
force_loadin = {datei: False for datei in in_dates}
interpol_indexes = {}
total_lengths = {}
for datei in out_dates:
# Special case if only one date
if np.size(out_dates[datei]) == 0:
continue
elif np.size(out_dates[datei]) == 1:
raise CifError(
f"Output date {out_dates[datei]} for {trid_out} should be an interval (start and end date), even if they are the same. Please check your YAML configuration file."
)
else:
out_dates_start = out_dates[datei]["start_date"]
out_dates_end = out_dates[datei]["end_date"]
# Initialize a flag to check that all output dates are covered
covered_output_dates = np.zeros_like(out_dates_start, dtype=bool)
# Loop over input periods and find corresponding output periods
tmp_indexes = {}
total_length = pd.Series(np.zeros(len(out_dates_start)))
length_per_input_period = []
ddi_masks = {}
for ddi in in_dates:
# Skip if empty input dates
if len(in_dates[ddi]) == 0:
continue
# Skip if input date interval do not include any output dates
if np.max(in_dates[ddi]["end_date"]) < np.min(out_dates_start) \
or np.min(in_dates[ddi]["start_date"]) > np.max(out_dates_end):
continue
in_dates_start = in_dates[ddi]["start_date"]
# Special case if only one date
if np.size(in_dates[ddi]) == 1:
in_dates_end = in_dates_start + datetime.timedelta(hours=1)
else:
in_dates_end = in_dates[ddi]["end_date"]
in_dates_all = in_dates[ddi].stack().reset_index(
drop=True).drop_duplicates()
in_dates_durations = \
pd.Series(in_dates_end - in_dates_start).dt.total_seconds()
# Crop out_dates to the date interval covered by input dates
ddi_mask = (
(out_dates_start <= in_dates_all.max())
& (out_dates_end >= in_dates_all.min())
)
ddi_masks[ddi] = ddi_mask
out_dates_end_tmp = out_dates_end[ddi_mask]
out_dates_start_tmp = out_dates_start[ddi_mask]
covered_output_dates[ddi_mask] = True
# Removing output dates duplicates to speed up the process
out_dates_tmp = pd.DataFrame(
data={'date_start': out_dates_start_tmp,
'date_end': out_dates_end_tmp,
'index': range(len(out_dates_end_tmp))
}
)
unique_index = out_dates_tmp.drop_duplicates(
['date_start', 'date_end'])['index'].values
out_dates_end_tmp = out_dates_end_tmp.iloc[unique_index]
out_dates_start_tmp = out_dates_start_tmp.iloc[unique_index]
# Find index of start dates
index_tmpstart = \
pd.concat([in_dates_all, out_dates_start_tmp]
).drop_duplicates().sort_values()
df_index = pd.Series(range(len(in_dates_all)),
index=in_dates_all)
df_index = df_index.reindex(index_tmpstart)
start_inside = df_index.interpolate(
method="index", limit_area="inside").loc[out_dates_start_tmp]
df_index = df_index.interpolate(
method="index", limit_direction="both")
out_index_start = df_index.loc[out_dates_start_tmp]
# Find index of end dates
index_tmpend = \
np.sort(np.unique(np.append(in_dates_all, out_dates_end_tmp)))
df_index = \
pd.Series(range(len(in_dates_all)), index=in_dates_all)
df_index = df_index.reindex(index_tmpend)
end_inside = df_index.interpolate(
method="index", limit_area="inside").loc[out_dates_end_tmp]
df_index = df_index.interpolate(
method="index", limit_direction="both")
out_index_end = df_index.loc[out_dates_end_tmp]
# Compute weights for each time step
# If outputs are snapshots (i.e., start and end dates are the same),
# define duration as 1 by default
out_durations = (
np.ceil(out_index_end.values)
- np.floor(out_index_start.values)).astype(int)
out_durations[out_dates_end_tmp == out_dates_start_tmp] = 1
out_durations[
np.isnan(start_inside).values
& np.isnan(end_inside).values] = 0
out_df = pd.DataFrame({"start": out_index_start.values,
"end": out_index_end.values,
"duration": out_durations})
out_df.index = unique_index
weights = copy.deepcopy(
out_df.loc[
out_df.index.repeat(
out_df["duration"])])
# Skip if weights are empty
if len(weights) == 0:
continue
indexes = np.zeros(len(weights))
start_index = np.array(
[0] + list(np.cumsum(
out_df["duration"].loc[out_df["duration"] > 0].values[:-1]))
).astype(int)
if len(weights) > 0:
indexes[start_index] = start_index
np.maximum.accumulate(indexes, out=indexes)
weights["indexes"] = \
np.arange(len(weights)) - indexes + np.floor(weights["start"])
weights["weights"] = np.maximum(
1,
np.minimum(weights["indexes"] + 1, weights["end"])
- np.maximum(weights["indexes"], weights["start"])
)
# Tweak index for snapshots at the end of the last input period
mask_tweak = (weights["indexes"] == len(in_dates_start))
weights.loc[mask_tweak, "indexes"] -= 1
# Scale weights by duration of each sub-period
mask_wgt = weights["weights"] != 0
weights.loc[mask_wgt, "weights"] *= \
in_dates_durations.iloc[
weights.loc[mask_wgt, "indexes"]].values
group_weights = weights["weights"].groupby(by=weights.index).sum()
total_length.loc[
np.where(ddi_mask)[0][group_weights.index]
] += group_weights.values
tmp_indexes[ddi] = weights.loc[:, ["indexes", "weights"]]
tmp_indexes[ddi]["total_weights"] = weights["weights"].sum()
tmp_indexes[ddi]["period"] = ddi
do_interpolation[datei] = True
force_loadin[ddi] = True
# Avoid recombining periods if asked
if not mapper.get("recombine_periods", True):
recombined = pd.concat(tmp_indexes.values())
indmax = \
recombined.reset_index().groupby(
"index").idxmax()["total_weights"]
recombined = recombined.iloc[indmax.values]
tmp_indexes = {
ddi: recombined.loc[recombined["period"] == ddi]
for ddi in tmp_indexes}
total_length = pd.concat(tmp_indexes.values())
total_length = \
total_length.groupby(total_length.index).sum()["weights"]
# Check whether the inputs and outputs have exactly the same index
# In that case, there is no need to do interpolation
same_periods = [
np.all(tmp_indexes[ddi].index.values
== tmp_indexes[ddi]["indexes"].values)
for ddi in tmp_indexes]
full_periods = [
len(out_dates_start) == len(in_dates[ddi])
and len(tmp_indexes[ddi]) != 0
for ddi in tmp_indexes]
do_interpolation[datei] = (
(tmp_indexes != {})
and (~np.all(same_periods) or (sum(full_periods) != 1))
)
# Check whether the transform is only a re-ordering of sub-periods
# Applies only if there is no interpolation to do
# (i.e., files can be used as a whole)
reorder = {
ddi: out_dates[datei].equals(in_dates[ddi])
if len(out_dates[datei]) == len(in_dates[ddi]) != 0
else len(out_dates[datei]) == len(in_dates[ddi])
for ddi in in_dates
}
reorder_periods[datei] = [
reorder[ddi]
and not do_interpolation[datei] for ddi in in_dates]
# If no interpolation to be done, but found no re-ordering, force interpolation
if not np.any(reorder_periods[datei]) and not do_interpolation[datei]:
do_interpolation[datei] = True
# Fix at 1 if total length is zero, to avoid nans in weights
# TODO: Make something cleaner for data not characterized by intervals
if np.any(total_length == 0):
warning("Setting zero lengths to one in temporal interpolation to "
"avoid nans in weights. Happens when interpolating to "
"snapshots data contrary to interval data.")
total_length.loc[total_length == 0] = 1
# Normalize weights by total length
for ddi in in_dates:
if ddi not in tmp_indexes:
continue
if "weights" in tmp_indexes.get(ddi, {}):
tmp_indexes[ddi]["weights"] /= total_length.iloc[
np.where(ddi_masks[ddi])[0][tmp_indexes[ddi].index]
].values
if len(tmp_indexes[ddi]) == 0:
del tmp_indexes[ddi]
interpol_indexes[datei] = tmp_indexes
# Raise exception if any output is not covered
if not np.all(covered_output_dates):
raise CifError(
f"Some output dates are not covered by the inputs for {trid_out}.\n"
"Please check your Yml.\n"
f"Missing dates are: "
f"{out_dates_start[~covered_output_dates]}\n"
"Whereas input dates are: \n" +
("\n".join(
[f"- {ddi}\n"
+ "\n".join([f" - {di}" for di in in_dates[ddi]])
for ddi in in_dates
]
))
)
# If interpolation is required for no date, no need to force reading
in_dict["force_loadin"] = force_loadin
mapper["do_interpolation"] = do_interpolation
mapper["reorder_periods"] = reorder_periods
mapper["interpol_indexes"] = interpol_indexes
