import copy
import numpy as np
import tracemalloc
from ......utils.classes.setup import Setup
from ......utils.classes.transforms import Transform
from ......utils.mappers import safe_deepcopy
from logging import debug
from . import connect_pipes
from . import propagate_attributes
from . import init_default_transformations
from ......utils.check.errclass import CifTypeError
[docs]
def add_default(
self,
transforms,
yml_dict,
position="last",
index=0,
init=False,
mapper={},
transform_type="state",
backup_comps={},
ref_transform="",
precursor=None,
successor=None,
transform_id=None,
do_pipe_entry=False
):
# Check usage of precursor and successor
assert (
(precursor is None or type(precursor) == dict)
and (successor is None or type(successor) == dict)
)
debug("Adding the following transform:")
debug(yml_dict)
# Initialize 'default_index' is necessary
if not hasattr(transforms, 'default_index'):
transforms.default_index = 0
# Create a random name to identify the current transformation
if transform_id is not None:
new_id = transform_id
elif 'plugin' in yml_dict and 'name' in yml_dict['plugin']:
plg = yml_dict['plugin']
new_id = f"{plg['name']}_{plg.get('version', 'std')}_{transforms.default_index:05d}"
else:
new_id = f"default_{transforms.default_index:05d}"
transforms.default_index += 1
new_transf = Setup.from_dict({new_id: yml_dict})
Setup.load_setup(new_transf, level=1)
new_transf = getattr(new_transf, new_id)
# Attach transform id to the transform itself for self identification
new_transf.transform_id = new_id
# Update overall transform pipe with current transform
setattr(transforms, new_id, new_transf)
# Initializes mapper
if init:
transf_mapper_loc = new_transf.ini_mapper(
backup_comps=backup_comps,
transforms_order=transforms.attributes,
ref_transform=ref_transform,
precursor=precursor,
successor=successor,
general_mapper=mapper,
all_transforms=transforms,
transform_name=new_id
)
# Clean input dates to make sure they are of proper format
transf_mapper_loc = new_transf.clean_input_dates(transf_mapper_loc)
# Deep-copy the mapper to avoid issues with references
transf_mapper_loc = safe_deepcopy(transf_mapper_loc)
new_transf.mapper = transf_mapper_loc
mapper[new_id] = transf_mapper_loc
update_successors_precursors(
new_id,
precursor,
successor,
transf_mapper_loc,
mapper
)
generate_internal_pipe(new_id, mapper)
if position in ["last", "end"]:
transforms.attributes.append(new_id)
elif position in ["start", "first"]:
transforms.attributes.insert(0, new_id)
elif position == "index":
transforms.attributes.insert(index, new_id)
# Check memory if requested
if self.monitor_memory:
current, peak = tracemalloc.get_traced_memory()
debug(f"Current memory usage is {current / 1024 ** 2}MB; "
f"Peak was {peak / 1024 ** 2}MB")
# Connecting new transformation to the rest of the pipeline
connect_pipes.connect_pipes(transforms, mapper, new_id)
# Propagate attributes backward and forward
propagate_attributes.propagate_attributes(
self, transforms, mapper, new_id,
backup_comps=backup_comps
)
# Initialize pipe_entry if no precursor
if do_pipe_entry:
init_pipe_entry(
self, transforms,
backup_comps, mapper,
new_id)
# Now initialize default transformations
# with precursors and successors if any mismatch
init_default_transformations.init_default_transformations(
self, transforms,
backup_comps, mapper,
new_id,
do_pipe_entry=do_pipe_entry
)
return new_transf, new_id
[docs]
def init_pipe_entry(
self, all_transforms,
backup_comps, mapper,
transform
):
debug(f"Initializing entry point for {transform}")
# Initialize required input variables
self.required_inputs = getattr(self, "required_inputs", {})
self.missing = getattr(self, "missing", {})
self.init_missing = getattr(self, "init_missing", {})
# Save required inputs and missing inputs
ignore_missing = hasattr(self, "init_inputs")
init_inputs = getattr(self, "init_inputs", None)
# Add default transformation "fromcontrol" when no predecessor is available
# Also include "unit_conversion" if required in configuration
transf_mapper = mapper[transform]
precursors = transf_mapper["precursors"]
for trid in precursors:
# Skip if there are precursors
if precursors[trid]:
continue
prm = trid[1]
cmp = trid[0]
# Save required inputs for debug dumping
if cmp not in self.required_inputs:
self.required_inputs[cmp] = {}
self.required_inputs[cmp][prm] = {"available": True}
# Fetch component from backup if necessary
# It is used to replace components names
components = self.datavect.components
comps = components.attributes
cmp_in = cmp if cmp in comps else backup_comps.get(cmp, None)
if cmp_in is None or not hasattr(components, cmp_in):
self.required_inputs[cmp][prm] = {"available": False}
if transform not in self.missing:
self.missing[transform] = {}
if cmp not in self.missing[transform]:
self.missing[transform][cmp] = []
self.missing[transform][cmp].append(prm)
# Check if missing parameter is needed for initialization
if ignore_missing:
if cmp in init_inputs.components.attributes:
list_params = getattr(init_inputs.components, cmp)
if list_params == []:
missing_params = self.missing[transform][cmp]
else:
missing_params = [
prm for prm in self.missing[transform][cmp]
if prm in list_params
]
if missing_params != []:
if transform not in self.init_missing:
self.init_missing[transform] = {}
self.init_missing[transform][cmp] = missing_params
continue
cmp_plg = getattr(components, cmp_in)
# Skip if component is observation and not data
if getattr(cmp_plg, "isobs", False):
continue
# Fetch parameters
# If no parameters, handle the component as a whole
if not hasattr(cmp_plg, "parameters"):
params = cmp_plg
parameters = [""]
else:
params = cmp_plg.parameters
parameters = params.attributes[:]
param = getattr(params, prm, cmp_plg)
if not hasattr(params, prm):
self.required_inputs[cmp][prm]["from_component"] = True
# TODO: split from control into from control and from datavect
# control variables are initialized with from control and if need
# to read data, from datavect should be added as precursor
ind_transform = all_transforms.attributes.index(transform)
yml_dict = {
"plugin": {
"name": "fromcontrol",
"version": "std",
"type": "transform",
"newplg": True,
},
"component": [cmp],
"parameter": [prm],
"orig_parameter_plg": param,
"orig_component_plg": cmp_plg,
}
ref_successor = {(cmp, prm): transform}
new_transf, fromcontrol_id = add_default(
self,
all_transforms,
yml_dict,
position="index",
index=ind_transform,
mapper=mapper,
init=True,
backup_comps=backup_comps,
# successor=ref_successor,
do_pipe_entry=True
)
# Rescaling if any
if hasattr(param, "unit_conversion"):
unit_conv = getattr(param, "unit_conversion")
yml_dict = {
"plugin": {
"name": "unit_conversion",
"version": "std",
"type": "transform",
},
"component": [cmp],
"parameter": [prm],
"orig_parameter_plg": param,
"orig_component_plg": cmp_plg,
**{attr: getattr(unit_conv, attr)
for attr in getattr(unit_conv, "attributes", []) if
attr != "plugin"}
}
ref_precursor = {(cmp, prm): fromcontrol_id}
ref_successor = mapper[fromcontrol_id]["successors"]
new_transf, new_id = add_default(
self,
all_transforms,
yml_dict,
position="index",
index=ind_transform + 1,
mapper=mapper,
init=True,
backup_comps=backup_comps,
precursor=ref_precursor,
successor=ref_successor
)
[docs]
def update_successors_precursors(
new_id,
precursors2add,
successor2add,
transf_mapper_loc,
mapper
):
# Force precursors
if "precursors" not in transf_mapper_loc:
transf_mapper_loc["precursors"] = {}
for trid in transf_mapper_loc["inputs"]:
if trid not in transf_mapper_loc["precursors"]:
transf_mapper_loc["precursors"][trid] = []
if precursors2add is None:
continue
if trid not in precursors2add:
continue
if type(precursors2add[trid]) == str:
transf_mapper_loc["precursors"][trid].append(
precursors2add[trid]
)
elif type(precursors2add[trid]) == list:
transf_mapper_loc["precursors"][trid].extend(
precursors2add[trid]
)
else:
raise CifTypeError(
f"Unexpexted type ({type(precursors2add[trid])} "
f"for precursors {precursors2add[trid]}) "
f"for transform {new_id}"
)
# Force successors
if "successors" not in transf_mapper_loc:
transf_mapper_loc["successors"] = {}
for trid in transf_mapper_loc["outputs"]:
if trid not in transf_mapper_loc["successors"]:
transf_mapper_loc["successors"][trid] = []
if successor2add is None:
continue
if trid not in successor2add:
continue
if type(successor2add[trid]) == str:
transf_mapper_loc["successors"][trid].append(
successor2add[trid]
)
elif type(successor2add[trid]) == list:
transf_mapper_loc["successors"][trid].extend(
successor2add[trid]
)
else:
raise CifTypeError(
f"Unexpexted type ({type(successor2add[trid])} "
f"for precursors {successor2add[trid]}) "
f"for transform {new_id}"
)
# If both precursors and successors are specified, hence update corresponding pipe
if successor2add is None or precursors2add is None:
return
for trid in successor2add:
# if trid not in both precursor and successor, can't force pipe
if trid not in precursors2add:
continue
# Loop over precursors and successors
successor_trid = [successor2add[trid]] if type(successor2add[trid]) == str \
else successor2add[trid]
precursor_trid = [precursors2add[trid]] if type(precursors2add[trid]) == str \
else precursors2add[trid]
for successor_tmp, precursor_tmp in zip(successor_trid, precursor_trid):
# if there was no direct pipeline, can't force it
# TODO: check whether this can reveal an error?
if trid not in mapper[precursor_tmp]["successors"]:
continue
if trid not in mapper[successor_tmp]["precursors"]:
continue
if successor_tmp not in mapper[precursor_tmp]["successors"][trid]:
continue
if precursor_tmp not in mapper[successor_tmp]["precursors"][trid]:
continue
mapper[precursor_tmp]["successors"][trid].remove(
successor_tmp)
mapper[precursor_tmp]["successors"][trid].append(new_id)
mapper[successor_tmp]["precursors"][trid].remove(
precursor_tmp)
mapper[successor_tmp]["precursors"][trid].append(new_id)
[docs]
def generate_internal_pipe(new_id, mapper):
transf_mapper = mapper[new_id]
# Generate the attribute outputs2inputs if not provided
# Then all inputs are supposed to be linked to all outputs
if "outputs2inputs" not in transf_mapper:
transf_mapper["outputs2inputs"] = {
trout: copy.deepcopy(list(transf_mapper["inputs"].keys()))
for trout in transf_mapper["outputs"]
}
# Now generate the corresponding path forwards
transf_mapper["inputs2outputs"] = \
Transform.generate_inputs2outputs(transf_mapper)
