Source code for openquake.baselib.node

# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright (C) 2014-2023 GEM Foundation
#
# OpenQuake is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OpenQuake is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.

"""
This module defines a Node class, together with a few conversion
functions which are able to convert NRML files into hierarchical
objects (DOM). That makes it easier to read and write XML from Python
and viceversa. Such features are used in the command-line conversion
tools. The Node class is kept intentionally similar to an
Element class, however it overcomes the limitation of ElementTree: in
particular a node can manage a lazy iterable of subnodes, whereas
ElementTree wants to keep everything in memory. Moreover the Node
class provides a convenient dot notation to access subnodes.

The Node class is instantiated with four arguments:

1. the node tag (a mandatory string)
2. the node attributes (a dictionary)
3. the node value (a string or None)
4. the subnodes (an iterable over nodes)

If a node has subnodes, its value should be None.

For instance, here is an example of instantiating a root node
with two subnodes a and b:

>>> from openquake.baselib.node import Node
>>> a = Node('a', {}, 'A1')
>>> b = Node('b', {'attrb': 'B'}, 'B1')
>>> root = Node('root', nodes=[a, b])
>>> root
<root {} None ...>

Node objects can be converted into nicely indented strings:

>>> print(root.to_str())
root
  a 'A1'
  b{attrb='B'} 'B1'
<BLANKLINE>

The subnodes can be retrieved with the dot notation:

>>> root.a
<a {} A1 >

The value of a node can be extracted with the `~` operator:

>>> ~root.a
'A1'

If there are multiple subnodes with the same name

>>> root.append(Node('a', {}, 'A2'))  # add another 'a' node

the dot notation will retrieve the first node.

It is possible to retrieve the other nodes from the ordinal
index:

>>> root[0], root[1], root[2]
(<a {} A1 >, <b {'attrb': 'B'} B1 >, <a {} A2 >)

The list of all subnodes with a given name can be retrieved
as follows:

>>> list(root.getnodes('a'))
[<a {} A1 >, <a {} A2 >]

It is also possible to delete a node given its index:

>>> del root[2]

A node is an iterable object yielding its subnodes:

>>> list(root)
[<a {} A1 >, <b {'attrb': 'B'} B1 >]

The attributes of a node can be retrieved with the square bracket notation:

>>> root.b['attrb']
'B'

It is possible to add and remove attributes freely:

>>> root.b['attr'] = 'new attr'
>>> del root.b['attr']

Node objects can be easily converted into ElementTree objects:

>>> node_to_elem(root)  #doctest: +ELLIPSIS
<Element 'root' at ...>

Then is trivial to generate the XML representation of a node:

>>> from xml.etree import ElementTree
>>> print(ElementTree.tostring(node_to_elem(root)).decode('utf-8'))
<root><a>A1</a><b attrb="B">B1</b></root>

Generating XML files larger than the available memory requires some
care. The trick is to use a node generator, such that it is not
necessary to keep the entire tree in memory. Here is an example:

>>> def gen_many_nodes(N):
...     for i in xrange(N):
...         yield Node('a', {}, 'Text for node %d' % i)

>>> lazytree = Node('lazytree', {}, nodes=gen_many_nodes(10))

The lazytree object defined here consumes no memory, because the
nodes are not created a instantiation time. They are created as
soon as you start iterating on the lazytree. In particular
list(lazytree) will generated all of them. If your goal is to
store the tree on the filesystem in XML format you should use
a writing routine converting a subnode at the time, without
requiring the full list of them. The routines provided by
ElementTree are no good, however baselib.writers
provide an StreamingXMLWriter just for that purpose.

Lazy trees should *not* be used unless it is absolutely necessary in
order to save memory; the problem is that if you use a lazy tree the
slice notation will not work (the underlying generator will not accept
it); moreover it will not be possible to iterate twice on the
subnodes, since the generator will be exhausted. Notice that even
accessing a subnode with the dot notation will avance the
generator. Finally, nodes containing lazy nodes will not be pickleable.
"""
import io
import sys
import copy
import types
import warnings
import itertools
import pprint as pp
import configparser
from contextlib import contextmanager
from openquake.baselib.python3compat import raise_, decode, encode
from xml.etree import ElementTree
from xml.sax.saxutils import escape, quoteattr
from xml.parsers.expat import ParserCreate, ExpatError, ErrorString

import numpy


