def read_configuration(cls, config_file):
"""read YAML configuration file"""
# load YAML events/measurements definition
f = open(config_file, 'r')
doc_yaml = yaml.compose(f)
f.close()
# split events & measurements definitions
measurements, events = list(), list()
for key, value in doc_yaml.value:
if value.tag == Measurements.yaml_tag:
measurements.append((key, value))
if value.tag == Events.yaml_tag:
events.append((key, value))
measurements_yaml = yaml.MappingNode(u'tag:yaml.org,2002:map',
measurements)
measurements_stream = yaml.serialize(measurements_yaml)
events_yaml = yaml.MappingNode(u'tag:yaml.org,2002:map', events)
events_stream = yaml.serialize(events_yaml)
# return event & measurements definition
return events_stream, measurements_stream
python类org()的实例源码
def _dump_yaml(applied):
"""
Dumps dict correctly processing multiline pre & postDescription string
:param applied: dict -> filled config ready to be converted to yaml
:return: str -> yaml config
"""
def str_presenter(dumper, data):
# check for multiline strings
if len(data.splitlines()) == 1 and data[-1] == '\n':
return dumper.represent_scalar(
'tag:yaml.org,2002:str', data, style='>')
if len(data.splitlines()) > 1:
return dumper.represent_scalar(
'tag:yaml.org,2002:str', data, style='|')
return dumper.represent_scalar(
'tag:yaml.org,2002:str', data.strip())
yaml.add_representer(unicode, str_presenter)
yaml.add_representer(str, str_presenter)
return yaml.dump(applied, default_flow_style=False, width=1000)
def __init__(self, stream):
"""Initialise Loader."""
try:
self._root = os.path.split(stream.name)[0]
except AttributeError:
self._root = os.path.curdir
super().__init__(stream)
self.add_implicit_resolver(
u'tag:yaml.org,2002:float',
re.compile(u'''^(?:
[-+]?(?:[0-9][0-9_]*)\\.[0-9_]*(?:[eE][-+]?[0-9]+)?
|[-+]?(?:[0-9][0-9_]*)(?:[eE][-+]?[0-9]+)
|\\.[0-9_]+(?:[eE][-+][0-9]+)?
|[-+]?[0-9][0-9_]*(?::[0-5]?[0-9])+\\.[0-9_]*
|[-+]?\\.(?:inf|Inf|INF)
|\\.(?:nan|NaN|NAN))$''', re.X),
list(u'-+0123456789.'))
self.filenames = [os.path.abspath(stream.name)]
def ordered_load(
stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
"""Load yaml as an ordered dict
This allows us to inspect the order of the file on disk to make
sure it was correct by our rules.
"""
class OrderedLoader(Loader):
pass
def construct_mapping(loader, node):
loader.flatten_mapping(node)
return object_pairs_hook(loader.construct_pairs(node))
OrderedLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
construct_mapping)
# for parameters.yaml we treat numbers (especially version
# numbers) as strings. So that microversion specification of 2.20
# and 2.2 don't get confused.
OrderedLoader.add_constructor(
u'tag:yaml.org,2002:float',
yaml.constructor.SafeConstructor.construct_yaml_str)
return yaml.load(stream, OrderedLoader)
def represent_stringish(dumper, data):
# Will crash on bytestrings with weird chars in them,
# because we can't tell if it's supposed to be e.g. utf-8 readable string
# or an arbitrary binary buffer, and former one *must* be pretty-printed
# PyYAML's Representer.represent_str does the guesswork and !!binary or !!python/str
# Explicit crash on any bytes object might be more sane, but also annoying
# Use something like base64 to encode such buffer values instead
# Having such binary stuff pretty much everywhere on unix (e.g. paths) kinda sucks
data = unicode(data) # read the comment above
# Try to use '|' style for multiline data,
# quoting it with 'literal' if lines are too long anyway,
# not sure if Emitter.analyze_scalar can also provide useful info here
style = dumper.pyaml_string_val_style
if not style:
style = 'plain'
if '\n' in data or not data or data == '-' or data[0] in '!&*[':
style = 'literal'
if '\n' in data[:-1]:
for line in data.splitlines():
if len(line) > dumper.best_width: break
else: style = '|'
return yaml.representer.ScalarNode('tag:yaml.org,2002:str', data, style=style)
def unicode_representer(_, data):
has_wide_lines = False
for line in data.splitlines():
if len(line) > 80:
has_wide_lines = True
break
if has_wide_lines:
return yaml.ScalarNode(
u'tag:yaml.org,2002:str', data, style='>')
if "\n" in data:
return yaml.ScalarNode(
u'tag:yaml.org,2002:str', data, style='|')
return yaml.ScalarNode(
u'tag:yaml.org,2002:str', data, style='')
def configure_yml(self):
def represent_ordereddict(dumper, data):
value = []
for item_key, item_value in data.items():
node_key = dumper.represent_data(item_key)
node_value = dumper.represent_data(item_value)
value.append((node_key, node_value))
return yaml.nodes.MappingNode(u'tag:yaml.org,2002:map', value)
self.collect_answers()
cfg = self.process_answers()
yaml.add_representer(OrderedDict, represent_ordereddict)
return yaml.dump(cfg, default_flow_style=False)
def __init__(self, *args, **kwds):
yaml.Loader.__init__(self, *args, **kwds)
self.add_constructor(u'tag:yaml.org,2002:map', type(self).construct_yaml_map)
self.add_constructor(u'tag:yaml.org,2002:omap', type(self).construct_yaml_map)
def string_presenter(self, dumper, data):
"""Presenter to force yaml.dump to use multi-line string style."""
if '\n' in data:
return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
else:
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
def represent_decimal(self, data):
return self.represent_scalar('tag:yaml.org,2002:str', str(data))
def represent_ordered_dict(self, data):
return self.represent_mapping('tag:yaml.org,2002:map', data.items())
def represent_decimal(self, data):
return self.represent_scalar('tag:yaml.org,2002:str', str(data))
def represent_ordered_dict(self, data):
return self.represent_mapping('tag:yaml.org,2002:map', data.items())
def format_node(cls, mapping, metric):
if mapping.tag in [
'tag:yaml.org,2002:str', Bytes2Kibibytes.yaml_tag,
Number.yaml_tag, StripExtraDash.yaml_tag]:
return yaml.ScalarNode(mapping.tag, mapping.value.format(**metric))
elif mapping.tag == 'tag:yaml.org,2002:map':
values = []
for key, value in mapping.value:
values.append((yaml.ScalarNode(key.tag, key.value),
cls.format_node(value, metric)))
return yaml.MappingNode(mapping.tag, values)
elif mapping.tag in [ArrayItem.yaml_tag, ValueItem.yaml_tag]:
values = []
for seq in mapping.value:
map_values = list()
for key, value in seq.value:
if key.value == 'SELECT':
map_values.append((yaml.ScalarNode(key.tag, key.value),
cls.format_node(value, metric)))
else:
map_values.append((yaml.ScalarNode(key.tag, key.value),
value))
values.append(yaml.MappingNode(seq.tag, map_values))
return yaml.SequenceNode(mapping.tag, values)
elif mapping.tag in [MapValue.yaml_tag]:
values = []
for key, value in mapping.value:
if key.value == 'VALUE':
values.append((yaml.ScalarNode(key.tag, key.value),
cls.format_node(value, metric)))
else:
values.append((yaml.ScalarNode(key.tag, key.value), value))
return yaml.MappingNode(mapping.tag, values)
return mapping
def represent_decimal(self, data):
return self.represent_scalar('tag:yaml.org,2002:str', str(data))
def represent_ordered_dict(self, data):
return self.represent_mapping('tag:yaml.org,2002:map', data.items())
def yaml_load(stream):
"""Load YAML document, but load all strings as unicode on py2."""
import yaml
class UnicodeLoader(yaml.SafeLoader):
"""Yaml SafeLoader Class, default encoding is UTF-8."""
pass
# NOTE:
# In [2]: yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
# Out[2]: 'tag:yaml.org,2002:map'
UnicodeLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_SCALAR_TAG,
UnicodeLoader.construct_scalar)
return yaml.load(stream, UnicodeLoader)
def pocketsphinx(self):
# NOTE: copy()
# Return a shallow copy of x.
# source: https://docs.python.org/3/library/copy.html
"""Get pocketsphinx speech to text settings."""
return self._data.get('pocketsphinx', {}).copy()
def features_enabled(self):
"""Get features settings."""
return self._data.get('features', [])
# In [2]: yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
# Out[2]: 'tag:yaml.org,2002:map'
# In [3]:
def _read_yaml(filename):
yaml.Loader.add_constructor(u'tag:yaml.org,2002:str', construct_yaml_str)
yaml.SafeLoader.add_constructor(u'tag:yaml.org,2002:str', construct_yaml_str)
with open(filename, 'r') as f:
return yaml.safe_load(f.read())
def setup_yaml():
""" http://stackoverflow.com/a/8661021 """
represent_dict_order = lambda self, data: self.represent_mapping(
'tag:yaml.org,2002:map', data.items())
yaml.add_representer(collections.OrderedDict, represent_dict_order)
def unicode_representer(dumper, uni):
node = yaml.ScalarNode(tag=u'tag:yaml.org,2002:str', value=str(uni))
return node
def represent_decimal(self, data):
return self.represent_scalar('tag:yaml.org,2002:str', str(data))
def represent_ordered_dict(self, data):
return self.represent_mapping('tag:yaml.org,2002:map', data.items())
def literal_unicode_representer(dumper, data):
"""
Use |- literal syntax for long strings
"""
if '\n' in data:
return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|')
else:
return dumper.represent_scalar(u'tag:yaml.org,2002:str', data)
def write_content(self, content=None):
if content:
self.content = content
self.__documents[self.__document_id] = self.content
def representer(dumper, data):
"""Represents a dict key started with '!' as a YAML tag
Assumes that there is only one !tag in the dict at the
current indent.
Python object:
{"!unknown_tag": ["some content", ]}
Resulting yaml:
!unknown_tag
- some content
"""
key = data.keys()[0]
if key.startswith("!"):
value = data[key]
if type(value) is dict:
node = dumper.represent_mapping(key, value)
elif type(value) is list:
node = dumper.represent_sequence(key, value)
else:
node = dumper.represent_scalar(key, value)
else:
node = dumper.represent_mapping(u'tag:yaml.org,2002:map', data)
return node
yaml.add_representer(dict, representer)
with self.__get_file("w") as file_obj:
yaml.dump_all(self.__documents, file_obj,
default_flow_style=self.default_flow_style,
default_style=self.default_style)
def _folded_unicode_representer(dumper, data: str):
return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='>')
def _literal_unicode_representer(dumper, data: str):
return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|')
def __init__(self, *args, **kwargs):
yaml.Loader.__init__(self, *args, **kwargs)
self.add_constructor(u'tag:yaml.org,2002:map', type(self).construct_yaml_map)
self.add_constructor(u'tag:yaml.org,2002:omap', type(self).construct_yaml_map)
def dump(self):
dumpdata = {}
dumpdata['hosts'] = self.dump_hosts()
dumpdata['hostgroup'] = self.dump_hostgroups()
dumpdata['architecture'] = self.dump_arch()
dumpdata['environment'] = self.dump_env()
dumpdata['os'] = self.dump_os()
dumpdata['model'] = self.dump_model()
dumpdata['media'] = self.dump_media()
dumpdata['domain'] = self.dump_domain()
dumpdata['settings'] = self.dump_settings()
dumpdata['subnet'] = self.dump_subnet()
dumpdata['smart-proxy'] = self.dump_smartproxy()
dumpdata['partition-table'] = self.dump_ptable()
dumpdata['provisioning-template'] = self.dump_provisioningtpl()
dumpdata['users'] = self.dump_users()
dumpdata['users'] = self.dump_users()
dumpdata['auth-source-ldap'] = self.dump_ldaps()
dumpdata['usergroups'] = self.dump_usergroups()
dumpdata['roles'] = self.dump_roles()
# print the result
fmyml = { 'foreman': dumpdata }
def str_presenter(dumper, data):
try:
dlen = len(data.splitlines())
except TypeError:
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
if (dlen > 1):
return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
yaml.add_representer(unicode, str_presenter)
yaml.add_representer(str, str_presenter)
yml = yaml.dump(fmyml, allow_unicode=True, default_flow_style=False )
print( (yml) )
