python - ruamel.yaml 自定义 CommentedMapping 用于自定义标签

Question

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

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”解析为类名。我的代码如下所示。

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”，但我被困在这里并且找不到任何工作。

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

摘要
如果有人能把我推向正确的方向，我会很高兴。我什至不知道这是否是正确的方法。

Answer 1

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

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

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

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