ruamel.yaml 用于自定义标签的自定义 CommentedMapping [英] ruamel.yaml custom CommentedMapping for custom tags

问题描述

我有一个带有自定义标签的 YAML 文件，如下所示:

I've got a YAML file with custom tags that looks like this:

flow123d_version: 3.1.0
problem: !modulek.Coupling_Sequential
  description: Simple dual porosity test - steady flow, simple transport
  mesh:
    mesh_file: ../00_mesh/square_1x1_40el.msh
  flow_equation: !Flow_Darcy_MH
    nonlinear_solver:
      linear_solver: !Petsc
        a_tol: 1.0e-07

到目前为止，我的代码可以加载它并将其转储回来.我的问题是我希望能够检查每个自定义 ! 并检查其他文件是否正确.让我们看看我的文件的第二行.你可以看到我的第一个自定义标签，它由 module.class_name 组成，我需要检查它们.我想将modulek"解析为模块，将Coupling_Sequential"解析为 class_name.我的代码看起来像这样.

and so far my code can load it and dump it back. My problem is that I would like to be able to check on every custom ! and check with other files if it is correct. Let's look at second line of my file. You can see there is my first custom tag and it consist of module.class_name and i need to check on both of them. I want to parse out 'modulek' as a module and 'Coupling_Sequential' as class_name. My code looks like this.

import types
import re

import ruamel.yaml as ruml
from ruamel.yaml.comments import CommentedMap, CommentedSeq

CommentsTag = ruml.comments.Tag


class CommentedScalar:
    """
    Class to store all scalars with their tags
    """
    original_constructors = {}

    def __str__(self):
        return str(self.value)

    def __repr__(self):
        return str(self.value)

    @classmethod
    def to_yaml(cls, dumper, data):
        representer = dumper.yaml_representers[type(data.value).__mro__[0]]
        node = representer(dumper, data.value)
        if data.tag.value is None:
            tag = node.tag
        elif data.tag.value.startswith(u'tag:yaml.org,2002'):
            tag = node.tag
        else:
            tag = data.tag.value
        # print("val: ", data.value, "repr: ", node.value, "tag: ", tag)
        return dumper.represent_scalar(tag, node.value)

    def __init__(self, tag, value):
        complete_tag = tag.split('.')
        self.tag.value = tag
        self.value = value
        # self.module.value = complete_tag
        self.module = '.'.join(complete_tag[:len(complete_tag) - 1])
        self.class_name = complete_tag[-1]

    @property
    def tag(self):
        # type: () -> Any
        if not hasattr(self, CommentsTag.attrib):
            setattr(self, CommentsTag.attrib, CommentsTag())
        return getattr(self, CommentsTag.attrib)

def construct_any_tag(self, tag_suffix, node):
    if tag_suffix is None:
        orig_tag = None
    else:
        orig_tag = "!" + tag_suffix
    if isinstance(node, ruml.ScalarNode):

        implicit_tag = self.composer.resolver.resolve(ruml.ScalarNode, node.value, (True, None))
        if implicit_tag in self.yaml_constructors:
            # constructor = CommentedScalar.original_constructors[implicit_tag]
            constructor = self.yaml_constructors[implicit_tag]
        else:
            constructor = self.construct_undefined

        data = constructor(self, node)
        if isinstance(data, types.GeneratorType):
            generator = data
            data = next(generator)  # type: ignore

        scal = CommentedScalar(orig_tag, data)
        yield scal

    elif isinstance(node, ruml.SequenceNode):
        for seq in self.construct_yaml_seq(node):
            seq.yaml_set_tag(orig_tag)
            yield seq
    elif isinstance(node, ruml.MappingNode):
        for map in self.construct_yaml_map(node):
            map.yaml_set_tag(orig_tag)
            yield map
    else:
        for dummy in self.construct_undefined(node):
            yield dummy


def represent_commented_seq(cls, data):
    if data.tag.value is None:
        tag = u'tag:yaml.org,2002:seq'
    else:
        tag = data.tag.value
    return cls.represent_sequence(tag, data)

