Skip to content

Machine-oriented binary syntax

The preserves.binary module implements the Preserves machine-oriented binary syntax.

The main entry points are functions encode, canonicalize, decode, and decode_with_annotations.

>>> encode(Record(Symbol('hi'), []))
b'\xb4\xb3\x02hi\x84'
>>> decode(b'\xb4\xb3\x02hi\x84')
#hi()

Annotated(item)

Bases: object

A Preserves Value along with a sequence of Values annotating it. Compares equal to the underlying Value, ignoring the annotations. See the specification document for more about annotations.

>>> import preserves
>>> a = preserves.parse('''
... # A comment
... [1 2 3]
... ''', include_annotations=True)
>>> a
@'A comment' (1, 2, 3)
>>> a.item
(1, 2, 3)
>>> a.annotations
['A comment']
>>> a == (1, 2, 3)
True
>>> a == preserves.parse('@xyz [1 2 3]', include_annotations=True)
True
>>> a[0]
Traceback (most recent call last):
  ...
TypeError: 'Annotated' object is not subscriptable
>>> a.item[0]
1
>>> type(a.item[0])
<class 'preserves.values.Annotated'>
>>> a.item[0].annotations
[]
>>> print(preserves.stringify(a))
@"A comment" [1 2 3]
>>> print(preserves.stringify(a, include_annotations=False))
[1 2 3]

Attributes:

Name Type Description
item Value

the underlying annotated Value
annotations list[Value]

the annotations attached to self.item
Source code in preserves/values.py 
456
457
458
def __init__(self, item):
    self.annotations = []
    self.item = item

peel()

Calls strip_annotations on self with depth=1.

Source code in preserves/values.py 
479
480
481
def peel(self):
    """Calls [strip_annotations][preserves.values.strip_annotations] on `self` with `depth=1`."""
    return strip_annotations(self, 1)

strip(depth=inf)

Calls strip_annotations on self and depth.

Source code in preserves/values.py 
475
476
477
def strip(self, depth=inf):
    """Calls [strip_annotations][preserves.values.strip_annotations] on `self` and `depth`."""
    return strip_annotations(self, depth)

DecodeError

Bases: ValueError

Raised whenever preserves.binary.Decoder or preserves.text.Parser detect invalid input.

Decoder(packet=b'', include_annotations=False, decode_embedded=lambda x: x)

Bases: BinaryCodec

Implementation of a decoder for the machine-oriented binary Preserves syntax.

Parameters:

Name Type Description Default
packet bytes

initial contents of the input buffer; may subsequently be extended by calling extend.

 b''
include_annotations bool

if True, wrap each value and subvalue in an Annotated object.

 False
decode_embedded

function accepting a Value and returning a possibly-decoded form of that value suitable for placing into an Embedded object.

 lambda x: x

Normal usage is to supply a buffer, and keep calling next until a ShortPacket exception is raised:

>>> d = Decoder(b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84')
>>> d.next()
123
>>> d.next()
'hello'
>>> d.next()
()
>>> d.next()
Traceback (most recent call last):
  ...
preserves.error.ShortPacket: Short packet

Alternatively, keep calling try_next until it yields None, which is not in the domain of Preserves Values:

>>> d = Decoder(b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84')
>>> d.try_next()
123
>>> d.try_next()
'hello'
>>> d.try_next()
()
>>> d.try_next()

For convenience, Decoder implements the iterator interface, backing it with try_next, so you can simply iterate over all complete values in an input:

>>> d = Decoder(b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84')
>>> list(d)
[123, 'hello', ()]

>>> for v in Decoder(b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84'):
...     print(repr(v))
123
'hello'
()

Supply include_annotations=True to read annotations alongside the annotated values:

>>> d = Decoder(b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84', include_annotations=True)
>>> list(d)
[123, 'hello', @#x ()]

If you are incrementally reading from, say, a socket, you can use extend to add new input as if comes available:

>>> d = Decoder(b'\xb0\x01{\xb1\x05he')
>>> d.try_next()
123
>>> d.try_next() # returns None because the input is incomplete
>>> d.extend(b'llo')
>>> d.try_next()
'hello'
>>> d.try_next()

Attributes:

Name Type Description
packet bytes

buffered input waiting to be processed
index int

read position within packet
Source code in preserves/binary.py 
127
128
129
130
131
132
def __init__(self, packet=b'', include_annotations=False, decode_embedded=lambda x: x):
    super(Decoder, self).__init__()
    self.packet = packet
    self.index = 0
    self.include_annotations = include_annotations
    self.decode_embedded = decode_embedded

extend(data)

Appends data to the remaining bytes in self.packet, trimming already-processed bytes from the front of self.packet and resetting self.index to zero.

Source code in preserves/binary.py 
134
135
136
137
138
def extend(self, data):
    """Appends `data` to the remaining bytes in `self.packet`, trimming already-processed
    bytes from the front of `self.packet` and resetting `self.index` to zero."""
    self.packet = self.packet[self.index:] + data
    self.index = 0

next()

Reads the next complete Value from the internal buffer, raising ShortPacket if too few bytes are available, or DecodeError if the input is invalid somehow.

Source code in preserves/binary.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def next(self):
    """Reads the next complete `Value` from the internal buffer, raising
    [ShortPacket][preserves.error.ShortPacket] if too few bytes are available, or
    [DecodeError][preserves.error.DecodeError] if the input is invalid somehow.

    """
    tag = self.nextbyte()
    if tag == 0x80: return self.wrap(False)
    if tag == 0x81: return self.wrap(True)
    if tag == 0x84: raise DecodeError('Unexpected end-of-stream marker')
    if tag == 0x85:
        a = self.next()
        v = self.next()
        return self.unshift_annotation(a, v)
    if tag == 0x86:
        if self.decode_embedded is None:
            raise DecodeError('No decode_embedded function supplied')
        return self.wrap(Embedded(self.decode_embedded(self.next())))
    if tag == 0x87:
        count = self.nextbyte()
        if count == 8: return self.wrap(struct.unpack('>d', self.nextbytes(8))[0])
        raise DecodeError('Invalid IEEE754 size')
    if tag == 0xb0: return self.wrap(self.nextint(self.varint()))
    if tag == 0xb1: return self.wrap(self.nextbytes(self.varint()).decode('utf-8'))
    if tag == 0xb2: return self.wrap(self.nextbytes(self.varint()))
    if tag == 0xb3: return self.wrap(Symbol(self.nextbytes(self.varint()).decode('utf-8')))
    if tag == 0xb4:
        vs = self.nextvalues()
        if not vs: raise DecodeError('Too few elements in encoded record')
        return self.wrap(Record(vs[0], vs[1:]))
    if tag == 0xb5: return self.wrap(tuple(self.nextvalues()))
    if tag == 0xb6:
        vs = self.nextvalues()
        s = frozenset(vs)
        if len(s) != len(vs): raise DecodeError('Duplicate value')
        return self.wrap(s)
    if tag == 0xb7: return self.wrap(ImmutableDict.from_kvs(self.nextvalues()))
    raise DecodeError('Invalid tag: ' + hex(tag))

try_next()

Like next, but returns None instead of raising ShortPacket.

Source code in preserves/binary.py 
228
229
230
231
232
233
234
235
236
def try_next(self):
    """Like [next][preserves.binary.Decoder.next], but returns `None` instead of raising
    [ShortPacket][preserves.error.ShortPacket]."""
    start = self.index
    try:
        return self.next()
    except ShortPacket:
        self.index = start
        return None

Embedded(embeddedValue)

Representation of a Preserves Embedded value. For more on the meaning and use of embedded values, see the specification.

>>> import io
>>> e = Embedded(io.StringIO('some text'))
>>> e                                        # doctest: +ELLIPSIS
#:<_io.StringIO object at ...>
>>> e.embeddedValue                          # doctest: +ELLIPSIS
<_io.StringIO object at ...>

>>> import preserves
>>> print(preserves.stringify(Embedded(None)))
Traceback (most recent call last):
  ...
TypeError: Cannot preserves-format: None
>>> print(preserves.stringify(Embedded(None), format_embedded=lambda x: 'abcdef'))
#:"abcdef"

Attributes:

Name Type Description
embeddedValue

any Python value; could be a platform object, could be a representation of a Preserves Value, could be None, could be anything!
Source code in preserves/values.py 
601
602
def __init__(self, embeddedValue):
    self.embeddedValue = embeddedValue

EncodeError

Bases: ValueError

Raised whenever preserves.binary.Encoder or preserves.text.Formatter are unable to proceed.

Encoder(encode_embedded=lambda x: x, canonicalize=False, include_annotations=None)

Bases: BinaryCodec

Implementation of an encoder for the machine-oriented binary Preserves syntax.

>>> e = Encoder()
>>> e.append(123)
>>> e.append('hello')
>>> e.append(annotate([], Symbol('x')))
>>> e.contents()
b'\xb0\x01{\xb1\x05hello\x85\xb3\x01x\xb5\x84'

Parameters:

Name Type Description Default
encode_embedded

function accepting an Embedded.embeddedValue and returning a Value for serialization.

 lambda x: x
canonicalize bool

if True, ensures the serialized data are in canonical form. This is slightly more work than producing potentially-non-canonical output.

 False
include_annotations bool | None

if None, includes annotations in the output only when canonicalize is False, because canonical serialization of values demands omission of annotations. If explicitly True or False, however, annotations will be included resp. excluded no matter the canonicalize setting. This can be used to get canonical ordering (canonicalize=True) and annotations (include_annotations=True).

 None

Attributes:

Name Type Description
buffer bytearray

accumulator for the output of the encoder
Source code in preserves/binary.py 
298
299
300
301
302
303
304
305
306
307
308
309
def __init__(self,
             encode_embedded=lambda x: x,
             canonicalize=False,
             include_annotations=None):
    super(Encoder, self).__init__()
    self.buffer = bytearray()
    self._encode_embedded = encode_embedded
    self._canonicalize = canonicalize
    if include_annotations is None:
        self.include_annotations = not self._canonicalize
    else:
        self.include_annotations = include_annotations

append(v)

Extend self.buffer with an encoding of v.

Source code in preserves/binary.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def append(self, v):
    """Extend `self.buffer` with an encoding of `v`."""
    v = preserve(v)
    if hasattr(v, '__preserve_write_binary__'):
        v.__preserve_write_binary__(self)
    elif v is False:
        self.buffer.append(0x80)
    elif v is True:
        self.buffer.append(0x81)
    elif isinstance(v, float):
        self.buffer.append(0x87)
        self.buffer.append(8)
        self.buffer.extend(struct.pack('>d', v))
    elif isinstance(v, numbers.Number):
        self.encodeint(v)
    elif isinstance(v, bytes):
        self.encodebytes(0xb2, v)
    elif isinstance(v, basestring_):
        self.encodebytes(0xb1, v.encode('utf-8'))
    elif isinstance(v, list):
        self.encodevalues(0xb5, v)
    elif isinstance(v, tuple):
        self.encodevalues(0xb5, v)
    elif isinstance(v, set):
        self.encodeset(v)
    elif isinstance(v, frozenset):
        self.encodeset(v)
    elif isinstance(v, dict):
        self.encodedict(v)
    else:
        try:
            i = iter(v)
        except TypeError:
            i = None
        if i is None:
            self.cannot_encode(v)
        else:
            self.encodevalues(0xb5, i)

contents()

Returns a bytes constructed from the contents of self.buffer.

Source code in preserves/binary.py 
320
321
322
def contents(self):
    """Returns a `bytes` constructed from the contents of `self.buffer`."""
    return bytes(self.buffer)

reset()

Clears self.buffer to a fresh empty bytearray.

Source code in preserves/binary.py 
311
312
313
def reset(self):
    """Clears `self.buffer` to a fresh empty `bytearray`."""
    self.buffer = bytearray()

ImmutableDict(*args, **kwargs)

Bases: dict

A subclass of Python's built-in dict that overrides methods that could mutate the dictionary, causing them to raise TypeError('Immutable') if called.

Implements the __hash__ method, allowing ImmutableDict instances to be used whereever immutable data are permitted; in particular, as keys in other dictionaries.

>>> d = ImmutableDict([('a', 1), ('b', 2)])
>>> d
{'a': 1, 'b': 2}
>>> d['c'] = 3
Traceback (most recent call last):
  ...
TypeError: Immutable
>>> del d['b']
Traceback (most recent call last):
  ...
TypeError: Immutable
Source code in preserves/values.py 
340
341
342
343
def __init__(self, *args, **kwargs):
    if hasattr(self, '__hash'): raise TypeError('Immutable')
    super(ImmutableDict, self).__init__(*args, **kwargs)
    self.__hash = None

from_kvs(kvs) staticmethod

Constructs an ImmutableDict from a sequence of alternating keys and values; compare to the ImmutableDict constructor, which takes a sequence of key-value pairs.

>>> ImmutableDict.from_kvs(['a', 1, 'b', 2])
{'a': 1, 'b': 2}
>>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3
Traceback (most recent call last):
  ...
TypeError: Immutable
Source code in preserves/values.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@staticmethod
def from_kvs(kvs):
    """Constructs an [ImmutableDict][preserves.values.ImmutableDict] from a sequence of
    alternating keys and values; compare to the
    [ImmutableDict][preserves.values.ImmutableDict] constructor, which takes a sequence of
    key-value pairs.

    ```python
    >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])
    {'a': 1, 'b': 2}
    >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3
    Traceback (most recent call last):
      ...
    TypeError: Immutable

    ```

    """

    i = iter(kvs)
    result = ImmutableDict()
    result_proxy = super(ImmutableDict, result)
    try:
        while True:
            k = next(i)
            try:
                v = next(i)
            except StopIteration:
                raise DecodeError("Missing dictionary value")
            if k in result:
                raise DecodeError("Duplicate key: " + repr(k))
            result_proxy.__setitem__(k, v)
    except StopIteration:
        pass
    return result

Record(key, fields)

Bases: object

Representation of Preserves Records, which are a pair of a label Value and a sequence of field Values.

>>> r = Record(Symbol('label'), ['field1', ['field2item1', 'field2item2']])
>>> r
#label('field1', ['field2item1', 'field2item2'])
>>> r.key
#label
>>> r.fields
('field1', ['field2item1', 'field2item2'])
>>> import preserves
>>> preserves.stringify(r)
'<label "field1" ["field2item1" "field2item2"]>'
>>> r == preserves.parse('<label "field1" ["field2item1" "field2item2"]>')
True

Parameters:

Name Type Description Default
key Value

the Record's label

 required
fields iterable[Value]

the fields of the Record

 required

Attributes:

Name Type Description
key Value

the Record's label
fields tuple[Value]

the fields of the Record
Source code in preserves/values.py 
152
153
154
155
def __init__(self, key, fields):
    self.key = key
    self.fields = tuple(fields)
    self.__hash = None

makeBasicConstructor(label, fieldNames) staticmethod

Constructs and returns a "constructor" for Records having a certain label and number of fields.

Deprecated

Use preserves.schema definitions instead.

The "constructor" is a callable function that accepts len(fields) arguments and returns a Record with label as its label and the arguments to the constructor as field values.

In addition, the "constructor" has a constructorInfo attribute holding a RecordConstructorInfo object, an isClassOf attribute holding a unary function that returns True iff its argument is a Record with label label and arity len(fieldNames), and an ensureClassOf attribute that raises an Exception if isClassOf returns false on its argument and returns the argument otherwise.

Finally, for each field name f in fieldNames, the "constructor" object has an attribute _f that is a unary function that retrieves the f field from the passed in argument.

>>> c = Record.makeBasicConstructor(Symbol('date'), 'year month day')
>>> c(1969, 7, 16)
#date(1969, 7, 16)
>>> c.constructorInfo
#date/3
>>> c.isClassOf(c(1969, 7, 16))
True
>>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))
True
>>> c.isClassOf(Record(Symbol('date'), [1969]))
False
>>> c.ensureClassOf(c(1969, 7, 16))
#date(1969, 7, 16)
>>> c.ensureClassOf(Record(Symbol('date'), [1969]))
Traceback (most recent call last):
  ...
TypeError: Record: expected #date/3, got #date(1969)
>>> c._year(c(1969, 7, 16))
1969
>>> c._month(c(1969, 7, 16))
7
>>> c._day(c(1969, 7, 16))
16

Parameters:

Name Type Description Default
label Value

Label to use for constructed/matched Records

 required
fieldNames tuple[str] | list[str] | str

Names of the Record's fields

 required
Source code in preserves/values.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@staticmethod
def makeBasicConstructor(label, fieldNames):
    """Constructs and returns a "constructor" for `Record`s having a certain `label` and
    number of fields.

    Deprecated:
       Use [preserves.schema][] definitions instead.

    The "constructor" is a callable function that accepts `len(fields)` arguments and
    returns a [Record][preserves.values.Record] with `label` as its label and the arguments
    to the constructor as field values.

    In addition, the "constructor" has a `constructorInfo` attribute holding a
    [RecordConstructorInfo][preserves.values.RecordConstructorInfo] object, an `isClassOf`
    attribute holding a unary function that returns `True` iff its argument is a
    [Record][preserves.values.Record] with label `label` and arity `len(fieldNames)`, and
    an `ensureClassOf` attribute that raises an `Exception` if `isClassOf` returns false on
    its argument and returns the argument otherwise.

    Finally, for each field name `f` in `fieldNames`, the "constructor" object has an
    attribute `_f` that is a unary function that retrieves the `f` field from the passed in
    argument.

    ```python
    >>> c = Record.makeBasicConstructor(Symbol('date'), 'year month day')
    >>> c(1969, 7, 16)
    #date(1969, 7, 16)
    >>> c.constructorInfo
    #date/3
    >>> c.isClassOf(c(1969, 7, 16))
    True
    >>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))
    True
    >>> c.isClassOf(Record(Symbol('date'), [1969]))
    False
    >>> c.ensureClassOf(c(1969, 7, 16))
    #date(1969, 7, 16)
    >>> c.ensureClassOf(Record(Symbol('date'), [1969]))
    Traceback (most recent call last):
      ...
    TypeError: Record: expected #date/3, got #date(1969)
    >>> c._year(c(1969, 7, 16))
    1969
    >>> c._month(c(1969, 7, 16))
    7
    >>> c._day(c(1969, 7, 16))
    16

    ```

    Args:
        label (Value): Label to use for constructed/matched `Record`s
        fieldNames (tuple[str] | list[str] | str): Names of the `Record`'s fields

    """
    if type(fieldNames) == str:
        fieldNames = fieldNames.split()
    arity = len(fieldNames)
    def ctor(*fields):
        if len(fields) != arity:
            raise Exception("Record: cannot instantiate %r expecting %d fields with %d fields"%(
                label,
                arity,
                len(fields)))
        return Record(label, fields)
    ctor.constructorInfo = RecordConstructorInfo(label, arity)
    ctor.isClassOf = lambda v: \
                     isinstance(v, Record) and v.key == label and len(v.fields) == arity
    def ensureClassOf(v):
        if not ctor.isClassOf(v):
            raise TypeError("Record: expected %r/%d, got %r" % (label, arity, v))
        return v
    ctor.ensureClassOf = ensureClassOf
    for fieldIndex in range(len(fieldNames)):
        fieldName = fieldNames[fieldIndex]
        # Stupid python scoping bites again
        def getter(fieldIndex):
            return lambda v: ensureClassOf(v)[fieldIndex]
        setattr(ctor, '_' + fieldName, getter(fieldIndex))
    return ctor

makeConstructor(labelSymbolText, fieldNames) staticmethod

Equivalent to Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames).

Deprecated

Use preserves.schema definitions instead.

Source code in preserves/values.py 
190
191
192
193
194
195
196
197
198
@staticmethod
def makeConstructor(labelSymbolText, fieldNames):
    """
    Equivalent to `Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)`.

    Deprecated:
       Use [preserves.schema][] definitions instead.
    """
    return Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)

RecordConstructorInfo(key, arity)

Bases: object

Describes the shape of a Record constructor, namely its label and its arity (field count).

>>> RecordConstructorInfo(Symbol('label'), 3)
#label/3

Attributes:

Name Type Description
key Value

the label of matching Records
arity int

the number of fields in matching Records
Source code in preserves/values.py 
295
296
297
def __init__(self, key, arity):
    self.key = key
    self.arity = arity

ShortPacket

Bases: DecodeError

Raised whenever preserves.binary.Decoder or preserves.text.Parser discover that they want to read beyond the end of the currently-available input buffer in order to completely read an encoded value.

Symbol(name)

Bases: object

Representation of Preserves Symbols.

>>> Symbol('xyz')
#xyz
>>> Symbol('xyz').name
'xyz'
>>> repr(Symbol('xyz'))
'#xyz'
>>> str(Symbol('xyz'))
'xyz'
>>> import preserves
>>> preserves.stringify(Symbol('xyz'))
'xyz'
>>> preserves.stringify(Symbol('hello world'))
"'hello world'"
>>> preserves.parse('xyz')
#xyz
>>> preserves.parse("'hello world'")
#hello world

Attributes:

Name Type Description
name str | Symbol

The symbol's text label. If an existing Symbol is passed in, the existing Symbol's name is used as the name for the new Symbol.
Source code in preserves/values.py 
78
79
def __init__(self, name):
    self.name = name.name if isinstance(name, Symbol) else name

annotate(v, *anns)

Wraps v in an Annotated object, if it isn't already wrapped, and appends each of the anns to the Annotated's annotations sequence. NOTE: Does not recursively ensure that any parts of the argument v are themselves wrapped in Annotated objects!

>>> import preserves
>>> print(preserves.stringify(annotate(123, "A comment", "Another comment")))
@"A comment" @"Another comment" 123
Source code in preserves/values.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
def annotate(v, *anns):
    """Wraps `v` in an [Annotated][preserves.values.Annotated] object, if it isn't already
    wrapped, and appends each of the `anns` to the [Annotated][preserves.values.Annotated]'s
    `annotations` sequence. NOTE: Does not recursively ensure that any parts of the argument
    `v` are themselves wrapped in [Annotated][preserves.values.Annotated] objects!

    ```python
    >>> import preserves
    >>> print(preserves.stringify(annotate(123, "A comment", "Another comment")))
    @"A comment" @"Another comment" 123

    ```
    """
    if not is_annotated(v):
        v = Annotated(v)
    for a in anns:
        v.annotations.append(a)
    return v

canonicalize(v, **kwargs)

As encode, but sets canonicalize=True in the Encoder constructor.

Source code in preserves/binary.py 
436
437
438
439
440
441
def canonicalize(v, **kwargs):
    """As [encode][preserves.binary.encode], but sets `canonicalize=True` in the
    [Encoder][preserves.binary.Encoder] constructor.

    """
    return encode(v, canonicalize=True, **kwargs)

cmp_floats(a, b)

Implements the totalOrder predicate defined in section 5.10 of IEEE Std 754-2008.

Source code in preserves/values.py 
31
32
33
34
35
36
37
38
39
40
def cmp_floats(a, b):
    """Implements the `totalOrder` predicate defined in section 5.10 of [IEEE Std
    754-2008](https://dx.doi.org/10.1109/IEEESTD.2008.4610935).

    """
    a = float_to_int(a)
    b = float_to_int(b)
    if a & 0x8000000000000000: a = a ^ 0x7fffffffffffffff
    if b & 0x8000000000000000: b = b ^ 0x7fffffffffffffff
    return a - b

decode(bs, **kwargs)

Yields the first complete encoded value from bs, passing kwargs through to the Decoder constructor. Raises exceptions as per next.

Parameters:

Name Type Description Default
bs bytes

encoded data to decode

 required
Source code in preserves/binary.py 
247
248
249
250
251
252
253
254
255
256
def decode(bs, **kwargs):
    """Yields the first complete encoded value from `bs`, passing `kwargs` through to the
    [Decoder][preserves.binary.Decoder] constructor. Raises exceptions as per
    [next][preserves.binary.Decoder.next].

    Args:
        bs (bytes): encoded data to decode

    """
    return Decoder(packet=bs, **kwargs).next()

decode_with_annotations(bs, **kwargs)

Like decode, but supplying include_annotations=True to the Decoder constructor.

Source code in preserves/binary.py 
258
259
260
261
def decode_with_annotations(bs, **kwargs):
    """Like [decode][preserves.binary.decode], but supplying `include_annotations=True` to the
    [Decoder][preserves.binary.Decoder] constructor."""
    return Decoder(packet=bs, include_annotations=True, **kwargs).next()

dict_kvs(d)

Generator function yielding a sequence of alternating keys and values from d. In some sense the inverse of ImmutableDict.from_kvs.

>>> list(dict_kvs({'a': 1, 'b': 2}))
['a', 1, 'b', 2]
Source code in preserves/values.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
def dict_kvs(d):
    """Generator function yielding a sequence of alternating keys and values from `d`. In some
    sense the inverse of [ImmutableDict.from_kvs][preserves.values.ImmutableDict.from_kvs].

    ```python
    >>> list(dict_kvs({'a': 1, 'b': 2}))
    ['a', 1, 'b', 2]

    ```
    """
    for k in d:
        yield k
        yield d[k]

encode(v, **kwargs)

Encode a single Value v to a byte string. Any supplied kwargs are passed on to the underlying Encoder constructor.

Source code in preserves/binary.py 
429
430
431
432
433
434
def encode(v, **kwargs):
    """Encode a single `Value` `v` to a byte string. Any supplied `kwargs` are passed on to the
    underlying [Encoder][preserves.binary.Encoder] constructor."""
    e = Encoder(**kwargs)
    e.append(v)
    return e.contents()

is_annotated(v)

True iff v is an instance of Annotated.

Source code in preserves/values.py 
495
496
497
def is_annotated(v):
    """`True` iff `v` is an instance of [Annotated][preserves.values.Annotated]."""
    return isinstance(v, Annotated)

preserve(v)

Converts v to a representation of a Preserves Value by (repeatedly) setting

v = v.__preserve__()

while v has a __preserve__ method. Parsed Schema values are able to render themselves to their serialized representations this way.

Source code in preserves/values.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def preserve(v):
    """Converts `v` to a representation of a Preserves `Value` by (repeatedly) setting

    ```python
    v = v.__preserve__()
    ```

    while `v` has a `__preserve__` method. Parsed [Schema][preserves.schema]
    values are able to render themselves to their serialized representations this way.

    """
    while hasattr(v, '__preserve__'):
        v = v.__preserve__()
    return v

strip_annotations(v, depth=inf)

Exposes depth layers of raw structure of potentially-Annotated Values. If depth==0 or v is not Annotated, just returns v. Otherwise, descends recursively into the structure of v.item.

>>> import preserves
>>> a = preserves.parse('@"A comment" [@a 1 @b 2 @c 3]', include_annotations=True)
>>> is_annotated(a)
True
>>> print(preserves.stringify(a))
@"A comment" [@a 1 @b 2 @c 3]
>>> print(preserves.stringify(strip_annotations(a)))
[1 2 3]
>>> print(preserves.stringify(strip_annotations(a, depth=1)))
[@a 1 @b 2 @c 3]
Source code in preserves/values.py 
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
def strip_annotations(v, depth=inf):
    """Exposes `depth` layers of raw structure of
    potentially-[Annotated][preserves.values.Annotated] `Value`s. If `depth==0` or `v` is not
    [Annotated][preserves.values.Annotated], just returns `v`. Otherwise, descends recursively
    into the structure of `v.item`.

    ```python
    >>> import preserves
    >>> a = preserves.parse('@"A comment" [@a 1 @b 2 @c 3]', include_annotations=True)
    >>> is_annotated(a)
    True
    >>> print(preserves.stringify(a))
    @"A comment" [@a 1 @b 2 @c 3]
    >>> print(preserves.stringify(strip_annotations(a)))
    [1 2 3]
    >>> print(preserves.stringify(strip_annotations(a, depth=1)))
    [@a 1 @b 2 @c 3]

    ```
    """

    if depth == 0: return v
    if not is_annotated(v): return v

    next_depth = depth - 1
    def walk(v):
        return strip_annotations(v, next_depth)

    v = v.item
    if isinstance(v, Record):
        return Record(strip_annotations(v.key, depth), tuple(walk(f) for f in v.fields))
    elif isinstance(v, list):
        return tuple(walk(f) for f in v)
    elif isinstance(v, tuple):
        return tuple(walk(f) for f in v)
    elif isinstance(v, set):
        return frozenset(walk(f) for f in v)
    elif isinstance(v, frozenset):
        return frozenset(walk(f) for f in v)
    elif isinstance(v, dict):
        return ImmutableDict.from_kvs(walk(f) for f in dict_kvs(v))
    elif is_annotated(v):
        raise ValueError('Improper annotation structure')
    else:
        return v