У меня есть 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)
Я думал о написании "CommentedMapping" аналогично CommentedScalar, но я застрял здесь и не смог найти ничего работающего.
@classmethod
def to_yaml(cls, dumper, data):
...
...
return ???
Резюме
Если бы кто-нибудь направил меня в правильном направлении, я был бы рад. Я даже не знаю, является ли это правильным способом сделать это.