Source code for pyrocko.guts

# https://pyrocko.org - GPLv3
#
# The Pyrocko Developers, 21st Century
# ---|P------/S----------~Lg----------

'''
Lightweight declarative YAML and XML data binding for Python.
'''

import datetime
import calendar
import re
import sys
import os
import types
import copy
import logging
import os.path as op
from collections import defaultdict
from base64 import b64decode, b64encode

from io import BytesIO

try:
    import numpy as num
except ImportError:
    num = None

import yaml
try:
    from yaml import CSafeLoader as SafeLoader, CSafeDumper as SafeDumper
except ImportError:
    from yaml import SafeLoader, SafeDumper

from .util import time_to_str, str_to_time, TimeStrError, hpfloat, \
    get_time_float


logger = logging.getLogger('pyrocko.guts')

ALLOW_INCLUDE = False


class GutsSafeDumper(SafeDumper):
    pass


class GutsSafeLoader(SafeLoader):
    pass


g_iprop = 0

g_deferred = {}
g_deferred_content = {}

g_tagname_to_class = {}
g_xmltagname_to_class = {}
g_guessable_xmlns = {}

guts_types = [
    'Object', 'SObject', 'String', 'Unicode', 'Int', 'Float',
    'Complex', 'Bool', 'Timestamp', 'DateTimestamp', 'StringPattern',
    'UnicodePattern', 'StringChoice', 'IntChoice', 'List', 'Dict', 'Tuple',
    'StringUnion', 'Choice', 'Any']

us_to_cc_regex = re.compile(r'([a-z])_([a-z])')


class literal(str):
    pass


class folded(str):
    pass


class singlequoted(str):
    pass


class doublequoted(str):
    pass


def make_str_presenter(style):
    def presenter(dumper, data):
        return dumper.represent_scalar(
            'tag:yaml.org,2002:str', str(data), style=style)

    return presenter


str_style_map = {
    None: lambda x: x,
    '|': literal,
    '>': folded,
    "'": singlequoted,
    '"': doublequoted}

for (style, cls) in str_style_map.items():
    if style:
        GutsSafeDumper.add_representer(cls, make_str_presenter(style))


class uliteral(str):
    pass


class ufolded(str):
    pass


class usinglequoted(str):
    pass


class udoublequoted(str):
    pass


def make_unicode_presenter(style):
    def presenter(dumper, data):
        return dumper.represent_scalar(
            'tag:yaml.org,2002:str', str(data), style=style)

    return presenter


unicode_style_map = {
    None: lambda x: x,
    '|': literal,
    '>': folded,
    "'": singlequoted,
    '"': doublequoted}

for (style, cls) in unicode_style_map.items():
    if style:
        GutsSafeDumper.add_representer(cls, make_unicode_presenter(style))


class blist(list):
    pass


class flist(list):
    pass


list_style_map = {
    None: list,
    'block': blist,
    'flow': flist}


def make_list_presenter(flow_style):
    def presenter(dumper, data):
        return dumper.represent_sequence(
            'tag:yaml.org,2002:seq', data, flow_style=flow_style)

    return presenter


GutsSafeDumper.add_representer(blist, make_list_presenter(False))
GutsSafeDumper.add_representer(flist, make_list_presenter(True))

if num:
    def numpy_float_presenter(dumper, data):
        return dumper.represent_float(float(data))

    def numpy_int_presenter(dumper, data):
        return dumper.represent_int(int(data))

    for dtype in (num.float64, num.float32):
        GutsSafeDumper.add_representer(dtype, numpy_float_presenter)

    for dtype in (num.int32, num.int64):
        GutsSafeDumper.add_representer(dtype, numpy_int_presenter)


def us_to_cc(s):
    return us_to_cc_regex.sub(lambda pat: pat.group(1)+pat.group(2).upper(), s)


cc_to_us_regex1 = re.compile(r'([a-z])([A-Z]+)([a-z]|$)')
cc_to_us_regex2 = re.compile(r'([A-Z])([A-Z][a-z])')


def cc_to_us(s):
    return cc_to_us_regex2.sub('\\1_\\2', cc_to_us_regex1.sub(
        '\\1_\\2\\3', s)).lower()


re_frac = re.compile(r'\.[1-9]FRAC')
frac_formats = dict([('.%sFRAC' % x, '%.'+x+'f') for x in '123456789'])


def encode_utf8(s):
    return s.encode('utf-8')


def no_encode(s):
    return s


def make_xmltagname_from_name(name):
    return us_to_cc(name)


def make_name_from_xmltagname(xmltagname):
    return cc_to_us(xmltagname)


def make_content_name(name):
    if name.endswith('_list'):
        return name[:-5]
    elif name.endswith('s'):
        return name[:-1]
    else:
        return name


def classnames(cls):
    if isinstance(cls, tuple):
        return '(%s)' % ', '.join(x.__name__ for x in cls)
    else:
        return cls.__name__


def expand_stream_args(mode):
    def wrap(f):
        '''
        Decorator to enhance functions taking stream objects.

        Wraps a function f(..., stream, ...) so that it can also be called as
        f(..., filename='myfilename', ...) or as f(..., string='mydata', ...).
        '''

        def g(*args, **kwargs):
            stream = kwargs.pop('stream', None)
            filename = kwargs.get('filename', None)
            if mode != 'r':
                filename = kwargs.pop('filename', None)
            string = kwargs.pop('string', None)

            assert sum(x is not None for x in (stream, filename, string)) <= 1

            if stream is not None:
                kwargs['stream'] = stream
                return f(*args, **kwargs)

            elif filename is not None:
                stream = open(filename, mode+'b')
                kwargs['stream'] = stream
                retval = f(*args, **kwargs)
                if isinstance(retval, types.GeneratorType):
                    def wrap_generator(gen):
                        try:
                            for x in gen:
                                yield x

                        except GeneratorExit:
                            pass

                        stream.close()

                    return wrap_generator(retval)

                else:
                    stream.close()
                    return retval

            elif string is not None:
                assert mode == 'r', \
                    'Keyword argument string=... cannot be used in dumper ' \
                    'function.'

                kwargs['stream'] = BytesIO(string.encode('utf-8'))
                return f(*args, **kwargs)

            else:
                assert mode == 'w', \
                    'Use keyword argument stream=... or filename=... in ' \
                    'loader function.'

                sout = BytesIO()
                f(stream=sout, *args, **kwargs)
                return sout.getvalue().decode('utf-8')

        g.__doc__ = f.__doc__
        return g

    return wrap


class Defer(object):
    def __init__(self, classname, *args, **kwargs):
        global g_iprop
        if kwargs.get('position', None) is None:
            kwargs['position'] = g_iprop

        g_iprop += 1

        self.classname = classname
        self.args = args
        self.kwargs = kwargs


