ruamel.yaml 自定义标记的自定义 CommentedMapping
ruamel.yaml custom CommentedMapping for custom tags
我有一个带有自定义标签的 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' 解析为 class_name。我的代码看起来像这样。
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，简单节点不需要。
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
我有一个带有自定义标签的 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' 解析为 class_name。我的代码看起来像这样。
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，简单节点不需要。
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