Python ruamel.yaml 模块，RoundTripLoader() 实例源码

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={})

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')