import numpy as np
import pandas as pd
from . import add_default
from ......utils.check.errclass import CifError
[docs]
def init_sparse(
self,
trid,
precursor_dict, ref_dict,
tr, transform,
param,
all_transforms,
mapper,
backup_comps,
precursors,
do_pipe_entry=False
):
"""Insert the transform needed to reconcile sparse/sampled data formats.
Compares the data representation of a precursor's output
(``precursor_dict``) against the target input (``ref_dict``) for a
given tracer, and inserts, at most, a single transform to bridge the
first mismatch found, in this priority order:
1. ``sparse2sample`` if the precursor is sparse but the target expects
gridded data.
2. ``array2sampled`` if the precursor is gridded but the target expects
sampled/sparse data (raises :class:`CifError` if the target is
flagged ``sampled``, which should never occur here).
3. ``time_interpolation`` if the precursor's ``input_dates`` do not
match the target's.
4. ``vertical_interpolation`` if neither side has a continuous
vertical domain and the precursor is gridded.
5. ``regrid`` if neither side has a continuous horizontal domain and
the precursor is gridded.
Each inserted transform is created via :func:`add_default.add_default`
and spliced between the precursor (``tr``) and ``transform``. As soon
as one transform is inserted, the function returns immediately: the
caller re-invokes the recursive default-initialization logic to
handle any remaining mismatches for other dimensions.
Args:
self: The parent object (obs operator) holding the transform pipe.
trid: ``(component, parameter)`` tuple identifying the tracer.
precursor_dict: Attribute dictionary of the precursor's output
(``sparse_data``, ``sampled``, ``domain``, ``input_dates``,
``continuous_hdomain``, ``continuous_vdomain``, etc.).
ref_dict: Attribute dictionary of the target input to reconcile
``precursor_dict`` against.
tr: Id of the precursor transform providing the data.
transform: Id of the transform whose input is being checked.
param: Object holding optional ``time_interpolation``,
``vertical_interpolation`` and ``regrid`` configuration blocks
used to parametrize the inserted transform.
all_transforms: Namespace holding all registered transform instances.
mapper: Dictionary mapping each transform id to its inputs/outputs
metadata, updated when a new transform is inserted.
backup_comps: Backup of components used to restore/compare state.
precursors: Precursors metadata, kept for interface consistency
with related init functions.
do_pipe_entry: Whether to run the pipe-entry initialization step
for the newly inserted transform.
Returns:
str or None: The id of the transform now directly feeding
``transform`` for this trid when a bridging transform was
inserted. Returns ``None`` if none of the mismatch conditions
apply (no bridging transform is needed).
Raises:
CifError: If ``ref_dict`` is flagged as ``sampled`` while the
precursor is neither sampled nor sparse (should never happen).
"""
cmp, prm = trid
precursor_id = tr
successor_id = None
# If precursor is sparse,
# project sparse to array
if (
precursor_dict.get("sparse_data", False)
and not ref_dict.get("sparse_data", False)
and ref_dict.get("domain", None) is not None
):
precursor_id = tr
yml_dict = {
"plugin": {
"name": "sparse2sample",
"version": "std",
"type": "transform",
},
"component": cmp,
"parameter": prm,
"successor": transform,
"precursor": tr,
}
ref_precursor = {(cmp, prm): tr}
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,
precursor=ref_precursor,
successor=ref_successor,
do_pipe_entry=do_pipe_entry
)
precursor_id = new_id
# Stop here as recursive init_default
# will take care of other dimensions
return precursor_id
# If successor is sampled, should not happen
if ref_dict.get("sampled", False):
raise CifError("Sparse to sample should not happen")
# Sample from complete field to sparse data
# Do it only if successor_id is not None,
# which means that indexing has been computed
if (
not precursor_dict.get("sampled", False)
and not precursor_dict.get("sparse_data", False)
and ref_dict.get("sparse_data", False)
and precursor_dict.get("domain", None) is not None
):
precursor_id = tr
yml_dict = {
"plugin": {
"name": "array2sampled",
"version": "std",
"type": "transform",
},
"component": cmp,
"parameter": prm,
"successor": transform,
"precursor": tr,
}
ref_precursor = {(cmp, prm): tr}
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,
precursor=ref_precursor,
successor=ref_successor,
do_pipe_entry=do_pipe_entry
)
precursor_id = new_id
# Stop here as recursive init_default
# will take care of other dimensions
return precursor_id
# Reindex dates
precursor_dates = precursor_dict.get("input_dates", {})
if precursor_dates is None:
precursor_dates = {}
target_dates = ref_dict["input_dates"]
if target_dates is None:
target_dates = {}
different_items = \
[k not in target_dates
or not precursor_dates.get(k, pd.DataFrame()).reset_index(drop=True)
.equals(target_dates.get(k, pd.DataFrame()).reset_index(drop=True))
if len(precursor_dates.get(k, pd.DataFrame())) == len(target_dates.get(k, pd.DataFrame())) != 0
else len(precursor_dates.get(k, pd.DataFrame())) != 0 or len(target_dates.get(k, pd.DataFrame())) != 0
for k in precursor_dates]
sparse_in = precursor_dict.get("sparse_data", False)
sparse_out = ref_dict.get("sparse_data", False)
if (np.sum(different_items) > 0) \
and precursor_dates != {}:
# Temporal re-indexing if any
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,
"sparse_in": False,
"sampled_in": False,
"sparse_out": False,
"sampled_out": True,
**{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,
precursor=ref_precursor,
successor=ref_successor,
do_pipe_entry=do_pipe_entry
)
precursor_id = new_id
if successor_id is None:
successor_id = new_id
# Stop here as recursive init_default
# will take care of other dimensions
return precursor_id
# Vertical interpolation if needed
do_vinterp = (
not precursor_dict.get("sparse_data", False)
and not precursor_dict.get("continuous_vdomain", False)
and not ref_dict.get("continuous_vdomain", False)
and precursor_dict.get("domain", None) is not None
)
if do_vinterp:
vinterp = getattr(param, "vertical_interpolation", None)
yml_dict = {
"plugin": {
"name": "vertical_interpolation",
"version": "std",
"type": "transform",
},
"method": getattr(vinterp, "method", "static-levels"),
"component": cmp,
"parameter": prm,
"successor": transform,
"precursor": precursor_id,
"sparse_in": False,
"sampled_in": False,
"sparse_out": False,
"sampled_out": True,
**{attr: getattr(vinterp, attr)
for attr in getattr(vinterp, "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,
precursor=ref_precursor,
successor=ref_successor,
do_pipe_entry=do_pipe_entry
)
precursor_id = new_id
if successor_id is None:
successor_id = new_id
# Stop here as recursive init_default
# will take care of other dimensions
return precursor_id
# Reprojects if domain is different
do_regrid = (
not precursor_dict.get("sparse_data", False)
and not precursor_dict.get("continuous_hdomain", False)
and not ref_dict.get("continuous_hdomain", False)
and precursor_dict.get("domain", None) is not None
)
if do_regrid:
regrid = getattr(
param, "regrid",
getattr(ref_dict.get("tracer", None), "regrid", None))
yml_dict = {
"plugin": {
"name": "regrid",
"version": "std",
"type": "transform",
},
"method": getattr(
regrid, "method",
"gridcell"
),
"component": [cmp],
"parameter": [prm],
"successor": transform,
"precursor": precursor_id,
"sparse_in": False,
"sampled_in": False,
"sparse_out": False,
"sampled_out": True,
**{attr: getattr(regrid, attr)
for attr in getattr(regrid, "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,
precursor=ref_precursor,
successor=ref_successor,
do_pipe_entry=do_pipe_entry
)
precursor_id = new_id
if successor_id is None:
successor_id = new_id
# Stop here as recursive init_default
# will take care of other dimensions
return precursor_id
# # Project sparse data to array
# if precursor_dict.get("sparse_data", False) \
# and not ref_dict.get("sparse_data", False):
# raise Exception("Is it still used?")
# if cmp == "stratosphere":
# print("HHHHHHHHHHHHHHHHHHHHH")
# print(__file__)
# import code
# code.interact(local=dict(locals(), **globals()))
# yml_dict = {
# "plugin": {
# "name": "sparse2sample",
# "version": "std",
# "type": "transform",
# },
# "component": [cmp],
# "parameter": [prm],
# "successor": transform,
# "precursor": precursor_id,
# }
# 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,
# precursor=ref_precursor,
# successor=ref_successor,
# do_pipe_entry=do_pipe_entry
# )
# precursor_id = new_id
# print("FFFFFFFFFFFFFFFF", tr, transform)
# return precursor_id
