Source code for wolframclient.serializers.encoders.numpy

# -*- coding: utf-8 -*-

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",
    numpy.int16: "Integer16",
    numpy.int32: "Integer32",
    numpy.int64: "Integer64",
    numpy.uint8: "UnsignedInteger8",
    numpy.uint16: "UnsignedInteger16",
    numpy.uint32: "UnsignedInteger32",
    numpy.uint64: "UnsignedInteger64",
    numpy.float32: "Real32",
    numpy.float64: "Real64",
    numpy.complex64: "ComplexReal32",
    numpy.complex128: "ComplexReal64",
}


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
[docs]@encoder.dispatch(numpy.ndarray) def encode_ndarray(serializer, o): try: wl_type = NUMPY_MAPPING[o.dtype.type] except KeyError: raise NotImplementedError( "NumPy serialization not implemented for %s. Choices are: %s" % (repr(o.dtype), ", ".join(map(repr, NUMPY_MAPPING.keys()))) ) o = to_little_endian(o) 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 serializer.serialize_numeric_array(data, o.shape, wl_type)
PACKEDARRAY_NUMPY_MAPPING = { numpy.int8: ("Integer8", None), numpy.int16: ("Integer16", None), numpy.int32: ("Integer32", None), numpy.int64: ("Integer64", None), numpy.uint8: ("Integer16", numpy.int16), numpy.uint16: ("Integer32", numpy.int32), numpy.uint32: ("Integer64", numpy.int64), # numpy.uint64: ("UnsignedInteger64", None), numpy.float32: ("Real32", None), numpy.float64: ("Real64", None), numpy.complex64: ("ComplexReal32", None), numpy.complex128: ("ComplexReal64", None), } """ Maps numpy dtype to appropriate wxf packed array type, eventually specifying the type to cast the data to. """
[docs]@encoder.dispatch(numpy.PackedArray) def encode_packed_array(serializer, o): try: wl_type, cast_to = PACKEDARRAY_NUMPY_MAPPING[o.dtype.type] except KeyError: raise NotImplementedError( "Packed array serialization not implemented for %s. Choices are: %s" % (repr(o.dtype), ", ".join(map(repr, PACKEDARRAY_NUMPY_MAPPING.keys()))) ) o = to_little_endian(o) if cast_to is not None: o = numpy.array(o, dtype=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 serializer.serialize_packed_array(data, o.shape, wl_type)
[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)