[docs]@contextmanager
def floatformat(fmt_string):
    """
    Context manager to change the default format string for the
    function :func:`openquake.baselib.writers.write_csv`.

    :param fmt_string: the format to use; for instance '%13.9E'
    """
    fmt_defaults = scientificformat.__defaults__
    scientificformat.__defaults__ = (fmt_string,) + fmt_defaults[1:]
    try:
        yield
    finally:
        scientificformat.__defaults__ = fmt_defaults


zeroset = set(['E', '-', '+', '.', '0'])


[docs]def scientificformat(value, fmt='%13.9E', sep=' ', sep2=':'):
    """
    :param value: the value to convert into a string
    :param fmt: the formatting string to use for float values
    :param sep: separator to use for vector-like values
    :param sep2: second separator to use for matrix-like values

    Convert a float or an array into a string by using the scientific notation
    and a fixed precision (by default 10 decimal digits). For instance:

    >>> scientificformat(-0E0)
    '0.000000000E+00'
    >>> scientificformat(-0.004)
    '-4.000000000E-03'
    >>> scientificformat([0.004])
    '4.000000000E-03'
    >>> scientificformat([0.01, 0.02], '%10.6E')
    '1.000000E-02 2.000000E-02'
    >>> scientificformat([[0.1, 0.2], [0.3, 0.4]], '%4.1E')
    '1.0E-01:2.0E-01 3.0E-01:4.0E-01'
    """
    if isinstance(value, numpy.bool_):
        return '1' if value else '0'
    elif isinstance(value, bytes):
        return value.decode('utf8')
    elif isinstance(value, str):
        return value
    elif hasattr(value, '__len__'):
        return sep.join((scientificformat(f, fmt, sep2) for f in value))
    elif isinstance(value, (float, numpy.float64, numpy.float32)):
        fmt_value = fmt % value
        if set(fmt_value) <= zeroset:
            # '-0.0000000E+00' is converted into '0.0000000E+00
            fmt_value = fmt_value.replace('-', '')
        return fmt_value
    return str(value)


[docs]def tostring(node, indent=4, nsmap=None):
    """
    Convert a node into an XML string by using the StreamingXMLWriter.
    This is useful for testing purposes.

    :param node: a node object (typically an ElementTree object)
    :param indent: the indentation to use in the XML (default 4 spaces)
    """
    out = io.BytesIO()
    writer = StreamingXMLWriter(out, indent, nsmap=nsmap)
    writer.serialize(node)
    return out.getvalue()


[docs]class StreamingXMLWriter(object):
    """
    A bynary stream XML writer. The typical usage is something like this::

        with StreamingXMLWriter(output_file) as writer:
            writer.start_tag('root')
            for node in nodegenerator():
                writer.serialize(node)
            writer.end_tag('root')
    """
    def __init__(self, bytestream, indent=4, encoding='utf-8', nsmap=None):
        """
        :param bytestream: the stream or file where to write the XML
        :param int indent: the indentation to use in the XML (default 4 spaces)
        """
        # guard against a common error, one must use io.BytesIO
        if isinstance(bytestream, (io.StringIO, io.TextIOWrapper)):
            raise TypeError('%r is not a byte stream' % bytestream)
        self.stream = bytestream
        self.indent = indent
        self.encoding = encoding
        self.indentlevel = 0
        self.nsmap = nsmap

[docs]    def shorten(self, tag):
        """
        Get the short representation of a fully qualified tag

        :param str tag: a (fully qualified or not) XML tag
        """
        if tag.startswith('{'):
            ns, _tag = tag.rsplit('}')
            tag = self.nsmap.get(ns[1:], '') + _tag
        return tag

    def _write(self, text):
        """Write text by respecting the current indentlevel"""
        spaces = ' ' * (self.indent * self.indentlevel)
        t = spaces + text.strip() + '\n'
        if hasattr(t, 'encode'):
            t = t.encode(self.encoding, 'xmlcharrefreplace')
        self.stream.write(t)  # expected bytes

[docs]    def emptyElement(self, name, attrs):
        """Add an empty element (may have attributes)"""
        attr = ' '.join('%s=%s' % (n, quoteattr(scientificformat(v)))
                        for n, v in sorted(attrs.items()))
        self._write('<%s %s/>' % (name, attr))

