# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
from wolframclient.exception import WolframLanguageException
from wolframclient.serializers.wxfencoder.constants import (
ARRAY_TYPES,
VALID_PACKED_ARRAY_TYPES,
WXF_CONSTANTS,
StructDouble,
StructFloat,
StructInt8LE,
StructInt16LE,
StructInt32LE,
StructInt64LE,
StructUInt8LE,
StructUInt16LE,
StructUInt32LE,
StructUInt64LE,
)
from wolframclient.utils import six
from wolframclient.utils.datastructures import Settings
if six.JYTHON:
import jarray
[docs]def write_varint(int_value, stream):
"""Serialize `int_value` into varint bytes and write them to
`stream`, return the stream.
"""
stream.write(varint_bytes(int_value))
[docs]def varint_bytes(int_value):
"""Serialize `int_value` into varint bytes and return them as a byetarray."""
buf = bytearray(9)
if int_value < 0:
raise TypeError("Negative values cannot be encoded as varint.")
count = 0
while True:
next = int_value & 0x7F
int_value >>= 7
if int_value:
buf[count] = next | 0x80
count += 1
else:
buf[count] = next
count += 1
break
return buf[:count]
_exceptions = {
0: (WXF_CONSTANTS.Integer8, 1),
-(1 << 7): (WXF_CONSTANTS.Integer8, 1),
-(1 << 15): (WXF_CONSTANTS.Integer16, 2),
-(1 << 31): (WXF_CONSTANTS.Integer32, 4),
-(1 << 63): (WXF_CONSTANTS.Integer64, 8),
}
_size = dict(
(j, (WXF_CONSTANTS["Integer%i" % ih], ih // 8))
for il, ih in ((1, 8), (9, 16), (17, 32), (33, 64))
for j in range(il, ih + 1)
)
[docs]def integer_size(value):
try:
return _exceptions.get(value, None) or _size[value.bit_length() + 1]
except KeyError:
raise ValueError("Value %i is not a machine-sized integer." % value)
_packing = {1: StructInt8LE, 2: StructInt16LE, 4: StructInt32LE, 8: StructInt64LE}
if six.JYTHON:
def integer_to_bytes(value, int_size):
buffer = jarray.zeros(8, "c")
_packing.get(int_size).pack_into(buffer, 0, value)
return buffer[:int_size].tostring()
elif six.PY2:
def integer_to_bytes(value, int_size):
buffer = bytearray(8)
_packing.get(int_size).pack_into(buffer, 0, value)
return buffer[:int_size]
else:
[docs] def integer_to_bytes(value, int_size):
return value.to_bytes(int_size, byteorder="little", signed=True)
if six.JYTHON:
def float_to_bytes(value):
buffer = jarray.zeros(8, "c")
StructDouble.pack_into(buffer, 0, value)
return buffer.tostring()
else:
[docs] def float_to_bytes(value):
buffer = bytearray(8)
StructDouble.pack_into(buffer, 0, value)
return buffer
[docs]def valid_dimension_or_fail(dimension):
if dimension <= 0:
raise WolframLanguageException(
"Invalid array dimensions: %s. Expecting strictly positive integer." % dimension
)
[docs]def array_to_wxf(wxf_token, data, dimensions, array_type_token):
yield wxf_token
yield array_type_token
yield varint_bytes(len(dimensions))
for dim in dimensions:
valid_dimension_or_fail(dim)
yield varint_bytes(dim)
yield data
[docs]def numeric_array_to_wxf(data, dimensions, wl_type):
return array_to_wxf(WXF_CONSTANTS.NumericArray, data, dimensions, ARRAY_TYPES[wl_type])
[docs]def packed_array_to_wxf(data, dimensions, wl_type):
array_type_token = ARRAY_TYPES[wl_type]
if array_type_token not in VALID_PACKED_ARRAY_TYPES:
raise ValueError("Invalid PackedArray type %s" % array_type_token)
return array_to_wxf(WXF_CONSTANTS.PackedArray, data, dimensions, array_type_token)
[docs]def array_to_list(data, dimensions, wl_type):
for dimension in dimensions:
valid_dimension_or_fail(dimension)
return _array_to_list(data, dimensions, wl_type)
if hasattr(memoryview, "cast"):
unpack_mapping = Settings(
Integer8="b",
UnsignedInteger8="B",
Integer16="h",
UnsignedInteger16="H",
Integer32="i",
UnsignedInteger32="I",
Integer64="q",
UnsignedInteger64="Q",
Real32="f",
Real64="d",
ComplexReal32="f",
ComplexReal64="d",
)
def _to_complex(array, max_depth, curr_depth):
# recursivelly traverse the array until the last (real) dimension is reached
# it correspond to an array of (fake) array of two elements (real and im parts).
if curr_depth < max_depth - 1:
for sub in array:
_to_complex(sub, max_depth, curr_depth + 1)
return
# iterate over the pairs
for index, complex_pair in enumerate(array):
array[index] = complex(*complex_pair)
def _array_to_list(data, shape, array_type):
view = memoryview(data)
if array_type == "ComplexReal32" or array_type == "ComplexReal64":
dimensions = list(shape)
array_depth = len(dimensions)
# In the given array, 2 reals give one complex,
# adding one last dimension to represent it.
dimensions.append(2)
as_list = view.cast(unpack_mapping[array_type], shape=dimensions).tolist()
_to_complex(as_list, array_depth, 0)
return as_list
else:
return view.cast(unpack_mapping[array_type], shape=shape).tolist()
else:
unpack_mapping = Settings(
Integer8=StructInt8LE,
UnsignedInteger8=StructUInt8LE,
Integer16=StructInt16LE,
UnsignedInteger16=StructUInt16LE,
Integer32=StructInt32LE,
UnsignedInteger32=StructUInt32LE,
Integer64=StructInt64LE,
UnsignedInteger64=StructUInt64LE,
Real32=StructFloat,
Real64=StructDouble,
ComplexReal32=StructFloat,
ComplexReal64=StructDouble,
)
def _array_to_list(data, shape, array_type):
value, _ = _build_array_from_bytes(data, 0, array_type, shape, 0)
return value
def _build_array_from_bytes(data, offset, array_type, dimensions, current_dim):
new_array = list()
if current_dim < len(dimensions) - 1:
for i in range(dimensions[current_dim]):
new_elem, offset = _build_array_from_bytes(
data, offset, array_type, dimensions, current_dim + 1
)
new_array.append(new_elem)
else:
struct = unpack_mapping[array_type]
# complex values, need two reals for each.
if array_type == "ComplexReal32" or array_type == "ComplexReal64":
for i in range(dimensions[-1]):
# this returns a tuple.
re = struct.unpack_from(data, offset=offset)
offset = offset + struct.size
im = struct.unpack_from(data, offset=offset)
offset = offset + struct.size
new_array.append(complex(re[0], im[0]))
else:
for i in range(dimensions[-1]):
# this returns a tuple.
value = struct.unpack_from(data, offset=offset)
offset = offset + struct.size
new_array.append(value[0])
return new_array, offset
