我有一个带有自定义标记的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\u 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 ???
摘要
如果有人能把我推向正确的方向,我会很高兴。我甚至不知道这样做是否正确
YAML输入中所有显式标记的节点都是映射节点,因此
CommentedScalar
永远不会被映射 创建(使用此输入)由于往返模式下的ruamel.yaml已经可以加载和卸载您的yaml,所以您最好只是步行 加载的数据并检查标记属性
但是,可以加载这些映射节点,但不能像您那样使用
yield
(并且 对于复杂节点(映射、序列),只需要yield
,而不是简单节点)其中:
相关问题 更多 >
编程相关推荐