def entry_point():
import six
from rqalpha.mod import SYSTEM_MOD_LIST
from rqalpha.utils.config import get_mod_config_path, load_mod_config
from rqalpha.utils.package_helper import import_mod
mod_config_path = get_mod_config_path()
mod_config = load_mod_config(mod_config_path, loader=yaml.RoundTripLoader)
for mod_name, config in six.iteritems(mod_config['mod']):
lib_name = "rqalpha_mod_{}".format(mod_name)
if not config['enabled']:
continue
if mod_name in SYSTEM_MOD_LIST:
# inject system mod
import_mod("rqalpha.mod." + lib_name)
else:
# inject third part mod
import_mod(lib_name)
cmd_cli(obj={})
python类RoundTripLoader()的实例源码
def generate_cwl_documentation(_):
cur_dir = os.path.abspath(os.path.dirname(__file__))
# find all cwl files
with WorkflowGenerator() as wf:
cwl_files = [step.run for step in wf.steps_library.steps.values()]
# sort alphabetically
cwl_files.sort()
tools_file = os.path.join(cur_dir, 'tools.rst')
tool_template = '\n{}\n{}\n\n{}\n'
with codecs.open(tools_file, 'wb', encoding='utf-8') as f:
f.write('Tools\n=====\n')
f.write('\n``nlppln`` contains the following tools:\n')
for cwl in cwl_files:
tool_name = os.path.basename(cwl)
plusses = '+'*len(tool_name)
with codecs.open(cwl) as c:
try:
cwl_yaml = yaml.load(c, Loader=yaml.RoundTripLoader)
doc = cwl_yaml.get('doc', 'No documentation')
f.write(tool_template.format(tool_name, plusses, doc))
except yaml.YAMLError:
pass
def from_yaml():
""" Load configuration from yaml source(s), cached to only run once """
default_yaml_str = snippets.get_snippet_content('hatchery.yml')
ret = yaml.load(default_yaml_str, Loader=yaml.RoundTripLoader)
for config_path in CONFIG_LOCATIONS:
config_path = os.path.expanduser(config_path)
if os.path.isfile(config_path):
with open(config_path) as config_file:
config_dict = yaml.load(config_file, Loader=yaml.RoundTripLoader)
if config_dict is None:
continue
for k, v in config_dict.items():
if k not in ret.keys():
raise ConfigError(
'found garbage key "{}" in {}'.format(k, config_path)
)
ret[k] = v
return ret
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 yaml_save(filename, data):
"""
***Converter Special ***
Save contents of an OrderedDict structure to a yaml file
:param filename: name of the yaml file to save to
:param data: OrderedDict to save
"""
sdata = convert_yaml(data)
print(", saving to '{}'".format(os.path.basename(filename)+'.yaml'))
if store_raw_output == True:
with open(filename+'_raw.yaml', 'w') as outfile:
outfile.write( sdata )
# Test if roundtrip gives the same result
data = yaml.load(sdata, yaml.RoundTripLoader)
_yaml_save_roundtrip(filename, data)
def uninstall(params):
"""
Uninstall third-party Mod
"""
from pip import main as pip_main
from pip.commands.uninstall import UninstallCommand
params = [param for param in params]
options, mod_list = UninstallCommand().parse_args(params)
params = ["uninstall"] + params
for mod_name in mod_list:
mod_name_index = params.index(mod_name)
if mod_name in system_mod:
print('System Mod can not be installed or uninstalled')
return
if "rqalpha_mod_" in mod_name:
lib_name = mod_name
mod_name = lib_name.replace("rqalpha_mod_", "")
else:
lib_name = "rqalpha_mod_" + mod_name
params[mod_name_index] = lib_name
# Uninstall Mod
pip_main(params)
# Remove Mod Config
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
for mod_name in mod_list:
if "rqalpha_mod_" in mod_name:
mod_name = mod_name.replace("rqalpha_mod_", "")
del config['mod'][mod_name]
dump_config(config_path, config)
def list():
"""
List all mod configuration
"""
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
print(yaml.dump(config['mod'], Dumper=yaml.RoundTripDumper))
def enable(mod_name):
"""
enable mod
"""
if mod_name not in system_mod and "rqalpha_mod_" in mod_name:
mod_name = mod_name.replace("rqalpha_mod_", "")
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
try:
config['mod'][mod_name]['enabled'] = True
dump_config(config_path, config)
except Exception as e:
pass
def disable(mod_name):
"""
disable mod
"""
if mod_name not in system_mod and "rqalpha_mod_" in mod_name:
mod_name = mod_name.replace("rqalpha_mod_", "")
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
try:
config['mod'][mod_name]['enabled'] = False
dump_config(config_path, config)
except Exception as e:
pass
def __init__(self, filename):
self.filename = filename
with open(filename, 'r') as f:
self.data = yaml.load(f, Loader=yaml.RoundTripLoader)
def _init():
parallelism = config.get('parallelism', 1)
logger.info('Using parallelism: %d', parallelism)
for sandbox in await create_sandboxes(parallelism):
_sandbox_pool.put_nowait(sandbox)
try:
with open(_LANGS_FILE) as file:
langs_config = yaml.load(file, Loader=yaml.RoundTripLoader)
except FileNotFoundError:
logger.error('Language file %s not found.', _LANGS_FILE)
exit(1)
for lang_name, lang_config in langs_config.items():
if lang_config['type'] == 'compiler':
compiler = Compiler(lang_config['compiler_file'],
shlex.split(lang_config['compiler_args']),
lang_config['code_file'],
lang_config['execute_file'],
shlex.split(lang_config['execute_args']))
_langs[lang_name] = partial(
_compiler_build, compiler,
time_limit_ns=lang_config.get('time_limit_ms', DEFAULT_TIME_MS) * 1000000,
memory_limit_bytes=lang_config.get('memory_limit_kb', DEFAULT_MEM_KB) * 1024,
process_limit=lang_config.get('process_limit', PROCESS_LIMIT))
elif lang_config['type'] == 'interpreter':
interpreter = Interpreter(lang_config['code_file'],
lang_config['execute_file'],
shlex.split(lang_config['execute_args']))
_langs[lang_name] = partial(_interpreter_build, interpreter)
else:
logger.error('Unknown type %s', lang_config['type'])
def _load_config():
try:
with open(_CONFIG_FILE, encoding='utf-8') as file:
return yaml.load(file, Loader=yaml.RoundTripLoader)
except FileNotFoundError:
logger.error('Config file %s not found.', _CONFIG_FILE)
exit(1)
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 from_string(cls, src):
return cls.from_kwargs(None, **yaml.safe_load(src, 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 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 read_parameter(parameter_file, parameter):
fr = open(parameter_file, "r")
param = yaml.load(fr, yaml.RoundTripLoader)
return merge_two_dicts(parameter,param)
def read_yaml(fpath):
return yaml.load(open(fpath).read(), Loader=yaml.RoundTripLoader, preserve_quotes=True) or {}
def load_yaml(file_handle):
"""Wrapper function to load yaml files.
"""
assert(yaml is not None), "\nError: ruamel yaml python package not found."
return yaml.load(file_handle, Loader=yaml.RoundTripLoader)
def install(params):
"""
Install third-party Mod
"""
from pip import main as pip_main
from pip.commands.install import InstallCommand
params = [param for param in params]
options, mod_list = InstallCommand().parse_args(params)
params = ["install"] + params
for mod_name in mod_list:
mod_name_index = params.index(mod_name)
if mod_name in system_mod:
print('System Mod can not be installed or uninstalled')
return
if "rqalpha_mod_" in mod_name:
lib_name = mod_name
mod_name = lib_name.replace("rqalpha_mod_", "")
else:
lib_name = "rqalpha_mod_" + mod_name
params[mod_name_index] = lib_name
# Install Mod
pip_main(params)
# Export config
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
for mod_name in mod_list:
if "rqalpha_mod_" in mod_name:
lib_name = mod_name
mod_name = lib_name.replace("rqalpha_mod_", "")
else:
lib_name = "rqalpha_mod_" + mod_name
mod = import_module(lib_name)
mod_config = yaml.load(mod.__mod_config__, yaml.RoundTripLoader)
config['mod'][mod_name] = mod_config
config['mod'][mod_name]['lib'] = lib_name
config['mod'][mod_name]['enabled'] = False
config['mod'][mod_name]['priority'] = 1000
dump_config(config_path, config)
def run_step(context):
"""Parses input yaml file and substitutes {tokens} from context.
Loads yaml into memory to do parsing, so be aware of big files.
Args:
context: pypyr.context.Context. Mandatory.
The following context keys expected:
- fileFormatYamlIn. mandatory. path-like.
Path to source file on disk.
- fileFormatYamlOut. mandatory. path-like. Write output file to
here. Will create directories in path for you.
Returns:
None.
Raises:
FileNotFoundError: take a guess
pypyr.errors.KeyNotInContextError: fileFormatYamlIn or
fileFormatYamlOut missing in context.
pypyr.errors.KeyInContextHasNoValueError: fileFormatYamlIn or
fileFormatYamlOut exists but is None.
"""
logger.debug("started")
context.assert_keys_have_values(__name__,
'fileFormatYamlIn',
'fileFormatYamlOut')
in_path = context.get_formatted('fileFormatYamlIn')
out_path = context.get_formatted('fileFormatYamlOut')
logger.debug(f"opening yaml source file: {in_path}")
with open(in_path) as infile:
payload = yaml.load(infile, Loader=yaml.RoundTripLoader)
logger.debug(f"opening destination file for writing: {out_path}")
os.makedirs(os.path.abspath(os.path.dirname(out_path)), exist_ok=True)
with open(out_path, 'w') as outfile:
formatted_iterable = context.get_formatted_iterable(payload)
yaml.dump(formatted_iterable,
outfile,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
width=50)
logger.info(
f"Read {in_path} yaml, formatted contents and wrote to {out_path}")
logger.debug("done")
def test_fileformatyaml_pass_with_substitutions():
"""Relative path to file should succeed.
Strictly speaking not a unit test.
"""
context = Context({
'k1': 'v1',
'k2': 'v2',
'k3': 'v3',
'k4': 'v4',
'k5': 'v5',
'fileFormatYamlIn': './tests/testfiles/testsubst.yaml',
'fileFormatYamlOut': './tests/testfiles/out/outsubst.yaml'})
fileformat.run_step(context)
assert context, "context shouldn't be None"
assert len(context) == 7, "context should have 7 items"
assert context['k1'] == 'v1'
assert context['fileFormatYamlIn'] == './tests/testfiles/testsubst.yaml'
assert context['fileFormatYamlOut'] == ('./tests/testfiles/out/'
'outsubst.yaml')
with open('./tests/testfiles/out/outsubst.yaml') as outfile:
outcontents = yaml.load(outfile, Loader=yaml.RoundTripLoader)
expected = {
'key': 'v1value1 !£$%# *',
'key2v2': 'blah',
# there is a comment here
'key3': [
'l1',
# and another
'!£$% * v3',
'l2', ['l31v4',
{'l32': ['l321',
'l322v5']
}
]
]
}
assert outcontents == expected
# atrociously lazy test clean-up
os.remove('./tests/testfiles/out/outsubst.yaml')
def test_fileformatyaml_pass_with_path_substitutions():
"""Relative path to file should succeed with path subsitutions.
Strictly speaking not a unit test.
"""
context = Context({
'k1': 'v1',
'k2': 'v2',
'k3': 'v3',
'k4': 'v4',
'k5': 'v5',
'pathIn': 'testsubst',
'pathOut': 'outsubst',
'fileFormatYamlIn': './tests/testfiles/{pathIn}.yaml',
'fileFormatYamlOut': './tests/testfiles/out/{pathOut}.yaml'})
fileformat.run_step(context)
assert context, "context shouldn't be None"
assert len(context) == 9, "context should have 9 items"
assert context['k1'] == 'v1'
assert context['fileFormatYamlIn'] == './tests/testfiles/{pathIn}.yaml'
assert context['fileFormatYamlOut'] == ('./tests/testfiles/out/'
'{pathOut}.yaml')
with open('./tests/testfiles/out/outsubst.yaml') as outfile:
outcontents = yaml.load(outfile, Loader=yaml.RoundTripLoader)
expected = {
'key': 'v1value1 !£$%# *',
'key2v2': 'blah',
# there is a comment here
'key3': [
'l1',
# and another
'!£$% * v3',
'l2', ['l31v4',
{'l32': ['l321',
'l322v5']
}
]
]
}
assert outcontents == expected
# atrociously lazy test clean-up
os.remove('./tests/testfiles/out/outsubst.yaml')
