Source code for wolframclient.serializers.wxfencoder.wxfexpr

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)