[docs]    def start_tag(self, name, attrs=None):
        """Open an XML tag"""
        if not attrs:
            self._write('<%s>' % name)
        else:
            self._write('<' + name)
            for (name, value) in sorted(attrs.items()):
                self._write(
                    ' %s=%s' % (name, quoteattr(scientificformat(value))))
            self._write('>')
        self.indentlevel += 1

[docs]    def end_tag(self, name):
        """Close an XML tag"""
        self.indentlevel -= 1
        self._write('</%s>' % name)

[docs]    def serialize(self, node):
        """Serialize a node object (typically an ElementTree object)"""
        if isinstance(node.tag, types.FunctionType):
            # this looks like a bug of ElementTree: comments are stored as
            # functions!?? see https://hg.python.org/sandbox/python2.7/file/tip/Lib/xml/etree/ElementTree.py#l458
            return
        if self.nsmap is not None:
            tag = self.shorten(node.tag)
        else:
            tag = node.tag
        with warnings.catch_warnings():  # unwanted ElementTree warning
            warnings.simplefilter('ignore')
            leafnode = not node
        # NB: we cannot use len(node) to identify leafs since nodes containing
        # an iterator have no length. They are always True, even if empty :-(
        if leafnode and node.text is None:
            self.emptyElement(tag, node.attrib)
            return
        self.start_tag(tag, node.attrib)
        if node.text is not None:
            if striptag(node.tag) == 'posList':
                # NOTE: by convention, posList must be a flat list of
                #   space-separated coordinates, so we need to flatten any
                #   nested lists or tuples, producing a single list of values
                obj = node.text
                while (isinstance(obj, (list, tuple))
                       and isinstance(obj[0], (list, tuple))):
                    obj = list(itertools.chain(*obj))
                txt = escape(scientificformat(obj).strip())
            else:
                txt = escape(scientificformat(node.text).strip())
            if txt:
                self._write(txt)
        for subnode in node:
            self.serialize(subnode)
        self.end_tag(tag)

    def __enter__(self):
        """Write the XML declaration"""
        self._write('<?xml version="1.0" encoding="%s"?>\n' %
                    self.encoding)
        return self

    def __exit__(self, etype, exc, tb):
        """Close the XML document"""
        pass


[docs]class SourceLineParser(ElementTree.XMLParser):
    """
    A custom parser managing line numbers: works for Python <= 3.3
    """
    def _start_list(self, tag, attrib_in):
        elem = super()._start_list(tag, attrib_in)
        elem.lineno = self.parser.CurrentLineNumber
        # there is also CurrentColumnNumber available, if wanted
        return elem


[docs]def fromstring(text):
    """Parse an XML string and return a tree"""
    return ElementTree.fromstring(text, SourceLineParser())


[docs]def parse(source, remove_comments=True, **kw):
    """Thin wrapper around ElementTree.parse"""
    return ElementTree.parse(source, SourceLineParser(), **kw)


[docs]def iterparse(source, events=('end',), remove_comments=True, **kw):
    """Thin wrapper around ElementTree.iterparse"""
    return ElementTree.iterparse(source, events, SourceLineParser(), **kw)


# ###################### utilities for the Node class ####################### #


def _displayattrs(attrib, expandattrs):
    """
    Helper function to display the attributes of a Node object in lexicographic
    order.

    :param attrib: dictionary with the attributes
    :param expandattrs: if True also displays the value of the attributes
    """
    if not attrib:
        return ''
    if expandattrs:
        alist = ['%s=%r' % item for item in sorted(attrib.items())]
    else:
        alist = list(attrib)
    return '{%s}' % ', '.join(alist)


def _display(node, indent, expandattrs, expandvals, output,
             striptags=True, shortentags=False, nsmap=None):
    """Core function to display a Node object"""
    attrs = _displayattrs(node.attrib, expandattrs)
    if node.text is None or not expandvals:
        val = ''
    elif isinstance(node.text, str):
        val = ' %s' % repr(node.text.strip())
    else:
        val = ' %s' % repr(node.text)  # node.text can be a tuple
    tag = node.tag
    if shortentags and nsmap:
        if tag.startswith('{'):
            ns, _tag = tag.rsplit('}')
            tag = '{' + nsmap.get(ns[1:], '') + '}' + _tag
    elif striptags:
        tag = striptag(node.tag)
    output.write(encode(indent + tag + attrs + val + '\n'))
    for sub_node in node:
        _display(sub_node, indent + '  ', expandattrs, expandvals, output,
                 striptags, shortentags, nsmap)


