Source code for wolframclient.serializers.encoders.numpy

from __future__ import absolute_import, print_function, unicode_literals

import sys

from wolframclient.utils.api import numpy
from wolframclient.utils.dispatch import Dispatch
from wolframclient.utils.functional import map

encoder = Dispatch()


NUMPY_MAPPING = {
    numpy.int8: ("Integer8", None),
    numpy.int16: ("Integer16", None),
    numpy.int32: ("Integer32", None),
    numpy.intc: ("Integer32", None),
    numpy.int64: ("Integer64", None),
    numpy.uint8: ("UnsignedInteger8", None),
    numpy.uint16: ("UnsignedInteger16", None),
    numpy.uint32: ("UnsignedInteger32", None),
    numpy.uintc: ("UnsignedInteger32", None),
    numpy.uint64: ("UnsignedInteger64", None),
    numpy.float32: ("Real32", None),
    numpy.float64: ("Real64", None),
    numpy.complex64: ("ComplexReal32", None),
    numpy.complex128: ("ComplexReal64", None),
}


PACKED_NUMPY_MAPPING = {
    numpy.int8: ("Integer8", None),
    numpy.int16: ("Integer16", None),
    numpy.int32: ("Integer32", None),
    numpy.intc: ("Integer32", None),
    numpy.int64: ("Integer64", None),
    numpy.uint8: ("Integer16", numpy.int16),
    numpy.uint16: ("Integer32", numpy.int32),
    numpy.uint32: ("Integer64", numpy.int64),
    numpy.uintc: ("Integer64", numpy.int64),
    # numpy.uint64: ("UnsignedInteger64", None),
    numpy.float32: ("Real32", None),
    numpy.float64: ("Real64", None),
    numpy.complex64: ("ComplexReal32", None),
    numpy.complex128: ("ComplexReal64", None),
}

SYS_IS_LE = sys.byteorder == "little"


[docs] def to_little_endian(array, inplace=False): """Return a numpy array of the same type with little endian byte ordering. Set `inplace` to `True` to mutate the input array.""" endianness = array.dtype.byteorder if endianness == ">" or (endianness == "=" and not SYS_IS_LE): return array.byteswap(inplace=inplace).newbyteorder() else: return array
def _iencode(serializer, o, mapping, processor): if not o.shape: return serializer.encode(o.item()) try: wl_type, cast_to = mapping[o.dtype.type] except KeyError: raise NotImplementedError( "Numpy serialization not implemented for {}. Choices are: {}".format( repr(o.dtype), ", ".join(map(repr, mapping.keys())) ) ) o = to_little_endian(o) if cast_to is not None: o = o.astype(cast_to) if hasattr(o, "tobytes"): # Numpy 1.9+ support array.tobytes, but previous versions don't and use tostring instead. data = o.tobytes() else: data = o.tostring() return processor(data, o.shape, wl_type) @encoder.dispatch(numpy.PackedArray) def encode_ndarray(serializer, o): return _iencode(serializer, o, PACKED_NUMPY_MAPPING, serializer.serialize_packed_array)
[docs] @encoder.dispatch(numpy.ndarray) def encode_ndarray(serializer, o): return _iencode(serializer, o, NUMPY_MAPPING, serializer.serialize_numeric_array)
[docs] @encoder.dispatch(numpy.integer) def encode_numpy_int(serializer, o): return serializer.serialize_int(int(o))
[docs] @encoder.dispatch(numpy.floating) def encode_numpy_floating(serializer, o): # mantissa, and base 2 exponent. mantissa, exp = numpy.frexp(o) return serializer.serialize_function( serializer.serialize_symbol(b"Times"), ( serializer.serialize_float(mantissa), serializer.serialize_function( serializer.serialize_symbol(b"Power"), (serializer.serialize_int(2), serializer.serialize_float(exp)), ), ), )
[docs] @encoder.dispatch((numpy.float16, numpy.float32, numpy.float64)) def encode_numpy_mp_float(serializer, o): return serializer.serialize_float(o)
[docs] @encoder.dispatch(numpy.complexfloating) def encode_complex(serializer, o): return serializer.serialize_complex(o)