Coverage for /usr/local/lib/python3.7/dist-packages/pyrocko/guts.py : 56%

# -*- coding: utf-8 -*- 

# http://pyrocko.org - GPLv3 

# 

# The Pyrocko Developers, 21st Century 

# ---|P------/S----------~Lg---------- 

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

from __future__ import absolute_import, print_function 

from builtins import str as newstr 

from builtins import range 

from future.utils import with_metaclass 

import datetime 

import calendar 

import re 

import sys 

import types 

import copy 

from collections import defaultdict 

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 

class GutsSafeDumper(SafeDumper): 

pass 

class GutsSafeLoader(SafeLoader): 

pass 

try: 

unicode 

except NameError: 

unicode = str 

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', 'List', 'Dict', 'Tuple', 'Union', 

'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 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 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.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') 

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 

class TBase(object): 

strict = False 

multivalued = None 

force_regularize = False 

propnames = [] 

@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 

if isinstance(self._default, DefaultMaker): 

self._default_cmp = self._default.make() 

else: 

self._default_cmp = self._default 

self.optional = optional 

self.name = None 

self._xmltagname = xmltagname 

self._xmlns = xmlns 

self.parent = None 

self.xmlstyle = xmlstyle 

self.help = help 

def default(self): 

return make_default(self._default) 

def is_default(self, val): 

if self._default_cmp is None: 

return val is None 

else: 

return self._default_cmp == 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, self.cls.__name__)) 

else: 

raise ValidationError( 

'%s: "%s" (type: %s) is not of type %s' % ( 

self.xname(), val, type(val), self.cls.__name__)) 

validator = self 

if type(val) != self.cls \ 

and isinstance(val, self.cls) and \ 

hasattr(val, 'T'): 

# derived classes only: validate with derived class validator 

validator = val.T.instance 

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 in guts_plain_dummy_types: 

return '``%s``' % self.cls.__name__ 

else: 

mod = self.cls.__module__ 

cls = self.cls.__name__ 

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 mod == '__builtin__': 

return '``%s``%s' % (cls, sadd) 

elif self.cls.__module__ == strip_module: 

return ':py:class:`%s`%s' % (cls, sadd) 

else: 

return ':py:class:`%s.%s`%s' % (mod, 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*') 

d = prop.default() 

if d is not None: 

descr.append('*default:* ``%s``' % repr(d)) 

hlp.append(' .. py:gattribute:: %s' % prop.name) 

hlp.append('')