问题描述
我有一个带有自定义标签的 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 nevercreated (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 walkingthe 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 youonly 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的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!