Source code for yggdrasil.communication.transforms.MapFieldsTransform
import copy
import numpy as np
from yggdrasil.communication.transforms.TransformBase import TransformBase
[docs]class MapFieldsTransform(TransformBase):
r"""Class for mapping a subset of the original fields in an object to
a different set of fields. Fields that don't exist in the map are preserved
unchanged.
Args:
map (dict): A mapping from original field name to new field names.
"""
_transformtype = 'map_fields'
_schema_required = ['map']
_schema_properties = {'map': {'type': 'object',
'additionalProperties': {'type': 'string'}}}
_schema_subtype_description = "Change the names of fields in a message"
[docs] def transform_datatype(self, datatype):
r"""Determine the datatype that will result from applying the transform
to the supplied datatype.
Args:
datatype (dict): Datatype to transform.
Returns:
dict: Transformed datatype.
"""
if (((datatype.get('type', None) == 'array')
and isinstance(datatype.get('items', None), list))):
datatype = copy.deepcopy(datatype)
for i, x in enumerate(datatype['items']):
if x.get('title', 'f%d' % i) in self.map:
x['title'] = self.map[x['title']]
elif datatype.get('type', None) == 'object':
datatype = copy.deepcopy(datatype)
for kold, knew in self.map.items():
datatype['properties'][knew] = datatype['properties'].pop(kold)
return datatype
[docs] def evaluate_transform(self, x, no_copy=False):
r"""Call transform on the provided message.
Args:
x (object): Message object to transform.
no_copy (bool, optional): If True, the transformation occurs in
place. Otherwise a copy is created and transformed. Defaults
to False.
Returns:
object: The transformed message.
"""
out = x
if isinstance(x, dict):
if not no_copy:
out = copy.deepcopy(x)
for kold, knew in self.map.items():
out[knew] = out.pop(kold)
elif isinstance(x, (list, tuple)):
pass
elif isinstance(x, np.ndarray):
if not no_copy:
out = copy.deepcopy(x)
new_names = list(x.dtype.names)
for kold, knew in self.map.items():
new_names[new_names.index(kold)] = knew
out.dtype.names = new_names
else:
raise TypeError("Cannot map fields from object of type '%s'" % type(x))
return out
[docs] @classmethod
def get_testing_options(cls, **kwargs):
r"""Get testing options for the transform class.
Returns:
list: Multiple dictionaries of keywords and messages before/after
pairs that will result from the transform created by the provided
keywords.
"""
return [{'kwargs': {'map': {'a': 'aa', 'c': 'cc'}},
'in/out': [(dict(zip('abc', range(3))),
{'aa': 0, 'b': 1, 'cc': 2})],
'in/out_t': [
({'type': 'object',
'properties': {
x: {'type': 'int'}
for x in 'abc'}},
{'type': 'object',
'properties': {
x: {'type': 'int'}
for x in ['aa', 'b', 'cc']}})]},
{'kwargs': {'map': {'a': 'aa', 'c': 'cc'}},
'in/out': [([0, 1, 2], [0, 1, 2])],
'in/out_t': [
({'type': 'array',
'items': [{'type': 'int', 'title': x}
for x in 'abc']},
{'type': 'array',
'items': [{'type': 'int', 'title': x}
for x in ['aa', 'b', 'cc']]})]},
{'kwargs': {'map': {'a': 'aa', 'c': 'cc'}},
'in/out': [(np.zeros(3, np.dtype({'names': ['a', 'b', 'c'],
'formats': ['i4', 'i4', 'i4']})),
np.zeros(3, np.dtype({'names': ['aa', 'b', 'cc'],
'formats': ['i4', 'i4', 'i4']})))]},
{'kwargs': {'map': {'a': 'aa', 'c': 'cc'}},
'in/out': [(None, TypeError)]}]