[docs]def node_display(root, expandattrs=False, expandvals=False, output=sys.stdout,
                 striptags=True, shortentags=False, nsmap=None):
    """
    Write an indented representation of the Node object on the output;
    this is intended for testing/debugging purposes.

    :param root: a Node object
    :param bool expandattrs: if True, the values of the attributes are
                             also printed, not only the names
    :param bool expandvals: if True, the values of the tags are also printed,
                            not only the names.
    :param output: stream where to write the string representation of the node
    :param bool striptags: do not display fully qualified tag names
    :param bool shortentags: display a shorter representation of the namespace
                             (overriding the striptags parameter)
    :param dict nsmap: map of namespaces (keys are full names, values are the
                       corresponding aliases)
    """
    _display(root, '', expandattrs, expandvals, output, striptags,
             shortentags, nsmap)


[docs]def striptag(tag):
    """
    Get the short representation of a fully qualified tag

    :param str tag: a (fully qualified or not) XML tag
    """
    if tag.startswith('{'):
        return tag.rsplit('}')[1]
    return tag


[docs]class Node(object):
    """
    A class to make it easy to edit hierarchical structures with attributes,
    such as XML files. Node objects must be pickleable and must consume as
    little memory as possible. Moreover they must be easily converted from
    and to ElementTree objects. The advantage over ElementTree objects
    is that subnodes can be lazily generated and that they can be accessed
    with the dot notation.
    """
    __slots__ = ('tag', 'attrib', 'text', 'nodes', 'lineno')

    def __init__(self, fulltag, attrib=None, text=None,
                 nodes=None, lineno=None):
        """
        :param str tag: the Node name
        :param dict attrib: the Node attributes
        :param str text: the Node text (default None)
        :param nodes: an iterable of subnodes (default empty list)
        :param lineno: line number where the tag was read in the source xml
        """
        self.tag = fulltag
        self.attrib = {} if attrib is None else attrib
        self.text = text
        self.nodes = [] if nodes is None else nodes
        self.lineno = lineno
        if self.nodes and self.text is not None:
            raise ValueError(
                'A branch node cannot have a value, got %r' % self.text)

    def __getattr__(self, name):
        if name.startswith('_'):
            # do the magic only for public names
            raise AttributeError(name)
        for node in self.nodes:
            if striptag(node.tag) == name:
                return node
        raise AttributeError("No subnode named '%s' found in '%s'" %
                             (name, striptag(self.tag)))

[docs]    def getnodes(self, name):
        "Return the direct subnodes with name 'name'"
        for node in self.nodes:
            if striptag(node.tag) == name:
                yield node

[docs]    def append(self, node):
        "Append a new subnode"
        if not isinstance(node, self.__class__):
            raise TypeError('Expected Node instance, got %r' % node)
        self.nodes.append(node)

[docs]    def get_nsmap(self):
        return {v: k for k, v in self.attrib.items() if k.startswith('xmlns')}

[docs]    def to_str(self, expandattrs=True, expandvals=True, striptags=True,
               shortentags=False):
        """
        Convert the node into a string, intended for testing/debugging purposes

        :param expandattrs:
          print the values of the attributes if True, else print only the names
        :param expandvals:
          print the values if True, else print only the tag names
        :param bool striptags: do not display fully qualified tag names
        :param bool shortentags: display a shorter representation of the
                                 namespace (overriding the striptags parameter)
        """
        out = io.BytesIO()
        node_display(self, expandattrs, expandvals, out, striptags,
                     shortentags, self.get_nsmap())
        return decode(out.getvalue())

    def __iter__(self):
        """Iterate over subnodes"""
        return iter(self.nodes)

    def __repr__(self):
        """A condensed representation for debugging purposes"""
        return '<%s %s %s %s>' % (striptag(self.tag), self.attrib, self.text,
                                  '' if not self.nodes else '...')

    def __getitem__(self, i):
        """
        Retrieve a subnode, if i is an integer, or an attribute, if i
        is a string.
        """
        if isinstance(i, str):
            return self.attrib[i]
        else:  # assume an integer or a slice
            return self.nodes[i]

