def load_config_spec(config_spec, config_sections, repl_vars, language):
config_split = config_spec.strip().split(':')
config_path = config_split[0]
if len(config_split) > 1:
config_sections = config_split[1].split('|')
with open(config_path) as config_file:
all_config_data = yaml.load(config_file, Loader=yaml.Loader)
# Make a list of the appropriate configuration sections (just the ones
# we are actually using) from the YAML file.
segments = [all_config_data[i] for i in config_sections]
segments.append(all_config_data.get(language, {}))
# Merge all of the segments of data into a single config dictionary.
config = merge(*segments)
# Perform final replacements.
return replace_vars(config, repl_vars)
python类Loader()的实例源码
def _ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
"""
Ordered yaml loader
Use this instead ot yaml.loader/yaml.saveloader to get an Ordereddict
:param stream: stream to read from
:param Loader: yaml-loader to use
:object_pairs_hook: ...
:return: OrderedDict structure
"""
# usage example: ordered_load(stream, yaml.SafeLoader)
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)
return yaml.load(stream, OrderedLoader)
def yaml_load_roundtrip(filename):
"""
Load contents of a yaml file into an dict structure for editing (using Roundtrip Loader)
:param filename: name of the yaml file to load
:return: data structure loaded from file
"""
if not EDITING_ENABLED:
return None
y = None
try:
with open(filename+YAML_FILE, 'r') as stream:
sdata = stream.read()
sdata = sdata.replace('\n', '\n\n')
y = yaml.load(sdata, yaml.RoundTripLoader)
except Exception as e:
logger.error("yaml_load_roundtrip: YAML-file load error: '%s'" % (e))
y = {}
return y
def import_cwl(self, cwl_path):
"""
Load content of cwl into the :class:`cwlgen.CommandLineTool` object.
:param cwl_path: Path of the CWL tool to be loaded.
:type cwl_path: STRING
:return: CWL tool content in cwlgen model.
:rtype: :class:`cwlgen.CommandLineTool`
"""
with open(cwl_path) as yaml_file:
cwl_dict = ryaml.load(yaml_file, Loader=ryaml.Loader)
tool = self._init_tool(cwl_dict)
for key, element in cwl_dict.items():
try:
getattr(self, '_load_{}'.format(key))(tool, element)
except AttributeError:
logger.warning(key + " content is not processed (yet).")
return tool
def read_user_config(flags):
"""Read the user config from disk and return it.
Args:
flags (argparse.Namespace): The flags from sys.argv.
Returns:
dict: The user config.
"""
# Load the user configuration if it exists and save a dictionary.
user_config = {}
user_config_file = os.path.realpath(os.path.expanduser(flags.user_config))
if os.path.isfile(user_config_file):
with io.open(user_config_file) as ucf:
user_config = yaml.load(ucf.read(), Loader=yaml.Loader) or {}
# Sanity check: Is there a configuration? If not, abort.
if not user_config:
setup_logging(INFO)
logger.critical('No user configuration found.')
logger.warn('This is probably your first time running Artman.')
logger.warn('Run `configure-artman` to get yourself set up.')
sys.exit(64)
# Done; return the user config.
return user_config
def execute(self, gapic_code_dir, grpc_code_dir, proto_code_dir, gapic_api_yaml):
with open(gapic_api_yaml[0]) as f:
gapic_config = yaml.load(f, Loader=yaml.Loader)
package_name = gapic_config.get('language_settings').get('csharp').get('package_name')
package_root = '{0}/{1}'.format(gapic_code_dir, package_name)
prod_dir = '{0}/{1}'.format(package_root, package_name)
# Copy proto/grpc .cs files into prod directory
self.exec_command(['sh', '-c', 'cp {0}/*.cs {1}'.format(proto_code_dir, prod_dir)])
self.exec_command(['sh', '-c', 'cp {0}/*.cs {1}'.format(grpc_code_dir, prod_dir)])
def execute(self, src_proto_path, import_proto_path, common_protos_yaml,
organization_name):
self._organization_name = organization_name
with io.open(common_protos_yaml) as file_:
common_protos_data = yaml.load(file_, Loader=yaml.Loader)
# Treat google.protobuf, google.iam as a common proto package, even
# though they are not included in the common-protos we generate.
#
# TODO (geigerj): remove 'google.iam' when it is included in the common
# protos package.
common_protos = ['google.protobuf', 'google.iam']
for package in common_protos_data['packages']:
common_protos.append('google.' + package['name'].replace('/', '.'))
tmpdir = os.path.join(
tempfile.gettempdir(), 'artman-python', str(int(time.time())))
new_proto_dir = os.path.join(tmpdir, 'proto')
new_src_path = set()
new_import_path = [new_proto_dir]
self._copy_and_transform_directories(
src_proto_path, new_proto_dir, common_protos, paths=new_src_path)
self._copy_and_transform_directories(
import_proto_path, new_proto_dir, common_protos)
# Update src_proto_path, import_proto_path
return list(new_src_path), new_import_path
def load_config(config_path, loader=yaml.Loader, verify_version=True):
if not os.path.exists(config_path):
system_log.error(_("config.yml not found in {config_path}").format(config_path))
return False
with codecs.open(config_path, encoding="utf-8") as stream:
config = yaml.load(stream, loader)
if verify_version:
config = config_version_verify(config, config_path)
return config
def load_config(config_path, loader=yaml.Loader):
if config_path is None:
return {}
if not os.path.exists(config_path):
system_log.error(_(u"config.yml not found in {config_path}").format(config_path))
return False
with codecs.open(config_path, encoding="utf-8") as stream:
config = yaml.load(stream, loader)
return config
def load_mod_config(config_path, loader=yaml.Loader):
mod_config = load_config(config_path, loader)
if mod_config is None or "mod" not in mod_config:
import os
os.remove(config_path)
config_path = get_mod_config_path()
return load_mod_config(config_path, loader)
else:
return mod_config
def __init__(self, stream, version=None, **kwargs):
"""Initialise Loader."""
try:
self._root = os.path.dirname(stream.name)
except AttributeError:
self._root = os.path.curdir
yaml.Loader.__init__(self, stream, version=None, **kwargs)
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 prepare_input(argv=None):
"""
Get, parse and prepare input file.
"""
p = ArgumentParser(description='InsiliChem Ommprotocol: '
'easy to deploy MD protocols for OpenMM')
p.add_argument('input', metavar='INPUT FILE', type=extant_file,
help='YAML input file')
p.add_argument('--version', action='version', version='%(prog)s v{}'.format(__version__))
p.add_argument('-c', '--check', action='store_true',
help='Validate input file only')
args = p.parse_args(argv if argv else sys.argv[1:])
# Load config file
with open(args.input) as f:
cfg = yaml.load(f, Loader=YamlLoader)
# Paths and dirs
cfg['_path'] = os.path.abspath(args.input)
cfg['system_options'] = prepare_system_options(cfg)
cfg['outputpath'] = sanitize_path_for_file(cfg.get('outputpath', '.'), args.input)
if not args.check:
with ignored_exceptions(OSError):
os.makedirs(cfg['outputpath'])
handler = prepare_handler(cfg)
return handler, cfg, args
def _create_config(self, api_name, api_version, api_full_name, output_dir,
package_dependencies_yaml, package_defaults_yaml, proto_deps,
proto_test_deps, language, local_paths, src_proto_path, package_type,
gapic_api_yaml, release_level=None,packaging='single-artifact',
generated_package_version=None):
googleapis_dir = local_paths['googleapis']
googleapis_path = os.path.commonprefix(
[os.path.relpath(p, googleapis_dir) for p in src_proto_path])
with open(package_dependencies_yaml) as dep_file:
package_dependencies = yaml.load(dep_file, Loader=yaml.Loader)
with open(package_defaults_yaml) as dep_file:
package_defaults = yaml.load(dep_file, Loader=yaml.Loader)
if release_level is not None:
package_defaults['release_level'][language] = (
release_level)
# Apply package version and development status overrides if specified
# in the artman config
if generated_package_version is not None:
release_version_type = package_defaults['release_level'][language]
if release_version_type != 'ga':
package_defaults['generated_package_version'][language] = (
generated_package_version)
else:
package_defaults['generated_ga_package_version'][language] = (
generated_package_version)
gapic_config_name = ''
if len(gapic_api_yaml) > 0:
gapic_config_name = os.path.basename(gapic_api_yaml[0])
dependency_type = 'local'
if packaging == 'single-artifact':
dependency_type = 'release'
config = {
'short_name': api_name,
'major_version': api_version,
'proto_path': googleapis_path,
'package_name': {
'default': api_full_name,
},
'proto_deps': proto_deps,
'package_type': package_type,
'dependency_type': dependency_type,
'gapic_config_name': gapic_config_name,
}
if proto_test_deps:
config['proto_test_deps'] = proto_test_deps
config.update(package_dependencies)
config.update(package_defaults)
return config
# Separated so that this can be mocked for testing
