Records as their annotatedValuesThe path expression // [.^ [= Test + = NondeterministicTest]] [. 1 rec] proceeds in three steps:
pip install preserves\nThis package (preserves on pypi.org) implements Preserves for Python 3.x. It provides the core semantics as well as both the human-readable text syntax (a superset of JSON) and machine-oriented binary format (including canonicalization) for Preserves. It also implements Preserves Schema and Preserves Path.
Preserves is a data model, with associated serialization formats.
It supports records with user-defined labels, embedded references, and the usual suite of atomic and compound data types, including binary data as a distinct type from text strings. Its annotations allow separation of data from metadata such as comments, trace information, and provenance information.
Preserves departs from many other data languages in defining how to compare two values. Comparison is based on the data model, not on syntax or on data structures of any particular implementation language.
"},{"location":"#mapping-between-preserves-values-and-python-values","title":"Mapping between Preserves values and Python values","text":"Preserves Values are categorized in the following way:
Value = Atom\n | Compound\n | Embedded\n\n Atom = Boolean\n | Float\n | Double\n | SignedInteger\n | String\n | ByteString\n | Symbol\n\n Compound = Record\n | Sequence\n | Set\n | Dictionary\nPython's strings, byte strings, integers, booleans, and double-precision floats stand directly for their Preserves counterparts. Wrapper objects for Float and Symbol complete the suite of atomic types.
Python's lists and tuples correspond to Preserves Sequences, and dicts and sets to Dictionary and Set values, respectively. Preserves Records are represented by Record objects. Finally, embedded values are represented by Embedded objects.
import preserves\nThe main package re-exports a subset of the exports of its constituent modules:
From preserves.values:
From preserves.error:
From preserves.binary:
From preserves.text:
From preserves.compare:
From preserves.merge:
It also exports the compare and fold modules themselves, permitting patterns like
>>> from preserves import *\n>>> compare.cmp(123, 234)\n-1\nFinally, it provides a few utility aliases for common tasks:
"},{"location":"api/#preserves.dumps","title":"dumps = stringify module-attribute","text":"This alias for stringify provides a familiar pythonesque name for converting a Preserves Value to a string.
loads = parse module-attribute","text":"This alias for parse provides a familiar pythonesque name for converting a string to a Preserves Value.
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'), []))\nb'\\xb4\\xb3\\x02hi\\x84'\n>>> decode(b'\\xb4\\xb3\\x02hi\\x84')\n#hi()\nDecoder(packet=b'', include_annotations=False, decode_embedded=lambda x: x)","text":" Bases: BinaryCodec
Implementation of a decoder for the machine-oriented binary Preserves syntax.
Parameters:
Name Type Description Defaultpacket 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'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> d.next()\n123\n>>> d.next()\n'hello'\n>>> d.next()\n()\n>>> d.next()\nTraceback (most recent call last):\n ...\npreserves.error.ShortPacket: Short packet\nAlternatively, keep calling try_next until it yields None, which is not in the domain of Preserves Values:
>>> d = Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> d.try_next()\n123\n>>> d.try_next()\n'hello'\n>>> d.try_next()\n()\n>>> d.try_next()\nFor 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'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> list(d)\n[123, 'hello', ()]\n>>> for v in Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84'):\n... print(repr(v))\n123\n'hello'\n()\nSupply include_annotations=True to read annotations alongside the annotated values:
>>> d = Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84', include_annotations=True)\n>>> list(d)\n[123, 'hello', @#x ()]\nIf you are incrementally reading from, say, a socket, you can use extend to add new input as if comes available:
>>> d = Decoder(b'\\xa0{\\xb1\\x05he')\n>>> d.try_next()\n123\n>>> d.try_next() # returns None because the input is incomplete\n>>> d.extend(b'llo')\n>>> d.try_next()\n'hello'\n>>> d.try_next()\nAttributes:
Name Type Descriptionpacket bytes buffered input waiting to be processed
index int read position within packet
preserves/binary.py def __init__(self, packet=b'', include_annotations=False, decode_embedded=lambda x: x):\n super(Decoder, self).__init__()\n self.packet = packet\n self.index = 0\n self.include_annotations = include_annotations\n self.decode_embedded = decode_embedded\nextend(data)","text":"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.
preserves/binary.py def extend(self, data):\n\"\"\"Appends `data` to the remaining bytes in `self.packet`, trimming already-processed\n bytes from the front of `self.packet` and resetting `self.index` to zero.\"\"\"\n self.packet = self.packet[self.index:] + data\n self.index = 0\nnext()","text":"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.
preserves/binary.py def next(self):\n\"\"\"Reads the next complete `Value` from the internal buffer, raising\n [ShortPacket][preserves.error.ShortPacket] if too few bytes are available, or\n [DecodeError][preserves.error.DecodeError] if the input is invalid somehow.\n\n \"\"\"\n tag = self.nextbyte()\n if tag == 0x80: return self.wrap(False)\n if tag == 0x81: return self.wrap(True)\n if tag == 0x82: return self.wrap(Float.from_bytes(self.nextbytes(4)))\n if tag == 0x83: return self.wrap(struct.unpack('>d', self.nextbytes(8))[0])\n if tag == 0x84: raise DecodeError('Unexpected end-of-stream marker')\n if tag == 0x85:\n a = self.next()\n v = self.next()\n return self.unshift_annotation(a, v)\n if tag == 0x86:\n if self.decode_embedded is None:\n raise DecodeError('No decode_embedded function supplied')\n return self.wrap(Embedded(self.decode_embedded(self.next())))\n if tag >= 0x90 and tag <= 0x9f: return self.wrap(tag - (0xa0 if tag > 0x9c else 0x90))\n if tag >= 0xa0 and tag <= 0xaf: return self.wrap(self.nextint(tag - 0xa0 + 1))\n if tag == 0xb0: return self.wrap(self.nextint(self.varint()))\n if tag == 0xb1: return self.wrap(self.nextbytes(self.varint()).decode('utf-8'))\n if tag == 0xb2: return self.wrap(self.nextbytes(self.varint()))\n if tag == 0xb3: return self.wrap(Symbol(self.nextbytes(self.varint()).decode('utf-8')))\n if tag == 0xb4:\n vs = self.nextvalues()\n if not vs: raise DecodeError('Too few elements in encoded record')\n return self.wrap(Record(vs[0], vs[1:]))\n if tag == 0xb5: return self.wrap(tuple(self.nextvalues()))\n if tag == 0xb6: return self.wrap(frozenset(self.nextvalues()))\n if tag == 0xb7: return self.wrap(ImmutableDict.from_kvs(self.nextvalues()))\n raise DecodeError('Invalid tag: ' + hex(tag))\ntry_next()","text":"Like next, but returns None instead of raising ShortPacket.
preserves/binary.py def try_next(self):\n\"\"\"Like [next][preserves.binary.Decoder.next], but returns `None` instead of raising\n [ShortPacket][preserves.error.ShortPacket].\"\"\"\n start = self.index\n try:\n return self.next()\n except ShortPacket:\n self.index = start\n return None\nEncoder(encode_embedded=lambda x: x, canonicalize=False, include_annotations=None)","text":" Bases: BinaryCodec
Implementation of an encoder for the machine-oriented binary Preserves syntax.
>>> e = Encoder()\n>>> e.append(123)\n>>> e.append('hello')\n>>> e.append(annotate([], Symbol('x')))\n>>> e.contents()\nb'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84'\nParameters:
Name Type Description Defaultencode_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 Descriptionbuffer bytearray accumulator for the output of the encoder
Source code inpreserves/binary.py def __init__(self,\n encode_embedded=lambda x: x,\n canonicalize=False,\n include_annotations=None):\n super(Encoder, self).__init__()\n self.buffer = bytearray()\n self._encode_embedded = encode_embedded\n self._canonicalize = canonicalize\n if include_annotations is None:\n self.include_annotations = not self._canonicalize\n else:\n self.include_annotations = include_annotations\nappend(v)","text":"Extend self.buffer with an encoding of v.
preserves/binary.py def append(self, v):\n\"\"\"Extend `self.buffer` with an encoding of `v`.\"\"\"\n v = preserve(v)\n if hasattr(v, '__preserve_write_binary__'):\n v.__preserve_write_binary__(self)\n elif v is False:\n self.buffer.append(0x80)\n elif v is True:\n self.buffer.append(0x81)\n elif isinstance(v, float):\n self.buffer.append(0x83)\n self.buffer.extend(struct.pack('>d', v))\n elif isinstance(v, numbers.Number):\n if v >= -3 and v <= 12:\n self.buffer.append(0x90 + (v if v >= 0 else v + 16))\n else:\n self.encodeint(v)\n elif isinstance(v, bytes):\n self.encodebytes(2, v)\n elif isinstance(v, basestring_):\n self.encodebytes(1, v.encode('utf-8'))\n elif isinstance(v, list):\n self.encodevalues(5, v)\n elif isinstance(v, tuple):\n self.encodevalues(5, v)\n elif isinstance(v, set):\n self.encodeset(v)\n elif isinstance(v, frozenset):\n self.encodeset(v)\n elif isinstance(v, dict):\n self.encodedict(v)\n else:\n try:\n i = iter(v)\n except TypeError:\n i = None\n if i is None:\n self.cannot_encode(v)\n else:\n self.encodevalues(5, i)\ncontents()","text":"Returns a bytes constructed from the contents of self.buffer.
preserves/binary.py def contents(self):\n\"\"\"Returns a `bytes` constructed from the contents of `self.buffer`.\"\"\"\n return bytes(self.buffer)\nreset()","text":"Clears self.buffer to a fresh empty bytearray.
preserves/binary.py def reset(self):\n\"\"\"Clears `self.buffer` to a fresh empty `bytearray`.\"\"\"\n self.buffer = bytearray()\ncanonicalize(v, **kwargs)","text":"As encode, but sets canonicalize=True in the Encoder constructor.
preserves/binary.py def canonicalize(v, **kwargs):\n\"\"\"As [encode][preserves.binary.encode], but sets `canonicalize=True` in the\n [Encoder][preserves.binary.Encoder] constructor.\n\n \"\"\"\n return encode(v, canonicalize=True, **kwargs)\ndecode(bs, **kwargs)","text":"Yields the first complete encoded value from bs, passing kwargs through to the Decoder constructor. Raises exceptions as per next.
Parameters:
Name Type Description Defaultbs bytes encoded data to decode
required Source code inpreserves/binary.py def decode(bs, **kwargs):\n\"\"\"Yields the first complete encoded value from `bs`, passing `kwargs` through to the\n [Decoder][preserves.binary.Decoder] constructor. Raises exceptions as per\n [next][preserves.binary.Decoder.next].\n\n Args:\n bs (bytes): encoded data to decode\n\n \"\"\"\n return Decoder(packet=bs, **kwargs).next()\ndecode_with_annotations(bs, **kwargs)","text":"Like decode, but supplying include_annotations=True to the Decoder constructor.
preserves/binary.py def decode_with_annotations(bs, **kwargs):\n\"\"\"Like [decode][preserves.binary.decode], but supplying `include_annotations=True` to the\n [Decoder][preserves.binary.Decoder] constructor.\"\"\"\n return Decoder(packet=bs, include_annotations=True, **kwargs).next()\nencode(v, **kwargs)","text":"Encode a single Value v to a byte string. Any supplied kwargs are passed on to the underlying Encoder constructor.
preserves/binary.py def encode(v, **kwargs):\n\"\"\"Encode a single `Value` `v` to a byte string. Any supplied `kwargs` are passed on to the\n underlying [Encoder][preserves.binary.Encoder] constructor.\"\"\"\n e = Encoder(**kwargs)\n e.append(v)\n return e.contents()\nPreserves specifies a total ordering and an equivalence between terms. The preserves.compare module implements the ordering and equivalence relations.
>>> cmp(\"bzz\", \"c\")\n-1\n>>> cmp(True, [])\n-1\n>>> lt(\"bzz\", \"c\")\nTrue\n>>> eq(\"bzz\", \"c\")\nFalse\nNote that the ordering relates more values than Python's built-in ordering:
>>> [1, 2, 2] < [1, 2, \"3\"]\nTraceback (most recent call last):\n ..\nTypeError: '<' not supported between instances of 'int' and 'str'\n\n>>> lt([1, 2, 2], [1, 2, \"3\"])\nTrue\ncmp(a, b)","text":"Returns -1 if a < b, or 0 if a = b, or 1 if a > b according to the Preserves total order.
preserves/compare.py def cmp(a, b):\n\"\"\"Returns `-1` if `a` < `b`, or `0` if `a` = `b`, or `1` if `a` > `b` according to the\n [Preserves total order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return _cmp(preserve(a), preserve(b))\neq(a, b)","text":"Returns True iff a = b according to the Preserves equivalence relation.
preserves/compare.py def eq(a, b):\n\"\"\"Returns `True` iff `a` = `b` according to the [Preserves equivalence\n relation](https://preserves.dev/preserves.html#equivalence).\"\"\"\n return _eq(preserve(a), preserve(b))\nle(a, b)","text":"Returns True iff a \u2264 b according to the Preserves total order.
preserves/compare.py def le(a, b):\n\"\"\"Returns `True` iff `a` \u2264 `b` according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return cmp(a, b) <= 0\nlt(a, b)","text":"Returns True iff a < b according to the Preserves total order.
preserves/compare.py def lt(a, b):\n\"\"\"Returns `True` iff `a` < `b` according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return cmp(a, b) < 0\nsorted(iterable, *, key=lambda x: x, reverse=False)","text":"Returns a sorted list built from iterable, extracting a sort key using key, and ordering according to the Preserves total order. Directly analogous to the built-in Python sorted routine, except uses the Preserves order instead of Python's less-than relation.
preserves/compare.py def sorted(iterable, *, key=lambda x: x, reverse=False):\n\"\"\"Returns a sorted list built from `iterable`, extracting a sort key using `key`, and\n ordering according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order). Directly analogous to the\n [built-in Python `sorted`\n routine](https://docs.python.org/3/library/functions.html#sorted), except uses the\n Preserves order instead of Python's less-than relation.\n\n \"\"\"\n return _sorted(iterable, key=lambda x: _key(key(x)), reverse=reverse)\nsorted_items(d)","text":"Given a dictionary d, yields a list of (key, value) tuples sorted by key.
preserves/compare.py def sorted_items(d):\n\"\"\"Given a dictionary `d`, yields a list of `(key, value)` tuples sorted by `key`.\"\"\"\n return sorted(d.items(), key=_item_key)\nThe preserves.error module exports various Error subclasses.
DecodeError","text":" Bases: ValueError
Raised whenever preserves.binary.Decoder or preserves.text.Parser detect invalid input.
"},{"location":"error/#preserves.error.EncodeError","title":"EncodeError","text":" Bases: ValueError
Raised whenever preserves.binary.Encoder or preserves.text.Formatter are unable to proceed.
"},{"location":"error/#preserves.error.ShortPacket","title":"ShortPacket","text":" 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.
"},{"location":"fold/","title":"Traversing values","text":"The preserves.fold module exports various utilities for traversing compound Values.
map_embeddeds(f, v)","text":"Returns an equivalent copy of v, except where each contained Embedded value is replaced by f applied to the Embedded's embeddedValue attribute.
>>> map_embeddeds(lambda w: Embedded(f'w={w}'), ['a', Embedded(123), {'z': 6.0}])\n('a', #!'w=123', {'z': 6.0})\npreserves/fold.py def map_embeddeds(f, v):\n\"\"\"Returns an [equivalent][preserves.compare.eq] copy of `v`, except where each contained\n [Embedded][preserves.values.Embedded] value is replaced by `f` applied to the Embedded's\n `embeddedValue` attribute.\n\n ```python\n >>> map_embeddeds(lambda w: Embedded(f'w={w}'), ['a', Embedded(123), {'z': 6.0}])\n ('a', #!'w=123', {'z': 6.0})\n\n ```\n \"\"\"\n def walk(v):\n if isinstance(v, Embedded):\n return f(v.embeddedValue)\n elif isinstance(v, (list, tuple)):\n return tuple(walk(w) for w in v)\n elif isinstance(v, (set, frozenset)):\n return frozenset(walk(w) for w in v)\n elif isinstance(v, dict):\n return ImmutableDict.from_kvs(walk(w) for w in dict_kvs(v))\n elif isinstance(v, Record):\n return Record(walk(v.key), walk(v.fields))\n else:\n return v\n return walk(v)\nThe preserves.merge module exports various utilities for merging Values.
merge(v0, *vs, merge_embedded=None)","text":"Repeatedly merges v0 with each element in vs using merge2, returning the final result. The merge_embedded parameter is passed on to merge2.
preserves/merge.py def merge(v0, *vs, merge_embedded=None):\n\"\"\"Repeatedly merges `v0` with each element in `vs` using [merge2][preserves.merge.merge2],\n returning the final result. The `merge_embedded` parameter is passed on to merge2.\"\"\"\n v = v0\n for vN in vs:\n v = merge2(v, vN, merge_embedded=merge_embedded)\n return v\nmerge2(a, b, merge_embedded=None)","text":"Merges a and b, returning the result. Raises ValueError if, during the merge, a pair of incompatible values is discovered.
If a and b are Embedded objects, their embeddedValues are merged using merge_embedded, and the result is again wrapped in an Embedded object.
>>> merge2(123, 234)\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2(123, 123)\n123\n>>> merge2('hi', 0)\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2([1, 2], [1, 2])\n[1, 2]\n>>> merge2([1, 2], [1, 3])\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2({'a': 1, 'b': 2}, {'a': 1, 'c': 3})\n{'a': 1, 'b': 2, 'c': 3}\n>>> merge2({'a': 1, 'b': 2}, {'a': 10, 'c': 3})\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2(Record('a', [1, {'x': 2}]), Record('a', [1, {'y': 3}]))\na(1, {'x': 2, 'y': 3})\npreserves/merge.py def merge2(a, b, merge_embedded=None):\n\"\"\"Merges `a` and `b`, returning the result. Raises `ValueError` if, during the merge, a\n pair of incompatible values is discovered.\n\n If `a` and `b` are [Embedded][preserves.values.Embedded] objects, their `embeddedValue`s\n are merged using `merge_embedded`, and the result is again wrapped in an\n [Embedded][preserves.values.Embedded] object.\n\n ```python\n >>> merge2(123, 234)\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2(123, 123)\n 123\n >>> merge2('hi', 0)\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2([1, 2], [1, 2])\n [1, 2]\n >>> merge2([1, 2], [1, 3])\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2({'a': 1, 'b': 2}, {'a': 1, 'c': 3})\n {'a': 1, 'b': 2, 'c': 3}\n >>> merge2({'a': 1, 'b': 2}, {'a': 10, 'c': 3})\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2(Record('a', [1, {'x': 2}]), Record('a', [1, {'y': 3}]))\n a(1, {'x': 2, 'y': 3})\n\n ```\n\n \"\"\"\n if a == b:\n return a\n if isinstance(a, (list, tuple)) and isinstance(b, (list, tuple)):\n return merge_seq(a, b)\n if isinstance(a, (set, frozenset)) and isinstance(b, (set, frozenset)):\n _die()\n if isinstance(a, dict) and isinstance(b, dict):\n r = {}\n for (ak, av) in a.items():\n bv = b.get(ak, None)\n r[ak] = av if bv is None else merge2(av, bv, merge_embedded=merge_embedded)\n for (bk, bv) in b.items():\n if bk not in r:\n r[bk] = bv\n return r\n if isinstance(a, Record) and isinstance(b, Record):\n return Record(merge2(a.key, b.key, merge_embedded=merge_embedded),\n merge_seq(a.fields, b.fields, merge_embedded=merge_embedded))\n if isinstance(a, Embedded) and isinstance(b, Embedded):\n m = (merge_embedded or merge_embedded_id)(a.embeddedValue, b.embeddedValue)\n if m is None: _die()\n return Embedded(m)\n _die()\nThe preserves.path module implements Preserves Path.
Preserves Path is roughly analogous to XPath, but for Preserves values: just as XPath selects portions of an XML document, a Preserves Path uses path expressions to select portions of a Value.
Use parse to compile a path expression, and then use the exec method on the result to apply it to a given input:
parse(PATH_EXPRESSION_STRING).exec(PRESERVES_VALUE)\n -> SEQUENCE_OF_PRESERVES_VALUES\nWhen preserves.path is run as a __main__ module, sys.argv[1] is parsed, interpreted as a path expression, and run against human-readable values read from standard input. Each matching result is passed to stringify and printed to standard output.
The following examples use testdata:
>>> with open('tests/samples.bin', 'rb') as f:\n... testdata = decode_with_annotations(f.read())\nRecall that samples.bin contains a binary-syntax form of the human-readable [samples.pr](https://preserves.dev/tests/samples.pr) test data file, intended to exercise most of the features of Preserves. In particular, the rootValue` in the file has a number of annotations (for documentation and other purposes).
The path expression .annotations ^ Documentation . 0 / string proceeds in five steps:
.annotations selects each annotation on the root document^ Documentation retains only those values (each an annotation of the root) that are Records with label equal to the symbol Documentation. 0 moves into the first child (the first field) of each such Record, which in our case is a list of other Values/ selects all immediate children of these listsstring retains only those values that are stringsThe result of evaluating it on testdata is as follows:
>>> selector = parse('.annotations ^ Documentation . 0 / string')\n>>> for result in selector.exec(testdata):\n... print(stringify(result))\n\"Individual test cases may be any of the following record types:\"\n\"In each test, let value = strip(annotatedValue),\"\n\" forward = value,\"\n\" back = value,\"\n\"except where test-case-specific values of `forward` and/or `back`\"\n\"are provided by the executing harness, and check the following\"\n\"numbered expectations according to the table above:\"\n\"Implementations may vary in their treatment of the difference between expectations\"\n\"13/14 and 16/17, depending on how they wish to treat end-of-stream conditions.\"\nRecords as their annotatedValues","text":"The path expression // [.^ [= Test + = NondeterministicTest]] [. 1 rec] proceeds in three steps:
// recursively decomposes the input, yielding all direct and indirect descendants of each input value
[.^ [= Test + = NondeterministicTest]] retains only those inputs (each a descendant of the root) that yield more than zero results when executed against the expression within the brackets:
.^ selects only labels of values that are Records, filtering by type and transforming in a single step[= Test + = NondeterministicTest] again filters by a path expression:+ operator takes the union of matches of its arguments= Test selects values (remember, record labels) equal to the symbol Test= NondeterministicTest selects values equal to NondeterministicTestThe result is thus all Records anywhere inside testdata that have either Test or NondeterministicTest as their labels.
[. 1 rec] filters these Records by another path expression:
. 1 selects their second field (fields are numbered from 0)rec retains only values that are RecordsEvaluating the expression against testdata yields the following:
>>> selector = parse('// [.^ [= Test + = NondeterministicTest]] [. 1 rec]')\n>>> for result in selector.exec(testdata):\n... print(stringify(result))\n<Test #[tLMHY2FwdHVyZbSzB2Rpc2NhcmSEhA==] <capture <discard>>>\n<Test #[tLMHb2JzZXJ2ZbSzBXNwZWFrtLMHZGlzY2FyZIS0swdjYXB0dXJltLMHZGlzY2FyZISEhIQ=] <observe <speak <discard> <capture <discard>>>>>\n<Test #[tLWzBnRpdGxlZLMGcGVyc29ukrMFdGhpbmeRhKBlsQlCbGFja3dlbGy0swRkYXRloQcdkpOEsQJEcoQ=] <[titled person 2 thing 1] 101 \"Blackwell\" <date 1821 2 3> \"Dr\">>\n<Test #[tLMHZGlzY2FyZIQ=] <discard>>\n<Test #[tJe1hIQ=] <7 []>>\n<Test #[tLMHZGlzY2FyZLMIc3VycHJpc2WE] <discard surprise>>\n<Test #[tLEHYVN0cmluZ5OUhA==] <\"aString\" 3 4>>\n<Test #[tLSzB2Rpc2NhcmSEk5SE] <<discard> 3 4>>\n<Test #[hbMCYXK0swFShbMCYWazAWaE] @ar <R @af f>>\n<Test #[tIWzAmFyswFShbMCYWazAWaE] <@ar R @af f>>\nPredicate = syntax.Predicate module-attribute","text":"Schema definition for representing a Preserves Path Predicate.
Selector = syntax.Selector module-attribute","text":"Schema definition for representing a sequence of Preserves Path Steps.
syntax = load_schema_file(pathlib.Path(__file__).parent / 'path.prb').path module-attribute","text":"This value is a Python representation of a Preserves Schema definition for the Preserves Path expression language. The language is defined in the file path.prs.
"},{"location":"path/#preserves.path.exec","title":"exec(self, v)","text":"WARNING: This is not a function: it is a method on Selector, Predicate, and so on.
>>> sel = parse('/ [.length gt 1]')\n>>> sel.exec(['', 'a', 'ab', 'abc', 'abcd', 'bcd', 'cd', 'd', ''])\n('ab', 'abc', 'abcd', 'bcd', 'cd')\npreserves/path.py @extend(syntax.Function)\ndef exec(self, v):\n\"\"\"WARNING: This is not a *function*: it is a *method* on\n [Selector][preserves.path.Selector], [Predicate][preserves.path.Predicate], and so on.\n\n ```python\n >>> sel = parse('/ [.length gt 1]')\n >>> sel.exec(['', 'a', 'ab', 'abc', 'abcd', 'bcd', 'cd', 'd', ''])\n ('ab', 'abc', 'abcd', 'bcd', 'cd')\n\n ```\n\n \"\"\"\n return (len(self.selector.exec(v)),)\nparse(s)","text":"Parse s as a Preserves Path path expression, yielding a Selector object. Selectors (and Predicates etc.) have an exec method defined on them.
Raises ValueError if s is not a valid path expression.
preserves/path.py def parse(s):\n\"\"\"Parse `s` as a Preserves Path path expression, yielding a\n [Selector][preserves.path.Selector] object. Selectors (and Predicates etc.) have an\n [exec][preserves.path.exec] method defined on them.\n\n Raises `ValueError` if `s` is not a valid path expression.\n\n \"\"\"\n return parse_selector(Parser(s))\nThis is an implementation of Preserves Schema for Python 3.
TODO
"},{"location":"schema/#preserves.schema.meta","title":"meta = load_schema_file(__metaschema_filename).schema module-attribute","text":"TODO
"},{"location":"schema/#preserves.schema.Compiler","title":"Compiler()","text":"TODO
Source code inpreserves/schema.py def __init__(self):\n self.root = Namespace(())\nDefinition(*args, **kwargs)","text":" Bases: SchemaObject
TODO
Source code inpreserves/schema.py def __init__(self, *args, **kwargs):\n self._fields = args\n if self.SIMPLE:\n if self.EMPTY:\n if len(args) != 0:\n raise TypeError('%s takes no arguments' % (self._constructor_name(),))\n else:\n if len(args) != 1:\n raise TypeError('%s needs exactly one argument' % (self._constructor_name(),))\n self.value = args[0]\n else:\n i = 0\n for arg in args:\n if i >= len(self.FIELD_NAMES):\n raise TypeError('%s given too many positional arguments' % (self._constructor_name(),))\n setattr(self, self.SAFE_FIELD_NAMES[i], arg)\n i = i + 1\n for (argname, arg) in kwargs.items():\n if hasattr(self, argname):\n raise TypeError('%s given duplicate attribute: %r' % (self._constructor_name, argname))\n if argname not in self.SAFE_FIELD_NAMES:\n raise TypeError('%s given unknown attribute: %r' % (self._constructor_name, argname))\n setattr(self, argname, arg)\n i = i + 1\n if i != len(self.FIELD_NAMES):\n raise TypeError('%s needs argument(s) %r' % (self._constructor_name(), self.FIELD_NAMES))\nEnumeration()","text":" Bases: SchemaObject
TODO
Source code inpreserves/schema.py def __init__(self):\n raise TypeError('Cannot create instance of Enumeration')\nNamespace(prefix)","text":"TODO
Source code inpreserves/schema.py def __init__(self, prefix):\n self._prefix = prefix\nSchemaDecodeFailed(cls, p, v, failures=None)","text":" Bases: ValueError
TODO
Source code inpreserves/schema.py def __init__(self, cls, p, v, failures=None):\n super().__init__()\n self.cls = cls\n self.pattern = p\n self.value = v\n self.failures = [] if failures is None else failures\nSchemaObject","text":"TODO
"},{"location":"schema/#preserves.schema.SchemaObject.__preserve__","title":"__preserve__()","text":"TODO
Source code inpreserves/schema.py def __preserve__(self):\n\"\"\"TODO\"\"\"\n raise NotImplementedError('Subclass responsibility')\ndecode(v) classmethod","text":"TODO
Source code inpreserves/schema.py @classmethod\ndef decode(cls, v):\n\"\"\"TODO\"\"\"\n raise NotImplementedError('Subclass responsibility')\ntry_decode(v) classmethod","text":"TODO
Source code inpreserves/schema.py @classmethod\ndef try_decode(cls, v):\n\"\"\"TODO\"\"\"\n try:\n return cls.decode(v)\n except SchemaDecodeFailed:\n return None\nencode(p, v)","text":"TODO
Source code inpreserves/schema.py def encode(p, v):\n\"\"\"TODO\"\"\"\n if hasattr(v, '__escape_schema__'):\n return preserve(v.__escape_schema__())\n if p == ANY:\n return v\n if p.key == NAMED:\n return encode(p[1], safegetattr(v, p[0].name))\n if p.key == ATOM:\n return v\n if p.key == EMBEDDED:\n return Embedded(v)\n if p.key == LIT:\n return p[0]\n if p.key == SEQOF:\n return tuple(encode(p[0], w) for w in v)\n if p.key == SETOF:\n return set(encode(p[0], w) for w in v)\n if p.key == DICTOF:\n return dict((encode(p[0], k), encode(p[1], w)) for (k, w) in v.items())\n if p.key == REF:\n return preserve(v)\n if p.key == REC:\n return Record(encode(p[0], v), encode(p[1], v))\n if p.key == TUPLE:\n return tuple(encode(pp, v) for pp in p[0])\n if p.key == TUPLE_PREFIX:\n return tuple(encode(pp, v) for pp in p[0]) + encode(p[1], v)\n if p.key == DICT:\n return dict((k, encode(pp, v)) for (k, pp) in p[0].items())\n if p.key == AND:\n return merge(*[encode(pp, v) for pp in p[0]])\n raise ValueError(f'Bad schema {p}')\nextend(cls)","text":"TODO
Source code inpreserves/schema.py def extend(cls):\n\"\"\"TODO\"\"\"\n @wraps(cls)\n def extender(f):\n setattr(cls, f.__name__, f)\n return f\n return extender\nload_schema_file(filename)","text":"TODO
Source code inpreserves/schema.py def load_schema_file(filename):\n\"\"\"TODO\"\"\"\n c = Compiler()\n c.load(filename)\n return c.root\nThe preserves.text module implements the Preserves human-readable text syntax.
The main entry points are functions stringify, parse, and parse_with_annotations.
>>> stringify(Record(Symbol('hi'), [1, [2, 3]]))\n'<hi 1 [2 3]>'\n>>> parse('<hi 1 [2 3]>')\n#hi(1, (2, 3))\nFormatter(format_embedded=lambda x: x, indent=None, with_commas=False, trailing_comma=False, include_annotations=True)","text":" Bases: TextCodec
Printer (and indenting pretty-printer) for producing human-readable syntax from Preserves Values.
>>> f = Formatter()\n>>> f.append({'a': 1, 'b': 2})\n>>> f.append(Record(Symbol('label'), ['field1', ['field2item1', 'field2item2']]))\n>>> print(f.contents())\n{\"a\": 1 \"b\": 2} <label \"field1\" [\"field2item1\" \"field2item2\"]>\n\n>>> f = Formatter(indent=4)\n>>> f.append({'a': 1, 'b': 2})\n>>> f.append(Record(Symbol('label'), ['field1', ['field2item1', 'field2item2']]))\n>>> print(f.contents())\n{\n \"a\": 1\n \"b\": 2\n}\n<label \"field1\" [\n \"field2item1\"\n \"field2item2\"\n]>\nParameters:
Name Type Description Defaultformat_embedded function accepting an Embedded.embeddedValue and returning a Value for serialization.
lambda x: x indent int | None None disables indented pretty-printing; otherwise, an int specifies indentation per nesting-level.
None with_commas bool True causes commas to separate sequence and set items and dictionary entries; False omits commas.
False trailing_comma bool True causes a comma to be printed after the final item or entry in a sequence, set or dictionary; False omits this trailing comma
False include_annotations bool True causes annotations to be included in the output; False causes them to be omitted.
True Attributes:
Name Type Descriptionindent_delta int indentation per nesting-level
chunks list[str] fragments of output
Source code inpreserves/text.py def __init__(self,\n format_embedded=lambda x: x,\n indent=None,\n with_commas=False,\n trailing_comma=False,\n include_annotations=True):\n super(Formatter, self).__init__()\n self.indent_delta = 0 if indent is None else indent\n self.indent_distance = 0\n self.nesting = 0\n self.with_commas = with_commas\n self.trailing_comma = trailing_comma\n self.chunks = []\n self._format_embedded = format_embedded\n self.include_annotations = include_annotations\nappend(v)","text":"Extend self.chunks with at least one chunk, together making up the text representation of v.
preserves/text.py def append(self, v):\n\"\"\"Extend `self.chunks` with at least one chunk, together making up the text\n representation of `v`.\"\"\"\n if self.chunks and self.nesting == 0:\n self.write_indent_space()\n try:\n self.nesting += 1\n self._append(v)\n finally:\n self.nesting -= 1\ncontents()","text":"Returns a str constructed from the join of the chunks in self.chunks.
preserves/text.py def contents(self):\n\"\"\"Returns a `str` constructed from the join of the chunks in `self.chunks`.\"\"\"\n return u''.join(self.chunks)\nis_indenting()","text":"Returns True iff this Formatter is in pretty-printing indenting mode.
preserves/text.py def is_indenting(self):\n\"\"\"Returns `True` iff this [Formatter][preserves.text.Formatter] is in pretty-printing\n indenting mode.\"\"\"\n return self.indent_delta > 0\nParser(input_buffer='', include_annotations=False, parse_embedded=lambda x: x)","text":" Bases: TextCodec
Parser for the human-readable Preserves text syntax.
Parameters:
Name Type Description Defaultinput_buffer str initial contents of the input buffer; may subsequently be extended by calling extend.
'' include_annotations bool if True, wrap each value and subvalue in an Annotated object.
False parse_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 input text, and keep calling next until a ShortPacket exception is raised:
>>> d = Parser('123 \"hello\" @x []')\n>>> d.next()\n123\n>>> d.next()\n'hello'\n>>> d.next()\n()\n>>> d.next()\nTraceback (most recent call last):\n ...\npreserves.error.ShortPacket: Short input buffer\nAlternatively, keep calling try_next until it yields None, which is not in the domain of Preserves Values:
>>> d = Parser('123 \"hello\" @x []')\n>>> d.try_next()\n123\n>>> d.try_next()\n'hello'\n>>> d.try_next()\n()\n>>> d.try_next()\nFor convenience, Parser implements the iterator interface, backing it with try_next, so you can simply iterate over all complete values in an input:
>>> d = Parser('123 \"hello\" @x []')\n>>> list(d)\n[123, 'hello', ()]\n>>> for v in Parser('123 \"hello\" @x []'):\n... print(repr(v))\n123\n'hello'\n()\nSupply include_annotations=True to read annotations alongside the annotated values:
>>> d = Parser('123 \"hello\" @x []', include_annotations=True)\n>>> list(d)\n[123, 'hello', @#x ()]\nIf you are incrementally reading from, say, a socket, you can use extend to add new input as if comes available:
>>> d = Parser('123 \"he')\n>>> d.try_next()\n123\n>>> d.try_next() # returns None because the input is incomplete\n>>> d.extend('llo\"')\n>>> d.try_next()\n'hello'\n>>> d.try_next()\nAttributes:
Name Type Descriptioninput_buffer str buffered input waiting to be processed
index int read position within input_buffer
preserves/text.py def __init__(self, input_buffer=u'', include_annotations=False, parse_embedded=lambda x: x):\n super(Parser, self).__init__()\n self.input_buffer = input_buffer\n self.index = 0\n self.include_annotations = include_annotations\n self.parse_embedded = parse_embedded\nextend(text)","text":"Appends text to the remaining contents of self.input_buffer, trimming already-processed text from the front of self.input_buffer and resetting self.index to zero.
preserves/text.py def extend(self, text):\n\"\"\"Appends `text` to the remaining contents of `self.input_buffer`, trimming already-processed\n text from the front of `self.input_buffer` and resetting `self.index` to zero.\"\"\"\n self.input_buffer = self.input_buffer[self.index:] + text\n self.index = 0\nnext()","text":"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.
preserves/text.py def next(self):\n\"\"\"Reads the next complete `Value` from the internal buffer, raising\n [ShortPacket][preserves.error.ShortPacket] if too few bytes are available, or\n [DecodeError][preserves.error.DecodeError] if the input is invalid somehow.\n\n \"\"\"\n self.skip_whitespace()\n c = self.peek()\n if c == '\"':\n self.skip()\n return self.wrap(self.read_string('\"'))\n if c == '|':\n self.skip()\n return self.wrap(Symbol(self.read_string('|')))\n if c in ';@':\n annotations = self.gather_annotations()\n v = self.next()\n if self.include_annotations:\n v.annotations = annotations + v.annotations\n return v\n if c == ':':\n raise DecodeError('Unexpected key/value separator between items')\n if c == '#':\n self.skip()\n c = self.nextchar()\n if c == 'f': return self.wrap(False)\n if c == 't': return self.wrap(True)\n if c == '{': return self.wrap(frozenset(self.upto('}')))\n if c == '\"': return self.wrap(self.read_literal_binary())\n if c == 'x':\n c = self.nextchar()\n if c == '\"': return self.wrap(self.read_hex_binary())\n if c == 'f': return self.wrap(self.read_hex_float(4))\n if c == 'd': return self.wrap(self.read_hex_float(8))\n raise DecodeError('Invalid #x syntax')\n if c == '[': return self.wrap(self.read_base64_binary())\n if c == '=':\n old_ann = self.include_annotations\n self.include_annotations = True\n bs_val = self.next()\n self.include_annotations = old_ann\n if len(bs_val.annotations) > 0:\n raise DecodeError('Annotations not permitted after #=')\n bs_val = bs_val.item\n if not isinstance(bs_val, bytes):\n raise DecodeError('ByteString must follow #=')\n return self.wrap(Decoder(bs_val, include_annotations = self.include_annotations).next())\n if c == '!':\n if self.parse_embedded is None:\n raise DecodeError('No parse_embedded function supplied')\n return self.wrap(Embedded(self.parse_embedded(self.next())))\n raise DecodeError('Invalid # syntax')\n if c == '<':\n self.skip()\n vs = self.upto('>')\n if len(vs) == 0:\n raise DecodeError('Missing record label')\n return self.wrap(Record(vs[0], vs[1:]))\n if c == '[':\n self.skip()\n return self.wrap(self.upto(']'))\n if c == '{':\n self.skip()\n return self.wrap(self.read_dictionary())\n if c in '>]}':\n raise DecodeError('Unexpected ' + c)\n self.skip()\n return self.wrap(self.read_raw_symbol_or_number([c]))\ntry_next()","text":"Like next, but returns None instead of raising ShortPacket.
preserves/text.py def try_next(self):\n\"\"\"Like [next][preserves.text.Parser.next], but returns `None` instead of raising\n [ShortPacket][preserves.error.ShortPacket].\"\"\"\n start = self.index\n try:\n return self.next()\n except ShortPacket:\n self.index = start\n return None\nparse(text, **kwargs)","text":"Yields the first complete encoded value from text, passing kwargs through to the Parser constructor. Raises exceptions as per next.
Parameters:
Name Type Description Defaulttext str encoded data to decode
required Source code inpreserves/text.py def parse(text, **kwargs):\n\"\"\"Yields the first complete encoded value from `text`, passing `kwargs` through to the\n [Parser][preserves.text.Parser] constructor. Raises exceptions as per\n [next][preserves.text.Parser.next].\n\n Args:\n text (str): encoded data to decode\n\n \"\"\"\n return Parser(input_buffer=text, **kwargs).next()\nparse_with_annotations(bs, **kwargs)","text":"Like parse, but supplying include_annotations=True to the Parser constructor.
preserves/text.py def parse_with_annotations(bs, **kwargs):\n\"\"\"Like [parse][preserves.text.parse], but supplying `include_annotations=True` to the\n [Parser][preserves.text.Parser] constructor.\"\"\"\n return Parser(input_buffer=bs, include_annotations=True, **kwargs).next()\nstringify(v, **kwargs)","text":"Convert a single Value v to a string. Any supplied kwargs are passed on to the underlying Formatter constructor.
preserves/text.py def stringify(v, **kwargs):\n\"\"\"Convert a single `Value` `v` to a string. Any supplied `kwargs` are passed on to the\n underlying [Formatter][preserves.text.Formatter] constructor.\"\"\"\n e = Formatter(**kwargs)\n e.append(v)\n return e.contents()\nPython's strings, byte strings, integers, booleans, and double-precision floats stand directly for their Preserves counterparts. Wrapper objects for Float and Symbol complete the suite of atomic types.
Python's lists and tuples correspond to Preserves Sequences, and dicts and sets to Dictionary and Set values, respectively. Preserves Records are represented by Record objects. Finally, embedded values are represented by Embedded objects.
The preserves.values module implements the core representations of Preserves Values as Python values.
Annotated(item)","text":" 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\n>>> a = preserves.parse('''\n... ; A comment\n... [1 2 3]\n... ''', include_annotations=True)\n>>> a\n@' A comment' (1, 2, 3)\n>>> a.item\n(1, 2, 3)\n>>> a.annotations\n[' A comment']\n>>> a == (1, 2, 3)\nTrue\n>>> a == preserves.parse('@xyz [1 2 3]', include_annotations=True)\nTrue\n>>> a[0]\nTraceback (most recent call last):\n ...\nTypeError: 'Annotated' object is not subscriptable\n>>> a.item[0]\n1\n>>> type(a.item[0])\n<class 'preserves.values.Annotated'>\n>>> a.item[0].annotations\n[]\n>>> print(preserves.stringify(a))\n@\" A comment\" [1 2 3]\n>>> print(preserves.stringify(a, include_annotations=False))\n[1 2 3]\nAttributes:
Name Type Descriptionitem Value the underlying annotated Value
annotations list[Value] the annotations attached to self.item
preserves/values.py def __init__(self, item):\n self.annotations = []\n self.item = item\npeel()","text":"Calls strip_annotations on self with depth=1.
preserves/values.py def peel(self):\n\"\"\"Calls [strip_annotations][preserves.values.strip_annotations] on `self` with `depth=1`.\"\"\"\n return strip_annotations(self, 1)\nstrip(depth=inf)","text":"Calls strip_annotations on self and depth.
preserves/values.py def strip(self, depth=inf):\n\"\"\"Calls [strip_annotations][preserves.values.strip_annotations] on `self` and `depth`.\"\"\"\n return strip_annotations(self, depth)\nEmbedded(embeddedValue)","text":"Representation of a Preserves Embedded value. For more on the meaning and use of embedded values, see the specification.
>>> import io\n>>> e = Embedded(io.StringIO('some text'))\n>>> e # doctest: +ELLIPSIS\n#!<_io.StringIO object at ...>\n>>> e.embeddedValue # doctest: +ELLIPSIS\n<_io.StringIO object at ...>\n>>> import preserves\n>>> print(preserves.stringify(Embedded(None)))\nTraceback (most recent call last):\n ...\nTypeError: Cannot preserves-format: None\n>>> print(preserves.stringify(Embedded(None), format_embedded=lambda x: 'abcdef'))\n#!\"abcdef\"\nAttributes:
Name Type DescriptionembeddedValue any Python value; could be a platform object, could be a representation of a Preserves Value, could be None, could be anything!
preserves/values.py def __init__(self, embeddedValue):\n self.embeddedValue = embeddedValue\nFloat(value)","text":" Bases: object
Wrapper for treating a Python double-precision floating-point value as a single-precision (32-bit) float, from Preserves' perspective. (Python lacks native single-precision floating point support.)
>>> Float(3.45)\nFloat(3.45)\n>>> import preserves\n>>> preserves.stringify(Float(3.45))\n'3.45f'\n>>> preserves.stringify(3.45)\n'3.45'\n>>> preserves.parse('3.45f')\nFloat(3.45)\n>>> preserves.parse('3.45')\n3.45\n>>> preserves.encode(Float(3.45))\nb'\\x82@\\\\\\xcc\\xcd'\n>>> preserves.encode(3.45)\nb'\\x83@\\x0b\\x99\\x99\\x99\\x99\\x99\\x9a'\nAttributes:
Name Type Descriptionvalue float the double-precision representation of intended single-precision value
Source code inpreserves/values.py def __init__(self, value):\n self.value = value\nfrom_bytes(bs) staticmethod","text":"Converts a 4-byte-long byte string to a 32-bit single-precision floating point value wrapped in a Float instance. Takes care to preserve the quiet/signalling bit-pattern of NaN values, unlike its struct.unpack('>f', ...) equivalent.
>>> Float.from_bytes(b'\\x7f\\x80\\x00{')\nFloat(nan)\n>>> Float.from_bytes(b'\\x7f\\x80\\x00{').to_bytes()\nb'\\x7f\\x80\\x00{'\n\n>>> struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]\nnan\n>>> Float(struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]).to_bytes()\nb'\\x7f\\xc0\\x00{'\n>>> struct.pack('>f', struct.unpack('>f', b'\\x7f\\x80\\x00{')[0])\nb'\\x7f\\xc0\\x00{'\n(Note well the difference between 7f80007b and 7fc0007b!)
preserves/values.py @staticmethod\ndef from_bytes(bs):\n\"\"\"Converts a 4-byte-long byte string to a 32-bit single-precision floating point value\n wrapped in a [Float][preserves.values.Float] instance. Takes care to preserve the\n quiet/signalling bit-pattern of NaN values, unlike its `struct.unpack('>f', ...)`\n equivalent.\n\n ```python\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{')\n Float(nan)\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{').to_bytes()\n b'\\\\x7f\\\\x80\\\\x00{'\n\n >>> struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]\n nan\n >>> Float(struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]).to_bytes()\n b'\\\\x7f\\\\xc0\\\\x00{'\n >>> struct.pack('>f', struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0])\n b'\\\\x7f\\\\xc0\\\\x00{'\n\n ```\n\n (Note well the difference between `7f80007b` and `7fc0007b`!)\n\n \"\"\"\n vf = struct.unpack('>I', bs)[0]\n if (vf & 0x7f800000) == 0x7f800000:\n # NaN or inf. Preserve quiet/signalling bit by manually expanding to double-precision.\n sign = vf >> 31\n payload = vf & 0x007fffff\n dbs = struct.pack('>Q', (sign << 63) | 0x7ff0000000000000 | (payload << 29))\n return Float(struct.unpack('>d', dbs)[0])\n else:\n return Float(struct.unpack('>f', bs)[0])\nto_bytes()","text":"Converts this 32-bit single-precision floating point value to its binary32 format, taking care to preserve the quiet/signalling bit-pattern of NaN values, unlike its struct.pack('>f', ...) equivalent.
>>> Float.from_bytes(b'\\x7f\\x80\\x00{')\nFloat(nan)\n>>> Float.from_bytes(b'\\x7f\\x80\\x00{').to_bytes()\nb'\\x7f\\x80\\x00{'\n\n>>> struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]\nnan\n>>> Float(struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]).to_bytes()\nb'\\x7f\\xc0\\x00{'\n>>> struct.pack('>f', struct.unpack('>f', b'\\x7f\\x80\\x00{')[0])\nb'\\x7f\\xc0\\x00{'\n(Note well the difference between 7f80007b and 7fc0007b!)
preserves/values.py def to_bytes(self):\n\"\"\"Converts this 32-bit single-precision floating point value to its binary32 format,\n taking care to preserve the quiet/signalling bit-pattern of NaN values, unlike its\n `struct.pack('>f', ...)` equivalent.\n\n ```python\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{')\n Float(nan)\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{').to_bytes()\n b'\\\\x7f\\\\x80\\\\x00{'\n\n >>> struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]\n nan\n >>> Float(struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]).to_bytes()\n b'\\\\x7f\\\\xc0\\\\x00{'\n >>> struct.pack('>f', struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0])\n b'\\\\x7f\\\\xc0\\\\x00{'\n\n ```\n\n (Note well the difference between `7f80007b` and `7fc0007b`!)\n\n \"\"\"\n\n if math.isnan(self.value) or math.isinf(self.value):\n dbs = struct.pack('>d', self.value)\n vd = struct.unpack('>Q', dbs)[0]\n sign = vd >> 63\n payload = (vd >> 29) & 0x007fffff\n vf = (sign << 31) | 0x7f800000 | payload\n return struct.pack('>I', vf)\n else:\n return struct.pack('>f', self.value)\nImmutableDict(*args, **kwargs)","text":" 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)])\n>>> d\n{'a': 1, 'b': 2}\n>>> d['c'] = 3\nTraceback (most recent call last):\n ...\nTypeError: Immutable\n>>> del d['b']\nTraceback (most recent call last):\n ...\nTypeError: Immutable\npreserves/values.py def __init__(self, *args, **kwargs):\n if hasattr(self, '__hash'): raise TypeError('Immutable')\n super(ImmutableDict, self).__init__(*args, **kwargs)\n self.__hash = None\nfrom_kvs(kvs) staticmethod","text":"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])\n{'a': 1, 'b': 2}\n>>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3\nTraceback (most recent call last):\n ...\nTypeError: Immutable\npreserves/values.py @staticmethod\ndef from_kvs(kvs):\n\"\"\"Constructs an [ImmutableDict][preserves.values.ImmutableDict] from a sequence of\n alternating keys and values; compare to the\n [ImmutableDict][preserves.values.ImmutableDict] constructor, which takes a sequence of\n key-value pairs.\n\n ```python\n >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])\n {'a': 1, 'b': 2}\n >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3\n Traceback (most recent call last):\n ...\n TypeError: Immutable\n\n ```\n\n \"\"\"\n\n i = iter(kvs)\n result = ImmutableDict()\n result_proxy = super(ImmutableDict, result)\n try:\n while True:\n k = next(i)\n try:\n v = next(i)\n except StopIteration:\n raise DecodeError(\"Missing dictionary value\")\n result_proxy.__setitem__(k, v)\n except StopIteration:\n pass\n return result\nRecord(key, fields)","text":" 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']])\n>>> r\n#label('field1', ['field2item1', 'field2item2'])\n>>> r.key\n#label\n>>> r.fields\n('field1', ['field2item1', 'field2item2'])\n>>> import preserves\n>>> preserves.stringify(r)\n'<label \"field1\" [\"field2item1\" \"field2item2\"]>'\n>>> r == preserves.parse('<label \"field1\" [\"field2item1\" \"field2item2\"]>')\nTrue\nParameters:
Name Type Description Defaultkey Value the Record's label
fields iterable[Value] the fields of the Record
Attributes:
Name Type Descriptionkey Value the Record's label
fields tuple[Value] the fields of the Record
preserves/values.py def __init__(self, key, fields):\n self.key = key\n self.fields = tuple(fields)\n self.__hash = None\nmakeBasicConstructor(label, fieldNames) staticmethod","text":"Constructs and returns a \"constructor\" for Records having a certain label and number of fields.
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')\n>>> c(1969, 7, 16)\n#date(1969, 7, 16)\n>>> c.constructorInfo\n#date/3\n>>> c.isClassOf(c(1969, 7, 16))\nTrue\n>>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))\nTrue\n>>> c.isClassOf(Record(Symbol('date'), [1969]))\nFalse\n>>> c.ensureClassOf(c(1969, 7, 16))\n#date(1969, 7, 16)\n>>> c.ensureClassOf(Record(Symbol('date'), [1969]))\nTraceback (most recent call last):\n ...\nTypeError: Record: expected #date/3, got #date(1969)\n>>> c._year(c(1969, 7, 16))\n1969\n>>> c._month(c(1969, 7, 16))\n7\n>>> c._day(c(1969, 7, 16))\n16\nParameters:
Name Type Description Defaultlabel Value Label to use for constructed/matched Records
fieldNames tuple[str] | list[str] | str Names of the Record's fields
preserves/values.py @staticmethod\ndef makeBasicConstructor(label, fieldNames):\n\"\"\"Constructs and returns a \"constructor\" for `Record`s having a certain `label` and\n number of fields.\n\n Deprecated:\n Use [preserves.schema][] definitions instead.\n\n The \"constructor\" is a callable function that accepts `len(fields)` arguments and\n returns a [Record][preserves.values.Record] with `label` as its label and the arguments\n to the constructor as field values.\n\n In addition, the \"constructor\" has a `constructorInfo` attribute holding a\n [RecordConstructorInfo][preserves.values.RecordConstructorInfo] object, an `isClassOf`\n attribute holding a unary function that returns `True` iff its argument is a\n [Record][preserves.values.Record] with label `label` and arity `len(fieldNames)`, and\n an `ensureClassOf` attribute that raises an `Exception` if `isClassOf` returns false on\n its argument and returns the argument otherwise.\n\n Finally, for each field name `f` in `fieldNames`, the \"constructor\" object has an\n attribute `_f` that is a unary function that retrieves the `f` field from the passed in\n argument.\n\n ```python\n >>> c = Record.makeBasicConstructor(Symbol('date'), 'year month day')\n >>> c(1969, 7, 16)\n #date(1969, 7, 16)\n >>> c.constructorInfo\n #date/3\n >>> c.isClassOf(c(1969, 7, 16))\n True\n >>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))\n True\n >>> c.isClassOf(Record(Symbol('date'), [1969]))\n False\n >>> c.ensureClassOf(c(1969, 7, 16))\n #date(1969, 7, 16)\n >>> c.ensureClassOf(Record(Symbol('date'), [1969]))\n Traceback (most recent call last):\n ...\n TypeError: Record: expected #date/3, got #date(1969)\n >>> c._year(c(1969, 7, 16))\n 1969\n >>> c._month(c(1969, 7, 16))\n 7\n >>> c._day(c(1969, 7, 16))\n 16\n\n ```\n\n Args:\n label (Value): Label to use for constructed/matched `Record`s\n fieldNames (tuple[str] | list[str] | str): Names of the `Record`'s fields\n\n \"\"\"\n if type(fieldNames) == str:\n fieldNames = fieldNames.split()\n arity = len(fieldNames)\n def ctor(*fields):\n if len(fields) != arity:\n raise Exception(\"Record: cannot instantiate %r expecting %d fields with %d fields\"%(\n label,\n arity,\n len(fields)))\n return Record(label, fields)\n ctor.constructorInfo = RecordConstructorInfo(label, arity)\n ctor.isClassOf = lambda v: \\\n isinstance(v, Record) and v.key == label and len(v.fields) == arity\n def ensureClassOf(v):\n if not ctor.isClassOf(v):\n raise TypeError(\"Record: expected %r/%d, got %r\" % (label, arity, v))\n return v\n ctor.ensureClassOf = ensureClassOf\n for fieldIndex in range(len(fieldNames)):\n fieldName = fieldNames[fieldIndex]\n # Stupid python scoping bites again\n def getter(fieldIndex):\n return lambda v: ensureClassOf(v)[fieldIndex]\n setattr(ctor, '_' + fieldName, getter(fieldIndex))\n return ctor\nmakeConstructor(labelSymbolText, fieldNames) staticmethod","text":"Equivalent to Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames).
Use preserves.schema definitions instead.
Source code inpreserves/values.py @staticmethod\ndef makeConstructor(labelSymbolText, fieldNames):\n\"\"\"\n Equivalent to `Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)`.\n\n Deprecated:\n Use [preserves.schema][] definitions instead.\n \"\"\"\n return Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)\nRecordConstructorInfo(key, arity)","text":" Bases: object
Describes the shape of a Record constructor, namely its label and its arity (field count).
>>> RecordConstructorInfo(Symbol('label'), 3)\n#label/3\nAttributes:
Name Type Descriptionkey Value the label of matching Records
arity int the number of fields in matching Records
preserves/values.py def __init__(self, key, arity):\n self.key = key\n self.arity = arity\nSymbol(name)","text":" Bases: object
Representation of Preserves Symbols.
>>> Symbol('xyz')\n#xyz\n>>> Symbol('xyz').name\n'xyz'\n>>> import preserves\n>>> preserves.stringify(Symbol('xyz'))\n'xyz'\n>>> preserves.stringify(Symbol('hello world'))\n'|hello world|'\n>>> preserves.parse('xyz')\n#xyz\n>>> preserves.parse('|hello world|')\n#hello world\nAttributes:
Name Type Descriptionname str the symbol's text label
Source code inpreserves/values.py def __init__(self, name):\n self.name = name.name if isinstance(name, Symbol) else name\nannotate(v, *anns)","text":"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\n>>> print(preserves.stringify(annotate(123, \"A comment\", \"Another comment\")))\n@\"A comment\" @\"Another comment\" 123\npreserves/values.py def annotate(v, *anns):\n\"\"\"Wraps `v` in an [Annotated][preserves.values.Annotated] object, if it isn't already\n wrapped, and appends each of the `anns` to the [Annotated][preserves.values.Annotated]'s\n `annotations` sequence. NOTE: Does not recursively ensure that any parts of the argument\n `v` are themselves wrapped in [Annotated][preserves.values.Annotated] objects!\n\n ```python\n >>> import preserves\n >>> print(preserves.stringify(annotate(123, \"A comment\", \"Another comment\")))\n @\"A comment\" @\"Another comment\" 123\n\n ```\n \"\"\"\n if not is_annotated(v):\n v = Annotated(v)\n for a in anns:\n v.annotations.append(a)\n return v\ncmp_floats(a, b)","text":"Implements the totalOrder predicate defined in section 5.10 of IEEE Std 754-2008.
preserves/values.py def cmp_floats(a, b):\n\"\"\"Implements the `totalOrder` predicate defined in section 5.10 of [IEEE Std\n 754-2008](https://dx.doi.org/10.1109/IEEESTD.2008.4610935).\n\n \"\"\"\n a = float_to_int(a)\n b = float_to_int(b)\n if a & 0x8000000000000000: a = a ^ 0x7fffffffffffffff\n if b & 0x8000000000000000: b = b ^ 0x7fffffffffffffff\n return a - b\ndict_kvs(d)","text":"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}))\n['a', 1, 'b', 2]\npreserves/values.py def dict_kvs(d):\n\"\"\"Generator function yielding a sequence of alternating keys and values from `d`. In some\n sense the inverse of [ImmutableDict.from_kvs][preserves.values.ImmutableDict.from_kvs].\n\n ```python\n >>> list(dict_kvs({'a': 1, 'b': 2}))\n ['a', 1, 'b', 2]\n\n ```\n \"\"\"\n for k in d:\n yield k\n yield d[k]\nis_annotated(v)","text":"True iff v is an instance of Annotated.
preserves/values.py def is_annotated(v):\n\"\"\"`True` iff `v` is an instance of [Annotated][preserves.values.Annotated].\"\"\"\n return isinstance(v, Annotated)\npreserve(v)","text":"Converts v to a representation of a Preserves Value by (repeatedly) setting
v = v.__preserve__()\nwhile v has a __preserve__ method. Parsed Schema values are able to render themselves to their serialized representations this way.
preserves/values.py def preserve(v):\n\"\"\"Converts `v` to a representation of a Preserves `Value` by (repeatedly) setting\n\n ```python\n v = v.__preserve__()\n ```\n\n while `v` has a `__preserve__` method. Parsed [Schema][preserves.schema]\n values are able to render themselves to their serialized representations this way.\n\n \"\"\"\n while hasattr(v, '__preserve__'):\n v = v.__preserve__()\n return v\nstrip_annotations(v, depth=inf)","text":"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\n>>> a = preserves.parse('@\"A comment\" [@a 1 @b 2 @c 3]', include_annotations=True)\n>>> is_annotated(a)\nTrue\n>>> print(preserves.stringify(a))\n@\"A comment\" [@a 1 @b 2 @c 3]\n>>> print(preserves.stringify(strip_annotations(a)))\n[1 2 3]\n>>> print(preserves.stringify(strip_annotations(a, depth=1)))\n[@a 1 @b 2 @c 3]\npreserves/values.py def strip_annotations(v, depth=inf):\n\"\"\"Exposes `depth` layers of raw structure of\n potentially-[Annotated][preserves.values.Annotated] `Value`s. If `depth==0` or `v` is not\n [Annotated][preserves.values.Annotated], just returns `v`. Otherwise, descends recursively\n into the structure of `v.item`.\n\n ```python\n >>> import preserves\n >>> a = preserves.parse('@\"A comment\" [@a 1 @b 2 @c 3]', include_annotations=True)\n >>> is_annotated(a)\n True\n >>> print(preserves.stringify(a))\n @\"A comment\" [@a 1 @b 2 @c 3]\n >>> print(preserves.stringify(strip_annotations(a)))\n [1 2 3]\n >>> print(preserves.stringify(strip_annotations(a, depth=1)))\n [@a 1 @b 2 @c 3]\n\n ```\n \"\"\"\n\n if depth == 0: return v\n if not is_annotated(v): return v\n\n next_depth = depth - 1\n def walk(v):\n return strip_annotations(v, next_depth)\n\n v = v.item\n if isinstance(v, Record):\n return Record(strip_annotations(v.key, depth), tuple(walk(f) for f in v.fields))\n elif isinstance(v, list):\n return tuple(walk(f) for f in v)\n elif isinstance(v, tuple):\n return tuple(walk(f) for f in v)\n elif isinstance(v, set):\n return frozenset(walk(f) for f in v)\n elif isinstance(v, frozenset):\n return frozenset(walk(f) for f in v)\n elif isinstance(v, dict):\n return ImmutableDict.from_kvs(walk(f) for f in dict_kvs(v))\n elif is_annotated(v):\n raise ValueError('Improper annotation structure')\n else:\n return v\npip install preserves\nThis package (preserves on pypi.org) implements Preserves for Python 3.x. It provides the core semantics as well as both the human-readable text syntax (a superset of JSON) and machine-oriented binary format (including canonicalization) for Preserves. It also implements Preserves Schema and Preserves Path.
Preserves is a data model, with associated serialization formats.
It supports records with user-defined labels, embedded references, and the usual suite of atomic and compound data types, including binary data as a distinct type from text strings. Its annotations allow separation of data from metadata such as comments, trace information, and provenance information.
Preserves departs from many other data languages in defining how to compare two values. Comparison is based on the data model, not on syntax or on data structures of any particular implementation language.
"},{"location":"#mapping-between-preserves-values-and-python-values","title":"Mapping between Preserves values and Python values","text":"Preserves Values are categorized in the following way:
Value = Atom\n | Compound\n | Embedded\n\n Atom = Boolean\n | Float\n | Double\n | SignedInteger\n | String\n | ByteString\n | Symbol\n\n Compound = Record\n | Sequence\n | Set\n | Dictionary\nPython's strings, byte strings, integers, booleans, and double-precision floats stand directly for their Preserves counterparts. Wrapper objects for Float and Symbol complete the suite of atomic types.
Python's lists and tuples correspond to Preserves Sequences, and dicts and sets to Dictionary and Set values, respectively. Preserves Records are represented by Record objects. Finally, embedded values are represented by Embedded objects.
import preserves\nThe main package re-exports a subset of the exports of its constituent modules:
From preserves.values:
From preserves.error:
From preserves.binary:
From preserves.text:
From preserves.compare:
From preserves.merge:
It also exports the compare and fold modules themselves, permitting patterns like
>>> from preserves import *\n>>> compare.cmp(123, 234)\n-1\nFinally, it provides a few utility aliases for common tasks:
"},{"location":"api/#preserves.dumps","title":"dumps = stringify module-attribute","text":"This alias for stringify provides a familiar pythonesque name for converting a Preserves Value to a string.
loads = parse module-attribute","text":"This alias for parse provides a familiar pythonesque name for converting a string to a Preserves Value.
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'), []))\nb'\\xb4\\xb3\\x02hi\\x84'\n>>> decode(b'\\xb4\\xb3\\x02hi\\x84')\n#hi()\nDecoder(packet=b'', include_annotations=False, decode_embedded=lambda x: x)","text":" Bases: BinaryCodec
Implementation of a decoder for the machine-oriented binary Preserves syntax.
Parameters:
Name Type Description Defaultpacket 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'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> d.next()\n123\n>>> d.next()\n'hello'\n>>> d.next()\n()\n>>> d.next()\nTraceback (most recent call last):\n ...\npreserves.error.ShortPacket: Short packet\nAlternatively, keep calling try_next until it yields None, which is not in the domain of Preserves Values:
>>> d = Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> d.try_next()\n123\n>>> d.try_next()\n'hello'\n>>> d.try_next()\n()\n>>> d.try_next()\nFor 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'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84')\n>>> list(d)\n[123, 'hello', ()]\n>>> for v in Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84'):\n... print(repr(v))\n123\n'hello'\n()\nSupply include_annotations=True to read annotations alongside the annotated values:
>>> d = Decoder(b'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84', include_annotations=True)\n>>> list(d)\n[123, 'hello', @#x ()]\nIf you are incrementally reading from, say, a socket, you can use extend to add new input as if comes available:
>>> d = Decoder(b'\\xa0{\\xb1\\x05he')\n>>> d.try_next()\n123\n>>> d.try_next() # returns None because the input is incomplete\n>>> d.extend(b'llo')\n>>> d.try_next()\n'hello'\n>>> d.try_next()\nAttributes:
Name Type Descriptionpacket bytes buffered input waiting to be processed
index int read position within packet
preserves/binary.py def __init__(self, packet=b'', include_annotations=False, decode_embedded=lambda x: x):\n super(Decoder, self).__init__()\n self.packet = packet\n self.index = 0\n self.include_annotations = include_annotations\n self.decode_embedded = decode_embedded\nextend(data)","text":"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.
preserves/binary.py def extend(self, data):\n\"\"\"Appends `data` to the remaining bytes in `self.packet`, trimming already-processed\n bytes from the front of `self.packet` and resetting `self.index` to zero.\"\"\"\n self.packet = self.packet[self.index:] + data\n self.index = 0\nnext()","text":"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.
preserves/binary.py def next(self):\n\"\"\"Reads the next complete `Value` from the internal buffer, raising\n [ShortPacket][preserves.error.ShortPacket] if too few bytes are available, or\n [DecodeError][preserves.error.DecodeError] if the input is invalid somehow.\n\n \"\"\"\n tag = self.nextbyte()\n if tag == 0x80: return self.wrap(False)\n if tag == 0x81: return self.wrap(True)\n if tag == 0x82: return self.wrap(Float.from_bytes(self.nextbytes(4)))\n if tag == 0x83: return self.wrap(struct.unpack('>d', self.nextbytes(8))[0])\n if tag == 0x84: raise DecodeError('Unexpected end-of-stream marker')\n if tag == 0x85:\n a = self.next()\n v = self.next()\n return self.unshift_annotation(a, v)\n if tag == 0x86:\n if self.decode_embedded is None:\n raise DecodeError('No decode_embedded function supplied')\n return self.wrap(Embedded(self.decode_embedded(self.next())))\n if tag >= 0x90 and tag <= 0x9f: return self.wrap(tag - (0xa0 if tag > 0x9c else 0x90))\n if tag >= 0xa0 and tag <= 0xaf: return self.wrap(self.nextint(tag - 0xa0 + 1))\n if tag == 0xb0: return self.wrap(self.nextint(self.varint()))\n if tag == 0xb1: return self.wrap(self.nextbytes(self.varint()).decode('utf-8'))\n if tag == 0xb2: return self.wrap(self.nextbytes(self.varint()))\n if tag == 0xb3: return self.wrap(Symbol(self.nextbytes(self.varint()).decode('utf-8')))\n if tag == 0xb4:\n vs = self.nextvalues()\n if not vs: raise DecodeError('Too few elements in encoded record')\n return self.wrap(Record(vs[0], vs[1:]))\n if tag == 0xb5: return self.wrap(tuple(self.nextvalues()))\n if tag == 0xb6: return self.wrap(frozenset(self.nextvalues()))\n if tag == 0xb7: return self.wrap(ImmutableDict.from_kvs(self.nextvalues()))\n raise DecodeError('Invalid tag: ' + hex(tag))\ntry_next()","text":"Like next, but returns None instead of raising ShortPacket.
preserves/binary.py def try_next(self):\n\"\"\"Like [next][preserves.binary.Decoder.next], but returns `None` instead of raising\n [ShortPacket][preserves.error.ShortPacket].\"\"\"\n start = self.index\n try:\n return self.next()\n except ShortPacket:\n self.index = start\n return None\nEncoder(encode_embedded=lambda x: x, canonicalize=False, include_annotations=None)","text":" Bases: BinaryCodec
Implementation of an encoder for the machine-oriented binary Preserves syntax.
>>> e = Encoder()\n>>> e.append(123)\n>>> e.append('hello')\n>>> e.append(annotate([], Symbol('x')))\n>>> e.contents()\nb'\\xa0{\\xb1\\x05hello\\x85\\xb3\\x01x\\xb5\\x84'\nParameters:
Name Type Description Defaultencode_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 Descriptionbuffer bytearray accumulator for the output of the encoder
Source code inpreserves/binary.py def __init__(self,\n encode_embedded=lambda x: x,\n canonicalize=False,\n include_annotations=None):\n super(Encoder, self).__init__()\n self.buffer = bytearray()\n self._encode_embedded = encode_embedded\n self._canonicalize = canonicalize\n if include_annotations is None:\n self.include_annotations = not self._canonicalize\n else:\n self.include_annotations = include_annotations\nappend(v)","text":"Extend self.buffer with an encoding of v.
preserves/binary.py def append(self, v):\n\"\"\"Extend `self.buffer` with an encoding of `v`.\"\"\"\n v = preserve(v)\n if hasattr(v, '__preserve_write_binary__'):\n v.__preserve_write_binary__(self)\n elif v is False:\n self.buffer.append(0x80)\n elif v is True:\n self.buffer.append(0x81)\n elif isinstance(v, float):\n self.buffer.append(0x83)\n self.buffer.extend(struct.pack('>d', v))\n elif isinstance(v, numbers.Number):\n if v >= -3 and v <= 12:\n self.buffer.append(0x90 + (v if v >= 0 else v + 16))\n else:\n self.encodeint(v)\n elif isinstance(v, bytes):\n self.encodebytes(2, v)\n elif isinstance(v, basestring_):\n self.encodebytes(1, v.encode('utf-8'))\n elif isinstance(v, list):\n self.encodevalues(5, v)\n elif isinstance(v, tuple):\n self.encodevalues(5, v)\n elif isinstance(v, set):\n self.encodeset(v)\n elif isinstance(v, frozenset):\n self.encodeset(v)\n elif isinstance(v, dict):\n self.encodedict(v)\n else:\n try:\n i = iter(v)\n except TypeError:\n i = None\n if i is None:\n self.cannot_encode(v)\n else:\n self.encodevalues(5, i)\ncontents()","text":"Returns a bytes constructed from the contents of self.buffer.
preserves/binary.py def contents(self):\n\"\"\"Returns a `bytes` constructed from the contents of `self.buffer`.\"\"\"\n return bytes(self.buffer)\nreset()","text":"Clears self.buffer to a fresh empty bytearray.
preserves/binary.py def reset(self):\n\"\"\"Clears `self.buffer` to a fresh empty `bytearray`.\"\"\"\n self.buffer = bytearray()\ncanonicalize(v, **kwargs)","text":"As encode, but sets canonicalize=True in the Encoder constructor.
preserves/binary.py def canonicalize(v, **kwargs):\n\"\"\"As [encode][preserves.binary.encode], but sets `canonicalize=True` in the\n [Encoder][preserves.binary.Encoder] constructor.\n\n \"\"\"\n return encode(v, canonicalize=True, **kwargs)\ndecode(bs, **kwargs)","text":"Yields the first complete encoded value from bs, passing kwargs through to the Decoder constructor. Raises exceptions as per next.
Parameters:
Name Type Description Defaultbs bytes encoded data to decode
required Source code inpreserves/binary.py def decode(bs, **kwargs):\n\"\"\"Yields the first complete encoded value from `bs`, passing `kwargs` through to the\n [Decoder][preserves.binary.Decoder] constructor. Raises exceptions as per\n [next][preserves.binary.Decoder.next].\n\n Args:\n bs (bytes): encoded data to decode\n\n \"\"\"\n return Decoder(packet=bs, **kwargs).next()\ndecode_with_annotations(bs, **kwargs)","text":"Like decode, but supplying include_annotations=True to the Decoder constructor.
preserves/binary.py def decode_with_annotations(bs, **kwargs):\n\"\"\"Like [decode][preserves.binary.decode], but supplying `include_annotations=True` to the\n [Decoder][preserves.binary.Decoder] constructor.\"\"\"\n return Decoder(packet=bs, include_annotations=True, **kwargs).next()\nencode(v, **kwargs)","text":"Encode a single Value v to a byte string. Any supplied kwargs are passed on to the underlying Encoder constructor.
preserves/binary.py def encode(v, **kwargs):\n\"\"\"Encode a single `Value` `v` to a byte string. Any supplied `kwargs` are passed on to the\n underlying [Encoder][preserves.binary.Encoder] constructor.\"\"\"\n e = Encoder(**kwargs)\n e.append(v)\n return e.contents()\nPreserves specifies a total ordering and an equivalence between terms. The preserves.compare module implements the ordering and equivalence relations.
>>> cmp(\"bzz\", \"c\")\n-1\n>>> cmp(True, [])\n-1\n>>> lt(\"bzz\", \"c\")\nTrue\n>>> eq(\"bzz\", \"c\")\nFalse\nNote that the ordering relates more values than Python's built-in ordering:
>>> [1, 2, 2] < [1, 2, \"3\"]\nTraceback (most recent call last):\n ..\nTypeError: '<' not supported between instances of 'int' and 'str'\n\n>>> lt([1, 2, 2], [1, 2, \"3\"])\nTrue\ncmp(a, b)","text":"Returns -1 if a < b, or 0 if a = b, or 1 if a > b according to the Preserves total order.
preserves/compare.py def cmp(a, b):\n\"\"\"Returns `-1` if `a` < `b`, or `0` if `a` = `b`, or `1` if `a` > `b` according to the\n [Preserves total order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return _cmp(preserve(a), preserve(b))\neq(a, b)","text":"Returns True iff a = b according to the Preserves equivalence relation.
preserves/compare.py def eq(a, b):\n\"\"\"Returns `True` iff `a` = `b` according to the [Preserves equivalence\n relation](https://preserves.dev/preserves.html#equivalence).\"\"\"\n return _eq(preserve(a), preserve(b))\nle(a, b)","text":"Returns True iff a \u2264 b according to the Preserves total order.
preserves/compare.py def le(a, b):\n\"\"\"Returns `True` iff `a` \u2264 `b` according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return cmp(a, b) <= 0\nlt(a, b)","text":"Returns True iff a < b according to the Preserves total order.
preserves/compare.py def lt(a, b):\n\"\"\"Returns `True` iff `a` < `b` according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order).\"\"\"\n return cmp(a, b) < 0\nsorted(iterable, *, key=lambda x: x, reverse=False)","text":"Returns a sorted list built from iterable, extracting a sort key using key, and ordering according to the Preserves total order. Directly analogous to the built-in Python sorted routine, except uses the Preserves order instead of Python's less-than relation.
preserves/compare.py def sorted(iterable, *, key=lambda x: x, reverse=False):\n\"\"\"Returns a sorted list built from `iterable`, extracting a sort key using `key`, and\n ordering according to the [Preserves total\n order](https://preserves.dev/preserves.html#total-order). Directly analogous to the\n [built-in Python `sorted`\n routine](https://docs.python.org/3/library/functions.html#sorted), except uses the\n Preserves order instead of Python's less-than relation.\n\n \"\"\"\n return _sorted(iterable, key=lambda x: _key(key(x)), reverse=reverse)\nsorted_items(d)","text":"Given a dictionary d, yields a list of (key, value) tuples sorted by key.
preserves/compare.py def sorted_items(d):\n\"\"\"Given a dictionary `d`, yields a list of `(key, value)` tuples sorted by `key`.\"\"\"\n return sorted(d.items(), key=_item_key)\nThe preserves.error module exports various Error subclasses.
DecodeError","text":" Bases: ValueError
Raised whenever preserves.binary.Decoder or preserves.text.Parser detect invalid input.
"},{"location":"error/#preserves.error.EncodeError","title":"EncodeError","text":" Bases: ValueError
Raised whenever preserves.binary.Encoder or preserves.text.Formatter are unable to proceed.
"},{"location":"error/#preserves.error.ShortPacket","title":"ShortPacket","text":" 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.
"},{"location":"fold/","title":"Traversing values","text":"The preserves.fold module exports various utilities for traversing compound Values.
map_embeddeds(f, v)","text":"Returns an equivalent copy of v, except where each contained Embedded value is replaced by f applied to the Embedded's embeddedValue attribute.
>>> map_embeddeds(lambda w: Embedded(f'w={w}'), ['a', Embedded(123), {'z': 6.0}])\n('a', #!'w=123', {'z': 6.0})\npreserves/fold.py def map_embeddeds(f, v):\n\"\"\"Returns an [equivalent][preserves.compare.eq] copy of `v`, except where each contained\n [Embedded][preserves.values.Embedded] value is replaced by `f` applied to the Embedded's\n `embeddedValue` attribute.\n\n ```python\n >>> map_embeddeds(lambda w: Embedded(f'w={w}'), ['a', Embedded(123), {'z': 6.0}])\n ('a', #!'w=123', {'z': 6.0})\n\n ```\n \"\"\"\n def walk(v):\n if isinstance(v, Embedded):\n return f(v.embeddedValue)\n elif isinstance(v, (list, tuple)):\n return tuple(walk(w) for w in v)\n elif isinstance(v, (set, frozenset)):\n return frozenset(walk(w) for w in v)\n elif isinstance(v, dict):\n return ImmutableDict.from_kvs(walk(w) for w in dict_kvs(v))\n elif isinstance(v, Record):\n return Record(walk(v.key), walk(v.fields))\n else:\n return v\n return walk(v)\nThe preserves.merge module exports various utilities for merging Values.
merge(v0, *vs, merge_embedded=None)","text":"Repeatedly merges v0 with each element in vs using merge2, returning the final result. The merge_embedded parameter is passed on to merge2.
preserves/merge.py def merge(v0, *vs, merge_embedded=None):\n\"\"\"Repeatedly merges `v0` with each element in `vs` using [merge2][preserves.merge.merge2],\n returning the final result. The `merge_embedded` parameter is passed on to merge2.\"\"\"\n v = v0\n for vN in vs:\n v = merge2(v, vN, merge_embedded=merge_embedded)\n return v\nmerge2(a, b, merge_embedded=None)","text":"Merges a and b, returning the result. Raises ValueError if, during the merge, a pair of incompatible values is discovered.
If a and b are Embedded objects, their embeddedValues are merged using merge_embedded, and the result is again wrapped in an Embedded object.
>>> merge2(123, 234)\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2(123, 123)\n123\n>>> merge2('hi', 0)\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2([1, 2], [1, 2])\n[1, 2]\n>>> merge2([1, 2], [1, 3])\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2({'a': 1, 'b': 2}, {'a': 1, 'c': 3})\n{'a': 1, 'b': 2, 'c': 3}\n>>> merge2({'a': 1, 'b': 2}, {'a': 10, 'c': 3})\nTraceback (most recent call last):\n ...\nValueError: Cannot merge items\n>>> merge2(Record('a', [1, {'x': 2}]), Record('a', [1, {'y': 3}]))\na(1, {'x': 2, 'y': 3})\npreserves/merge.py def merge2(a, b, merge_embedded=None):\n\"\"\"Merges `a` and `b`, returning the result. Raises `ValueError` if, during the merge, a\n pair of incompatible values is discovered.\n\n If `a` and `b` are [Embedded][preserves.values.Embedded] objects, their `embeddedValue`s\n are merged using `merge_embedded`, and the result is again wrapped in an\n [Embedded][preserves.values.Embedded] object.\n\n ```python\n >>> merge2(123, 234)\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2(123, 123)\n 123\n >>> merge2('hi', 0)\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2([1, 2], [1, 2])\n [1, 2]\n >>> merge2([1, 2], [1, 3])\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2({'a': 1, 'b': 2}, {'a': 1, 'c': 3})\n {'a': 1, 'b': 2, 'c': 3}\n >>> merge2({'a': 1, 'b': 2}, {'a': 10, 'c': 3})\n Traceback (most recent call last):\n ...\n ValueError: Cannot merge items\n >>> merge2(Record('a', [1, {'x': 2}]), Record('a', [1, {'y': 3}]))\n a(1, {'x': 2, 'y': 3})\n\n ```\n\n \"\"\"\n if a == b:\n return a\n if isinstance(a, (list, tuple)) and isinstance(b, (list, tuple)):\n return merge_seq(a, b)\n if isinstance(a, (set, frozenset)) and isinstance(b, (set, frozenset)):\n _die()\n if isinstance(a, dict) and isinstance(b, dict):\n r = {}\n for (ak, av) in a.items():\n bv = b.get(ak, None)\n r[ak] = av if bv is None else merge2(av, bv, merge_embedded=merge_embedded)\n for (bk, bv) in b.items():\n if bk not in r:\n r[bk] = bv\n return r\n if isinstance(a, Record) and isinstance(b, Record):\n return Record(merge2(a.key, b.key, merge_embedded=merge_embedded),\n merge_seq(a.fields, b.fields, merge_embedded=merge_embedded))\n if isinstance(a, Embedded) and isinstance(b, Embedded):\n m = (merge_embedded or merge_embedded_id)(a.embeddedValue, b.embeddedValue)\n if m is None: _die()\n return Embedded(m)\n _die()\nThe preserves.path module implements Preserves Path.
Preserves Path is roughly analogous to XPath, but for Preserves values: just as XPath selects portions of an XML document, a Preserves Path uses path expressions to select portions of a Value.
Use parse to compile a path expression, and then use the exec method on the result to apply it to a given input:
parse(PATH_EXPRESSION_STRING).exec(PRESERVES_VALUE)\n -> SEQUENCE_OF_PRESERVES_VALUES\nWhen preserves.path is run as a __main__ module, sys.argv[1] is parsed, interpreted as a path expression, and run against human-readable values read from standard input. Each matching result is passed to stringify and printed to standard output.
The following examples use testdata:
>>> with open('tests/samples.bin', 'rb') as f:\n... testdata = decode_with_annotations(f.read())\nRecall that samples.bin contains a binary-syntax form of the human-readable [samples.pr](https://preserves.dev/tests/samples.pr) test data file, intended to exercise most of the features of Preserves. In particular, the rootValue` in the file has a number of annotations (for documentation and other purposes).
The path expression .annotations ^ Documentation . 0 / string proceeds in five steps:
.annotations selects each annotation on the root document^ Documentation retains only those values (each an annotation of the root) that are Records with label equal to the symbol Documentation. 0 moves into the first child (the first field) of each such Record, which in our case is a list of other Values/ selects all immediate children of these listsstring retains only those values that are stringsThe result of evaluating it on testdata is as follows:
>>> selector = parse('.annotations ^ Documentation . 0 / string')\n>>> for result in selector.exec(testdata):\n... print(stringify(result))\n\"Individual test cases may be any of the following record types:\"\n\"In each test, let value = strip(annotatedValue),\"\n\" forward = value,\"\n\" back = value,\"\n\"except where test-case-specific values of `forward` and/or `back`\"\n\"are provided by the executing harness, and check the following\"\n\"numbered expectations according to the table above:\"\n\"Implementations may vary in their treatment of the difference between expectations\"\n\"13/14 and 16/17, depending on how they wish to treat end-of-stream conditions.\"\nThe path expression // [.^ [= Test + = NondeterministicTest]] [. 1 rec] proceeds in three steps:
// recursively decomposes the input, yielding all direct and indirect descendants of each input value
[.^ [= Test + = NondeterministicTest]] retains only those inputs (each a descendant of the root) that yield more than zero results when executed against the expression within the brackets:
.^ selects only labels of values that are Records, filtering by type and transforming in a single step[= Test + = NondeterministicTest] again filters by a path expression:+ operator takes the union of matches of its arguments= Test selects values (remember, record labels) equal to the symbol Test= NondeterministicTest selects values equal to NondeterministicTestThe result is thus all Records anywhere inside testdata that have either Test or NondeterministicTest as their labels.
[. 1 rec] filters these Records by another path expression:
. 1 selects their second field (fields are numbered from 0)rec retains only values that are RecordsEvaluating the expression against testdata yields the following:
>>> selector = parse('// [.^ [= Test + = NondeterministicTest]] [. 1 rec]')\n>>> for result in selector.exec(testdata):\n... print(stringify(result))\n<Test #[tLMHY2FwdHVyZbSzB2Rpc2NhcmSEhA==] <capture <discard>>>\n<Test #[tLMHb2JzZXJ2ZbSzBXNwZWFrtLMHZGlzY2FyZIS0swdjYXB0dXJltLMHZGlzY2FyZISEhIQ=] <observe <speak <discard> <capture <discard>>>>>\n<Test #[tLWzBnRpdGxlZLMGcGVyc29ukrMFdGhpbmeRhKBlsQlCbGFja3dlbGy0swRkYXRloQcdkpOEsQJEcoQ=] <[titled person 2 thing 1] 101 \"Blackwell\" <date 1821 2 3> \"Dr\">>\n<Test #[tLMHZGlzY2FyZIQ=] <discard>>\n<Test #[tJe1hIQ=] <7 []>>\n<Test #[tLMHZGlzY2FyZLMIc3VycHJpc2WE] <discard surprise>>\n<Test #[tLEHYVN0cmluZ5OUhA==] <\"aString\" 3 4>>\n<Test #[tLSzB2Rpc2NhcmSEk5SE] <<discard> 3 4>>\n<Test #[hbMCYXK0swFShbMCYWazAWaE] @ar <R @af f>>\n<Test #[tIWzAmFyswFShbMCYWazAWaE] <@ar R @af f>>\nPredicate = syntax.Predicate module-attribute","text":"Schema definition for representing a Preserves Path Predicate.
Selector = syntax.Selector module-attribute","text":"Schema definition for representing a sequence of Preserves Path Steps.
syntax = load_schema_file(pathlib.Path(__file__).parent / 'path.prb').path module-attribute","text":"This value is a Python representation of a Preserves Schema definition for the Preserves Path expression language. The language is defined in the file path.prs.
"},{"location":"path/#preserves.path.exec","title":"exec(self, v)","text":"WARNING: This is not a function: it is a method on Selector, Predicate, and so on.
>>> sel = parse('/ [.length gt 1]')\n>>> sel.exec(['', 'a', 'ab', 'abc', 'abcd', 'bcd', 'cd', 'd', ''])\n('ab', 'abc', 'abcd', 'bcd', 'cd')\npreserves/path.py @extend(syntax.Function)\ndef exec(self, v):\n\"\"\"WARNING: This is not a *function*: it is a *method* on\n [Selector][preserves.path.Selector], [Predicate][preserves.path.Predicate], and so on.\n\n ```python\n >>> sel = parse('/ [.length gt 1]')\n >>> sel.exec(['', 'a', 'ab', 'abc', 'abcd', 'bcd', 'cd', 'd', ''])\n ('ab', 'abc', 'abcd', 'bcd', 'cd')\n\n ```\n\n \"\"\"\n return (len(self.selector.exec(v)),)\nparse(s)","text":"Parse s as a Preserves Path path expression, yielding a Selector object. Selectors (and Predicates etc.) have an exec method defined on them.
Raises ValueError if s is not a valid path expression.
preserves/path.py def parse(s):\n\"\"\"Parse `s` as a Preserves Path path expression, yielding a\n [Selector][preserves.path.Selector] object. Selectors (and Predicates etc.) have an\n [exec][preserves.path.exec] method defined on them.\n\n Raises `ValueError` if `s` is not a valid path expression.\n\n \"\"\"\n return parse_selector(Parser(s))\nThis is an implementation of Preserves Schema for Python 3.
TODO
"},{"location":"schema/#preserves.schema.meta","title":"meta = load_schema_file(__metaschema_filename).schema module-attribute","text":"TODO
"},{"location":"schema/#preserves.schema.Compiler","title":"Compiler()","text":"TODO
Source code inpreserves/schema.py def __init__(self):\n self.root = Namespace(())\nDefinition(*args, **kwargs)","text":" Bases: SchemaObject
TODO
Source code inpreserves/schema.py def __init__(self, *args, **kwargs):\n self._fields = args\n if self.SIMPLE:\n if self.EMPTY:\n if len(args) != 0:\n raise TypeError('%s takes no arguments' % (self._constructor_name(),))\n else:\n if len(args) != 1:\n raise TypeError('%s needs exactly one argument' % (self._constructor_name(),))\n self.value = args[0]\n else:\n i = 0\n for arg in args:\n if i >= len(self.FIELD_NAMES):\n raise TypeError('%s given too many positional arguments' % (self._constructor_name(),))\n setattr(self, self.SAFE_FIELD_NAMES[i], arg)\n i = i + 1\n for (argname, arg) in kwargs.items():\n if hasattr(self, argname):\n raise TypeError('%s given duplicate attribute: %r' % (self._constructor_name, argname))\n if argname not in self.SAFE_FIELD_NAMES:\n raise TypeError('%s given unknown attribute: %r' % (self._constructor_name, argname))\n setattr(self, argname, arg)\n i = i + 1\n if i != len(self.FIELD_NAMES):\n raise TypeError('%s needs argument(s) %r' % (self._constructor_name(), self.FIELD_NAMES))\nEnumeration()","text":" Bases: SchemaObject
TODO
Source code inpreserves/schema.py def __init__(self):\n raise TypeError('Cannot create instance of Enumeration')\nNamespace(prefix)","text":"TODO
Source code inpreserves/schema.py def __init__(self, prefix):\n self._prefix = prefix\nSchemaDecodeFailed(cls, p, v, failures=None)","text":" Bases: ValueError
TODO
Source code inpreserves/schema.py def __init__(self, cls, p, v, failures=None):\n super().__init__()\n self.cls = cls\n self.pattern = p\n self.value = v\n self.failures = [] if failures is None else failures\nSchemaObject","text":"TODO
"},{"location":"schema/#preserves.schema.SchemaObject.__preserve__","title":"__preserve__()","text":"TODO
Source code inpreserves/schema.py def __preserve__(self):\n\"\"\"TODO\"\"\"\n raise NotImplementedError('Subclass responsibility')\ndecode(v) classmethod","text":"TODO
Source code inpreserves/schema.py @classmethod\ndef decode(cls, v):\n\"\"\"TODO\"\"\"\n raise NotImplementedError('Subclass responsibility')\ntry_decode(v) classmethod","text":"TODO
Source code inpreserves/schema.py @classmethod\ndef try_decode(cls, v):\n\"\"\"TODO\"\"\"\n try:\n return cls.decode(v)\n except SchemaDecodeFailed:\n return None\nencode(p, v)","text":"TODO
Source code inpreserves/schema.py def encode(p, v):\n\"\"\"TODO\"\"\"\n if hasattr(v, '__escape_schema__'):\n return preserve(v.__escape_schema__())\n if p == ANY:\n return v\n if p.key == NAMED:\n return encode(p[1], safegetattr(v, p[0].name))\n if p.key == ATOM:\n return v\n if p.key == EMBEDDED:\n return Embedded(v)\n if p.key == LIT:\n return p[0]\n if p.key == SEQOF:\n return tuple(encode(p[0], w) for w in v)\n if p.key == SETOF:\n return set(encode(p[0], w) for w in v)\n if p.key == DICTOF:\n return dict((encode(p[0], k), encode(p[1], w)) for (k, w) in v.items())\n if p.key == REF:\n return preserve(v)\n if p.key == REC:\n return Record(encode(p[0], v), encode(p[1], v))\n if p.key == TUPLE:\n return tuple(encode(pp, v) for pp in p[0])\n if p.key == TUPLE_PREFIX:\n return tuple(encode(pp, v) for pp in p[0]) + encode(p[1], v)\n if p.key == DICT:\n return dict((k, encode(pp, v)) for (k, pp) in p[0].items())\n if p.key == AND:\n return merge(*[encode(pp, v) for pp in p[0]])\n raise ValueError(f'Bad schema {p}')\nextend(cls)","text":"TODO
Source code inpreserves/schema.py def extend(cls):\n\"\"\"TODO\"\"\"\n @wraps(cls)\n def extender(f):\n setattr(cls, f.__name__, f)\n return f\n return extender\nload_schema_file(filename)","text":"TODO
Source code inpreserves/schema.py def load_schema_file(filename):\n\"\"\"TODO\"\"\"\n c = Compiler()\n c.load(filename)\n return c.root\nThe preserves.text module implements the Preserves human-readable text syntax.
The main entry points are functions stringify, parse, and parse_with_annotations.
>>> stringify(Record(Symbol('hi'), [1, [2, 3]]))\n'<hi 1 [2 3]>'\n>>> parse('<hi 1 [2 3]>')\n#hi(1, (2, 3))\nFormatter(format_embedded=lambda x: x, indent=None, with_commas=False, trailing_comma=False, include_annotations=True)","text":" Bases: TextCodec
Printer (and indenting pretty-printer) for producing human-readable syntax from Preserves Values.
>>> f = Formatter()\n>>> f.append({'a': 1, 'b': 2})\n>>> f.append(Record(Symbol('label'), ['field1', ['field2item1', 'field2item2']]))\n>>> print(f.contents())\n{\"a\": 1 \"b\": 2} <label \"field1\" [\"field2item1\" \"field2item2\"]>\n\n>>> f = Formatter(indent=4)\n>>> f.append({'a': 1, 'b': 2})\n>>> f.append(Record(Symbol('label'), ['field1', ['field2item1', 'field2item2']]))\n>>> print(f.contents())\n{\n \"a\": 1\n \"b\": 2\n}\n<label \"field1\" [\n \"field2item1\"\n \"field2item2\"\n]>\nParameters:
Name Type Description Defaultformat_embedded function accepting an Embedded.embeddedValue and returning a Value for serialization.
lambda x: x indent int | None None disables indented pretty-printing; otherwise, an int specifies indentation per nesting-level.
None with_commas bool True causes commas to separate sequence and set items and dictionary entries; False omits commas.
False trailing_comma bool True causes a comma to be printed after the final item or entry in a sequence, set or dictionary; False omits this trailing comma
False include_annotations bool True causes annotations to be included in the output; False causes them to be omitted.
True Attributes:
Name Type Descriptionindent_delta int indentation per nesting-level
chunks list[str] fragments of output
Source code inpreserves/text.py def __init__(self,\n format_embedded=lambda x: x,\n indent=None,\n with_commas=False,\n trailing_comma=False,\n include_annotations=True):\n super(Formatter, self).__init__()\n self.indent_delta = 0 if indent is None else indent\n self.indent_distance = 0\n self.nesting = 0\n self.with_commas = with_commas\n self.trailing_comma = trailing_comma\n self.chunks = []\n self._format_embedded = format_embedded\n self.include_annotations = include_annotations\nappend(v)","text":"Extend self.chunks with at least one chunk, together making up the text representation of v.
preserves/text.py def append(self, v):\n\"\"\"Extend `self.chunks` with at least one chunk, together making up the text\n representation of `v`.\"\"\"\n if self.chunks and self.nesting == 0:\n self.write_indent_space()\n try:\n self.nesting += 1\n self._append(v)\n finally:\n self.nesting -= 1\ncontents()","text":"Returns a str constructed from the join of the chunks in self.chunks.
preserves/text.py def contents(self):\n\"\"\"Returns a `str` constructed from the join of the chunks in `self.chunks`.\"\"\"\n return u''.join(self.chunks)\nis_indenting()","text":"Returns True iff this Formatter is in pretty-printing indenting mode.
preserves/text.py def is_indenting(self):\n\"\"\"Returns `True` iff this [Formatter][preserves.text.Formatter] is in pretty-printing\n indenting mode.\"\"\"\n return self.indent_delta > 0\nParser(input_buffer='', include_annotations=False, parse_embedded=lambda x: x)","text":" Bases: TextCodec
Parser for the human-readable Preserves text syntax.
Parameters:
Name Type Description Defaultinput_buffer str initial contents of the input buffer; may subsequently be extended by calling extend.
'' include_annotations bool if True, wrap each value and subvalue in an Annotated object.
False parse_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 input text, and keep calling next until a ShortPacket exception is raised:
>>> d = Parser('123 \"hello\" @x []')\n>>> d.next()\n123\n>>> d.next()\n'hello'\n>>> d.next()\n()\n>>> d.next()\nTraceback (most recent call last):\n ...\npreserves.error.ShortPacket: Short input buffer\nAlternatively, keep calling try_next until it yields None, which is not in the domain of Preserves Values:
>>> d = Parser('123 \"hello\" @x []')\n>>> d.try_next()\n123\n>>> d.try_next()\n'hello'\n>>> d.try_next()\n()\n>>> d.try_next()\nFor convenience, Parser implements the iterator interface, backing it with try_next, so you can simply iterate over all complete values in an input:
>>> d = Parser('123 \"hello\" @x []')\n>>> list(d)\n[123, 'hello', ()]\n>>> for v in Parser('123 \"hello\" @x []'):\n... print(repr(v))\n123\n'hello'\n()\nSupply include_annotations=True to read annotations alongside the annotated values:
>>> d = Parser('123 \"hello\" @x []', include_annotations=True)\n>>> list(d)\n[123, 'hello', @#x ()]\nIf you are incrementally reading from, say, a socket, you can use extend to add new input as if comes available:
>>> d = Parser('123 \"he')\n>>> d.try_next()\n123\n>>> d.try_next() # returns None because the input is incomplete\n>>> d.extend('llo\"')\n>>> d.try_next()\n'hello'\n>>> d.try_next()\nAttributes:
Name Type Descriptioninput_buffer str buffered input waiting to be processed
index int read position within input_buffer
preserves/text.py def __init__(self, input_buffer=u'', include_annotations=False, parse_embedded=lambda x: x):\n super(Parser, self).__init__()\n self.input_buffer = input_buffer\n self.index = 0\n self.include_annotations = include_annotations\n self.parse_embedded = parse_embedded\nextend(text)","text":"Appends text to the remaining contents of self.input_buffer, trimming already-processed text from the front of self.input_buffer and resetting self.index to zero.
preserves/text.py def extend(self, text):\n\"\"\"Appends `text` to the remaining contents of `self.input_buffer`, trimming already-processed\n text from the front of `self.input_buffer` and resetting `self.index` to zero.\"\"\"\n self.input_buffer = self.input_buffer[self.index:] + text\n self.index = 0\nnext()","text":"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.
preserves/text.py def next(self):\n\"\"\"Reads the next complete `Value` from the internal buffer, raising\n [ShortPacket][preserves.error.ShortPacket] if too few bytes are available, or\n [DecodeError][preserves.error.DecodeError] if the input is invalid somehow.\n\n \"\"\"\n self.skip_whitespace()\n c = self.peek()\n if c == '\"':\n self.skip()\n return self.wrap(self.read_string('\"'))\n if c == '|':\n self.skip()\n return self.wrap(Symbol(self.read_string('|')))\n if c in ';@':\n annotations = self.gather_annotations()\n v = self.next()\n if self.include_annotations:\n v.annotations = annotations + v.annotations\n return v\n if c == ':':\n raise DecodeError('Unexpected key/value separator between items')\n if c == '#':\n self.skip()\n c = self.nextchar()\n if c == 'f': return self.wrap(False)\n if c == 't': return self.wrap(True)\n if c == '{': return self.wrap(frozenset(self.upto('}')))\n if c == '\"': return self.wrap(self.read_literal_binary())\n if c == 'x':\n c = self.nextchar()\n if c == '\"': return self.wrap(self.read_hex_binary())\n if c == 'f': return self.wrap(self.read_hex_float(4))\n if c == 'd': return self.wrap(self.read_hex_float(8))\n raise DecodeError('Invalid #x syntax')\n if c == '[': return self.wrap(self.read_base64_binary())\n if c == '=':\n old_ann = self.include_annotations\n self.include_annotations = True\n bs_val = self.next()\n self.include_annotations = old_ann\n if len(bs_val.annotations) > 0:\n raise DecodeError('Annotations not permitted after #=')\n bs_val = bs_val.item\n if not isinstance(bs_val, bytes):\n raise DecodeError('ByteString must follow #=')\n return self.wrap(Decoder(bs_val, include_annotations = self.include_annotations).next())\n if c == '!':\n if self.parse_embedded is None:\n raise DecodeError('No parse_embedded function supplied')\n return self.wrap(Embedded(self.parse_embedded(self.next())))\n raise DecodeError('Invalid # syntax')\n if c == '<':\n self.skip()\n vs = self.upto('>')\n if len(vs) == 0:\n raise DecodeError('Missing record label')\n return self.wrap(Record(vs[0], vs[1:]))\n if c == '[':\n self.skip()\n return self.wrap(self.upto(']'))\n if c == '{':\n self.skip()\n return self.wrap(self.read_dictionary())\n if c in '>]}':\n raise DecodeError('Unexpected ' + c)\n self.skip()\n return self.wrap(self.read_raw_symbol_or_number([c]))\ntry_next()","text":"Like next, but returns None instead of raising ShortPacket.
preserves/text.py def try_next(self):\n\"\"\"Like [next][preserves.text.Parser.next], but returns `None` instead of raising\n [ShortPacket][preserves.error.ShortPacket].\"\"\"\n start = self.index\n try:\n return self.next()\n except ShortPacket:\n self.index = start\n return None\nparse(text, **kwargs)","text":"Yields the first complete encoded value from text, passing kwargs through to the Parser constructor. Raises exceptions as per next.
Parameters:
Name Type Description Defaulttext str encoded data to decode
required Source code inpreserves/text.py def parse(text, **kwargs):\n\"\"\"Yields the first complete encoded value from `text`, passing `kwargs` through to the\n [Parser][preserves.text.Parser] constructor. Raises exceptions as per\n [next][preserves.text.Parser.next].\n\n Args:\n text (str): encoded data to decode\n\n \"\"\"\n return Parser(input_buffer=text, **kwargs).next()\nparse_with_annotations(bs, **kwargs)","text":"Like parse, but supplying include_annotations=True to the Parser constructor.
preserves/text.py def parse_with_annotations(bs, **kwargs):\n\"\"\"Like [parse][preserves.text.parse], but supplying `include_annotations=True` to the\n [Parser][preserves.text.Parser] constructor.\"\"\"\n return Parser(input_buffer=bs, include_annotations=True, **kwargs).next()\nstringify(v, **kwargs)","text":"Convert a single Value v to a string. Any supplied kwargs are passed on to the underlying Formatter constructor.
preserves/text.py def stringify(v, **kwargs):\n\"\"\"Convert a single `Value` `v` to a string. Any supplied `kwargs` are passed on to the\n underlying [Formatter][preserves.text.Formatter] constructor.\"\"\"\n e = Formatter(**kwargs)\n e.append(v)\n return e.contents()\nPython's strings, byte strings, integers, booleans, and double-precision floats stand directly for their Preserves counterparts. Wrapper objects for Float and Symbol complete the suite of atomic types.
Python's lists and tuples correspond to Preserves Sequences, and dicts and sets to Dictionary and Set values, respectively. Preserves Records are represented by Record objects. Finally, embedded values are represented by Embedded objects.
The preserves.values module implements the core representations of Preserves Values as Python values.
Annotated(item)","text":" 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\n>>> a = preserves.parse('''\n... ; A comment\n... [1 2 3]\n... ''', include_annotations=True)\n>>> a\n@' A comment' (1, 2, 3)\n>>> a.item\n(1, 2, 3)\n>>> a.annotations\n[' A comment']\n>>> a == (1, 2, 3)\nTrue\n>>> a == preserves.parse('@xyz [1 2 3]', include_annotations=True)\nTrue\n>>> a[0]\nTraceback (most recent call last):\n ...\nTypeError: 'Annotated' object is not subscriptable\n>>> a.item[0]\n1\n>>> type(a.item[0])\n<class 'preserves.values.Annotated'>\n>>> a.item[0].annotations\n[]\n>>> print(preserves.stringify(a))\n@\" A comment\" [1 2 3]\n>>> print(preserves.stringify(a, include_annotations=False))\n[1 2 3]\nAttributes:
Name Type Descriptionitem Value the underlying annotated Value
annotations list[Value] the annotations attached to self.item
preserves/values.py def __init__(self, item):\n self.annotations = []\n self.item = item\npeel()","text":"Calls strip_annotations on self with depth=1.
preserves/values.py def peel(self):\n\"\"\"Calls [strip_annotations][preserves.values.strip_annotations] on `self` with `depth=1`.\"\"\"\n return strip_annotations(self, 1)\nstrip(depth=inf)","text":"Calls strip_annotations on self and depth.
preserves/values.py def strip(self, depth=inf):\n\"\"\"Calls [strip_annotations][preserves.values.strip_annotations] on `self` and `depth`.\"\"\"\n return strip_annotations(self, depth)\nEmbedded(embeddedValue)","text":"Representation of a Preserves Embedded value. For more on the meaning and use of embedded values, see the specification.
>>> import io\n>>> e = Embedded(io.StringIO('some text'))\n>>> e # doctest: +ELLIPSIS\n#!<_io.StringIO object at ...>\n>>> e.embeddedValue # doctest: +ELLIPSIS\n<_io.StringIO object at ...>\n>>> import preserves\n>>> print(preserves.stringify(Embedded(None)))\nTraceback (most recent call last):\n ...\nTypeError: Cannot preserves-format: None\n>>> print(preserves.stringify(Embedded(None), format_embedded=lambda x: 'abcdef'))\n#!\"abcdef\"\nAttributes:
Name Type DescriptionembeddedValue any Python value; could be a platform object, could be a representation of a Preserves Value, could be None, could be anything!
preserves/values.py def __init__(self, embeddedValue):\n self.embeddedValue = embeddedValue\nFloat(value)","text":" Bases: object
Wrapper for treating a Python double-precision floating-point value as a single-precision (32-bit) float, from Preserves' perspective. (Python lacks native single-precision floating point support.)
>>> Float(3.45)\nFloat(3.45)\n>>> import preserves\n>>> preserves.stringify(Float(3.45))\n'3.45f'\n>>> preserves.stringify(3.45)\n'3.45'\n>>> preserves.parse('3.45f')\nFloat(3.45)\n>>> preserves.parse('3.45')\n3.45\n>>> preserves.encode(Float(3.45))\nb'\\x82@\\\\\\xcc\\xcd'\n>>> preserves.encode(3.45)\nb'\\x83@\\x0b\\x99\\x99\\x99\\x99\\x99\\x9a'\nAttributes:
Name Type Descriptionvalue float the double-precision representation of intended single-precision value
Source code inpreserves/values.py def __init__(self, value):\n self.value = value\nfrom_bytes(bs) staticmethod","text":"Converts a 4-byte-long byte string to a 32-bit single-precision floating point value wrapped in a Float instance. Takes care to preserve the quiet/signalling bit-pattern of NaN values, unlike its struct.unpack('>f', ...) equivalent.
>>> Float.from_bytes(b'\\x7f\\x80\\x00{')\nFloat(nan)\n>>> Float.from_bytes(b'\\x7f\\x80\\x00{').to_bytes()\nb'\\x7f\\x80\\x00{'\n\n>>> struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]\nnan\n>>> Float(struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]).to_bytes()\nb'\\x7f\\xc0\\x00{'\n>>> struct.pack('>f', struct.unpack('>f', b'\\x7f\\x80\\x00{')[0])\nb'\\x7f\\xc0\\x00{'\n(Note well the difference between 7f80007b and 7fc0007b!)
preserves/values.py @staticmethod\ndef from_bytes(bs):\n\"\"\"Converts a 4-byte-long byte string to a 32-bit single-precision floating point value\n wrapped in a [Float][preserves.values.Float] instance. Takes care to preserve the\n quiet/signalling bit-pattern of NaN values, unlike its `struct.unpack('>f', ...)`\n equivalent.\n\n ```python\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{')\n Float(nan)\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{').to_bytes()\n b'\\\\x7f\\\\x80\\\\x00{'\n\n >>> struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]\n nan\n >>> Float(struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]).to_bytes()\n b'\\\\x7f\\\\xc0\\\\x00{'\n >>> struct.pack('>f', struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0])\n b'\\\\x7f\\\\xc0\\\\x00{'\n\n ```\n\n (Note well the difference between `7f80007b` and `7fc0007b`!)\n\n \"\"\"\n vf = struct.unpack('>I', bs)[0]\n if (vf & 0x7f800000) == 0x7f800000:\n # NaN or inf. Preserve quiet/signalling bit by manually expanding to double-precision.\n sign = vf >> 31\n payload = vf & 0x007fffff\n dbs = struct.pack('>Q', (sign << 63) | 0x7ff0000000000000 | (payload << 29))\n return Float(struct.unpack('>d', dbs)[0])\n else:\n return Float(struct.unpack('>f', bs)[0])\nto_bytes()","text":"Converts this 32-bit single-precision floating point value to its binary32 format, taking care to preserve the quiet/signalling bit-pattern of NaN values, unlike its struct.pack('>f', ...) equivalent.
>>> Float.from_bytes(b'\\x7f\\x80\\x00{')\nFloat(nan)\n>>> Float.from_bytes(b'\\x7f\\x80\\x00{').to_bytes()\nb'\\x7f\\x80\\x00{'\n\n>>> struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]\nnan\n>>> Float(struct.unpack('>f', b'\\x7f\\x80\\x00{')[0]).to_bytes()\nb'\\x7f\\xc0\\x00{'\n>>> struct.pack('>f', struct.unpack('>f', b'\\x7f\\x80\\x00{')[0])\nb'\\x7f\\xc0\\x00{'\n(Note well the difference between 7f80007b and 7fc0007b!)
preserves/values.py def to_bytes(self):\n\"\"\"Converts this 32-bit single-precision floating point value to its binary32 format,\n taking care to preserve the quiet/signalling bit-pattern of NaN values, unlike its\n `struct.pack('>f', ...)` equivalent.\n\n ```python\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{')\n Float(nan)\n >>> Float.from_bytes(b'\\\\x7f\\\\x80\\\\x00{').to_bytes()\n b'\\\\x7f\\\\x80\\\\x00{'\n\n >>> struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]\n nan\n >>> Float(struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0]).to_bytes()\n b'\\\\x7f\\\\xc0\\\\x00{'\n >>> struct.pack('>f', struct.unpack('>f', b'\\\\x7f\\\\x80\\\\x00{')[0])\n b'\\\\x7f\\\\xc0\\\\x00{'\n\n ```\n\n (Note well the difference between `7f80007b` and `7fc0007b`!)\n\n \"\"\"\n\n if math.isnan(self.value) or math.isinf(self.value):\n dbs = struct.pack('>d', self.value)\n vd = struct.unpack('>Q', dbs)[0]\n sign = vd >> 63\n payload = (vd >> 29) & 0x007fffff\n vf = (sign << 31) | 0x7f800000 | payload\n return struct.pack('>I', vf)\n else:\n return struct.pack('>f', self.value)\nImmutableDict(*args, **kwargs)","text":" 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)])\n>>> d\n{'a': 1, 'b': 2}\n>>> d['c'] = 3\nTraceback (most recent call last):\n ...\nTypeError: Immutable\n>>> del d['b']\nTraceback (most recent call last):\n ...\nTypeError: Immutable\npreserves/values.py def __init__(self, *args, **kwargs):\n if hasattr(self, '__hash'): raise TypeError('Immutable')\n super(ImmutableDict, self).__init__(*args, **kwargs)\n self.__hash = None\nfrom_kvs(kvs) staticmethod","text":"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])\n{'a': 1, 'b': 2}\n>>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3\nTraceback (most recent call last):\n ...\nTypeError: Immutable\npreserves/values.py @staticmethod\ndef from_kvs(kvs):\n\"\"\"Constructs an [ImmutableDict][preserves.values.ImmutableDict] from a sequence of\n alternating keys and values; compare to the\n [ImmutableDict][preserves.values.ImmutableDict] constructor, which takes a sequence of\n key-value pairs.\n\n ```python\n >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])\n {'a': 1, 'b': 2}\n >>> ImmutableDict.from_kvs(['a', 1, 'b', 2])['c'] = 3\n Traceback (most recent call last):\n ...\n TypeError: Immutable\n\n ```\n\n \"\"\"\n\n i = iter(kvs)\n result = ImmutableDict()\n result_proxy = super(ImmutableDict, result)\n try:\n while True:\n k = next(i)\n try:\n v = next(i)\n except StopIteration:\n raise DecodeError(\"Missing dictionary value\")\n result_proxy.__setitem__(k, v)\n except StopIteration:\n pass\n return result\nRecord(key, fields)","text":" 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']])\n>>> r\n#label('field1', ['field2item1', 'field2item2'])\n>>> r.key\n#label\n>>> r.fields\n('field1', ['field2item1', 'field2item2'])\n>>> import preserves\n>>> preserves.stringify(r)\n'<label \"field1\" [\"field2item1\" \"field2item2\"]>'\n>>> r == preserves.parse('<label \"field1\" [\"field2item1\" \"field2item2\"]>')\nTrue\nParameters:
Name Type Description Defaultkey Value the Record's label
fields iterable[Value] the fields of the Record
Attributes:
Name Type Descriptionkey Value the Record's label
fields tuple[Value] the fields of the Record
preserves/values.py def __init__(self, key, fields):\n self.key = key\n self.fields = tuple(fields)\n self.__hash = None\nmakeBasicConstructor(label, fieldNames) staticmethod","text":"Constructs and returns a \"constructor\" for Records having a certain label and number of fields.
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')\n>>> c(1969, 7, 16)\n#date(1969, 7, 16)\n>>> c.constructorInfo\n#date/3\n>>> c.isClassOf(c(1969, 7, 16))\nTrue\n>>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))\nTrue\n>>> c.isClassOf(Record(Symbol('date'), [1969]))\nFalse\n>>> c.ensureClassOf(c(1969, 7, 16))\n#date(1969, 7, 16)\n>>> c.ensureClassOf(Record(Symbol('date'), [1969]))\nTraceback (most recent call last):\n ...\nTypeError: Record: expected #date/3, got #date(1969)\n>>> c._year(c(1969, 7, 16))\n1969\n>>> c._month(c(1969, 7, 16))\n7\n>>> c._day(c(1969, 7, 16))\n16\nParameters:
Name Type Description Defaultlabel Value Label to use for constructed/matched Records
fieldNames tuple[str] | list[str] | str Names of the Record's fields
preserves/values.py @staticmethod\ndef makeBasicConstructor(label, fieldNames):\n\"\"\"Constructs and returns a \"constructor\" for `Record`s having a certain `label` and\n number of fields.\n\n Deprecated:\n Use [preserves.schema][] definitions instead.\n\n The \"constructor\" is a callable function that accepts `len(fields)` arguments and\n returns a [Record][preserves.values.Record] with `label` as its label and the arguments\n to the constructor as field values.\n\n In addition, the \"constructor\" has a `constructorInfo` attribute holding a\n [RecordConstructorInfo][preserves.values.RecordConstructorInfo] object, an `isClassOf`\n attribute holding a unary function that returns `True` iff its argument is a\n [Record][preserves.values.Record] with label `label` and arity `len(fieldNames)`, and\n an `ensureClassOf` attribute that raises an `Exception` if `isClassOf` returns false on\n its argument and returns the argument otherwise.\n\n Finally, for each field name `f` in `fieldNames`, the \"constructor\" object has an\n attribute `_f` that is a unary function that retrieves the `f` field from the passed in\n argument.\n\n ```python\n >>> c = Record.makeBasicConstructor(Symbol('date'), 'year month day')\n >>> c(1969, 7, 16)\n #date(1969, 7, 16)\n >>> c.constructorInfo\n #date/3\n >>> c.isClassOf(c(1969, 7, 16))\n True\n >>> c.isClassOf(Record(Symbol('date'), [1969, 7, 16]))\n True\n >>> c.isClassOf(Record(Symbol('date'), [1969]))\n False\n >>> c.ensureClassOf(c(1969, 7, 16))\n #date(1969, 7, 16)\n >>> c.ensureClassOf(Record(Symbol('date'), [1969]))\n Traceback (most recent call last):\n ...\n TypeError: Record: expected #date/3, got #date(1969)\n >>> c._year(c(1969, 7, 16))\n 1969\n >>> c._month(c(1969, 7, 16))\n 7\n >>> c._day(c(1969, 7, 16))\n 16\n\n ```\n\n Args:\n label (Value): Label to use for constructed/matched `Record`s\n fieldNames (tuple[str] | list[str] | str): Names of the `Record`'s fields\n\n \"\"\"\n if type(fieldNames) == str:\n fieldNames = fieldNames.split()\n arity = len(fieldNames)\n def ctor(*fields):\n if len(fields) != arity:\n raise Exception(\"Record: cannot instantiate %r expecting %d fields with %d fields\"%(\n label,\n arity,\n len(fields)))\n return Record(label, fields)\n ctor.constructorInfo = RecordConstructorInfo(label, arity)\n ctor.isClassOf = lambda v: \\\n isinstance(v, Record) and v.key == label and len(v.fields) == arity\n def ensureClassOf(v):\n if not ctor.isClassOf(v):\n raise TypeError(\"Record: expected %r/%d, got %r\" % (label, arity, v))\n return v\n ctor.ensureClassOf = ensureClassOf\n for fieldIndex in range(len(fieldNames)):\n fieldName = fieldNames[fieldIndex]\n # Stupid python scoping bites again\n def getter(fieldIndex):\n return lambda v: ensureClassOf(v)[fieldIndex]\n setattr(ctor, '_' + fieldName, getter(fieldIndex))\n return ctor\nmakeConstructor(labelSymbolText, fieldNames) staticmethod","text":"Equivalent to Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames).
Use preserves.schema definitions instead.
Source code inpreserves/values.py @staticmethod\ndef makeConstructor(labelSymbolText, fieldNames):\n\"\"\"\n Equivalent to `Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)`.\n\n Deprecated:\n Use [preserves.schema][] definitions instead.\n \"\"\"\n return Record.makeBasicConstructor(Symbol(labelSymbolText), fieldNames)\nRecordConstructorInfo(key, arity)","text":" Bases: object
Describes the shape of a Record constructor, namely its label and its arity (field count).
>>> RecordConstructorInfo(Symbol('label'), 3)\n#label/3\nAttributes:
Name Type Descriptionkey Value the label of matching Records
arity int the number of fields in matching Records
preserves/values.py def __init__(self, key, arity):\n self.key = key\n self.arity = arity\nSymbol(name)","text":" Bases: object
Representation of Preserves Symbols.
>>> Symbol('xyz')\n#xyz\n>>> Symbol('xyz').name\n'xyz'\n>>> import preserves\n>>> preserves.stringify(Symbol('xyz'))\n'xyz'\n>>> preserves.stringify(Symbol('hello world'))\n'|hello world|'\n>>> preserves.parse('xyz')\n#xyz\n>>> preserves.parse('|hello world|')\n#hello world\nAttributes:
Name Type Descriptionname str the symbol's text label
Source code inpreserves/values.py def __init__(self, name):\n self.name = name.name if isinstance(name, Symbol) else name\nannotate(v, *anns)","text":"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\n>>> print(preserves.stringify(annotate(123, \"A comment\", \"Another comment\")))\n@\"A comment\" @\"Another comment\" 123\npreserves/values.py def annotate(v, *anns):\n\"\"\"Wraps `v` in an [Annotated][preserves.values.Annotated] object, if it isn't already\n wrapped, and appends each of the `anns` to the [Annotated][preserves.values.Annotated]'s\n `annotations` sequence. NOTE: Does not recursively ensure that any parts of the argument\n `v` are themselves wrapped in [Annotated][preserves.values.Annotated] objects!\n\n ```python\n >>> import preserves\n >>> print(preserves.stringify(annotate(123, \"A comment\", \"Another comment\")))\n @\"A comment\" @\"Another comment\" 123\n\n ```\n \"\"\"\n if not is_annotated(v):\n v = Annotated(v)\n for a in anns:\n v.annotations.append(a)\n return v\ncmp_floats(a, b)","text":"Implements the totalOrder predicate defined in section 5.10 of IEEE Std 754-2008.
preserves/values.py def cmp_floats(a, b):\n\"\"\"Implements the `totalOrder` predicate defined in section 5.10 of [IEEE Std\n 754-2008](https://dx.doi.org/10.1109/IEEESTD.2008.4610935).\n\n \"\"\"\n a = float_to_int(a)\n b = float_to_int(b)\n if a & 0x8000000000000000: a = a ^ 0x7fffffffffffffff\n if b & 0x8000000000000000: b = b ^ 0x7fffffffffffffff\n return a - b\ndict_kvs(d)","text":"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}))\n['a', 1, 'b', 2]\npreserves/values.py def dict_kvs(d):\n\"\"\"Generator function yielding a sequence of alternating keys and values from `d`. In some\n sense the inverse of [ImmutableDict.from_kvs][preserves.values.ImmutableDict.from_kvs].\n\n ```python\n >>> list(dict_kvs({'a': 1, 'b': 2}))\n ['a', 1, 'b', 2]\n\n ```\n \"\"\"\n for k in d:\n yield k\n yield d[k]\nis_annotated(v)","text":"True iff v is an instance of Annotated.
preserves/values.py def is_annotated(v):\n\"\"\"`True` iff `v` is an instance of [Annotated][preserves.values.Annotated].\"\"\"\n return isinstance(v, Annotated)\npreserve(v)","text":"Converts v to a representation of a Preserves Value by (repeatedly) setting
v = v.__preserve__()\nwhile v has a __preserve__ method. Parsed Schema values are able to render themselves to their serialized representations this way.
preserves/values.py def preserve(v):\n\"\"\"Converts `v` to a representation of a Preserves `Value` by (repeatedly) setting\n\n ```python\n v = v.__preserve__()\n ```\n\n while `v` has a `__preserve__` method. Parsed [Schema][preserves.schema]\n values are able to render themselves to their serialized representations this way.\n\n \"\"\"\n while hasattr(v, '__preserve__'):\n v = v.__preserve__()\n return v\nstrip_annotations(v, depth=inf)","text":"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\n>>> a = preserves.parse('@\"A comment\" [@a 1 @b 2 @c 3]', include_annotations=True)\n>>> is_annotated(a)\nTrue\n>>> print(preserves.stringify(a))\n@\"A comment\" [@a 1 @b 2 @c 3]\n>>> print(preserves.stringify(strip_annotations(a)))\n[1 2 3]\n>>> print(preserves.stringify(strip_annotations(a, depth=1)))\n[@a 1 @b 2 @c 3]\npreserves/values.py def strip_annotations(v, depth=inf):\n\"\"\"Exposes `depth` layers of raw structure of\n potentially-[Annotated][preserves.values.Annotated] `Value`s. If `depth==0` or `v` is not\n [Annotated][preserves.values.Annotated], just returns `v`. Otherwise, descends recursively\n into the structure of `v.item`.\n\n ```python\n >>> import preserves\n >>> a = preserves.parse('@\"A comment\" [@a 1 @b 2 @c 3]', include_annotations=True)\n >>> is_annotated(a)\n True\n >>> print(preserves.stringify(a))\n @\"A comment\" [@a 1 @b 2 @c 3]\n >>> print(preserves.stringify(strip_annotations(a)))\n [1 2 3]\n >>> print(preserves.stringify(strip_annotations(a, depth=1)))\n [@a 1 @b 2 @c 3]\n\n ```\n \"\"\"\n\n if depth == 0: return v\n if not is_annotated(v): return v\n\n next_depth = depth - 1\n def walk(v):\n return strip_annotations(v, next_depth)\n\n v = v.item\n if isinstance(v, Record):\n return Record(strip_annotations(v.key, depth), tuple(walk(f) for f in v.fields))\n elif isinstance(v, list):\n return tuple(walk(f) for f in v)\n elif isinstance(v, tuple):\n return tuple(walk(f) for f in v)\n elif isinstance(v, set):\n return frozenset(walk(f) for f in v)\n elif isinstance(v, frozenset):\n return frozenset(walk(f) for f in v)\n elif isinstance(v, dict):\n return ImmutableDict.from_kvs(walk(f) for f in dict_kvs(v))\n elif is_annotated(v):\n raise ValueError('Improper annotation structure')\n else:\n return v\n