[docs]    def get(self, attr, value=None):
        """
        Get the given `attr`; if missing, returns `value` or `None`.
        """
        return self.attrib.get(attr, value)

    def __setitem__(self, i, value):
        """
        Update a subnode, if i is an integer, or an attribute, if i
        is a string.
        """
        if isinstance(i, str):
            self.attrib[i] = value
        else:  # assume an integer or a slice
            self.nodes[i] = value

    def __delitem__(self, i):
        """
        Remove a subnode, if i is an integer, or an attribute, if i
        is a string.
        """
        if isinstance(i, str):
            del self.attrib[i]
        else:  # assume an integer or a slice
            del self.nodes[i]

    def __invert__(self):
        """
        Return the value of a leaf; raise a TypeError if the node is not a leaf
        """
        if self:
            raise TypeError('%s is a composite node, not a leaf' % self)
        return self.text

    def __len__(self):
        """Return the number of subnodes"""
        return len(self.nodes)

    def __bool__(self):
        """
        Return True if there are subnodes; it does not iter on the
        subnodes, so for lazy nodes it returns True even if the
        generator is empty.
        """
        return bool(self.nodes)

    def __deepcopy__(self, memo):
        new = object.__new__(self.__class__)
        new.tag = self.tag
        new.attrib = self.attrib.copy()
        new.text = copy.copy(self.text)
        new.nodes = [copy.deepcopy(n, memo) for n in self.nodes]
        new.lineno = self.lineno
        return new

    def __getstate__(self):
        return dict((slot, getattr(self, slot))
                    for slot in self.__class__.__slots__)

    def __setstate__(self, state):
        for slot in self.__class__.__slots__:
            setattr(self, slot, state[slot])

    def __eq__(self, other):
        assert other is not None
        return all(getattr(self, slot) == getattr(other, slot)
                   for slot in self.__class__.__slots__)

    def __ne__(self, other):
        return not self.__eq__(other)


[docs]def to_literal(self):
    """
    Convert the node into a literal Python object
    """
    if not self.nodes:
        return (self.tag, self.attrib, self.text, [])
    else:
        return (self.tag, self.attrib, self.text,
                list(map(to_literal, self.nodes)))


[docs]def pprint(self, stream=None, indent=1, width=80, depth=None):
    """
    Pretty print the underlying literal Python object
    """
    pp.pprint(to_literal(self), stream, indent, width, depth)


[docs]def node_from_dict(dic, nodefactory=Node):
    """
    Convert a (nested) dictionary into a Node object.
    """
    [(tag, dic)] = dic.items()
    if isinstance(dic, dict):
        dic = dic.copy()
        text = dic.pop('text', None)
        attrib = {n[1:]: dic.pop(n) for n in sorted(dic) if n.startswith('_')}
    else:
        return nodefactory(tag, {}, dic)
    if not dic:
        return nodefactory(tag, attrib, text)
    [(k, vs)] = dic.items()
    if isinstance(vs, list):
        nodes = [node_from_dict({k: v}) for v in vs]
    else:
        nodes = [node_from_dict(dic)]
    return nodefactory(tag, attrib, nodes=nodes)


def _group(one_key_dicts):
    items = []
    for one_key_dict in one_key_dicts:
        [(k, v)] = one_key_dict.items()
        items.append((k, v))
    dic = {}
    for k, group in itertools.groupby(items, lambda item: item[0]):
        vs = []
        for k, v in group:
            vs.append(v)
        if len(vs) == 1:
            dic[k] = vs[0]
        else:
            dic[k] = vs
    return dic


[docs]def node_to_dict(node):
    """
    Convert a Node object into a (nested) dictionary
    with attributes tag, attrib, text, nodes.

    :param node: a Node-compatible object
    """
    tag = striptag(node.tag)
    dic = {}
    if node.attrib:
        for nam, val in node.attrib.items():
            dic['_' + nam] = (float(val)
                              if isinstance(val, numpy.float64) else val)
    if isinstance(node.text, str) and node.text.strip() == '':
        pass
    elif node.text is not None:
        if node.attrib:
            dic['text'] = node.text
        else:
            # TODO: ugly, dic sometimes is a dic and sometimes a scalar??
            dic = node.text
    if node.nodes:
        dic.update(_group([node_to_dict(n) for n in node]))
    return {tag: dic}


