import copy
import itertools
from logging import warning, debug
import pandas as pd
import numpy as np
from . import init_default_transformations
from .initiate_default_attributes import initiate_default_attributes
print_all = False
# attribute2print = "input_dates"
attribute2print = "sparse_data"
transfor2print = [
# 'transtest_2_after',
# 'dump2inputs_std_00009',
# 'time_interpolation_std_00021',
# 'loadfromoutputs_std_00050',
# 'run_model'
]
[docs]
def propagate_attributes(
self, all_transforms, mapper, transform_id,
backup_comps=None,
next_did_nothing=False,
previous_did_nothing=False,
only_backwards=False,
only_forwards=False,
parent_transform=None,
parent_trids=None
):
"""Propagate attributes backward and forward
when initializing a new transform."""
backup_comps = {} if backup_comps is None else backup_comps
# If only_backwards and only_forwards, just skip
if (only_backwards and only_forwards) or parent_trids == []:
return
# Initiate default values for some parameters if not explicitly specified in transforms
# (e.g., is_lbc, is_top)
initiate_default_attributes(all_transforms, mapper, transform_id)
# Propagate information about forward or backward propagation priority
to_propagate = [
"vdomain", "domain", "tracer", "sampled", "sparse_data",
"is_lbc", "is_top", "input_dates", "input_files", "recombine_periods",
"dirorig", "fileorig", "continuous_hdomain", "continuous_vdomain",
"has_control_ancestor", "has_obsvect_successor",
"all_successors_initialized"
]
for attribute in to_propagate:
propagate_attribute(
self, all_transforms, mapper,
[f"{attribute}_from_previous"],
transform_id,
only_backwards=True,
next_did_nothing=next_did_nothing,
parent_transform=parent_transform,
parent_trids=parent_trids
)
# Now propagate batches of attributes
# For each direction, provide here the list of attributes
# (some can go together)
# For each attribute list, three arguments to be provided:
# - whether or not to force propagation
# - list of arguments
# - how to deal with boolean:
# -> if "or", propagates True if any
# -> if None, will use the for_propagate argument
to_propagate = {
"only_forwards": [
(True, ["has_control_ancestor"], ["or"])
],
"only_backwards": [
(False, ["force_loadin"], ["or"]),
(True, ["has_obsvect_successor"], ["or"]),
(False, ["all_successors_initialized"], ["and"])
],
"both": [
(False, ["domain", "is_top", "is_lbc"], [None, None, None]),
(False, ["continuous_hdomain"], [None]),
(False, ["continuous_vdomain"], [None]),
(False, ["tracer"], [None]),
(False, ["sampled"], [None]),
(False, ["sparse_data"], [None]),
(False, ["input_dates"], [None]),
(False, ["input_files"], [None]),
(False, ["recombine_periods"], [None]),
(False, ["dirorig", "fileorig"], [None]),
]
}
did_nothing_forwards = True
did_nothing_backwards = True
list_updated_output_trids = []
list_updated_input_trids = []
for direction, to_propagate_dir in to_propagate.items():
# Some attributes are to be propagated
# in only one direction
only_backwards_tmp = only_backwards
only_forwards_tmp = only_forwards
if direction == "only_forwards":
only_forwards_tmp = True
elif direction == "only_backwards":
only_backwards_tmp = True
for force_propagate, attributes, deal_boolean in to_propagate_dir:
did_nothing_forwards_tmp, list_updated_output_trids_tmp, \
did_nothing_backwards_tmp, list_updated_input_trids_tmp = propagate_attribute(
self, all_transforms, mapper,
attributes,
transform_id,
backup_comps=backup_comps,
only_backwards=only_backwards_tmp,
only_forwards=only_forwards_tmp,
next_did_nothing=next_did_nothing,
previous_did_nothing=previous_did_nothing,
parent_transform=parent_transform,
parent_trids=parent_trids,
force_propagate=force_propagate,
deal_boolean=deal_boolean
)
did_nothing_forwards = did_nothing_forwards and did_nothing_forwards_tmp
did_nothing_backwards = did_nothing_backwards and did_nothing_backwards_tmp
list_updated_output_trids.extend(list_updated_output_trids_tmp)
list_updated_input_trids.extend(list_updated_input_trids_tmp)
list_updated_input_trids = list(set(list_updated_input_trids))
list_updated_output_trids = list(set(list_updated_output_trids))
# Propagates backwards to precursors
if not did_nothing_backwards or not next_did_nothing:
precursors = mapper[transform_id]["precursors"]
for trid in precursors:
for tr in precursors[trid]:
if tr == transform_id:
continue
propagate_attributes(
self, all_transforms, mapper,
tr,
only_backwards=True,
next_did_nothing=did_nothing_backwards,
previous_did_nothing=True,
parent_transform=transform_id,
parent_trids=list_updated_input_trids if parent_transform is not None else None,
)
# Propagates forwards to successors
if not did_nothing_forwards or not previous_did_nothing:
successors = mapper[transform_id]["successors"]
for trid in successors:
for tr in successors[trid]:
if tr == transform_id:
continue
propagate_attributes(
self, all_transforms, mapper,
tr,
only_forwards=True,
previous_did_nothing=did_nothing_forwards,
next_did_nothing=True,
parent_transform=transform_id,
parent_trids=list_updated_output_trids if parent_transform is not None else None,
)
# If propagated arguments, try to initiate reprojections, reindex, etc.
if not did_nothing_forwards or not did_nothing_backwards:
init_default_transformations.init_default_transformations(
self, all_transforms,
backup_comps, mapper,
transform_id,
trid_to_check=list_updated_output_trids
)
[docs]
def propagate_attribute(
self, all_transforms, mapper,
attributes,
transform_id,
only_backwards=False,
only_forwards=False,
next_did_nothing=False,
previous_did_nothing=False,
backup_comps=None,
parent_transform=None,
parent_trids=None,
force_propagate=False,
deal_boolean=None
):
backup_comps = {} if backup_comps is None else backup_comps
did_nothing_backwards = True
list_updated_input_trids = []
if not only_forwards:
did_nothing_backwards, list_updated_input_trids = propagate_backwards(
self, all_transforms, mapper, attributes,
transform_id,
next_did_nothing=next_did_nothing,
backup_comps=backup_comps,
parent_transform=parent_transform,
parent_trids=parent_trids,
force_propagate=force_propagate,
deal_boolean=deal_boolean
)
did_nothing = did_nothing_backwards
did_nothing_forwards = True
list_updated_output_trids = []
if not only_backwards:
did_nothing_forwards, list_updated_output_trids = propagate_forwards(
self, all_transforms, mapper, attributes,
transform_id,
previous_did_nothing=previous_did_nothing,
backup_comps=backup_comps,
parent_transform=parent_transform,
parent_trids=parent_trids,
force_propagate=force_propagate,
deal_boolean=deal_boolean
)
did_nothing = did_nothing_forwards
return did_nothing_forwards, list_updated_output_trids, \
did_nothing_backwards, list_updated_input_trids
[docs]
def propagate_backwards(
self, all_transforms, mapper, attributes,
transform_id,
next_did_nothing=False,
backup_comps=None,
parent_transform=None,
parent_trids=None,
force_propagate=False,
deal_boolean=None
):
backup_comps = {} if backup_comps is None else backup_comps
did_nothing = True
transf_mapper = mapper[transform_id]
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Backward propagation of {attribute2print} for {transform_id}: Before successors")
for trid in mapper[transform_id]["outputs"]:
if attribute2print in mapper[transform_id]["outputs"][trid]:
debug(mapper[transform_id]["outputs"]
[trid][attribute2print])
debug("\n\n\n")
# Reformat deal_boolean if not provided
if deal_boolean is None:
deal_boolean = [None for attr in attributes]
# If parent_transform, loop only on output trid with corresponding successors
list_trid_outputs = copy.deepcopy(list(transf_mapper["outputs"].keys()))
successors = transf_mapper["successors"]
if parent_transform is not None:
list_trid_outputs = [
trid for trid in transf_mapper["outputs"].keys()
if parent_transform in successors[trid]
]
# Propagate from successors
did_nothing_successor = True
list_updated_trid = []
for trid in list_trid_outputs:
# Skip here if parent_transform is not in successors
if parent_transform is not None:
if parent_transform not in successors[trid]:
continue
# If attributes already in outputs, skip
intersection = set(
transf_mapper["outputs"][trid].keys()).intersection(attributes)
if len(intersection) == len(attributes) \
and all([b is None for b in deal_boolean]):
continue
for tr in successors[trid]:
# Propagate only from parent transform
if tr != parent_transform and parent_transform is not None:
continue
# Propagate only parent trid if provided
if parent_trids is not None:
if trid not in parent_trids:
continue
# If not all attributes available in successor input, skip
in_inputs = [
attr not in mapper[tr]["inputs"][trid]
for attr in attributes
]
if any(in_inputs):
continue
# If only one successor, just propagate
ref_successor = transf_mapper["outputs"][trid].get(
f"{attributes[0]}_from_successor", transform_id)
if ref_successor == transform_id \
or ref_successor not in all_transforms.attributes:
for attr in attributes:
transf_mapper["outputs"][trid][attr] = mapper[tr]["inputs"][trid][attr]
transf_mapper["outputs"][trid][f"{attr}_from_successor"] = tr
did_nothing = False
did_nothing_successor = False
list_updated_trid.append(trid)
continue
# If multiple successors, check compatibility
different_input_output = [
not compare_attribute(
mapper[tr]["inputs"][trid][attr],
transf_mapper["outputs"][trid][attr],
deal_boolean=db
)
for attr, db in zip(attributes, deal_boolean)
]
# If not all values from successors are the same,
# propagate conflict
if any(different_input_output):
warning(
f"Could not propagate {attributes} backward from "
f"{tr} to {transform_id}, due to conflict with {ref_successor} "
f"being already used."
)
did_nothing = False
list_updated_trid.append(trid)
for attr in attributes:
transf_mapper["outputs"][trid][attr] = None
transf_mapper["outputs"][trid][f"{attr}_conflict_propagation"] = True
# From here, inputs and outputs are comparable
# with special treatment if any boolean
for db, attr in zip(deal_boolean, attributes):
if db is None:
continue
boolean_value = compare_attribute(
mapper[tr]["inputs"][trid][attr],
transf_mapper["outputs"][trid][attr],
deal_boolean=db,
return_boolean_value=True
)
transf_mapper["outputs"][trid][attr] = boolean_value
transf_mapper["outputs"][trid][f"{attr}_from_successor"] = tr
list_updated_trid.append(trid)
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Backward propagation of {attribute2print} for {transform_id}: After successors")
for trid in mapper[transform_id]["outputs"]:
if attribute2print in mapper[transform_id]["outputs"][trid]:
trid_dict = mapper[transform_id]["outputs"][trid]
debug(
f"\t- {trid}: {trid_dict[attribute2print]} / from_successor: {trid_dict.get(f'{attribute2print}_from_successor', 'no')}"
)
debug("\n\n\n")
# When parent_transform is not None,
# Only propagate output attributes to relevant inputs
list_trid_inputs = copy.deepcopy(list(transf_mapper["inputs"].keys()))
if parent_transform is not None:
list_trid_inputs = [
[
in_trid
for in_trid in transf_mapper.get("outputs2inputs", {}).get(out_trid, [out_trid])
if in_trid in transf_mapper["inputs"]
]
for out_trid in list_updated_trid
]
list_trid_inputs = list(itertools.chain(*list_trid_inputs))
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Backward propagation of {attribute2print} for {transform_id}: Before outputs")
for trid in mapper[transform_id]["inputs"]:
if attribute2print in mapper[transform_id]["inputs"][trid]:
debug(mapper[transform_id]["inputs"]
[trid][attribute2print])
debug("\n\n\n")
# Propagate outputs to inputs
did_nothing_inputs = True
list_updated_input_trids = []
for trid in set(list_trid_inputs):
# If already in inputs, skip
intersection = set(
transf_mapper["inputs"][trid].keys()).intersection(attributes)
if len(intersection) == len(attributes)\
and all([b is None for b in deal_boolean]):
anyNone = [transf_mapper["inputs"][trid][attr] is None
for attr in attributes]
if sum(anyNone) == 0:
continue
# Skip if value in input is not the default value
if not all([b is None for b in deal_boolean]) and len(intersection) == len(attributes):
default_values = [
transf_mapper["inputs"][trid].get(f"{attr}_default", False)
for attr in attributes
]
if not all(default_values):
continue
# Check whether trid is linked to some outputs
inputs2outputs = transf_mapper.get("inputs2outputs", {})[trid]
output_attributes = [
[
transf_mapper["outputs"][trid_out][attr]
if attr in transf_mapper["outputs"][trid_out]
else None
for attr in attributes
]
for trid_out in inputs2outputs
if trid_out in transf_mapper["outputs"]
]
output_trid = [
[
trid_out
if attr in transf_mapper["outputs"][trid_out]
else None
for attr in attributes
]
for trid_out in inputs2outputs
if trid_out in transf_mapper["outputs"]
]
output_initialized = [
transf_mapper["outputs"][trid_out].get(
"all_successors_initialized", False)
or trid_out == trid
for trid_out in inputs2outputs
if trid_out in transf_mapper["outputs"]
]
# Complement the list to merge if already in inputs
# Only for boolean
special_boolean_case = False
if not all([b is None for b in deal_boolean]) \
and len(intersection) == len(attributes):
special_boolean_case = True
output_attributes += [[
transf_mapper["inputs"][trid][attr]
for attr in attributes
]]
output_trid.append([trid])
skip, to_propagate, ind_to_propagate, anyNone = skip_attributes(
output_attributes, output_initialized,
force_propagate=self.force_propagate_attributes or force_propagate,
deal_boolean=deal_boolean
)
if skip:
# If outputs are not compatible,
# check if transform has special way to deal with it
getattr(all_transforms, transform_id).propagate_incompatible(
transf_mapper, attributes, trid, anyNone,
mode="backwards"
)
continue
# If ${attribute}_from_previous = True,
# then forbid backwards propagation
from_previous_out = [
transf_mapper["outputs"].get(output_trid[ind_to_propagate][k], {}).get(
attr + "_from_previous", False)
for k, attr in enumerate(attributes)
]
from_previous_in = [
transf_mapper["inputs"][trid].get(
attr + "_from_previous", False)
for attr in attributes
]
if any(from_previous_out) or any(from_previous_in):
continue
# In boolean case when merging with inputs as well,
# skip if using input value
if special_boolean_case:
if len(output_attributes) == 1:
continue
# If several values, merge them
for k, (db, attr) in enumerate(zip(deal_boolean, attributes)):
if db is None:
continue
boolean_value = copy.deepcopy(output_attributes[0][k])
for out_attr in output_attributes:
boolean_value = compare_attribute(
boolean_value,
out_attr[k],
deal_boolean=db,
return_boolean_value=True
)
if boolean_value != transf_mapper["inputs"][trid][attr]:
transf_mapper["inputs"][trid][attr] = boolean_value
transf_mapper["inputs"][trid][f"{attr}_default"] = False
list_updated_input_trids.append(trid)
did_nothing = False
did_nothing_inputs = False
continue
list_updated_input_trids.append(trid)
did_nothing = False
did_nothing_inputs = False
for value, attribute in zip(to_propagate, attributes):
transf_mapper["inputs"][trid][attribute] = value
transf_mapper["inputs"][trid][f"{attribute}_default"] = False
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Backward propagation of {attribute2print} for {transform_id}: After outputs")
for trid in mapper[transform_id]["inputs"]:
if attribute2print in mapper[transform_id]["inputs"][trid]:
debug(mapper[transform_id]["inputs"]
[trid][attribute2print])
debug("\n\n\n")
list_updated_input_trids = list(set(list_updated_input_trids))
if parent_transform is not None:
return did_nothing_inputs, list_updated_input_trids
else:
return did_nothing, list_updated_input_trids
[docs]
def propagate_forwards(
self, all_transforms, mapper, attributes,
transform_id,
previous_did_nothing=False,
backup_comps=None,
parent_transform=None,
parent_trids=None,
force_propagate=False,
deal_boolean=None
):
backup_comps = {} if backup_comps is None else backup_comps
did_nothing = True
transf_mapper = mapper[transform_id]
transform = getattr(all_transforms, transform_id)
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Forward propagation of {attribute2print} for {transform_id}: Before precursors")
for trid in mapper[transform_id]["inputs"]:
if attribute2print in mapper[transform_id]["inputs"][trid]:
trid_dict = mapper[transform_id]["inputs"][trid]
debug(
f"\t- {trid}: {trid_dict[attribute2print]} / from_successor: {trid_dict.get(f'{attribute2print}_from_successor', 'no')}"
)
debug("\n\n\n")
# Loop on inputs
# Limit the loop to precursors if provided
precursors = transf_mapper["precursors"]
list_inputs = copy.deepcopy(list(transf_mapper["inputs"].keys()))
if parent_transform is not None:
list_inputs = [
trid
for trid in transf_mapper["inputs"]
if parent_transform in precursors[trid]
]
# Propagate from precursors
list_updated_trid = []
for trid in list_inputs:
# If attribute already in inputs and not None, skip
intersection = set(
transf_mapper["inputs"][trid].keys()).intersection(attributes)
if len(intersection) == len(attributes):
if not any(
transf_mapper["inputs"][trid][attr] is None
for attr in attributes
):
continue
did_nothing_precursor = True
for tr in precursors[trid]:
# Propagate only from parent transform
if tr != parent_transform and parent_transform is not None:
continue
# If not all attributes available in precursor output, skip
not_in_outputs = [
attr not in mapper[tr]["outputs"][trid]
for attr in attributes
]
if any(not_in_outputs):
continue
# If only one precursor, just propagate
# Or if already from_precursor, just replace
ref_precursor = transf_mapper["inputs"][trid].get(
f"{attributes[0]}_from_precursor", transform_id)
if ref_precursor == transform_id \
or ref_precursor == tr \
or ref_precursor not in all_transforms.attributes:
for attr in attributes:
if (
transf_mapper["inputs"][trid].get(attr, None)
==
mapper[tr]["outputs"][trid][attr]
):
continue
if transf_mapper["inputs"][trid].get(attr, None) is not None:
continue
transf_mapper["inputs"][trid][attr] = mapper[tr]["outputs"][trid][attr]
transf_mapper["inputs"][trid][f"{attr}_from_precursor"] = tr
did_nothing = False
list_updated_trid.append(trid)
continue
same_input_output = [
compare_attribute(
mapper[tr]["outputs"][trid][attr],
transf_mapper["inputs"][trid][attr]
)
for attr in attributes
]
if sum(same_input_output) == len(attributes):
continue
warning(
f"Could not propagate {attributes} forward from "
f"{tr} to {transform_id}, due to conflict with {ref_precursor} "
f"being already used."
)
did_nothing = False
list_updated_trid.append(trid)
for attr in attributes:
transf_mapper["inputs"][trid][attr] = None
transf_mapper["inputs"][trid][
f"{attr}_conflict_propagation"] = True
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Forward propagation of {attribute2print} for {transform_id}: After precursors")
for trid in mapper[transform_id]["inputs"]:
if attribute2print in mapper[transform_id]["inputs"][trid]:
trid_dict = mapper[transform_id]["inputs"][trid]
debug(
f"\t- {trid}: {trid_dict[attribute2print]} / from_successor: {trid_dict.get(f'{attribute2print}_from_successor', 'no')}"
)
debug("\n\n\n")
# When parent_transform is not None,
# Only propagate output attributes to relevant inputs
list_trid_outputs = copy.deepcopy(list(transf_mapper["outputs"].keys()))
if parent_transform is not None:
list_trid_outputs = [
[
out_trid
for out_trid in transf_mapper.get("inputs2outputs", {}).get(in_trid, [in_trid])
if out_trid in transf_mapper["outputs"]
]
for in_trid in list_updated_trid
]
list_trid_outputs = list(itertools.chain(*list_trid_outputs))
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Forward propagation of {attribute2print} for {transform_id}: Before inputs")
for trid in mapper[transform_id]["outputs"]:
if attribute2print in mapper[transform_id]["outputs"][trid]:
trid_dict = mapper[transform_id]["outputs"][trid]
debug(
f"\t- {trid}: {trid_dict[attribute2print]} / from_successor: {trid_dict.get(f'{attribute2print}_from_successor', 'no')}"
)
debug("\n\n\n")
# Propagate inputs to outputs
did_nothing_inputs = True
list_updated_output_trids = []
for trid in set(list_trid_outputs):
# If all already in outputs, skip
intersection = set(
transf_mapper["outputs"][trid].keys()).intersection(attributes)
if len(intersection) == len(attributes):
if not any(
transf_mapper["outputs"][trid][attr] is None
for attr in attributes
):
continue
# Check if output trid is linked to some inputs
outputs2inputs = transf_mapper.get("outputs2inputs", {})[trid]
input_attributes = [
[
transf_mapper["inputs"][trid_in].get(attr, None)
for attr in attributes
]
for trid_in in outputs2inputs
if trid_in in transf_mapper["inputs"]
]
skip, to_propagate, ind_to_propagate, any_none = skip_attributes(
input_attributes,
force_propagate=self.force_propagate_attributes or force_propagate
)
if skip:
continue
did_nothing = False
did_nothing_inputs = False
list_updated_output_trids.append(trid)
for value, attribute in zip(to_propagate, attributes):
transf_mapper["outputs"][trid][attribute] = value
# Print debug if need
if transform_id in transfor2print:
if attribute2print in attributes:
debug(
f"Forward propagation of {attribute2print} for {transform_id}: After inputs")
for trid in mapper[transform_id]["outputs"]:
if attribute2print in mapper[transform_id]["outputs"][trid]:
trid_dict = mapper[transform_id]["outputs"][trid]
debug(
f"\t- {trid}: {trid_dict[attribute2print]} / from_successor: {trid_dict.get(f'{attribute2print}_from_successor', 'no')}"
)
debug("\n\n\n")
list_updated_output_trids = list(set(list_updated_output_trids))
if parent_transform is not None:
return did_nothing_inputs, list_updated_output_trids
else:
return did_nothing, list_updated_output_trids
[docs]
def skip_attributes(
list_attributes,
list_initialized=None,
force_propagate=False,
deal_boolean=None
):
"""Check compatibility of attributes and determines whether to skip
Args:
list_attributes (_type_): _description_
list_initialized (_type_): _description_
force_propagate (bool, optional): _description_. Defaults to False.
Returns:
bool, list, bool:
- List of attributes are not compatible
- List of attributes to propagate
- True if all Nones
"""
# If empty, skip
if len(list_attributes) == 0:
return True, [], None, True
# Check that all successors are fully initialized
if list_initialized is not None:
if not all(list_initialized):
return True, [], None, True
# Initiate deal_boolean with all Nones if not provided
if deal_boolean is None:
deal_boolean = [None for a in list_attributes[0]]
# Check for the presence of None when no indication on initialized
list_none = list(
itertools.chain(
*[
[attr is None
for attr in lattrs]
for lattrs in list_attributes
]
)
)
any_none = any(list_none)
all_none = all(list_none)
if (any_none and not force_propagate and list_initialized is None) or all_none:
return True, [], None, any_none
# If length is one and not default, skip if None, otherwise do not skip
if len(list_attributes) == 1:
if all(
lattrs is None
for lattrs in list_attributes[0]
):
return True, [], None, any_none
else:
return False, list_attributes[0], 0, any_none
# Check that all the same
all_same = all(
all(
compare_attribute(
l, list_attributes[0][k],
deal_boolean=deal_boolean[k]
) and l is not None
for k, l in enumerate(lattrs)
)
for lattrs in list_attributes
)
if all_same:
return False, list_attributes[0], 0, any_none
# Stop here if not force_propagate
if not force_propagate and list_initialized is None:
return True, [], None, any_none
# Check that all not None values are the same if force_propagate
ind_ref = [
ind
for ind, lattributes in enumerate(list_attributes)
if not any(l is None for l in lattributes) and len(lattributes) > 0
][0]
all_same = all([
all([
compare_attribute(
l, list_attributes[ind_ref][k],
deal_boolean=deal_boolean[k]
)
for k, l in enumerate(lattrs)
if l is not None and list_attributes[ind_ref][k] is not None
])
for lattrs in list_attributes
])
if all_same:
to_propagate = [
[l for l in lattributes if l is not None]
for lattributes in list_attributes
]
ind_to_propagate = [
ind
for ind, lattributes in enumerate(to_propagate)
if not any(l is None for l in lattributes) and len(lattributes) > 0
]
to_propagate = [
lattributes
for lattributes in to_propagate
if not any(l is None for l in lattributes) and len(lattributes) > 0
]
if len(to_propagate) > 0:
to_propagate = to_propagate[0]
ind_to_propagate = ind_to_propagate[0]
return False, to_propagate, ind_to_propagate, any_none
# Even in force_propagate, don't propagate if ambiguity
return True, [], None, any_none
[docs]
def compare_attribute(
attr1, attr2,
deal_boolean=None,
return_boolean_value=False
):
# Start with special treatment for boolean
if deal_boolean is not None:
if deal_boolean not in ["or", "and"]:
return ValueError(
f"{deal_boolean}is not recognized. "
"Only 'or' is implemented so far"
)
# If does not return the value,
# Just returns True to indicate that the elements are comparable
if not return_boolean_value:
return True
# Deal with Nones if any
if attr1 is None:
return attr2
elif attr2 is None:
return attr1
# Now deal with the different cases
if type(attr1) == type(attr2) == bool:
# Either returns value,
# or indicates that elements are comparable
if return_boolean_value:
if deal_boolean == "or":
return attr1 or attr2
else:
return attr1 and attr2
else:
return True
# Now deal with dictionaries
# If only one of the two is a dictionay, expands the other
if isinstance(attr1, dict) and type(attr2) == bool:
attr2 = {
k: attr2 for k in attr1
}
if isinstance(attr2, dict) and type(attr1) == bool:
attr1 = {
k: attr1 for k in attr2
}
return {
k: attr1.get(k, False) or attr2.get(k, False)
for k in set(list(attr1.keys()) + list(attr2.keys()))
}
# Now deal with regular cases
if type(attr1) != type(attr2):
return False
elif isinstance(attr1, dict):
if attr1.keys() != attr2.keys():
return False
same_values = all(
compare_attribute(attr1[key], attr2[key])
for key in attr1.keys()
)
return same_values
elif isinstance(attr1, (list, tuple, np.ndarray)):
if len(attr1) != len(attr2):
return False
return np.all(np.asarray(attr1) == np.asarray(attr2))
elif isinstance(attr1, pd.DataFrame):
return attr1.equals(attr2)
# For other types, compare directly
return attr1 == attr2
