"""
Read and write the JSON-able values as a subset of XML-RPC.
JSON prescribes no XML representation of its own, so we borrow the one of
XML-RPC for it: a value is written as a single discriminator element --
``<nil>``, ``<boolean>``, ``<double>``, ``<string>``, ``<array>`` or
``<struct>`` -- which says what the value is. XML-RPC itself knows no null,
so we follow the widespread extension, and write a JSON null as an empty
``<nil/>``.
This module knows nothing of the meta-model beyond the JSON-able types, and
it drives no document of its own. The caller hands over the very stream of
``(event, element)`` tuples, or the very
:py:class:`aas_core3_2.xmlcommon.Writer`, with which
the enclosing document is being read or written, so that a JSON-able property
is read and written in the same pass as everything around it.
These elements reside in no namespace at all, as the XML-RPC specification
prescribes, and not in the namespace of the enclosing document. The outermost
one therefore undeclares the default namespace with ``xmlns=""``, and
the elements nested in it inherit that.
"""
# This code has been automatically generated by aas-core-codegen.
# Do NOT edit or append.
import collections.abc
import math
import re
from typing import Any, Dict, Iterator, List, Tuple
import aas_core3_2.common as our_common
import aas_core3_2.types as our_types
from aas_core3_2.xmlcommon import (
DeserializationException,
Element,
ElementSegment,
IndexSegment,
KeySegment,
SerializationException,
Writer,
collapse_whitespace,
parse_unqualified_element_tag,
raise_if_has_tail_or_attrib,
raise_if_non_whitespace_text,
read_end_element,
read_next_start_element,
read_str_from_element_text,
read_text_from_element,
)
# region De-serialization
[docs]def read_value_content(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> our_types.JsonValue:
"""
Read the content of :paramref:`element` as a JSON-able value.
The content is a single discriminator element -- ``<nil>``,
``<boolean>``, ``<double>``, ``<string>``, ``<array>`` or ``<struct>`` --
which says what the value is.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element enclosing the discriminator
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed JSON-able value
"""
raise_if_has_tail_or_attrib(element)
raise_if_non_whitespace_text(element, "a discriminator element")
discriminator = read_next_start_element(
iterator,
"a discriminator "
"(one of <nil>, <boolean>, <double>, <string>, <array> or <struct>)",
)
result = _read_discriminator(discriminator, iterator)
read_end_element(element, iterator)
return result
#: Match a numeral of the ``<double>`` lexical space.
#:
#: This is the numeric part of the lexical space of ``xs:double``, and
#: deliberately not its three named literals -- ``INF``, ``-INF`` and ``NaN``
#: -- since a JSON number can be none of them.
#:
#: Mind the explicit ``[0-9]``: ``\d`` would match a digit of any script, so
#: the Arabic-Indic ``۵`` would pass, and :py:func:`float` would read it
#: as 5.
#:
#: See: https://www.w3.org/TR/xmlschema-2/#double
_DOUBLE_RE = re.compile(r"(\+|-)?([0-9]+(\.[0-9]*)?|\.[0-9]+)([Ee](\+|-)?[0-9]+)?")
def _read_discriminator(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> our_types.JsonValue:
"""
Read :paramref:`element` as one of the six XML-RPC discriminators.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element of the discriminator
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed JSON-able value
"""
tag_wo_ns = parse_unqualified_element_tag(element)
# NOTE (mristin):
# The discriminator is an element which we actually enter, so it is a step
# of the path. The one case which gets no step is the element which is no
# discriminator at all, raised below and outside of this ``try``: it is that
# very element which does not belong here, so naming it in the path as well
# as in the message would say nothing more.
try:
if tag_wo_ns == "nil":
read_end_element(element, iterator)
# NOTE (mristin):
# The text of an element is only complete at its end, so we can check
# it only now.
if element.text is not None and len(element.text.strip()) != 0:
raise DeserializationException(
f"Expected no text in a <nil> element, but got: {element.text!r}"
)
return None
if tag_wo_ns == "boolean":
# NOTE (mristin):
# Real XML-RPC tooling writes and expects a strict ``1``/``0``, and not
# the ``true``/``false`` which ``xs:boolean`` and the xmlization use.
text = collapse_whitespace(read_text_from_element(element, iterator))
if text == "1":
return True
if text == "0":
return False
raise DeserializationException(
f'Expected "0" or "1" as the text of a <boolean> element, '
f"but got: {text!r}"
)
if tag_wo_ns == "double":
text = collapse_whitespace(read_text_from_element(element, iterator))
# NOTE (mristin):
# The lexical form is matched before the text is converted. ``float``
# reads far more than we admit here: a hexadecimal significand, so
# "0x10" would come out as 16, an underscore separator, and
# the spellings "inf", "infinity" and "nan".
#
# Mind that the numeral excludes "INF", "-INF" and "NaN" on purpose,
# unlike ``xs:double``, which names all three. A <double> carries
# a JSON number, and JSON knows neither an infinity nor
# a not-a-number, so there is no JSON-able value for such a text
# to de-serialize into.
if _DOUBLE_RE.fullmatch(text) is None:
raise DeserializationException(
f"Expected a number as the text of a <double> element, "
f"but got: {text!r}"
)
number = float(text)
# NOTE (mristin):
# A literal too large for a ``float`` gives an infinity, which is no
# JSON-able value either, so it is refused rather than rounded.
if not math.isfinite(number):
raise DeserializationException(
f"Expected a number representable as a JSON-able value as "
f"the text of a <double> element, but got a value which "
f"rounds to an infinity: {text!r}"
)
return number
if tag_wo_ns == "string":
# NOTE (mristin):
# ``xs:string`` is ``preserve`` and not ``collapse``, so a <string>
# keeps its whitespace, unlike a <boolean> or a <double>.
return read_str_from_element_text(element, iterator)
if tag_wo_ns == "array":
return read_array_body(element, iterator)
if tag_wo_ns == "struct":
return read_struct_body(element, iterator)
except DeserializationException as exception:
exception.path._prepend(ElementSegment(element))
raise
raise DeserializationException(
f"Expected a discriminator element "
f"(one of <nil>, <boolean>, <double>, <string>, <array> or <struct>), "
f"but got: {tag_wo_ns!r}"
)
[docs]def read_array_body(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> our_types.JsonArray:
"""
Read the content of :paramref:`element` as a ``<data>`` of ``<value>``'s.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element enclosing the ``<data>``
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed JSON-able array
"""
raise_if_has_tail_or_attrib(element)
raise_if_non_whitespace_text(element, "a <data> element")
data_element = read_next_start_element(iterator, "a <data> element")
if parse_unqualified_element_tag(data_element) != "data":
raise DeserializationException(
f"Expected a start element <data>, " f"but got element {data_element.tag!r}"
)
# NOTE (mristin):
# The <data> element is a step of the path, as we enter it, and so is
# the position of an item. The item's own <value> element is not: the index
# already names it, so a step of its own would say nothing more.
try:
raise_if_non_whitespace_text(data_element, "<value> elements")
result: List[Any] = []
while True:
# NOTE (mristin):
# We pull the next item element here instead of delegating it to
# a helper, as this loop runs once for every item of every array.
next_event_element = next(iterator, None)
if next_event_element is None:
raise DeserializationException(
f"Expected a <value> element or the end element corresponding "
f"to {data_element.tag}, but got the end-of-input"
)
next_event, item_element = next_event_element
if next_event == "end" and item_element.tag == data_element.tag:
# We reached the end element enclosing the items.
break
if next_event != "start":
raise DeserializationException(
f"Expected a start element corresponding to an item, "
f"but got event {next_event!r} "
f"and element {item_element.tag!r}"
)
try:
item = read_value_element(item_element, iterator)
except DeserializationException as exception:
exception.path._prepend(IndexSegment(item_element, len(result)))
raise
result.append(item)
except DeserializationException as exception:
exception.path._prepend(ElementSegment(data_element))
raise
read_end_element(element, iterator)
return result
[docs]def read_value_element(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> our_types.JsonValue:
"""
Read :paramref:`element`, which has to be a ``<value>``, as a JSON-able value.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element, expected to be a ``<value>``
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed JSON-able value
"""
tag_wo_ns = parse_unqualified_element_tag(element)
if tag_wo_ns != "value":
raise DeserializationException(
f"Expected a start element <value>, but got: {tag_wo_ns!r}"
)
return read_value_content(element, iterator)
[docs]def read_struct_body(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> our_types.JsonObject:
"""
Read the content of :paramref:`element` as a sequence of ``<member>``'s.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element enclosing the members
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed JSON-able object
"""
raise_if_has_tail_or_attrib(element)
raise_if_non_whitespace_text(element, "<member> elements")
result: Dict[str, Any] = {}
while True:
next_event_element = next(iterator, None)
if next_event_element is None:
raise DeserializationException(
f"Expected a <member> element or the end element corresponding "
f"to {element.tag}, but got the end-of-input"
)
next_event, member_element = next_event_element
if next_event == "end" and member_element.tag == element.tag:
break
if (
next_event != "start"
or parse_unqualified_element_tag(member_element) != "member"
):
raise DeserializationException(
f"Expected a start element <member>, but got "
f"event {next_event!r} and element {member_element.tag!r}"
)
key, value = _read_member(member_element, iterator)
# NOTE (mristin):
# A repeated <member> name is refused, just as a repeated property
# element is refused in the xmlization. Letting the later member win
# would silently accept a document which says two different things
# about the same key.
if key in result:
exception = DeserializationException("The member occurred more than once")
exception.path._prepend(KeySegment(member_element, key))
raise exception
result[key] = value
return result
def _read_member(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> Tuple[str, our_types.JsonValue]:
"""
Read :paramref:`element`, a ``<member>``, as a key and a JSON-able value.
The end element corresponding to :paramref:`element` will be read as well.
:param element: start element of the ``<member>``
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: the member's name and its parsed value
"""
raise_if_has_tail_or_attrib(element)
raise_if_non_whitespace_text(element, "a <name> and a <value> element")
name_element = read_next_start_element(iterator, "a <name> element")
if parse_unqualified_element_tag(name_element) != "name":
raise DeserializationException(
f"Expected a start element <name>, " f"but got element {name_element.tag!r}"
)
key = read_str_from_element_text(name_element, iterator)
# NOTE (mristin):
# From here on the key is known, so every failure can name the member it
# belongs to. A <struct> carries the key of a member in a <name> child
# element instead of in an attribute, so the key segment renders as
# a predicate on that child element, and the <member> element itself gets
# no step of its own.
try:
value_element = read_next_start_element(iterator, "a <value> element")
try:
value = read_value_element(value_element, iterator)
except DeserializationException as exception:
exception.path._prepend(ElementSegment(value_element))
raise
except DeserializationException as exception:
exception.path._prepend(KeySegment(element, key))
raise
read_end_element(element, iterator)
return key, value
# endregion
# region Serialization
[docs]def write_discriminator(value: our_types.JsonValue, writer: Writer) -> None:
"""
Write :paramref:`value` as one of the six XML-RPC discriminators.
:param value: to be serialized
:param writer: to write to
:raise: :py:class:`SerializationException` if the value could not be written
"""
if value is None:
writer.write_empty_element_in_no_namespace("nil")
return
# NOTE (mristin):
# A ``bool`` is an ``int`` in Python, so it has to be checked first --
# otherwise every boolean would be written as a number.
if isinstance(value, bool):
# NOTE (mristin):
# Real XML-RPC tooling writes and expects a strict ``1``/``0``, and not
# the ``true``/``false`` which ``xs:boolean`` and the xmlization use.
writer.write_start_element_in_no_namespace("boolean")
writer.stream.write("1" if value else "0")
writer.write_end_element_in_no_namespace("boolean")
return
if isinstance(value, int):
# NOTE (mristin):
# An ``int`` is written out as it stands, free of the fraction which
# a ``float`` carries. JSON knows a single numeric type, though, so
# it has to be one which a ``float`` can hold exactly -- otherwise
# a reader, which reads every <double> as a ``float``, would get
# a different number back.
if our_common.try_to_convert_int_to_float(value) is None:
raise SerializationException(
f"Expected a JSON-able value, but got the integer {value}, "
f"which is not exactly representable as a JSON number"
)
writer.write_start_element_in_no_namespace("double")
writer.stream.write(str(value))
writer.write_end_element_in_no_namespace("double")
return
if isinstance(value, float):
# NOTE (mristin):
# JSON knows neither an infinity nor a not-a-number, so neither is
# a JSON-able value, and ``_read_discriminator`` refuses to read
# either back.
if not math.isfinite(value):
raise SerializationException(
f"Expected a JSON-able value, but got the number {value}, "
f"which is neither finite nor representable in JSON"
)
writer.write_start_element_in_no_namespace("double")
# NOTE (mristin):
# ``str`` spells a negative zero as "-0.0", but only if the sign is
# looked at: ``value == 0`` holds for both zeroes.
if value == 0 and math.copysign(1.0, value) < 0.0:
writer.stream.write("-0.0")
else:
writer.stream.write(str(value))
writer.write_end_element_in_no_namespace("double")
return
if isinstance(value, str):
writer.write_start_element_in_no_namespace("string")
writer.stream.write(
value.replace("&", "&").replace("<", "<").replace(">", ">")
)
writer.write_end_element_in_no_namespace("string")
return
if isinstance(value, (dict, collections.abc.Mapping)):
writer.write_start_element_in_no_namespace("struct")
write_struct_body(value, writer)
writer.write_end_element_in_no_namespace("struct")
return
array_like = our_common.try_to_cast_to_array_like(value)
if array_like is not None:
writer.write_start_element_in_no_namespace("array")
write_array_body(array_like, writer)
writer.write_end_element_in_no_namespace("array")
return
raise SerializationException(
f"Expected a JSON-able value (a null, a boolean, a number, a string, "
f"an array or an object), but got: {type(value)}"
)
[docs]def write_array_body(value: our_types.JsonArray, writer: Writer) -> None:
"""
Write the ``<data>`` of the JSON-able array :paramref:`value`.
The path of a failure points into :paramref:`value`, and not into
the document, which is only half written when the failure is raised: it is
the value which the caller holds and can look at.
:param value: to be serialized
:param writer: to write to
:raise: :py:class:`SerializationException` if the value could not be written
"""
writer.write_start_element_in_no_namespace("data")
for i, item in enumerate(value):
writer.write_start_element_in_no_namespace("value")
try:
write_discriminator(item, writer)
except SerializationException as exception:
exception._prepend_index(i)
raise
writer.write_end_element_in_no_namespace("value")
writer.write_end_element_in_no_namespace("data")
[docs]def write_struct_body(value: our_types.JsonObject, writer: Writer) -> None:
"""
Write the ``<member>``'s of the JSON-able object :paramref:`value`.
The path of a failure points into :paramref:`value`, and not into
the document, which is only half written when the failure is raised: it is
the value which the caller holds and can look at.
:param value: to be serialized
:param writer: to write to
:raise: :py:class:`SerializationException` if the value could not be written
"""
for key, item in value.items():
if not isinstance(key, str):
# NOTE (mristin):
# A key which is no string names nothing which the path could
# descend into, so the path stops at the object which holds it.
raise SerializationException(
f"Expected only string keys in a JSON-able object, but got "
f"a key of type: {type(key)}"
)
writer.write_start_element_in_no_namespace("member")
writer.write_start_element_in_no_namespace("name")
writer.stream.write(
key.replace("&", "&").replace("<", "<").replace(">", ">")
)
writer.write_end_element_in_no_namespace("name")
writer.write_start_element_in_no_namespace("value")
try:
write_discriminator(item, writer)
except SerializationException as exception:
exception._prepend_key(key)
raise
writer.write_end_element_in_no_namespace("value")
writer.write_end_element_in_no_namespace("member")
# endregion