[docs]def node_from_elem(elem, nodefactory=Node, lazy=()):
    """
    Convert (recursively) an ElementTree object into a Node object.
    """
    children = list(elem)
    lineno = getattr(elem, 'lineno', None)
    if not children:
        return nodefactory(elem.tag, dict(elem.attrib), elem.text,
                           lineno=lineno)
    if striptag(elem.tag) in lazy:
        nodes = (node_from_elem(ch, nodefactory, lazy) for ch in children)
    else:
        nodes = [node_from_elem(ch, nodefactory, lazy) for ch in children]
    return nodefactory(elem.tag, dict(elem.attrib), nodes=nodes, lineno=lineno)


# taken from https://gist.github.com/651801, which comes for the effbot
[docs]def node_to_elem(root):
    """
    Convert (recursively) a Node object into an ElementTree object.
    """
    def generate_elem(append, node, level):
        var = "e" + str(level)
        arg = repr(node.tag)
        if node.attrib:
            arg += ", **%r" % node.attrib
        if level == 1:
            append("e1 = Element(%s)" % arg)
        else:
            append("%s = SubElement(e%d, %s)" % (var, level - 1, arg))
        if not node.nodes:
            append("%s.text = %r" % (var, node.text))
        for x in node:
            generate_elem(append, x, level + 1)
    # generate code to create a tree
    output = []
    generate_elem(output.append, root, 1)  # print "\n".join(output)
    namespace = {"Element": ElementTree.Element,
                 "SubElement": ElementTree.SubElement}
    exec("\n".join(output), globals(), namespace)
    return namespace["e1"]


[docs]def read_nodes(fname, filter_elem, nodefactory=Node, remove_comments=True):
    """
    Convert an XML file into a lazy iterator over Node objects
    satifying the given specification, i.e. a function element -> boolean.

    :param fname: file name of file object
    :param filter_elem: element specification

    In case of errors, add the file name to the error message.
    """
    try:
        for _, el in iterparse(fname, remove_comments=remove_comments):
            if filter_elem(el):
                yield node_from_elem(el, nodefactory)
                el.clear()  # save memory
    except Exception:
        etype, exc, tb = sys.exc_info()
        msg = str(exc)
        if str(fname) not in msg:
            msg = '%s in %s' % (msg, fname)
        raise_(etype, msg, tb)


[docs]def node_from_xml(xmlfile, nodefactory=Node):
    """
    Convert a .xml file into a Node object.

    :param xmlfile: a file name or file object open for reading
    """
    root = parse(xmlfile).getroot()
    return node_from_elem(root, nodefactory)


[docs]def node_to_xml(node, output, nsmap=None):
    """
    Convert a Node object into a pretty .xml file without keeping
    everything in memory. If you just want the string representation
    use tostring(node).

    :param node: a Node-compatible object (ElementTree nodes are fine)
    :param output: a binary output file
    :param nsmap: if given, shorten the tags with aliases

    """
    if nsmap:
        for ns, prefix in nsmap.items():
            if prefix:
                node['xmlns:' + prefix[:-1]] = ns
            else:
                node['xmlns'] = ns
    with StreamingXMLWriter(output, nsmap=nsmap) as w:
        w.serialize(node)


[docs]def node_from_ini(ini_file, nodefactory=Node, root_name='ini'):
    """
    Convert a .ini file into a Node object.

    :param ini_file: a filename or a file like object in read mode
    """
    fileobj = open(ini_file) if isinstance(ini_file, str) else ini_file
    cfp = configparser.ConfigParser(interpolation=None)
    cfp.read_file(fileobj)
    root = nodefactory(root_name)
    sections = cfp.sections()
    for section in sections:
        params = dict(cfp.items(section))
        root.append(Node(section, params))
    return root


[docs]def node_to_ini(node, output=sys.stdout):
    """
    Convert a Node object with the right structure into a .ini file.

    :params node: a Node object
    :params output: a file-like object opened in write mode
    """
    for subnode in node:
        output.write(u'\n[%s]\n' % subnode.tag)
        for name, value in sorted(subnode.attrib.items()):
            output.write(u'%s=%s\n' % (name, value))
    output.flush()


[docs]def node_copy(node, nodefactory=Node):
    """Make a deep copy of the node"""
    return nodefactory(node.tag, node.attrib.copy(), node.text,
                       [node_copy(n, nodefactory) for n in node])