def get_yaml_serializer():
    """
    Get YAML serialization/deserialization object with proper
    configuration for conversion.
    :return: Confugured instance of ruamel.yaml.YAML.
    """
    yml = ruml.YAML(typ='rt')
    yml.indent(mapping=2, sequence=4, offset=2)
    yml.width = 120
    yml.representer.add_representer(CommentedScalar, CommentedScalar.to_yaml)
    yml.representer.add_representer(CommentedSeq, represent_commented_seq)
    yml.representer.add_representer(CommentedMap, CommentedMapping.to_yaml)
    yml.constructor.add_multi_constructor("!", construct_any_tag)
    return yml


def get_node_tag(node):
    if hasattr(node, "tag"):
        tag = node.tag.value
        if tag and len(tag) > 1 and tag[0] == '!' and tag[1] != '!':
            return tag
    return ""


def load_yaml(path):
    yml = get_yaml_serializer()
    with open(path, 'r') as stream:
        data = yml.load(stream)
    return data


def write_yaml(data, path):
    yml = get_yaml_serializer()
    with open(path, 'w')as stream:
        yml.dump(data, stream)

我正在考虑编写类似于 CommentedScalar 的CommentedMapping"，但我被困在这里，找不到任何工作.

I was thinking about writing "CommentedMapping" similarly to CommentedScalar but i got stuck here and could not find anything working.

@classmethod
def to_yaml(cls, dumper, data):
    ...
    ...
    return ???

总结
如果有人将我推向正确的方向，我会很高兴.我什至不知道这样做是否正确.

Summary
If someone would push me in right direction I'd be glad. I don't even know if this is the right way to do it.

推荐答案

YAML 输入中所有显式标记的节点都是映射节点，因此您的 CommentedScalar 永远不会创建(使用此输入).

All explicitly tagged nodes in your YAML input are mapping nodes, so your CommentedScalar is never created (with this input).

由于往返模式下的 ruamel.yaml 已经可以加载和转储您的 YAML，因此您最好步行加载的数据并检查标签属性.

Since ruamel.yaml in round-trip mode can already load and dump your YAML, you might be better of just walking the loaded data and checking for tag attributes.

但是，可以加载这些映射节点，但不能像您那样使用 yield(并且您复杂节点(映射、序列)只需要 yield 而不是简单节点.

However it is possible to load those mapping nodes, but not using yield in the way you do (and you only need yield for complex nodes (mappings, sequences) not for simple nodes).

import sys
import ruamel.yaml

yaml_str = """\
flow123d_version: 3.1.0
problem: !modulek.Coupling_Sequential
  description: Simple dual porosity test - steady flow, simple transport
  mesh:
    mesh_file: ../00_mesh/square_1x1_40el.msh
  flow_equation: !Flow_Darcy_MH
    nonlinear_solver:
      linear_solver: !Petsc
        a_tol: 1.0e-07
"""

yaml = ruamel.yaml.YAML()

@yaml.register_class
class MyMap(ruamel.yaml.comments.CommentedMap):
    def __init__(self, tag):
        ruamel.yaml.comments.CommentedMap.__init__(self)
        self._tag = tag + "@@"  # just to make clear things were changed here

    @classmethod
    def to_yaml(cls, representer, data):
        return representer.represent_mapping(data._tag, data)


def construct_any_tag(self, tag_suffix, node):
    if tag_suffix is None:
        orig_tag = None
    else:
        orig_tag = "!" + tag_suffix
    if isinstance(node, ruamel.yaml.nodes.MappingNode):
        data = MyMap(orig_tag)
        yield data
        state = ruamel.yaml.constructor.SafeConstructor.construct_mapping(self, node, deep=True)
        data.update(state)
    else:
        raise NotImplementedError


yaml = ruamel.yaml.YAML()
yaml.constructor.add_multi_constructor("!", construct_any_tag)
data = yaml.load(yaml_str)
yaml.dump(data, sys.stdout)

给出:

flow123d_version: 3.1.0
problem: !modulek.Coupling_Sequential@@
  description: Simple dual porosity test - steady flow, simple transport
  mesh:
    mesh_file: ../00_mesh/square_1x1_40el.msh
  flow_equation: !Flow_Darcy_MH@@
    nonlinear_solver:
      linear_solver: !Petsc@@
        a_tol: 1.0e-07

这篇关于ruamel.yaml 用于自定义标签的自定义 CommentedMapping的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！