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
):
"""Instantiate a new transform from a YAML-like config and wire it in.
Creates and registers a transform described by ``yml_dict`` (via
:class:`Setup`), gives it a unique id (either ``transform_id``, or a
name derived from its plugin name/version, or a generic
``default_{index}``), and inserts it into ``transforms.attributes``
at the requested ``position``. If ``init`` is True, also initializes
its mapper entry (:meth:`Transform.ini_mapper`), forces the
precursors/successors given in ``precursor``/``successor``
(:func:`update_successors_precursors`), and builds its internal
inputs/outputs paths (:func:`generate_internal_pipe`).
After creation, connects the new transform to the rest of the pipe
(:func:`connect_pipes.connect_pipes`), propagates attributes
backward/forward (:func:`propagate_attributes.propagate_attributes`),
optionally initializes its pipe entry when it has no precursor
(:func:`init_pipe_entry`), and finally lets
:func:`init_default_transformations.init_default_transformations`
insert any further default transforms needed to reconcile input/output
formats.
Args:
self: The parent object (obs operator), exposing ``monitor_memory``.
transforms: Namespace holding all registered transform instances,
mutated in place with the new transform.
yml_dict: Dictionary describing the transform to create, in the
same format as a transform entry in the YAML configuration
(must include a ``"plugin"`` block).
position: Where to insert the new transform id in
``transforms.attributes``: ``"last"``/``"end"``,
``"start"``/``"first"``, or ``"index"`` (uses ``index``).
index: Insertion index used when ``position == "index"``.
init: Whether to initialize the transform's mapper entry.
mapper: Dictionary mapping each transform id to its inputs/outputs/
precursors/successors metadata, updated with the new entry.
transform_type: Type passed through to the transform setup,
currently unused directly in this function.
backup_comps: Backup of components used to restore/compare state.
ref_transform: Id of a reference transform used by
:meth:`Transform.ini_mapper` to resolve relative attributes.
precursor: Optional ``{trid: precursor_id(s)}`` mapping forcing the
new transform's precursors for the given input trids.
successor: Optional ``{trid: successor_id(s)}`` mapping forcing the
new transform's successors for the given output trids.
transform_id: Optional explicit id to use instead of an
auto-generated one.
do_pipe_entry: Whether to initialize the pipe entry (default
"fromcontrol"/"unit_conversion" transforms) for the new
transform.
Returns:
tuple: ``(new_transf, new_id)``, the newly created transform
instance and its id in ``transforms``.
"""
# 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
):
"""Add default entry transforms for the inputs of ``transform`` with no precursor.
For every input trid of ``transform`` that has no precursor yet,
resolves the corresponding component/parameter in
``self.datavect.components`` and:
- If the component/parameter cannot be resolved, records it in
``self.missing`` (and, if it is required for initialization per
``self.init_inputs``, in ``self.init_missing`` too) so it can be
reported later, and skips inserting a transform for it.
- If the component is an observation (``isobs``), skips it (it is
not data to be initialized from control).
- Otherwise, inserts a ``fromcontrol`` transform as new precursor via
:func:`add_default`, plus a ``unit_conversion`` transform right
after it if the parameter defines a ``unit_conversion`` block.
Along the way, ``self.required_inputs`` is populated for every
encountered component/parameter for later debug dumping.
Args:
self: The obs operator, exposing/receiving ``required_inputs``,
``missing``, ``init_missing``, ``datavect`` and optionally
``init_inputs``.
all_transforms: Namespace holding all registered transform instances.
backup_comps: Backup of components, used to resolve component names
that were renamed/replaced.
mapper: Dictionary mapping each transform id to its inputs/outputs/
precursors metadata, updated with any inserted transform.
transform: Id of the transform whose entry (precursor-less) inputs
are being initialized.
"""
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 explicit precursors/successors on a newly created transform.
Ensures every input/output trid of the new transform has a
``"precursors"``/``"successors"`` list (creating empty ones if
absent), then appends the ids given in ``precursors2add``/
``successor2add`` (a string or list of strings per trid) to the
matching lists.
When both ``precursors2add`` and ``successor2add`` are provided and
share a trid whose precursor/successor pair was already directly
connected (i.e. the precursor already lists the successor, and vice
versa), that direct link is rewired through the new transform: the
new transform id replaces the old successor in the precursor's
``successors`` list, and replaces the old precursor in the
successor's ``precursors`` list.
Args:
new_id: Id of the newly created transform being spliced in.
precursors2add: Optional ``{trid: precursor_id(s)}`` mapping of
precursors to force for the new transform's inputs.
successor2add: Optional ``{trid: successor_id(s)}`` mapping of
successors to force for the new transform's outputs.
transf_mapper_loc: Mapper entry of the new transform, mutated in
place with the forced precursors/successors.
mapper: Dictionary mapping each transform id to its precursors/
successors metadata, mutated in place when rewiring direct
links through the new transform.
Raises:
CifTypeError: If a value in ``precursors2add`` or ``successor2add``
is neither a string nor a list.
"""
# 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):
"""Build the internal input/output linkage of a transform's mapper entry.
If not already present, sets ``outputs2inputs`` so that every output
trid is linked to all input trids (the default assumption that any
output may depend on any input). Then derives the reverse mapping
``inputs2outputs`` from it via
:meth:`Transform.generate_inputs2outputs`.
Args:
new_id: Id of the transform whose internal pipe is being generated.
mapper: Dictionary mapping each transform id to its mapper entry;
``mapper[new_id]`` is mutated in place with ``outputs2inputs``
and ``inputs2outputs``.
"""
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)
