# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
from wolframclient.serializers.wxfencoder.constants import (
VALID_PACKED_ARRAY_TYPES,
WXF_CONSTANTS,
)
from wolframclient.serializers.wxfencoder.serializer import WXFSerializerException
from wolframclient.serializers.wxfencoder.utils import (
float_to_bytes,
integer_size,
integer_to_bytes,
write_varint,
)
from wolframclient.utils import six
__all__ = [
"WXFExprFunction",
"WXFExprInteger",
"WXFExprString",
"WXFExprSymbol",
"WXFExprBinaryString",
"WXF_CONSTANTS",
]
class WXFExpr(object):
"""Represent a WXF expression.
This is an abstract class. Only its children must be used.
"""
__slots__ = "wxf_type"
def __init__(self, wxf_type):
self.wxf_type = wxf_type
def _serialize_to_wxf(self, context, stream):
""" Write the serialized form of a given WXFExpr. """
raise NotImplementedError
[docs]class WXFExprFunction(WXFExpr):
"""Functions have a length representing the number of parts (including zero).
Each function has a head which is itself a expr, usually a `WXFExprSymbol` which
is not accounted in the length.
"""
__slots__ = "length"
def __init__(self, length):
super(WXFExprFunction, self).__init__(WXF_CONSTANTS.Function)
self.length = length
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
# Function has a head which account for one part element contrary to association
context.step_into_new_function(self.length)
write_varint(self.length, stream)
[docs]class WXFExprInteger(WXFExpr):
""" Integers have various length, from one byte up to eight and are signed
values. Values above 2^63-1 are represented with `WXFExprBigInteger`.
Internally WXF uses the two's complement representation of integer values.
The endianness is system independent and is always little-endian.
"""
__slots__ = "value", "int_size"
def __init__(self, value):
if not isinstance(value, six.integer_types):
raise TypeError("WXFExprInteger must be initialize with an integer value.")
wxf_type, self.int_size = integer_size(value)
super(WXFExprInteger, self).__init__(wxf_type)
self.value = value
""" Encode the integer into bytes and return them in a `buffer`.
Note that the buffer is an bytearray in python 2.7 and an array in 3.x.
This method is only useful to hide the Python 2.7 implementation.
It is proxying int.to_bytes for version 3.4 and above.
"""
[docs] def to_bytes(self):
return integer_to_bytes(self.value, self.int_size)
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.add_part()
stream.write(self.to_bytes())
class WXFExprReal(WXFExpr):
""" Represent a floating point value. Internally WXF represents the value with
double float-point value in the IEEE 754 standard. """
__slots__ = "value"
def __init__(self, value):
if not isinstance(value, float):
raise TypeError("WXFExprReal must be initialized with a float.")
super(WXFExprReal, self).__init__(WXF_CONSTANTS.Real64)
self.value = value
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.add_part()
stream.write(float_to_bytes(self.value))
class _WXFExprStringLike(WXFExpr):
""" Parent class of all string based expressions.
Store a given string value as a utf-8 encoded binary string.
"""
__slots__ = "length", "value"
def __init__(self, wxf_type, value, allow_binary=False):
""" Initialize using a given string `value`.
If `allow_binary` is set to false, also accept binary types: `bytes` (py3) and `str` (py2)
"""
super(_WXFExprStringLike, self).__init__(wxf_type)
if allow_binary and isinstance(value, six.binary_type):
self.value = value
elif isinstance(value, six.text_type):
self.value = value.encode("utf-8")
else:
raise TypeError(
self.__class__.__name__
+ " must be initialize with a string. Was '"
+ value.__class__.__name__
+ "'."
)
self.length = len(self.value)
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.add_part()
write_varint(self.length, stream)
stream.write(self.value)
[docs]class WXFExprSymbol(_WXFExprStringLike):
""" A symbol represented by a string name. The name is always utf8 encoded."""
def __init__(self, value):
super(WXFExprSymbol, self).__init__(WXF_CONSTANTS.Symbol, value, allow_binary=True)
[docs]class WXFExprString(_WXFExprStringLike):
""" A string of unicode character. The string is always utf8 encoded.
Notabene: Python 3 does not allow utf8 encoding of the surrogate range
from `0xD800` to `0xDFFF`. Python 2 and the Wolfram Language on the other
hand handle those characters as any other unicode code points.
"""
def __init__(self, value):
super(WXFExprString, self).__init__(WXF_CONSTANTS.String, value)
class WXFExprBigInteger(_WXFExprStringLike):
""" A string of digits representing a big integer"""
def __init__(self, value):
super(WXFExprBigInteger, self).__init__(
WXF_CONSTANTS.BigInteger, value, allow_binary=True
)
class WXFExprBigReal(_WXFExprStringLike):
""" A string representation of a real value with arbitrary precision. The string format matches the one of the
:wl:`InputForm` string representation of the real in the Wolfram Language.
"""
def __init__(self, value):
super(WXFExprBigReal, self).__init__(WXF_CONSTANTS.BigReal, value, allow_binary=True)
[docs]class WXFExprBinaryString(WXFExpr):
"""A string of arbitrary bytes. Contrary to `WXFExprString` no encoding is
required."""
__slots__ = "data"
def __init__(self, data):
if isinstance(data, (six.binary_type, bytearray, six.buffer_types)):
self.data = data
else:
raise TypeError(
"WXFExprBinaryString must be initialized with binary data: bytes in Python 3, str in Python 2.7 or bytearray."
)
super(WXFExprBinaryString, self).__init__(WXF_CONSTANTS.BinaryString)
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.add_part()
write_varint(len(self.data), stream)
stream.write(self.data)
class _WXFExprArray(WXFExpr):
"""Arrays are multidimensional tables of machine-precision numeric values.
The `dimensions` is a list of strictly positive integers representing the
array shape. The data contains the flatten binary representation of the
values."""
__slots__ = "dimensions", "value_type", "data"
def __init__(self, wxf_type, dimensions, value_type, data=None):
super(_WXFExprArray, self).__init__(wxf_type)
if not isinstance(dimensions, (list, tuple)) or len(dimensions) == 0:
raise TypeError("Dimensions must be a non-empty list.")
if not all(isinstance(dim, six.integer_types) and dim > 0 for dim in dimensions):
raise ValueError("Dimensions must be positive integers.")
self.dimensions = dimensions
self.value_type = value_type
self.data = data
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.add_part()
stream.write(self.value_type)
write_varint(len(self.dimensions), stream)
for dim in self.dimensions:
write_varint(dim, stream)
if self.data is not None:
stream.write(self.data)
else:
raise WXFSerializerException("Missing array data.")
class WXFExprPackedArray(_WXFExprArray):
""" Packed array is a type of array that only supports a subset of all the
possible type of values: signed integers, reals, complexes. See `VALID_PACKED_ARRAY_TYPES`."""
def __init__(self, dimensions, value_type, data=None):
if value_type not in VALID_PACKED_ARRAY_TYPES:
raise Exception("Invalid packed array value type ({}).", value_type)
super(WXFExprPackedArray, self).__init__(
WXF_CONSTANTS.PackedArray, dimensions, value_type, data
)
class WXFExprNumericArray(_WXFExprArray):
""" Raw array is an array that supports many type of values:
signed and unsigned integers, reals, complexes. See `ARRAY_TYPES`."""
def __init__(self, dimensions, value_type, data=None):
super(WXFExprNumericArray, self).__init__(
WXF_CONSTANTS.NumericArray, dimensions, value_type, data
)
class WXFExprAssociation(WXFExpr):
""" Association is a key value store similar to `dict`. `WXFExprAssociation`
requires a length, the number of entries. Only `WXFExprRule` and `WXFExprRuleDelayed`
are valid entry types in an association.
"""
__slot__ = "length"
def __init__(self, length):
if not isinstance(length, six.integer_types) or length < 0:
raise TypeError("WXFExprAssociation must be instantiated with a length.")
super(WXFExprAssociation, self).__init__(WXF_CONSTANTS.Association)
self.length = length
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
context.step_into_new_assoc(self.length)
write_varint(self.length, stream)
class _WXFExprRule(WXFExpr):
def __init__(self, wxf_type):
super(_WXFExprRule, self).__init__(wxf_type)
def _serialize_to_wxf(self, stream, context):
stream.write(self.wxf_type)
# make sure those special tokens are correctly used inside an association.
if not context.is_rule_valid():
raise WXFSerializerException(
'WXF Rule and RuleDelayed must be part of an Association. Use a Function with head Symbol "Rule(Delayed)" outside associations.'
)
# rule always has two parts.
context.step_into_new_rule()
class WXFExprRule(_WXFExprRule):
""" Represent a rule in an association. Rule have two parts but no head.
Rules that are not part of an association (e.g list of rules) must be encoded
as a function with head `'Rule'`.
"""
def __init__(self):
super(WXFExprRule, self).__init__(WXF_CONSTANTS.Rule)
class WXFExprRuleDelayed(_WXFExprRule):
""" Represent a rule delayed in an association. See `WXFExprRule`"""
def __init__(self):
super(WXFExprRuleDelayed, self).__init__(WXF_CONSTANTS.RuleDelayed)