[docs]@contextmanager
def context(fname, node):
    """
    Context manager managing exceptions and adding line number of the
    current node and name of the current file to the error message.

    :param fname: the current file being processed
    :param node: the current node being processed
    """
    try:
        yield node
    except Exception:
        etype, exc, tb = sys.exc_info()
        msg = 'node %s: %s, line %s of %s' % (
            striptag(node.tag), exc, getattr(node, 'lineno', '?'), fname)
        raise_(etype, msg, tb)


[docs]class ValidatingXmlParser(object):
    """
    Validating XML Parser based on Expat. It has two methods `.parse_file`
    and `.parse_bytes` returning a validated :class:`Node` object.

    :param validators: a dictionary of validation functions
    :param stop: the tag where to stop the parsing (if any)
    """
[docs]    class Exit(Exception):
        """Raised when the parsing is stopped before the end on purpose"""

    def __init__(self, validators, stop=None):
        self.validators = validators
        self.stop = stop

    @contextmanager
    def _context(self):
        self.p = ParserCreate(namespace_separator='}')
        self.p.StartElementHandler = self._start_element
        self.p.EndElementHandler = self._end_element
        self.p.CharacterDataHandler = self._char_data
        self._ancestors = []
        self._root = None
        try:
            yield
        except ExpatError as err:
            msg = '%s: %s: %s' % (self.filename, err.lineno,
                                  ErrorString(err.code))
            e = ExpatError(msg)
            e.lineno = err.lineno
            e.offset = err.offset
            e.filename = self.filename
            raise e
        except ValueError as err:
            err.lineno = self.p.CurrentLineNumber
            err.offset = self.p.CurrentColumnNumber
            err.filename = self.filename
            raise err
        except self.Exit:
            pass

[docs]    def parse_bytes(self, bytestr, isfinal=True):
        """
        Parse a byte string. If the string is very large, split it in chuncks
        and parse each chunk with isfinal=False, then parse an empty chunk
        with isfinal=True.
        """
        with self._context():
            self.filename = None
            self.p.Parse(bytestr, isfinal)
        return self._root

[docs]    def parse_file(self, file_or_fname):
        """
        Parse a file or a filename
        """
        with self._context():
            if hasattr(file_or_fname, 'read'):
                self.filename = getattr(
                    file_or_fname, 'name', file_or_fname.__class__.__name__)
                self.p.ParseFile(file_or_fname)
            else:
                self.filename = file_or_fname
                with open(file_or_fname, 'rb') as f:
                    self.p.ParseFile(f)
        return self._root

    def _start_element(self, longname, attrs):
        try:
            xmlns, name = longname.split('}')
        except ValueError:  # no namespace in the longname
            name = tag = longname
        else:  # fix the tag with an opening brace
            tag = '{' + longname
        self._ancestors.append(
            Node(tag, attrs, lineno=self.p.CurrentLineNumber))
        if self.stop and name == self.stop:
            for anc in reversed(self._ancestors):
                self._end_element(anc.tag)
            raise self.Exit

    def _end_element(self, name):
        node = self._ancestors[-1]
        if isinstance(node.text, list):
            node.text = ''.join(node.text)
        with context(self.filename, node):
            self._root = self._literalnode(node)
        del self._ancestors[-1]
        if self._ancestors:
            self._ancestors[-1].append(self._root)

    def _char_data(self, data):
        if data:
            parent = self._ancestors[-1]
            if parent.text is None:
                parent.text = [data]
            else:
                parent.text.append(data)

    def _set_text(self, node, text, tag):
        if text is None:
            return
        try:
            val = self.validators[tag]
        except KeyError:
            return
        try:
            node.text = val(decode(text.strip()))
        except Exception as exc:
            raise ValueError('Could not convert %s->%s: %s' %
                             (tag, val.__name__, exc))

    def _set_attrib(self, node, n, tn, v):
        val = self.validators[tn]
        try:
            node.attrib[n] = val(decode(v))
        except Exception as exc:
            # NOTE: the line number and the file name are added by the
            # 'context' contextmanager
            raise ValueError(
                'Could not convert %s->%s: %s' %
                (tn, val.__name__, exc))

    def _literalnode(self, node):
        tag = striptag(node.tag)

        # cast the text
        self._set_text(node, node.text, tag)

        # cast the attributes
        for n, v in node.attrib.items():
            tn = '%s.%s' % (tag, n)
            if tn in self.validators:
                self._set_attrib(node, n, tn, v)
            elif n in self.validators:
                self._set_attrib(node, n, n, v)
        return node