[docs]class TBase(object): ''' Base class for Guts type definitions. :param default: Default value or :py:class:`DefaultMaker` object (see :py:meth:`Object.D`) to be used to generate a default. :param optional: If ``True``, the attribute is optional and may be set to ``None``. :type optional: bool :param xmlstyle: Controls how the attribute is serialized in XML. :Choices: ``'element'``, ``'attribute'``, ``'content'``. Only scalar and :py:class:`SObject` values may be serialized as ``'attribute'``. Only one attribute in a class may be serialized as ``'content'``. :type xmlstyle: str :param xmltagname: XML tag name to be used. By default, the attribute name converted to camel-case is used. :type xmltagname: str :param xmlns: XML namespace to be used for this attribute. :type xmlns: str :param help: Description of the attribute used in documentation and online help. :type help: rst formatted :py:class:`str` :param position: Position index to be used when the attribute should be output in non-default order. :type position: int ''' strict = False multivalued = None force_regularize = False propnames = [] _dummy_cls = None _cls = None _sphinx_doc_skip = False @classmethod def init_propertystuff(cls): cls.properties = [] cls.xmltagname_to_name = {} cls.xmltagname_to_name_multivalued = {} cls.xmltagname_to_class = {} cls.content_property = None def __init__( self, default=None, optional=False, xmlstyle='element', xmltagname=None, xmlns=None, help=None, position=None): global g_iprop if position is not None: self.position = position else: self.position = g_iprop g_iprop += 1 self._default = default self.optional = optional self.name = None self._xmltagname = xmltagname self._xmlns = xmlns self.parent = None self.xmlstyle = xmlstyle self.help = help self._sphinx_doc_skip = True def default(self): return make_default(self._default) def is_default(self, val): if self._default is None: return val is None else: return self._default == val def has_default(self): return self._default is not None def xname(self): if self.name is not None: return self.name elif self.parent is not None: return 'element of %s' % self.parent.xname() else: return '?' def set_xmlns(self, xmlns): if self._xmlns is None and not self.xmlns: self._xmlns = xmlns if self.multivalued: self.content_t.set_xmlns(xmlns) def get_xmlns(self): return self._xmlns or self.xmlns def get_xmltagname(self): if self._xmltagname is not None: return self.get_xmlns() + ' ' + self._xmltagname elif self.name: return self.get_xmlns() + ' ' \ + make_xmltagname_from_name(self.name) elif self.xmltagname: return self.get_xmlns() + ' ' + self.xmltagname else: assert False @classmethod def get_property(cls, name): for prop in cls.properties: if prop.name == name: return prop raise ValueError() @classmethod def remove_property(cls, name): prop = cls.get_property(name) if not prop.multivalued: del cls.xmltagname_to_class[prop.effective_xmltagname] del cls.xmltagname_to_name[prop.effective_xmltagname] else: del cls.xmltagname_to_class[prop.content_t.effective_xmltagname] del cls.xmltagname_to_name_multivalued[ prop.content_t.effective_xmltagname] if cls.content_property is prop: cls.content_property = None cls.properties.remove(prop) cls.propnames.remove(name) return prop @classmethod def add_property(cls, name, prop): prop.instance = prop prop.name = name prop.set_xmlns(cls.xmlns) if isinstance(prop, Choice.T): for tc in prop.choices: tc.effective_xmltagname = tc.get_xmltagname() cls.xmltagname_to_class[tc.effective_xmltagname] = tc._cls cls.xmltagname_to_name[tc.effective_xmltagname] = prop.name elif not prop.multivalued: prop.effective_xmltagname = prop.get_xmltagname() cls.xmltagname_to_class[prop.effective_xmltagname] = prop._cls cls.xmltagname_to_name[prop.effective_xmltagname] = prop.name else: prop.content_t.name = make_content_name(prop.name) prop.content_t.effective_xmltagname = \ prop.content_t.get_xmltagname() cls.xmltagname_to_class[ prop.content_t.effective_xmltagname] = prop.content_t._cls cls.xmltagname_to_name_multivalued[ prop.content_t.effective_xmltagname] = prop.name cls.properties.append(prop) cls.properties.sort(key=lambda x: x.position) cls.propnames = [p.name for p in cls.properties] if prop.xmlstyle == 'content': cls.content_property = prop @classmethod def ivals(cls, val): for prop in cls.properties: yield getattr(val, prop.name) @classmethod def ipropvals(cls, val): for prop in cls.properties: yield prop, getattr(val, prop.name) @classmethod def inamevals(cls, val): for prop in cls.properties: yield prop.name, getattr(val, prop.name) @classmethod def ipropvals_to_save(cls, val, xmlmode=False): for prop in cls.properties: v = getattr(val, prop.name) if v is not None and ( not (prop.optional or (prop.multivalued and not v)) or (not prop.is_default(v))): if xmlmode: yield prop, prop.to_save_xml(v) else: yield prop, prop.to_save(v) @classmethod def inamevals_to_save(cls, val, xmlmode=False): for prop, v in cls.ipropvals_to_save(val, xmlmode): yield prop.name, v @classmethod def translate_from_xml(cls, list_of_pairs, strict): d = {} for k, v in list_of_pairs: if k in cls.xmltagname_to_name_multivalued: k2 = cls.xmltagname_to_name_multivalued[k] if k2 not in d: d[k2] = [] d[k2].append(v) elif k in cls.xmltagname_to_name: k2 = cls.xmltagname_to_name[k] if k2 in d: raise ArgumentError( 'Unexpectedly found more than one child element "%s" ' 'within "%s".' % (k, cls.tagname)) d[k2] = v elif k is None: if cls.content_property: k2 = cls.content_property.name d[k2] = v else: if strict: raise ArgumentError( 'Unexpected child element "%s" found within "%s".' % ( k, cls.tagname)) return d def validate(self, val, regularize=False, depth=-1): if self.optional and val is None: return val is_derived = isinstance(val, self._cls) is_exact = type(val) == self._cls not_ok = not self.strict and not is_derived or \ self.strict and not is_exact if not_ok or self.force_regularize: if regularize: try: val = self.regularize_extra(val) except ValueError: raise ValidationError( '%s: could not convert "%s" to type %s' % ( self.xname(), val, classnames(self._cls))) else: raise ValidationError( '%s: "%s" (type: %s) is not of type %s' % ( self.xname(), val, type(val), classnames(self._cls))) validator = self if isinstance(self._cls, tuple): clss = self._cls else: clss = (self._cls,) for cls in clss: try: if type(val) != cls and isinstance(val, cls): validator = val.T.instance except AttributeError: pass validator.validate_extra(val) if depth != 0: val = validator.validate_children(val, regularize, depth) return val def regularize_extra(self, val): return self._cls(val) def validate_extra(self, val): pass def validate_children(self, val, regularize, depth): for prop, propval in self.ipropvals(val): newpropval = prop.validate(propval, regularize, depth-1) if regularize and (newpropval is not propval): setattr(val, prop.name, newpropval) return val def to_save(self, val): return val def to_save_xml(self, val): return self.to_save(val) def extend_xmlelements(self, elems, v): if self.multivalued: for x in v: elems.append((self.content_t.effective_xmltagname, x)) else: elems.append((self.effective_xmltagname, v)) def deferred(self): return [] def classname_for_help(self, strip_module=''): if self._dummy_cls is not self._cls: if self._dummy_cls.__module__ == strip_module: sadd = ' (:py:class:`%s`)' % ( self._dummy_cls.__name__) else: sadd = ' (:py:class:`%s.%s`)' % ( self._dummy_cls.__module__, self._dummy_cls.__name__) else: sadd = '' if self._dummy_cls in guts_plain_dummy_types: return '``%s``' % self._cls.__name__ elif self._dummy_cls.dummy_for_description: return '%s%s' % (self._dummy_cls.dummy_for_description, sadd) else: def sclass(cls): mod = cls.__module__ clsn = cls.__name__ if mod == '__builtin__' or mod == 'builtins': return '``%s``' % clsn elif mod == strip_module: return ':py:class:`%s`' % clsn else: return ':py:class:`%s.%s`' % (mod, clsn) if isinstance(self._cls, tuple): return '(%s)%s' % ( ' | '.join(sclass(cls) for cls in self._cls), sadd) else: return '%s%s' % (sclass(self._cls), sadd) @classmethod def props_help_string(cls): baseprops = [] for base in cls._dummy_cls.__bases__: if hasattr(base, 'T'): baseprops.extend(base.T.properties) hlp = [] hlp.append('') for prop in cls.properties: if prop in baseprops: continue descr = [ prop.classname_for_help( strip_module=cls._dummy_cls.__module__)] if prop.optional: descr.append('*optional*') if isinstance(prop._default, DefaultMaker): descr.append('*default:* ``%s``' % repr(prop._default)) else: d = prop.default() if d is not None: descr.append('*default:* ``%s``' % repr(d)) hlp.append(' .. py:gattribute:: %s' % prop.name) hlp.append('') hlp.append(' %s' % ', '.join(descr)) hlp.append(' ') if prop.help is not None: hlp.append(' %s' % prop.help) hlp.append('') return '\n'.join(hlp) @classmethod def class_help_string(cls): return cls._dummy_cls.__doc_template__ @classmethod def class_signature(cls): r = [] for prop in cls.properties: d = prop.default() if d is not None: arg = repr(d) elif prop.optional: arg = 'None' else: arg = '...' r.append('%s=%s' % (prop.name, arg)) return '(%s)' % ', '.join(r) @classmethod def help(cls): return cls.props_help_string()
class ObjectMetaClass(type): def __new__(meta, classname, bases, class_dict): classname = class_dict.get('class_name', classname) cls = type.__new__(meta, classname, bases, class_dict) if classname != 'Object': t_class_attr_name = '_%s__T' % classname if not hasattr(cls, t_class_attr_name): if hasattr(cls, 'T'): class T(cls.T): _sphinx_doc_skip = True T.__doc__ = cls.T.__doc__ else: class T(TBase): _sphinx_doc_skip = True T.__doc__ = TBase.__doc__ setattr(cls, t_class_attr_name, T) T = getattr(cls, t_class_attr_name) T.__name__ = 'T' T.__qualname__ = T.__qualname__.replace('__T', 'T') if cls.dummy_for is not None: T._cls = cls.dummy_for else: T._cls = cls T._dummy_cls = cls if hasattr(cls, 'xmltagname'): T.xmltagname = cls.xmltagname else: T.xmltagname = classname mod = sys.modules[cls.__module__] if hasattr(cls, 'xmlns'): T.xmlns = cls.xmlns elif hasattr(mod, 'guts_xmlns'): T.xmlns = mod.guts_xmlns else: T.xmlns = '' if T.xmlns and hasattr(cls, 'guessable_xmlns'): g_guessable_xmlns[T.xmltagname] = cls.guessable_xmlns if hasattr(mod, 'guts_prefix'): if mod.guts_prefix: T.tagname = mod.guts_prefix + '.' + classname else: T.tagname = classname else: if cls.__module__ != '__main__': T.tagname = cls.__module__ + '.' + classname else: T.tagname = classname T.classname = classname T.init_propertystuff() for k in dir(cls): prop = getattr(cls, k) if k.endswith('__'): k = k[:-2] if isinstance(prop, TBase): if prop.deferred(): for defer in prop.deferred(): g_deferred_content.setdefault( defer.classname[:-2], []).append((prop, defer)) g_deferred.setdefault( defer.classname[:-2], []).append((T, k, prop)) else: T.add_property(k, prop) elif isinstance(prop, Defer): g_deferred.setdefault(prop.classname[:-2], []).append( (T, k, prop)) if classname in g_deferred_content: for prop, defer in g_deferred_content[classname]: prop.process_deferred( defer, T(*defer.args, **defer.kwargs)) del g_deferred_content[classname] if classname in g_deferred: for (T_, k_, prop_) in g_deferred.get(classname, []): if isinstance(prop_, Defer): prop_ = T(*prop_.args, **prop_.kwargs) if not prop_.deferred(): T_.add_property(k_, prop_) del g_deferred[classname] g_tagname_to_class[T.tagname] = cls if hasattr(cls, 'xmltagname'): g_xmltagname_to_class[T.xmlns + ' ' + T.xmltagname] = cls cls.T = T T.instance = T() cls.__doc_template__ = cls.__doc__ cls.__doc__ = T.class_help_string() if cls.__doc__ is None: cls.__doc__ = 'Undocumented.' cls.__doc__ += '\n' + T.props_help_string() return cls
[docs]class ValidationError(ValueError): ''' Raised when an object is invalid according to its definition. ''' pass
[docs]class ArgumentError(ValueError): ''' Raised when invalid arguments would be used in an object's initialization. ''' pass
def make_default(x): if isinstance(x, DefaultMaker): return x.make() elif isinstance(x, Object): return clone(x) else: return x
[docs]class DefaultMaker(object): ''' Base class for default value factories. '''
[docs] def make(self): ''' Create a new object. ''' raise NotImplementedError
[docs]class ObjectDefaultMaker(DefaultMaker): ''' Default value factory for :py:class:`Object` derived classes. ''' def __init__(self, cls, args, kwargs): DefaultMaker.__init__(self) self._cls = cls self.args = args self.kwargs = kwargs self.instance = None def make(self): return self._cls( *[make_default(x) for x in self.args], **dict((k, make_default(v)) for (k, v) in self.kwargs.items())) def __eq__(self, other): if self.instance is None: self.instance = self.make() return self.instance == other def __repr__(self): sargs = [] for arg in self.args: sargs.append(repr(arg)) for k, v in self.kwargs.items(): sargs.append( '%s=%s' % ( k, ' '.join(line.strip() for line in repr(v).splitlines()))) return '%s(%s)' % (self._cls.__name__, ', '.join(sargs))
class TimestampDefaultMaker(DefaultMaker): def __init__(self, s, format='%Y-%m-%d %H:%M:%S.OPTFRAC'): DefaultMaker.__init__(self) self._stime = s self._format = format def make(self): return str_to_time(self._stime, self._format) def __repr__(self): return 'str_to_time(%s)' % repr(self._stime) def with_metaclass(meta, *bases): # inlined py2/py3 compat solution from python-future class metaclass(meta): __call__ = type.__call__ __init__ = type.__init__ def __new__(cls, name, this_bases, d): if this_bases is None: return type.__new__(cls, name, (), d) return meta(name, bases, d) return metaclass('temp', None, {})
[docs]class Object(with_metaclass(ObjectMetaClass, object)): ''' Base class for Guts objects. :cvar dummy_for: (class variable) If set, this indicates that the containing class is a dummy for another type. This can be used to hold native Python objects and non-Guts based objects as children of a Guts object. :cvar dummy_for_description: (class variable) Overrides the name shown in the "dummy for ..." documentation strings. ''' dummy_for = None dummy_for_description = None def __init__(self, **kwargs): if not kwargs.get('init_props', True): return for prop in self.T.properties: k = prop.name if k in kwargs: setattr(self, k, kwargs.pop(k)) else: if not prop.optional and not prop.has_default(): raise ArgumentError('Missing argument to %s: %s' % ( self.T.tagname, prop.name)) else: setattr(self, k, prop.default()) if kwargs: raise ArgumentError('Invalid argument to %s: %s' % ( self.T.tagname, ', '.join(list(kwargs.keys()))))
[docs] @classmethod def D(cls, *args, **kwargs): ''' Get a default value factory for this class, configured with specified arguments. :returns: Factory for default values. :rtype: :py:class:`ObjectDefaultMaker` object ''' return ObjectDefaultMaker(cls, args, kwargs)
[docs] def validate(self, regularize=False, depth=-1): ''' Validate this object. Raises :py:class:`ValidationError` when the object is invalid. :param depth: Maximum depth to descend into child objects. :type depth: int ''' self.T.instance.validate(self, regularize, depth)
[docs] def regularize(self, depth=-1): ''' Regularize this object. Regularization tries to convert child objects of invalid types to the expected types. Raises :py:class:`ValidationError` when the object is invalid and cannot be regularized. :param depth: Maximum depth to descend into child objects. :type depth: int ''' self.validate(regularize=True, depth=depth)
[docs] def dump(self, stream=None, filename=None, header=False): ''' Serialize to YAML. If neither ``stream`` nor ``filename`` is set, a string containing the serialized data is returned. :param stream: Output to stream. :param filename: Output to file of given name. :type filename: str :param header: File header to prepend to the output. :type header: str ''' return dump(self, stream=stream, filename=filename, header=header)
[docs] def dump_xml( self, stream=None, filename=None, header=False, ns_ignore=False): ''' Serialize to XML. If neither ``stream`` nor ``filename`` is set, a string containing the serialized data is returned. :param stream: Output to stream. :param filename: Output to file of given name. :type filename: str :param header: File header to prepend to the output. :type header: str :param ns_ignore: Whether to ignore the XML namespace. :type ns_ignore: bool ''' return dump_xml( self, stream=stream, filename=filename, header=header, ns_ignore=ns_ignore)
[docs] @classmethod def load(cls, stream=None, filename=None, string=None): ''' Deserialize from YAML. :param stream: Read input from stream. :param filename: Read input from file of given name. :type filename: str :param string: Read input from string. :type string: str ''' return load(stream=stream, filename=filename, string=string)
[docs] @classmethod def load_xml(cls, stream=None, filename=None, string=None, ns_hints=None, ns_ignore=False): ''' Deserialize from XML. :param stream: Read input from stream. :param filename: Read input from file of given name. :type filename: str :param string: Read input from string. :type string: str ''' if ns_hints is None: ns_hints = [cls.T.instance.get_xmlns()] return load_xml( stream=stream, filename=filename, string=string, ns_hints=ns_hints, ns_ignore=ns_ignore)
def __str__(self): return self.dump()
[docs]def to_dict(obj): ''' Get dict of guts object attributes. :param obj: :py:class`Object` object ''' return dict(obj.T.inamevals(obj))
[docs]class SObject(Object): ''' Base class for simple str-serializable Guts objects. Derived classes must support (de)serialization as in ``X(str(x))``. ''' class __T(TBase): def regularize_extra(self, val): if isinstance(val, str): return self._cls(val) return val def to_save(self, val): return str(val) def to_save_xml(self, val): return str(val)
[docs]class Any(Object): ''' Placeholder for any object. ''' class __T(TBase): def validate(self, val, regularize=False, depth=-1): if isinstance(val, Object): val.validate(regularize, depth) return val
[docs]class Int(Object): ''' Placeholder for :py:class:`int`. ''' dummy_for = int class __T(TBase): strict = True def to_save_xml(self, value): return repr(value)
[docs]class Float(Object): ''' Placeholder for :py:class:`float`. ''' dummy_for = float class __T(TBase): strict = True def to_save_xml(self, value): return repr(value)
[docs]class Complex(Object): ''' Placeholder for :py:class:`complex`. ''' dummy_for = complex class __T(TBase): strict = True def regularize_extra(self, val): if isinstance(val, list) or isinstance(val, tuple): assert len(val) == 2 val = complex(*val) elif not isinstance(val, complex): val = complex(val) return val def to_save(self, value): return repr(value) def to_save_xml(self, value): return repr(value)
[docs]class Bool(Object): ''' Placeholder for :py:class:`bool`. ''' dummy_for = bool class __T(TBase): strict = True def regularize_extra(self, val): if isinstance(val, str): if val.lower().strip() in ('0', 'false'): return False return bool(val) def to_save_xml(self, value): return repr(bool(value)).lower()
[docs]class String(Object): ''' Placeholder for :py:class:`str`. ''' dummy_for = str class __T(TBase): def __init__(self, *args, yamlstyle=None, **kwargs): TBase.__init__(self, *args, **kwargs) self.style_cls = str_style_map[yamlstyle] def to_save(self, val): return self.style_cls(val)
[docs]class Bytes(Object): ''' Placeholder for :py:class:`bytes`. ''' dummy_for = bytes class __T(TBase): def regularize_extra(self, val): if isinstance(val, str): val = b64decode(val) return val def to_save(self, val): return literal(b64encode(val).decode('utf-8'))
[docs]class Unicode(Object): ''' Placeholder for :py:class:`str`. ''' dummy_for = str class __T(TBase): def __init__(self, *args, yamlstyle=None, **kwargs): TBase.__init__(self, *args, **kwargs) self.style_cls = unicode_style_map[yamlstyle] def to_save(self, val): return self.style_cls(val)
guts_plain_dummy_types = (String, Unicode, Int, Float, Complex, Bool)
[docs]class Dict(Object): ''' Placeholder for :py:class:`dict`. ''' dummy_for = dict class __T(TBase): multivalued = dict def __init__(self, key_t=Any.T(), content_t=Any.T(), *args, **kwargs): TBase.__init__(self, *args, **kwargs) assert isinstance(key_t, TBase) assert isinstance(content_t, TBase) self.key_t = key_t self.content_t = content_t self.content_t.parent = self def default(self): if self._default is not None: return dict( (make_default(k), make_default(v)) for (k, v) in self._default.items()) if self.optional: return None else: return {} def has_default(self): return True def validate(self, val, regularize, depth): return TBase.validate(self, val, regularize, depth+1) def validate_children(self, val, regularize, depth): for key, ele in list(val.items()): newkey = self.key_t.validate(key, regularize, depth-1) newele = self.content_t.validate(ele, regularize, depth-1) if regularize: if newkey is not key or newele is not ele: del val[key] val[newkey] = newele return val def to_save(self, val): return dict((self.key_t.to_save(k), self.content_t.to_save(v)) for (k, v) in val.items()) def to_save_xml(self, val): raise NotImplementedError def classname_for_help(self, strip_module=''): return '``dict`` of %s objects' % \ self.content_t.classname_for_help(strip_module=strip_module)
[docs]class List(Object): ''' Placeholder for :py:class:`list`. ''' dummy_for = list class __T(TBase): multivalued = list def __init__(self, content_t=Any.T(), *args, yamlstyle=None, **kwargs): TBase.__init__(self, *args, **kwargs) assert isinstance(content_t, TBase) or isinstance(content_t, Defer) self.content_t = content_t self.content_t.parent = self self.style_cls = list_style_map[yamlstyle] def default(self): if self._default is not None: return [make_default(x) for x in self._default] if self.optional: return None else: return [] def has_default(self): return True def validate(self, val, regularize, depth): return TBase.validate(self, val, regularize, depth+1) def validate_children(self, val, regularize, depth): for i, ele in enumerate(val): newele = self.content_t.validate(ele, regularize, depth-1) if regularize and newele is not ele: val[i] = newele return val def to_save(self, val): return self.style_cls(self.content_t.to_save(v) for v in val) def to_save_xml(self, val): return [self.content_t.to_save_xml(v) for v in val] def deferred(self): if isinstance(self.content_t, Defer): return [self.content_t] return [] def process_deferred(self, defer, t_inst): if defer is self.content_t: self.content_t = t_inst def classname_for_help(self, strip_module=''): return '``list`` of %s objects' % \ self.content_t.classname_for_help(strip_module=strip_module)
def make_typed_list_class(t): class TL(List): class __T(List.T): def __init__(self, *args, **kwargs): List.T.__init__(self, content_t=t.T(), *args, **kwargs) return TL
[docs]class Tuple(Object): ''' Placeholder for :py:class:`tuple`. ''' dummy_for = tuple class __T(TBase): multivalued = tuple def __init__(self, n=None, content_t=Any.T(), *args, **kwargs): TBase.__init__(self, *args, **kwargs) assert isinstance(content_t, TBase) self.content_t = content_t self.content_t.parent = self self.n = n def default(self): if self._default is not None: return tuple( make_default(x) for x in self._default) elif self.optional: return None else: if self.n is not None: return tuple( self.content_t.default() for x in range(self.n)) else: return tuple() def has_default(self): return True def validate(self, val, regularize, depth): return TBase.validate(self, val, regularize, depth+1) def validate_extra(self, val): if self.n is not None and len(val) != self.n: raise ValidationError( '%s should have length %i' % (self.xname(), self.n)) def validate_children(self, val, regularize, depth): if not regularize: for ele in val: self.content_t.validate(ele, regularize, depth-1) return val else: newval = [] isnew = False for ele in val: newele = self.content_t.validate(ele, regularize, depth-1) newval.append(newele) if newele is not ele: isnew = True if isnew: return tuple(newval) else: return val def to_save(self, val): return tuple(self.content_t.to_save(v) for v in val) def to_save_xml(self, val): return [self.content_t.to_save_xml(v) for v in val] def classname_for_help(self, strip_module=''): if self.n is not None: return '``tuple`` of %i %s objects' % ( self.n, self.content_t.classname_for_help( strip_module=strip_module)) else: return '``tuple`` of %s objects' % ( self.content_t.classname_for_help( strip_module=strip_module))
duration_unit_factors = dict( s=1.0, m=60.0, h=3600.0, d=24*3600.0, y=365*24*3600.0) def parse_duration(s): unit = s[-1] if unit in duration_unit_factors: return float(s[:-1]) * duration_unit_factors[unit] else: return float(s) def str_duration(d): for k in 'ydhms': if abs(d) >= duration_unit_factors[k]: return '%g' % (d / duration_unit_factors[k]) + k return '%g' % d
[docs]class Duration(Object): ''' Placeholder for :py:class:`float` time duration [s] with human-readable (de)serialization. Examples: - ``'1s'`` -> 1 second - ``'1m'`` -> 1 minute - ``'1h'`` -> 1 hour - ``'1d'`` -> 1 day - ``'1y'`` -> about 1 year = 365*24*3600 seconds ''' dummy_for = float class __T(TBase): def regularize_extra(self, val): if isinstance(val, str): return parse_duration(val) return val def to_save(self, val): return str_duration(val) def to_save_xml(self, val): return str_duration(val)
re_tz = re.compile(r'(Z|([+-][0-2][0-9])(:?([0-5][0-9]))?)$')
[docs]class Timestamp(Object): ''' Placeholder for a UTC timestamp. ''' dummy_for = (hpfloat, float) dummy_for_description = 'pyrocko.util.get_time_float' class __T(TBase): def regularize_extra(self, val): time_float = get_time_float() if isinstance(val, datetime.datetime): tt = val.utctimetuple() val = time_float(calendar.timegm(tt)) + val.microsecond * 1e-6 elif isinstance(val, datetime.date): tt = val.timetuple() val = time_float(calendar.timegm(tt)) elif isinstance(val, str): val = val.strip() tz_offset = 0 m = re_tz.search(val) if m: sh = m.group(2) sm = m.group(4) tz_offset = (int(sh)*3600 if sh else 0) \ + (int(sm)*60 if sm else 0) val = re_tz.sub('', val) if len(val) > 10 and val[10] == 'T': val = val.replace('T', ' ', 1) try: val = str_to_time(val) - tz_offset except TimeStrError: raise ValidationError( '%s: cannot parse time/date: %s' % (self.xname(), val)) elif isinstance(val, (int, float)): val = time_float(val) else: raise ValidationError( '%s: cannot convert "%s" to type %s' % ( self.xname(), val, time_float)) return val def to_save(self, val): return time_to_str(val, format='%Y-%m-%d %H:%M:%S.9FRAC')\ .rstrip('0').rstrip('.') def to_save_xml(self, val): return time_to_str(val, format='%Y-%m-%dT%H:%M:%S.9FRAC')\ .rstrip('0').rstrip('.') + 'Z' @classmethod def D(self, s): return TimestampDefaultMaker(s)
[docs]class DateTimestamp(Object): ''' Placeholder for a UTC timestamp which (de)serializes as a date string. ''' dummy_for = (hpfloat, float) dummy_for_description = 'pyrocko.util.get_time_float' class __T(TBase): def regularize_extra(self, val): time_float = get_time_float() if isinstance(val, datetime.datetime): tt = val.utctimetuple() val = time_float(calendar.timegm(tt)) + val.microsecond * 1e-6 elif isinstance(val, datetime.date): tt = val.timetuple() val = time_float(calendar.timegm(tt)) elif isinstance(val, str): val = str_to_time(val, format='%Y-%m-%d') elif isinstance(val, int): val = time_float(val) return val def to_save(self, val): return time_to_str(val, format='%Y-%m-%d') def to_save_xml(self, val): return time_to_str(val, format='%Y-%m-%d') @classmethod def D(self, s): return TimestampDefaultMaker(s, format='%Y-%m-%d')
[docs]class StringPattern(String): ''' Any :py:class:`str` matching pattern ``%(pattern)s``. ''' dummy_for = str pattern = '.*' class __T(String.T): def __init__(self, pattern=None, *args, **kwargs): String.T.__init__(self, *args, **kwargs) if pattern is not None: self.pattern = pattern else: self.pattern = self._dummy_cls.pattern def validate_extra(self, val): pat = self.pattern if not re.search(pat, val): raise ValidationError('%s: "%s" does not match pattern %s' % ( self.xname(), val, repr(pat))) @classmethod def class_help_string(cls): dcls = cls._dummy_cls doc = dcls.__doc_template__ or StringPattern.__doc_template__ return doc % {'pattern': repr(dcls.pattern)}
[docs]class UnicodePattern(Unicode): ''' Any :py:class:`str` matching pattern ``%(pattern)s``. ''' dummy_for = str pattern = '.*' class __T(TBase): def __init__(self, pattern=None, *args, **kwargs): TBase.__init__(self, *args, **kwargs) if pattern is not None: self.pattern = pattern else: self.pattern = self._dummy_cls.pattern def validate_extra(self, val): pat = self.pattern if not re.search(pat, val, flags=re.UNICODE): raise ValidationError('%s: "%s" does not match pattern %s' % ( self.xname(), val, repr(pat))) @classmethod def class_help_string(cls): dcls = cls._dummy_cls doc = dcls.__doc_template__ or UnicodePattern.__doc_template__ return doc % {'pattern': repr(dcls.pattern)}
[docs]class StringChoice(String): ''' Any :py:class:`str` out of ``%(choices)s``. :cvar choices: Allowed choices (:py:class:`list` of :py:class:`str`). :cvar ignore_case: Whether to behave case-insensitive (:py:class:`bool`, default: ``False``). ''' dummy_for = str choices = [] ignore_case = False class __T(String.T): def __init__(self, choices=None, ignore_case=None, *args, **kwargs): String.T.__init__(self, *args, **kwargs) if choices is not None: self.choices = choices else: self.choices = self._dummy_cls.choices if ignore_case is not None: self.ignore_case = ignore_case else: self.ignore_case = self._dummy_cls.ignore_case if self.ignore_case: self.choices = [x.upper() for x in self.choices] def validate_extra(self, val): if self.ignore_case: val = val.upper() if val not in self.choices: raise ValidationError( '%s: "%s" is not a valid choice out of %s' % ( self.xname(), val, repr(self.choices))) @classmethod def class_help_string(cls): dcls = cls._dummy_cls doc = dcls.__doc_template__ or StringChoice.__doc_template__ return doc % {'choices': repr(dcls.choices)}
[docs]class IntChoice(Int): ''' Any :py:class:`int` out of ``%(choices)s``. ''' dummy_for = int choices = [] class __T(Int.T): def __init__(self, choices=None, *args, **kwargs): Int.T.__init__(self, *args, **kwargs) if choices is not None: self.choices = choices else: self.choices = self._dummy_cls.choices def validate_extra(self, val): if val not in self.choices: raise ValidationError( '%s: %i is not a valid choice out of %s' % ( self.xname(), val, repr(self.choices))) @classmethod def class_help_string(cls): dcls = cls._dummy_cls doc = dcls.__doc_template__ or IntChoice.__doc_template__ return doc % {'choices': repr(dcls.choices)}
# this will not always work...
[docs]class StringUnion(Object): ''' Any :py:class:`str` matching any of a set of constraints. :cvar members: List of constraints, e.g. :py:class:`StringChoice`, :py:class:`StringPattern`, ... (:py:class:`list` of :py:class:`TBase` derived objects). ''' members = [] dummy_for = str class __T(TBase): def __init__(self, members=None, *args, **kwargs): TBase.__init__(self, *args, **kwargs) if members is not None: self.members = members else: self.members = self._dummy_cls.members def validate(self, val, regularize=False, depth=-1): assert self.members e2 = None for member in self.members: try: return member.validate(val, regularize, depth=depth) except ValidationError as e: e2 = e raise e2
[docs]class Choice(Object): ''' Any out of a set of different types. :cvar choices: Allowed types (:py:class:`list` of :py:class:`TBase` derived objects). ''' choices = [] class __T(TBase): def __init__(self, choices=None, *args, **kwargs): TBase.__init__(self, *args, **kwargs) if choices is not None: self.choices = choices else: self.choices = self._dummy_cls.choices self.cls_to_xmltagname = dict( (t._cls, t.get_xmltagname()) for t in self.choices) def validate(self, val, regularize=False, depth=-1): if self.optional and val is None: return val t = None for tc in self.choices: is_derived = isinstance(val, tc._cls) is_exact = type(val) == tc._cls if not (not tc.strict and not is_derived or tc.strict and not is_exact): t = tc break if t is None: if regularize: ok = False for tc in self.choices: try: val = tc.regularize_extra(val) ok = True t = tc break except (ValidationError, ValueError): pass if not ok: raise ValidationError( '%s: could not convert "%s" to any type out of ' '(%s)' % (self.xname(), val, ','.join( classnames(x._cls) for x in self.choices))) else: raise ValidationError( '%s: "%s" (type: %s) is not of any type out of ' '(%s)' % (self.xname(), val, type(val), ','.join( classnames(x._cls) for x in self.choices))) validator = t if isinstance(t._cls, tuple): clss = t._cls else: clss = (t._cls,) for cls in clss: try: if type(val) != cls and isinstance(val, cls): validator = val.T.instance except AttributeError: pass validator.validate_extra(val) if depth != 0: val = validator.validate_children(val, regularize, depth) return val def extend_xmlelements(self, elems, v): elems.append(( self.cls_to_xmltagname[type(v)].split(' ', 1)[-1], v))
def _dump( object, stream, header=False, Dumper=GutsSafeDumper, _dump_function=yaml.dump): if not getattr(stream, 'encoding', None): enc = encode_utf8 else: enc = no_encode if header: stream.write(enc(u'%YAML 1.1\n')) if isinstance(header, str): banner = u'\n'.join('# ' + x for x in header.splitlines()) + '\n' stream.write(enc(banner)) _dump_function( object, stream=stream, encoding='utf-8', explicit_start=True, Dumper=Dumper) def _dump_all(object, stream, header=True, Dumper=GutsSafeDumper): _dump(object, stream=stream, header=header, _dump_function=yaml.dump_all) def _load(stream, Loader=GutsSafeLoader, allow_include=None, filename=None, included_files=None): class _Loader(Loader): _filename = filename _allow_include = allow_include _included_files = included_files or [] return yaml.load(stream=stream, Loader=_Loader) def _load_all(stream, Loader=GutsSafeLoader, allow_include=None, filename=None): class _Loader(Loader): _filename = filename _allow_include = allow_include return list(yaml.load_all(stream=stream, Loader=_Loader)) def _iload_all(stream, Loader=GutsSafeLoader, allow_include=None, filename=None): class _Loader(Loader): _filename = filename _allow_include = allow_include return yaml.load_all(stream=stream, Loader=_Loader) def multi_representer(dumper, data): node = dumper.represent_mapping( '!'+data.T.tagname, data.T.inamevals_to_save(data), flow_style=False) return node # hack for compatibility with early GF Store versions re_compatibility = re.compile( r'^pyrocko\.(trace|gf\.(meta|seismosizer)|fomosto\.' r'(dummy|poel|qseis|qssp))\.' ) def multi_constructor(loader, tag_suffix, node): tagname = str(tag_suffix) tagname = re_compatibility.sub('pf.', tagname) cls = g_tagname_to_class[tagname] kwargs = dict(iter(loader.construct_pairs(node, deep=True))) o = cls(**kwargs) o.validate(regularize=True, depth=1) return o def include_constructor(loader, node): allow_include = loader._allow_include \ if loader._allow_include is not None \ else ALLOW_INCLUDE if not allow_include: raise EnvironmentError( 'Not allowed to include YAML. Load with allow_include=True') if isinstance(node, yaml.nodes.ScalarNode): inc_file = loader.construct_scalar(node) else: raise TypeError('Unsupported YAML node %s' % repr(node)) if loader._filename is not None and not op.isabs(inc_file): inc_file = op.join(op.dirname(loader._filename), inc_file) if not op.isfile(inc_file): raise FileNotFoundError(inc_file) included_files = list(loader._included_files) if loader._filename is not None: included_files.append(op.abspath(loader._filename)) for included_file in loader._included_files: if op.samefile(inc_file, included_file): raise ImportError( 'Circular import of file "%s". Include path: %s' % ( op.abspath(inc_file), ' -> '.join('"%s"' % s for s in included_files))) with open(inc_file, 'rb') as f: return _load( f, Loader=loader.__class__, filename=inc_file, allow_include=True, included_files=included_files) def dict_noflow_representer(dumper, data): return dumper.represent_mapping( 'tag:yaml.org,2002:map', data, flow_style=False) yaml.add_multi_representer(Object, multi_representer, Dumper=GutsSafeDumper) yaml.add_constructor('!include', include_constructor, Loader=GutsSafeLoader) yaml.add_multi_constructor('!', multi_constructor, Loader=GutsSafeLoader) yaml.add_representer(dict, dict_noflow_representer, Dumper=GutsSafeDumper) def str_representer(dumper, data): return dumper.represent_scalar( 'tag:yaml.org,2002:str', str(data)) yaml.add_representer(str, str_representer, Dumper=GutsSafeDumper) class Constructor(object): def __init__(self, add_namespace_maps=False, strict=False, ns_hints=None, ns_ignore=False): self.stack = [] self.queue = [] self.namespaces = defaultdict(list) self.add_namespace_maps = add_namespace_maps self.strict = strict self.ns_hints = ns_hints self.ns_ignore = ns_ignore def start_element(self, ns_name, attrs): if self.ns_ignore: ns_name = ns_name.split(' ')[-1] if -1 == ns_name.find(' '): if self.ns_hints is None and ns_name in g_guessable_xmlns: self.ns_hints = g_guessable_xmlns[ns_name] if self.ns_hints: ns_names = [ ns_hint + ' ' + ns_name for ns_hint in self.ns_hints] elif self.ns_hints is None: ns_names = [' ' + ns_name] else: ns_names = [ns_name] for ns_name in ns_names: if self.stack and self.stack[-1][1] is not None: cls = self.stack[-1][1].T.xmltagname_to_class.get( ns_name, None) if isinstance(cls, tuple): cls = None else: if cls is not None and ( not issubclass(cls, Object) or issubclass(cls, SObject)): cls = None else: cls = g_xmltagname_to_class.get(ns_name, None) if cls: break self.stack.append((ns_name, cls, attrs, [], [])) def end_element(self, _): ns_name, cls, attrs, content2, content1 = self.stack.pop() ns = ns_name.split(' ', 1)[0] if cls is not None: content2.extend( (ns + ' ' + k if -1 == k.find(' ') else k, v) for (k, v) in attrs.items()) content2.append((None, ''.join(content1))) o = cls(**cls.T.translate_from_xml(content2, self.strict)) o.validate(regularize=True, depth=1) if self.add_namespace_maps: o.namespace_map = self.get_current_namespace_map() if self.stack and not all(x[1] is None for x in self.stack): self.stack[-1][-2].append((ns_name, o)) else: self.queue.append(o) else: content = [''.join(content1)] if self.stack: for c in content: self.stack[-1][-2].append((ns_name, c)) def characters(self, char_content): if self.stack: self.stack[-1][-1].append(char_content) def start_namespace(self, ns, uri): self.namespaces[ns].append(uri) def end_namespace(self, ns): self.namespaces[ns].pop() def get_current_namespace_map(self): return dict((k, v[-1]) for (k, v) in self.namespaces.items() if v) def get_queued_elements(self): queue = self.queue self.queue = [] return queue def _iload_all_xml( stream, bufsize=100000, add_namespace_maps=False, strict=False, ns_hints=None, ns_ignore=False): from xml.parsers.expat import ParserCreate from pyrocko import progress parser = ParserCreate('UTF-8', namespace_separator=' ') handler = Constructor( add_namespace_maps=add_namespace_maps, strict=strict, ns_hints=ns_hints, ns_ignore=ns_ignore) parser.StartElementHandler = handler.start_element parser.EndElementHandler = handler.end_element parser.CharacterDataHandler = handler.characters parser.StartNamespaceDeclHandler = handler.start_namespace parser.EndNamespaceDeclHandler = handler.end_namespace try: nbytes = os.fstat(stream.fileno()).st_size - stream.tell() except Exception: nbytes = None ibytes = 0 task = progress.task('Parsing XML', nbytes, logger=logger) try: while True: data = stream.read(bufsize) ibytes += len(data) parser.Parse(data, bool(not data)) for element in handler.get_queued_elements(): yield element task.update(ibytes) if not data: break except Exception: task.fail() raise finally: task.done() def _load_all_xml(*args, **kwargs): return list(_iload_all_xml(*args, **kwargs)) def _load_xml(*args, **kwargs): g = _iload_all_xml(*args, **kwargs) return next(g) def _dump_all_xml(objects, stream, root_element_name='root', header=True): if not getattr(stream, 'encoding', None): enc = encode_utf8 else: enc = no_encode _dump_xml_header(stream, header) beg = u'<%s>\n' % root_element_name end = u'</%s>\n' % root_element_name stream.write(enc(beg)) for ob in objects: _dump_xml(ob, stream=stream) stream.write(enc(end)) def _dump_xml_header(stream, banner=None): if not getattr(stream, 'encoding', None): enc = encode_utf8 else: enc = no_encode stream.write(enc(u'<?xml version="1.0" encoding="UTF-8" ?>\n')) if isinstance(banner, str): stream.write(enc(u'<!-- %s -->\n' % banner)) def _dump_xml( obj, stream, depth=0, ns_name=None, header=False, ns_map=[], ns_ignore=False): from xml.sax.saxutils import escape, quoteattr if not getattr(stream, 'encoding', None): enc = encode_utf8 else: enc = no_encode if depth == 0 and header: _dump_xml_header(stream, header) indent = ' '*depth*2 if ns_name is None: ns_name = obj.T.instance.get_xmltagname() if -1 != ns_name.find(' '): ns, name = ns_name.split(' ') else: ns, name = '', ns_name if isinstance(obj, Object): obj.validate(depth=1) attrs = [] elems = [] added_ns = False if not ns_ignore and ns and (not ns_map or ns_map[-1] != ns): attrs.append(('xmlns', ns)) ns_map.append(ns) added_ns = True for prop, v in obj.T.ipropvals_to_save(obj, xmlmode=True): if prop.xmlstyle == 'attribute': assert not prop.multivalued assert not isinstance(v, Object) attrs.append((prop.effective_xmltagname, v)) elif prop.xmlstyle == 'content': assert not prop.multivalued assert not isinstance(v, Object) elems.append((None, v)) else: prop.extend_xmlelements(elems, v) attr_str = '' if attrs: attr_str = ' ' + ' '.join( '%s=%s' % (k.split(' ')[-1], quoteattr(str(v))) for (k, v) in attrs) if not elems: stream.write(enc(u'%s<%s%s />\n' % (indent, name, attr_str))) else: oneline = len(elems) == 1 and elems[0][0] is None stream.write(enc(u'%s<%s%s>%s' % ( indent, name, attr_str, '' if oneline else '\n'))) for (k, v) in elems: if k is None: stream.write(enc(escape(str(v), {'\0': '&#00;'}))) else: _dump_xml(v, stream, depth+1, k, False, ns_map, ns_ignore) stream.write(enc(u'%s</%s>\n' % ( '' if oneline else indent, name))) if added_ns: ns_map.pop() else: stream.write(enc(u'%s<%s>%s</%s>\n' % ( indent, name, escape(str(obj), {'\0': '&#00;'}), name))) def walk(x, typ=None, path=()): if typ is None or isinstance(x, typ): yield path, x if isinstance(x, Object): for (prop, val) in x.T.ipropvals(x): if prop.multivalued: if val is not None: for iele, ele in enumerate(val): for y in walk(ele, typ, path=path + ((prop.name, iele),)): yield y else: for y in walk(val, typ, path=path+(prop.name,)): yield y
[docs]def clone(x, pool=None): ''' Clone guts object tree. Traverses guts object tree and recursively clones all guts attributes, falling back to :py:func:`copy.deepcopy` for non-guts objects. Objects deriving from :py:class:`Object` are instantiated using their respective init function. Multiply referenced objects in the source tree are multiply referenced also in the destination tree. This function can be used to clone guts objects ignoring any contained run-time state, i.e. any of their attributes not defined as a guts property. ''' if pool is None: pool = {} if id(x) in pool: x_copy = pool[id(x)][1] else: if isinstance(x, SObject): x_copy = x.__class__(str(x)) elif isinstance(x, Object): d = {} for (prop, y) in x.T.ipropvals(x): if y is not None: if not prop.multivalued: y_copy = clone(y, pool) elif prop.multivalued is dict: y_copy = dict( (clone(zk, pool), clone(zv, pool)) for (zk, zv) in y.items()) else: y_copy = type(y)(clone(z, pool) for z in y) else: y_copy = y d[prop.name] = y_copy x_copy = x.__class__(**d) else: x_copy = copy.deepcopy(x) pool[id(x)] = (x, x_copy) return x_copy
[docs]class YPathError(Exception): ''' This exception is raised for invalid ypath specifications. ''' pass
def _parse_yname(yname): ident = r'[a-zA-Z][a-zA-Z0-9_]*' rint = r'-?[0-9]+' m = re.match( r'^(%s)(\[((%s)?(:)(%s)?|(%s))\])?$' % (ident, rint, rint, rint), yname) if not m: raise YPathError('Syntax error in component: "%s"' % yname) d = dict( name=m.group(1)) if m.group(2): if m.group(5): istart = iend = None if m.group(4): istart = int(m.group(4)) if m.group(6): iend = int(m.group(6)) d['slice'] = (istart, iend) else: d['index'] = int(m.group(7)) return d def _decend(obj, ynames): if ynames: for sobj in iter_elements(obj, ynames): yield sobj else: yield obj
[docs]def iter_elements(obj, ypath): ''' Generator yielding elements matching a given ypath specification. :param obj: guts :py:class:`Object` instance :param ypath: Dot-separated object path (e.g. 'root.child.child'). To access list objects use slice notatation (e.g. 'root.child[:].child[1:3].child[1]'). Raises :py:exc:`YPathError` on failure. ''' try: if isinstance(ypath, str): ynames = ypath.split('.') else: ynames = ypath yname = ynames[0] ynames = ynames[1:] d = _parse_yname(yname) if d['name'] not in obj.T.propnames: raise AttributeError(d['name']) obj = getattr(obj, d['name']) if 'index' in d: sobj = obj[d['index']] for ssobj in _decend(sobj, ynames): yield ssobj elif 'slice' in d: for i in range(*slice(*d['slice']).indices(len(obj))): sobj = obj[i] for ssobj in _decend(sobj, ynames): yield ssobj else: for sobj in _decend(obj, ynames): yield sobj except (AttributeError, IndexError) as e: raise YPathError('Invalid ypath: "%s" (%s)' % (ypath, str(e)))
[docs]def get_elements(obj, ypath): ''' Get all elements matching a given ypath specification. :param obj: guts :py:class:`Object` instance :param ypath: Dot-separated object path (e.g. 'root.child.child'). To access list objects use slice notatation (e.g. 'root.child[:].child[1:3].child[1]'). Raises :py:exc:`YPathError` on failure. ''' return list(iter_elements(obj, ypath))
[docs]def set_elements(obj, ypath, value, validate=False, regularize=False): ''' Set elements matching a given ypath specification. :param obj: guts :py:class:`Object` instance :param ypath: Dot-separated object path (e.g. 'root.child.child'). To access list objects use slice notatation (e.g. 'root.child[:].child[1:3].child[1]'). :param value: All matching elements will be set to `value`. :param validate: Whether to validate affected subtrees. :param regularize: Whether to regularize affected subtrees. Raises :py:exc:`YPathError` on failure. ''' ynames = ypath.split('.') try: d = _parse_yname(ynames[-1]) if ynames[:-1]: it = iter_elements(obj, ynames[:-1]) else: it = [obj] for sobj in it: if d['name'] not in sobj.T.propnames: raise AttributeError(d['name']) if 'index' in d: ssobj = getattr(sobj, d['name']) ssobj[d['index']] = value elif 'slice' in d: ssobj = getattr(sobj, d['name']) for i in range(*slice(*d['slice']).indices(len(ssobj))): ssobj[i] = value else: setattr(sobj, d['name'], value) if regularize: sobj.regularize() if validate: sobj.validate() except (AttributeError, IndexError, YPathError) as e: raise YPathError('Invalid ypath: "%s" (%s)' % (ypath, str(e)))
def zip_walk(x, typ=None, path=(), stack=()): if typ is None or isinstance(x, typ): yield path, stack + (x,) if isinstance(x, Object): for (prop, val) in x.T.ipropvals(x): if prop.multivalued: if val is not None: for iele, ele in enumerate(val): for y in zip_walk( ele, typ, path=path + ((prop.name, iele),), stack=stack + (x,)): yield y else: for y in zip_walk(val, typ, path=path+(prop.name,), stack=stack + (x,)): yield y def path_element(x): if isinstance(x, tuple): if len(x) == 2: return '%s[%i]' % x elif len(x) == 3: return '%s[%i:%i]' % x else: return x def path_to_str(path): return '.'.join(path_element(x) for x in path)
[docs]@expand_stream_args('w') def dump(obj, stream, **kwargs): ''' Serialize to YAML. If neither ``stream`` nor ``filename`` is set, a string containing the serialized data is returned. :param obj: Object to be serialized. :type obj: :py:class:`Object` :param stream: Output to stream. :param filename: Output to file of given name. :type filename: str :param header: File header to prepend to the output. :type header: str ''' return _dump(obj, stream, **kwargs)
@expand_stream_args('r') def load(stream, **kwargs): return _load(stream, **kwargs) def load_string(s, **kwargs): return load(string=s, **kwargs) @expand_stream_args('w') def dump_all(obj, stream, **kwargs): return _dump_all(obj, stream, **kwargs) @expand_stream_args('r') def load_all(stream, **kwargs): return _load_all(stream, **kwargs) @expand_stream_args('r') def iload_all(stream, **kwargs): return _iload_all(stream, **kwargs)
[docs]@expand_stream_args('w') def dump_xml(obj, stream, **kwargs): ''' Serialize to XML. If neither ``stream`` nor ``filename`` is set, a string containing the serialized data is returned. :param obj: Object to be serialized. :type obj: :py:class:`Object` :param stream: Output to stream. :param filename: Output to file of given name. :type filename: str :param header: File header to prepend to the output. :type header: str :param ns_ignore: Whether to ignore the XML namespace. :type ns_ignore: bool ''' return _dump_xml(obj, stream, **kwargs)
@expand_stream_args('r') def load_xml(stream, **kwargs): kwargs.pop('filename', None) return _load_xml(stream, **kwargs) def load_xml_string(s, **kwargs): return load_xml(string=s, **kwargs) @expand_stream_args('w') def dump_all_xml(obj, stream, **kwargs): return _dump_all_xml(obj, stream, **kwargs) @expand_stream_args('r') def load_all_xml(stream, **kwargs): kwargs.pop('filename', None) return _load_all_xml(stream, **kwargs) @expand_stream_args('r') def iload_all_xml(stream, **kwargs): kwargs.pop('filename', None) return _iload_all_xml(stream, **kwargs) def _dump_all_spickle(objects, stream): import pickle header = b'SPICKLE'.ljust(512) stream.write(header, ) for obj in objects: pickle.dump(obj, stream) def _iload_all_spickle(stream): for obj, _ in _iload_all_spickle_internal(stream): yield obj def _load_one_spickle_internal(stream): import pickle fpos = stream.tell() return pickle.load(stream), fpos def _iload_all_spickle_internal(stream, offset=None): if offset is not None: stream.seek(offset, 0) else: header = stream.read(512) if not header.startswith(b'SPICKLE'): raise ValueError('Not a SPICKLE file.') while True: try: yield _load_one_spickle_internal(stream) except EOFError: break def _load_all_spickle(stream): return list(_iload_all_spickle(stream)) @expand_stream_args('w') def dump_all_spickle(objects, stream, **kwargs): return _dump_all_spickle(objects, stream) @expand_stream_args('r') def iload_all_spickle(stream, **kwargs): _iload_all_spickle(stream) @expand_stream_args('r') def load_all_spickle(stream, **kwargs): kwargs.pop('filename', None) return _load_all_spickle(stream, **kwargs) __all__ = guts_types + [ 'guts_types', 'TBase', 'ValidationError', 'ArgumentError', 'Defer', 'DefaultMaker', 'ObjectDefaultMaker', 'clone', 'dump', 'load', 'dump_all', 'load_all', 'iload_all', 'dump_xml', 'load_xml', 'dump_all_xml', 'load_all_xml', 'iload_all_xml', 'load_string', 'load_xml_string', 'make_typed_list_class', 'walk', 'zip_walk', 'path_to_str' ]