Source code for yggdrasil.metaschema.datatypes.ScalarMetaschemaType
import numpy as np
import copy
import warnings
import base64
from yggdrasil import units, constants
from yggdrasil.metaschema import data2dtype, definition2dtype
from yggdrasil.metaschema.datatypes.MetaschemaType import MetaschemaType
from yggdrasil.metaschema.datatypes.FixedMetaschemaType import (
create_fixed_type_class)
from yggdrasil.metaschema.properties import get_metaschema_property
[docs]class ScalarMetaschemaType(MetaschemaType):
r"""Type associated with a scalar.
Developer Notes:
|yggdrasil| defines scalars as an umbrella type encompassing int, uint,
float, bytes, and unicode.
"""
name = 'scalar'
description = 'A scalar value with or without units.'
properties = ['subtype', 'precision', 'units']
definition_properties = ['subtype']
metadata_properties = ['subtype', 'precision', 'units']
extract_properties = ['subtype', 'precision', 'units']
python_types = units.ALL_PYTHON_SCALARS_WITH_UNITS
[docs] @classmethod
def validate(cls, obj, raise_errors=False):
r"""Validate an object to check if it could be of this type.
Args:
obj (object): Object to validate.
raise_errors (bool, optional): If True, errors will be raised when
the object fails to be validated. Defaults to False.
Returns:
bool: True if the object could be of this type, False otherwise.
"""
if isinstance(obj, np.ndarray) and (obj.ndim == 0):
obj = obj.reshape((1, ))[0]
if super(ScalarMetaschemaType, cls).validate(units.get_data(obj),
raise_errors=raise_errors):
dtype = data2dtype(obj)
if cls.is_fixed and ('subtype' in cls.fixed_properties):
type_list = [
constants.VALID_TYPES[cls.fixed_properties['subtype']]]
else:
type_list = constants.NUMPY_TYPES
if dtype.name.startswith(tuple(type_list)):
return True
else:
if raise_errors:
raise ValueError(("dtype %s dosn't corresponding with any "
+ "of the accepted types: %s") %
(str(dtype), str(type_list)))
return False
# @classmethod
# def coerce_type(cls, obj, typedef=None, **kwargs):
# r"""Coerce objects of specific types to match the data type.
# Args:
# obj (object): Object to be coerced.
# typedef (dict, optional): Type defintion that object should be
# coerced to. Defaults to None.
# **kwargs: Additional keyword arguments are metadata entries that may
# aid in coercing the type.
# Returns:
# object: Coerced object.
# """
# if ((cls.is_fixed and ('subtype' in cls.fixed_properties)
# and (cls.fixed_properties['subtype'] in ['bytes', 'unicode']))):
# obj = cls.normalize(obj)
# return obj
[docs] @classmethod
def normalize(cls, obj):
r"""Normalize an object, if possible, to conform to this type.
Args:
obj (object): Object to normalize.
Returns:
object: Normalized object.
"""
if cls.is_fixed and ('subtype' in cls.fixed_properties):
if (cls.fixed_properties['subtype'] == 'bytes'):
if isinstance(obj, str):
obj = obj.encode("utf-8")
elif not isinstance(obj, bytes):
obj = str(obj).encode("utf-8")
elif (cls.fixed_properties['subtype'] == 'unicode'):
if isinstance(obj, bytes):
obj = obj.decode("utf-8")
else:
obj = str(obj)
else:
dtype = units.PYTHON_SCALARS_WITH_UNITS[
cls.fixed_properties['subtype']][0]
try:
obj = dtype(obj)
except (TypeError, ValueError):
pass
return obj
[docs] @classmethod
def encode_data(cls, obj, typedef):
r"""Encode an object's data.
Args:
obj (object): Object to encode.
typedef (dict): Type definition that should be used to encode the
object.
Returns:
string: Encoded object.
"""
arr = cls.to_array(obj)
out = base64.encodebytes(arr.tobytes()).decode('ascii')
return out
[docs] @classmethod
def encode_data_readable(cls, obj, typedef):
r"""Encode an object's data in a readable format that may not be
decoded in exactly the same way.
Args:
obj (object): Object to encode.
typedef (dict): Type definition that should be used to encode the
object.
Returns:
string: Encoded object.
"""
arr = cls.to_array(obj)
if isinstance(typedef, dict):
subtype = typedef.get('subtype', typedef.get('type', None))
else:
subtype_cls = get_metaschema_property('subtype')
subtype = subtype_cls.encode(obj)
if (cls.name in ['1darray', 'ndarray']):
return arr.tolist()
assert(arr.ndim > 0)
if subtype in ['int', 'uint']:
return int(arr[0])
elif subtype in ['float']:
return float(arr[0])
elif subtype in ['complex']:
return str(complex(arr[0]))
elif subtype in ['bytes', 'unicode']:
out = arr[0]
if isinstance(out, bytes):
out = out.decode("utf-8")
else:
out = str(out)
return out
else: # pragma: debug
warnings.warn(("No method for handling readable serialization of "
+ "subtype '%s', falling back to default.") % subtype)
return super(ScalarMetaschemaType, cls).encode_data_readable(obj, typedef)
[docs] @classmethod
def decode_data(cls, obj, typedef):
r"""Decode an object.
Args:
obj (string): Encoded object to decode.
typedef (dict): Type definition that should be used to decode the
object.
Returns:
object: Decoded object.
"""
bytes = base64.decodebytes(obj.encode('ascii'))
dtype = definition2dtype(typedef)
arr = np.frombuffer(bytes, dtype=dtype)
# arr = np.fromstring(bytes, dtype=dtype)
if 'shape' in typedef:
arr = arr.reshape(typedef['shape'])
out = cls.from_array(arr, unit_str=typedef.get('units', None),
dtype=dtype, typedef=typedef)
return out
[docs] @classmethod
def transform_type(cls, obj, typedef=None):
r"""Transform an object based on type info.
Args:
obj (object): Object to transform.
typedef (dict, optional): Type definition that should be used to
transform the object. Defaults to None.
Returns:
object: Transformed object.
"""
if typedef is None:
return obj
typedef0 = cls.encode_type(obj)
typedef1 = copy.deepcopy(typedef0)
typedef1.update(**typedef)
dtype = definition2dtype(typedef1)
arr = cls.to_array(obj).astype(dtype, casting='same_kind')
out = cls.from_array(arr, unit_str=typedef0.get('units', None),
dtype=dtype, typedef=typedef)
return units.convert_to(out, typedef1.get('units', None))
[docs] @classmethod
def to_array(cls, obj):
r"""Get np.array representation of the data.
Args:
obj (object): Object to get array for.
Returns:
np.ndarray: Array representation of object.
"""
obj_nounits = units.get_data(obj)
if isinstance(obj_nounits, np.ndarray):
arr = obj_nounits
else:
dtype = data2dtype(obj_nounits)
arr = np.array([obj_nounits], dtype=dtype)
return arr
[docs] @classmethod
def from_array(cls, arr, unit_str=None, dtype=None, typedef=None):
r"""Get object representation of the data.
Args:
arr (np.ndarray): Numpy array.
unit_str (str, optional): Units that should be added to returned
object.
dtype (np.dtype, optional): Numpy data type that should be maintained
as a base class when adding units. Defaults to None and is
determined from the object or typedef (if provided).
typedef (dict, optional): Type definition that should be used to
decode the object. Defaults to None and is determined from the
object or dtype (if provided).
Returns:
object: Object representation of the data in the input array.
"""
# if (typedef is None) and (dtype is not None):
# typedef = dtype2definition(dtype)
# elif (dtype is None) and (typedef is not None):
# dtype = definition2dtype(typedef)
if (cls.name not in ['1darray', 'ndarray']) and (arr.ndim > 0):
out = arr[0]
else:
out = arr
if typedef is not None:
# Cast numpy type to native python type if they are equivalent
out = cls.as_python_type(out, typedef)
if unit_str is not None:
if dtype is None:
dtype = data2dtype(out)
out = units.add_units(out, unit_str, dtype=dtype)
return out
[docs] @classmethod
def get_extract_properties(cls, metadata):
r"""Get the list of properties that should be kept when extracting a
typedef from message metadata.
Args:
metadata (dict): Metadata that typedef is being extracted from.
Returns:
list: Keywords that should be kept in the typedef.
"""
out = super(ScalarMetaschemaType, cls).get_extract_properties(metadata)
dtype = metadata.get('subtype', metadata['type'])
if (((dtype in constants.FLEXIBLE_TYPES)
and (metadata['type'] not in ['1darray', 'ndarray'])
and (not metadata.get('fixed_precision', False)))):
out.remove('precision')
if units.is_null_unit(metadata.get('units', '')):
out.remove('units')
return out
[docs] @classmethod
def as_python_type(cls, obj, typedef):
r"""Convert a possible numpy type into a native Python type if possible.
Args:
obj (object): Object to convert.
typedef (dict): Type definition for the object.
Returns:
object: Native Python version of input object if conversion possible.
"""
if ((isinstance(typedef, dict)
and (typedef.get('type', '1darray') not in ['1darray', 'ndarray']))):
stype = typedef.get('subtype', typedef.get('type', None))
py_type = units.PYTHON_SCALARS_WITH_UNITS[stype][0]
if np.dtype(py_type) == type(obj):
obj = py_type(obj)
return obj
@classmethod
def _generate_data(cls, typedef, numeric_value=None):
r"""Generate mock data for the specified type.
Args:
typedef (dict): Type definition.
Returns:
object: Python object of the specified type.
"""
dtype = definition2dtype(typedef)
subtype = typedef.get('subtype', typedef['type'])
if subtype in ['bytes', 'unicode']:
if subtype == 'bytes':
value = b'x' * int(typedef['precision'] / 8)
else:
value = 'x' * int(typedef['precision'] / 32)
elif numeric_value is not None:
value = numeric_value
else:
value = 1.0
if typedef['type'] == '1darray':
out = np.repeat(np.array([value], dtype),
typedef.get('length', 2))
elif typedef['type'] == 'ndarray':
out = np.tile(np.array([value], dtype),
typedef.get('shape', (4, 5)))
else:
out = np.array([value], dtype)[0]
out = units.add_units(out, typedef.get('units', ''))
return out
[docs] @classmethod
def get_test_data(cls, typedef=None):
r"""object: Test data."""
if typedef is None:
typedef = {'type': cls.name}
if not hasattr(cls, 'fixed_properties'):
typedef.setdefault('subtype', 'float')
subtype = typedef.get('subtype', typedef['type'])
if subtype in ['bytes', 'unicode']:
typedef.setdefault('precision', 3 * 32)
elif subtype == 'complex':
typedef.setdefault('precision', 64)
else:
typedef.setdefault('precision', 32)
return super(ScalarMetaschemaType, cls).get_test_data(typedef)
# Dynamically create explicity scalar classes for shorthand
for t in constants.VALID_TYPES.keys():
short_doc = 'A %s value with or without units.' % t
long_doc = ('%s\n\n'
' Developer Notes:\n'
' Precision X is preserved.\n\n') % short_doc
kwargs = {'target_globals': globals(),
'__doc__': long_doc,
'__module__': ScalarMetaschemaType.__module__,
'python_types': units.PYTHON_SCALARS_WITH_UNITS[t]}
create_fixed_type_class(t, short_doc, ScalarMetaschemaType,
{'subtype': t}, **kwargs)