def __enter__(self):
self._file = open(self.filename, 'rU', encoding="utf-8")
return yaml.load_all(self._file)
python类load_all()的实例源码
def load(cls, config_file):
""" Load experiment parameters from a YAML configuration file
Parameters
----------
config_file: string
path to a YAML configuration file
Returns
-------
dictionary (or list of dictionaries) of parameters to pass to a behavior
"""
try:
import yaml
except ImportError:
raise ImportError("Pyyaml is required to use a .yaml configuration file")
parameters = list()
with open(config_file, "rb") as config:
for val in yaml.load_all(config):
parameters.append(val)
if len(parameters) == 1:
parameters = parameters[0]
return parameters
def test():
def spec_test(name, spec):
"Make a test function for a case from specification.yml"
def test(*args):
exc, res = None, None
try:
res = jsone.render(spec['template'], spec['context'])
except jsone.JSONTemplateError as e:
if 'error' not in spec:
raise
exc = e
if 'error' in spec:
assert exc, "expected exception"
expected = spec['error']
if expected is True: # no specific expectation
return
eq_(str(exc), expected)
else:
assert not exc
assert res == spec['result'], \
'{!r} != {!r}'.format(res, spec['result'])
return test
with open(os.path.join(os.path.dirname(__file__), '../specification.yml')) as f:
with freeze_time('2017-01-19T16:27:20.974Z'):
for spec in yaml.load_all(f):
if 'section' in spec:
section = spec['section']
continue
name = '{}: {}'.format(section, spec['title'])
t = spec_test(name, spec)
yield (t, name)
def onPaste(self):
try:
data = yaml.load_all(unicode(self.clipboard.text()))
if isinstance(data, dict):
data = [data]
except yaml.YAMLError as e:
msgBox(Lang.value('MSG_YAML_failed') % e)
return
for entry in data:
entry = model.makeSafe(entry)
Mainframe.model.insert(entry, True, Mainframe.model.current + 1)
self.rebuild()
Mainframe.sigWrapper.sigModelChanged.emit()
def load_info(path, qt_parent=None):
path_base, path_extension = _ospath.splitext(path)
filename = path_base + '.yaml'
try:
with open(filename, 'r') as info_file:
info = list(_yaml.load_all(info_file))
except FileNotFoundError as e:
print('\nAn error occured. Could not find metadata file:\n{}'.format(filename))
if qt_parent is not None:
_QMessageBox.critical(qt_parent, 'An error occured', 'Could not find metadata file:\n{}'.format(filename))
raise NoMetadataFileError(e)
return info
def yml_2_json():
json_text.delete(1.0, tk.END)
error_tips['text']=tips_str
txt = yml_text.get(1.0, tk.END)
if len(txt) <= 1:
return
try:
yml_doc = yaml.load_all(txt)
j = ""
for obj in yml_doc:
j += json.dumps(obj, indent=4)
json_text.insert(1.0, j)
except:
error_tips['text']=tips_str + 'YAML????'
def headers(self):
try:
headers = next(yaml.load_all(self.read(body=False)))
except StopIteration as e:
raise ValueError("YAML header is missing. Please ensure that the top of your post has a header of the following form:\n" + SAMPLE_HEADER)
except yaml.YAMLError as e:
raise ValueError(
"YAML header is incorrectly formatted or missing. The following information may be useful:\n{}\nIf you continue to have difficulties, try pasting your YAML header into an online parser such as http://yaml-online-parser.appspot.com/.".format(str(e)))
for key, value in headers.copy().items():
if isinstance(value, datetime.date):
headers[key] = datetime.datetime.combine(value, datetime.time(0))
if key == 'tags' and isinstance(value, list):
headers[key] = [str(v) if six.PY3 else unicode(v) for v in value]
return headers
def load_all(*args, **kwargs):
"""Delegate to yaml loadall.
"""
if kwargs is None:
kwargs = {}
kwargs['Loader'] = Loader
return yaml.load_all(*args, **kwargs)
def load(stream, safe=False, many=False):
Loader = SafeCustomLoader if safe else CustomLoader
if many:
return tuple(yaml.load_all(stream, Loader))
else:
return yaml.load(stream, Loader)
def generate_yaml_tests():
file_path = join(YAML_TESTS_ROOT, "macros.yaml")
with open(file_path) as f:
configs = load_all(f)
for config in configs:
yield (config['name'], _generate_test(config))
def bulk_create_problem_yml(self, path):
"""Create new problem entries in database from yml document stream.
Can create a yml file with individual problems delimited into documents
using the `---` ... `---` document stream syntax.
Parameters
----------
path: str or path-like
Path to yml file
"""
with open(path, "r") as f:
obj_all = yaml.load_all(f)
for obj in obj_all:
self.create_problem(**obj)
def load_all(filename):
"""Generate objects contained in ``filename``."""
if filename.endswith('.yaml') or filename.endswith('.yml'):
with io.open(filename, encoding='utf-8') as handle:
for obj in yaml.load_all(handle):
yield obj
else:
raise UsageError("Unsupported file type (extension) in '{}'!".format(filename))
def read_merged_files(cfgfiles):
"""Read a list of hierachical config files, and merge their keys."""
result = {}
for cfgfile in cfgfiles:
for obj in load_all(cfgfile):
merge_objects(result, obj)
return result
def parse_file(file_name):
with open(file_name) as f:
theYAML = list(yaml.load_all(f, Loader=Loader))
try:
theYAML.remove(None)
except ValueError:
pass
return theYAML
def test_duplicates(self):
mydir = os.path.dirname(__file__)
path = os.path.join(mydir, 'dups_good')
buildfilepath = os.path.join(path, 'build.yml')
if os.path.exists(buildfilepath):
os.remove(buildfilepath)
build.generate_build_file(path)
assert os.path.exists(buildfilepath)
with open(buildfilepath) as fd:
docs = yaml.load_all(fd)
data = next(docs, None)
contents = data['contents']
assert contents == {
# File extensions added to "a" due to conflicts
'a_txt': {'file': 'a.txt'},
'a_csv': {'file': 'a.csv'},
# "a" dir stays the same - but it's fine cause other "a"s got renamed
'a': {},
# Directories don't actually have extensions, so include them even with no conficts
'dir_ext': {},
# Weird characters replaced with a single "_"
'a_b_c': {'file': 'a%%b___c'},
# Prepend "n" to files that start with a number
'n1': {'file': '1'},
# ... even if there used to be an underscore there
'n123': {'file': '_123'},
# Handle conflicts with numbers, too
'n1_txt': {'file': '1.txt'},
}
os.remove(buildfilepath)
def read_yml_file(fn):
mydir = os.path.dirname(__file__)
filepath = os.path.join(mydir, fn)
with open(filepath) as fd:
return next(yaml.load_all(fd), None)
def load_inputs(filename: str) -> list:
"""Load inputs configuration from `filename`."""
if not filename:
filename = _find_config_file("inputs.yaml")
_LOG.debug("loading inputs from %s", filename)
if not os.path.isfile(filename):
_LOG.error("loading inputs file error: '%s' not found", filename)
return None
try:
with open(filename) as fin:
inps = [doc for doc in yaml.load_all(fin)
if doc and doc.get("enable", True)]
_LOG.debug("loading inputs - found %d enabled inputs",
len(inps))
if not inps:
_LOG.error("loading inputs error: no valid/enabled "
"inputs found")
return inps
except IOError as err:
_LOG.error("loading inputs from file %s error: %s", filename,
err)
except yaml.error.YAMLError as err:
_LOG.error("loading inputs from file %s - invalid YAML: %s",
filename, err)
return None
def process_all_files(result_file, file_names):
for source_path in (Path(n) for n in file_names):
detail_log.info("read {}".format(source_path))
with source_path.open() as source_file:
game_iter = yaml.load_all(source_file)
statistics = gather_stats(game_iter)
result_file.write(
yaml.dump(dict(statistics), explicit_start=True)
)
def __init__(self, filename):
with open(filename) as infile:
try:
text = infile.read()
except:
raise ZettelLoaderError(
"%s: I/O error - Encoding must be UTF-8" % filename)
try:
self.ydocs = yaml.load_all(text)
except:
raise ZettelLoaderError("%s: YAML load failure" % filename)
