def save(data, stream=None, Dumper=yaml.SafeDumper,
default_flow_style=False,
encoding='utf-8',
**kwds):
class OrderedDumper(Dumper):
pass
def _dict_representer(dumper, data):
return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
data.items())
OrderedDumper.add_representer(OrderedDict, _dict_representer)
OrderedDumper.add_representer(np.float64,
lambda dumper, data: dumper.represent_float(float(data)))
OrderedDumper.add_representer(complex,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.complex128,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.ndarray,
lambda dumper, data: dumper.represent_list(list(data)))
# I added the following two lines to make pyrpl compatible with pyinstruments. In principle they can be erased
if isinstance(data, dict) and not isinstance(data, OrderedDict):
data = OrderedDict(data)
return yaml.dump(data,
stream=stream,
Dumper=OrderedDumper,
default_flow_style=default_flow_style,
encoding=encoding,
**kwds)
# usage example:
# load(stream, yaml.SafeLoader)
# save(data, stream=f, Dumper=yaml.SafeDumper)
python类load()的实例源码
def _set_yml(self, yml_content):
"""
:param yml_content: sets the branch to yml_content
:return: None
"""
branch = load(yml_content)
self._parent._data[self._branch] = branch
self._save()
def _load(self):
""" loads data from file """
if self._filename is None:
# if no file is used, just ignore this call
return
logger.debug("Loading config file %s", self._filename)
# read file from disc
with open(self._filename) as f:
self._data = load(f)
# store the modification time of this file version
self._mtime = os.path.getmtime(self._filename)
# make sure that reload timeout starts from this moment
self._lastreload = time()
# empty file gives _data=None
if self._data is None:
self._data = OrderedDict()
# update dict of the MemoryTree object
to_remove = []
# remove all obsolete entries
for name in self.__dict__:
if not name.startswith('_') and name not in self._data:
to_remove.append(name)
for name in to_remove:
self.__dict__.pop(name)
# insert the branches into the object __dict__ for auto-completion
self.__dict__.update(self._data)
def from_workspace(cls, path):
filename = os.path.join(path, '.shipmaster.yaml')
if not os.path.exists(filename):
return None
with open(filename, 'r') as file:
return cls.from_kwargs(path, **yaml.load(file, yaml.RoundTripLoader))
def test_fileformatyaml_pass_no_substitutions():
"""Relative path to file should succeed.
Strictly speaking not a unit test.
"""
context = Context({
'ok1': 'ov1',
'fileFormatYamlIn': './tests/testfiles/test.yaml',
'fileFormatYamlOut': './tests/testfiles/out/out.yaml'})
fileformat.run_step(context)
assert context, "context shouldn't be None"
assert len(context) == 3, "context should have 2 items"
assert context['ok1'] == 'ov1'
assert context['fileFormatYamlIn'] == './tests/testfiles/test.yaml'
assert context['fileFormatYamlOut'] == './tests/testfiles/out/out.yaml'
with open('./tests/testfiles/out/out.yaml') as outfile:
outcontents = yaml.load(outfile, Loader=yaml.RoundTripLoader)
assert len(outcontents) == 3
assert outcontents['key'] == 'value1 !£$%# *'
assert outcontents['key2'] == 'blah'
assert outcontents['key3'] == ['l1',
'!£$% *',
'l2',
[
'l31',
{'l32': ['l321', 'l322']}
]
]
# atrociously lazy test clean-up
os.remove('./tests/testfiles/out/out.yaml')
def load(config_path):
with config_path.open() as fp:
config = yaml.load(fp.read(), yaml.RoundTripLoader)
envs = dict(_parse_environments(config.get('environments', {})))
zones = dict(_parse_zones(config.get('zones', {})))
return envs, zones
def construct_include(self, node):
"""Include file referenced at node."""
filename = os.path.join(self._root, self.construct_scalar(node))
filename = os.path.abspath(filename)
extension = os.path.splitext(filename)[1].lstrip('.')
with open(filename, 'r') as f:
if extension in ('yaml', 'yml'):
return yaml.load(f, Loader=self)
else:
return ''.join(f.readlines())
def load(cls, path, *args, **kwargs):
name, ext = os.path.splitext(path)
try:
return cls._loaders(ext.lstrip('.'))(path, *args, **kwargs)
except KeyError:
logger.error('! Unknown loader for format %s. '
'Trying with ParmEd as fallback', ext)
return cls.from_parmed(path, *args, **kwargs)
except IOError:
raise IOError('Could not access file {}'.format(path))
def __init__(self, master=None, **kwargs):
InputContainer.__init__(self, **kwargs)
if isinstance(master, str):
raise ValueError('To instantiate from file, use .load() or '
'one of the .from_*() methods.')
self.master = master
self._path = kwargs.get('path')
def prepare_handler(cfg):
"""
Load all files into single object.
"""
positions, velocities, box = None, None, None
_path = cfg['_path']
forcefield = cfg.pop('forcefield', None)
topology = sanitize_path_for_file(cfg.pop('topology'), _path)
if 'checkpoint' in cfg:
restart_path = sanitize_path_for_file(cfg['checkpoint'], _path)
restart = Restart.load(restart_path)
positions = restart.positions
velocities = restart.velocities
box = restart.box
if 'positions' in cfg:
positions_path = sanitize_path_for_file(cfg.pop('positions'), _path)
positions = Positions.load(positions_path)
box = BoxVectors.load(positions_path)
if 'velocities' in cfg:
velocities_path = sanitize_path_for_file(cfg.pop('velocities'), _path)
velocities = Velocities.load(velocities_path)
if 'box' in cfg:
box_path = sanitize_path_for_file(cfg.pop('box'), _path)
box = BoxVectors.load(box_path)
options = {}
for key in 'positions velocities box forcefield'.split():
value = locals()[key]
if value is not None:
options[key] = value
return SystemHandler.load(topology, **options)
def statexml2pdb(topology, state, output=None):
"""
Given an OpenMM xml file containing the state of the simulation,
generate a PDB snapshot for easy visualization.
"""
state = Restart.from_xml(state)
system = SystemHandler.load(topology, positions=state.positions)
if output is None:
output = topology + '.pdb'
system.write_pdb(output)
def assertParametersConverted(self, actual, expected):
print(yaml.load(expected, Loader=yaml.RoundTripLoader)['parameters'])
print(lose_parameters_to_full(yaml.load(actual, Loader=yaml.RoundTripLoader)['parameters']))
self.assertEquals(
yaml.load(expected, Loader=yaml.RoundTripLoader),
{'parameters': lose_parameters_to_full(yaml.load(actual, Loader=yaml.RoundTripLoader)['parameters'])}
)
def load(self, vb=False):
try:
with open('config.yaml', 'r') as fp:
self.doc = yaml.load(fp, Loader=yaml.RoundTripLoader)
except IOError as e:
log.err("Could not open config.yaml: " + str(e))
except Exception as e:
log.err("An unexcpected exception of type: "
+ type(e).__name__
+ "has occurred: " + str(e))
else:
return self
def _from_json(json_string=None, filename=None,
encoding="utf-8", errors="strict", multiline=False, **kwargs):
if filename:
with open(filename, 'r', encoding=encoding, errors=errors) as f:
if multiline:
data = [json.loads(line.strip(), **kwargs) for line in f
if line.strip() and not line.strip().startswith("#")]
else:
data = json.load(f, **kwargs)
elif json_string:
data = json.loads(json_string, **kwargs)
else:
raise BoxError('from_json requires a string or filename')
return data
def _from_yaml(yaml_string=None, filename=None,
encoding="utf-8", errors="strict",
**kwargs):
if filename:
with open(filename, 'r',
encoding=encoding, errors=errors) as f:
data = yaml.load(f, **kwargs)
elif yaml_string:
data = yaml.load(yaml_string, **kwargs)
else:
raise BoxError('from_yaml requires a string or filename')
return data
# Helper functions
def from_json(cls, json_string=None, filename=None, encoding="utf-8",
errors="strict", multiline=False, **kwargs):
"""
Transform a json object string into a BoxList object. If the incoming
json is a dict, you must use Box.from_json.
:param json_string: string to pass to `json.loads`
:param filename: filename to open and pass to `json.load`
:param encoding: File encoding
:param errors: How to handle encoding errors
:param multiline: One object per line
:param kwargs: parameters to pass to `Box()` or `json.loads`
:return: BoxList object from json data
"""
bx_args = {}
for arg in kwargs.copy():
if arg in BOX_PARAMETERS:
bx_args[arg] = kwargs.pop(arg)
data = _from_json(json_string, filename=filename, encoding=encoding,
errors=errors, multiline=multiline, **kwargs)
if not isinstance(data, list):
raise BoxError('json data not returned as a list, '
'but rather a {0}'.format(type(data).__name__))
return cls(data, **bx_args)
def read_parameter(parameter_file, parameter):
fr = open(parameter_file, "r")
param = yaml.load(fr, yaml.RoundTripLoader)
return merge_two_dicts(parameter,param)
def get_and_validate_mailer_config(args):
with open(args.config) as fh:
config = yaml.load(fh.read(), Loader=yaml.SafeLoader)
jsonschema.validate(config, CONFIG_SCHEMA)
utils.setup_defaults(config)
return config
def yaml_load_fromstring(string, ordered=False):
"""
Load contents of a string into an dict/OrderedDict structure. The string has to be valid yaml
:param string: name of the yaml file to load
:type string: str
:param ordered: load to an OrderedDict? Default=False
:type ordered: bool
:return: configuration data loaded from the file (or None if an error occured)
:rtype: Dict | OrderedDict | None
"""
dict_type = 'dict'
if ordered:
dict_type = 'OrderedDict'
logger.info("Loading '{}' to '{}'".format(string, dict_type))
y = None
estr = ''
try:
sdata = string
# sdata = sdata.replace('\n', '\n\n')
if ordered:
y = _ordered_load(sdata, yaml.SafeLoader)
else:
y = yaml.load(sdata, yaml.SafeLoader)
except Exception as e:
estr = str(e)
if "found character '\\t'" in estr:
estr = estr[estr.find('line'):]
estr = 'TABs are not allowed in YAML files, use spaces for indentation instead!\nError in ' + estr
if ("while scanning a simple key" in estr) and ("could not found expected ':'" in estr):
estr = estr[estr.find('column'):estr.find('could not')]
estr = 'The colon (:) following a key has to be followed by a space. The space is missing!\nError in ' + estr
return y, estr
def load(self):
"""
load the contents of the yaml-file to the data-structure
"""
self.data = yaml_load_roundtrip(self.filename)
