def create_entity_skeleton(self):
print('Creating Template for Entities')
template = os.path.join(self.template_dir, 'entities.yaml')
message = """# Template file for entities\n\n"""
skeleton = {}
for intent in self.assist._intent_action_funcs:
entity_map = self.assist._intent_mappings.get(intent)
action_func = self.assist._intent_action_funcs[intent][0]
args = inspect.getargspec(action_func).args
# dont add API 'sys' entities to the template
if entity_map:
args = [a for a in args if 'sys.' not in entity_map.get(a, [])]
for param in [p for p in args if p not in skeleton]:
skeleton[param] = [None, None]
with open(template, 'w') as f:
f.write(message)
f.write('#Format as below\n\n')
f.write("# entity_name:\n")
f.write("# - entry1: list of synonyms \n")
f.write("# - entry2: list of synonyms \n\n")
f.write("#For example:\n\n")
f.write("# drink:\n")
f.write("# - water: ['aqua', 'h20'] \n")
f.write("# - coffee: ['joe', 'caffeine', 'espresso', 'late'] \n")
f.write("# - soda: ['pop', 'coke']\n\n\n\n")
yaml.dump(skeleton, f, default_flow_style=False, Dumper=yaml.RoundTripDumper)
python类dump()的实例源码
def _to_yaml(obj, filename=None, default_flow_style=False,
encoding="utf-8", errors="strict",
**yaml_kwargs):
if filename:
with open(filename, 'w',
encoding=encoding, errors=errors) as f:
yaml.dump(obj, stream=f,
default_flow_style=default_flow_style,
**yaml_kwargs)
else:
return yaml.dump(obj,
default_flow_style=default_flow_style,
**yaml_kwargs)
def to_json(self, filename=None,
encoding="utf-8", errors="strict", **json_kwargs):
"""
Transform the Box object into a JSON string.
:param filename: If provided will save to file
:param encoding: File encoding
:param errors: How to handle encoding errors
:param json_kwargs: additional arguments to pass to json.dump(s)
:return: string of JSON or return of `json.dump`
"""
return _to_json(self.to_dict(), filename=filename,
encoding=encoding, errors=errors, **json_kwargs)
def yaml_save(filename, data):
"""
Save contents of an OrderedDict structure to a yaml file
:param filename: name of the yaml file to save to
:type filename: str
:param data: configuration data to to save
:type filename: str
:type data: OrderedDict
:returns: Nothing
"""
ordered = (type(data).__name__ == 'OrderedDict')
dict_type = 'dict'
if ordered:
dict_type = 'OrderedDict'
logger.info("Saving '{}' to '{}'".format(dict_type, filename))
if ordered:
sdata = _ordered_dump(data, Dumper=yaml.SafeDumper, indent=4, width=768, allow_unicode=True, default_flow_style=False)
else:
sdata = yaml.dump(data, Dumper=yaml.SafeDumper, indent=4, width=768, allow_unicode=True, default_flow_style=False)
sdata = _format_yaml_dump( sdata )
with open(filename, 'w') as outfile:
outfile.write( sdata )
# ==================================================================================
def _ordered_dump(data, stream=None, Dumper=yaml.Dumper, **kwds):
"""
Ordered yaml dumper
Use this instead ot yaml.Dumper/yaml.SaveDumper to get an Ordereddict
:param stream: stream to write to
:param Dumper: yaml-dumper to use
:**kwds: Additional keywords
:return: OrderedDict structure
"""
# usage example: ordered_dump(data, Dumper=yaml.SafeDumper)
class OrderedDumper(Dumper):
pass
def _dict_representer(dumper, data):
return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
data.items())
OrderedDumper.add_representer(OrderedDict, _dict_representer)
return yaml.dump(data, stream, OrderedDumper, **kwds)
# ==================================================================================
# Routines to handle editing of yaml files
#
def _yaml_save_roundtrip(filename, data):
"""
Dump yaml using the RoundtripDumper and correct linespacing in output file
"""
sdata = yaml.dump(data, Dumper=yaml.RoundTripDumper, version=yaml_version, indent=indent_spaces, block_seq_indent=2, width=12288, allow_unicode=True)
ldata = sdata.split('\n')
rdata = []
for index, line in enumerate(ldata):
# Fix for ruamel.yaml handling: Reinsert empty line before comment of next section
if len(line.lstrip()) > 0 and line.lstrip()[0] == '#':
indentcomment = len(line) - len(line.lstrip(' '))
indentprevline = len(ldata[index-1]) - len(ldata[index-1].lstrip(' '))
if indentprevline - indentcomment >= 2*indent_spaces:
rdata.append('')
rdata.append(line)
# Fix for ruamel.yaml handling: Remove empty line with spaces that have been inserted
elif line.strip() == '' and line != '':
if ldata[index-1] != '':
rdata.append(line)
else:
rdata.append(line)
sdata = '\n'.join(rdata)
if sdata[0] == '\n':
sdata =sdata[1:]
with open(filename+'.yaml', 'w') as outfile:
outfile.write( sdata )
def yaml_format(obj):
class MyDumper(yaml.Dumper):
def represent_mapping(self, tag, mapping, flow_style=False):
return yaml.Dumper.represent_mapping(self, tag, mapping, flow_style)
return yaml.dump(obj, Dumper=MyDumper).strip()
def yaml_format(obj):
class MyDumper(yaml.Dumper):
def represent_mapping(self, tag, mapping, flow_style=False):
return yaml.Dumper.represent_mapping(self, tag, mapping, flow_style)
return yaml.dump(obj, Dumper=MyDumper).strip()
def yaml_format(obj):
class MyDumper(yaml.Dumper):
def represent_mapping(self, tag, mapping, flow_style=False):
return yaml.Dumper.represent_mapping(self, tag, mapping, flow_style)
return yaml.dump(obj, Dumper=MyDumper).strip()
def yaml_format(obj):
class MyDumper(yaml.Dumper):
def represent_mapping(self, tag, mapping, flow_style=False):
return yaml.Dumper.represent_mapping(self, tag, mapping, flow_style)
return yaml.dump(obj, Dumper=MyDumper).strip()
def _write_config(filename, cfg, roundtrip=False):
try:
with open(filename, 'w') as f:
if roundtrip:
f.write(yaml.round_trip_dump(dict(cfg), indent=4))
else:
f.write(yaml.dump(cfg, indent=4))
except Exception as ex:
raise ConfigWriteError(filename, errors=[ex])
return cfg
def serialize_message(obj):
if isinstance(obj, Message):
serial = obj.__dict__
return serial
elif isinstance(obj, datetime.datetime):
serial = obj.isoformat()
return serial
elif isinstance(obj, Exception):
return yaml.dump(obj)
else:
raise TypeError(
"Message object is not serializable")
def write_yaml(content, fpath):
yaml.dump(content, open(fpath, 'w'), Dumper=yaml.RoundTripDumper)
def __init__(self, runDir, fileMode, hltMode = None):
self.runDir = runDir
self.runNumber = int(runDir.replace("Run", ""))
self.prettyName = "Run {0}".format(self.runNumber)
# Need to rework the qa container
#self.qaContainer = qa.qaFunctionContainer
self.mode = fileMode
self.subsystems = BTrees.OOBTree.BTree()
self.hltMode = hltMode
# Try to retrieve the HLT mode if it was not passed
runInfoFilePath = os.path.join(processingParameters["dirPrefix"], self.runDir, "runInfo.yaml")
if not hltMode:
try:
with open(runInfoFilePath, "rb") as f:
runInfo = yaml.load(f.read())
self.hltMode = runInfo["hltMode"]
except IOError as e:
# File does not exist
# HLT mode will have to be unknown
self.hltMode = "U"
# Run Information
# Since this is only information to save, only write it if the file doesn't exist
if not os.path.exists(runInfoFilePath):
runInfo = {}
# "U" for unknown
runInfo["hltMode"] = hltMode if hltMode else "U"
# Write information
if not os.path.exists(os.path.dirname(runInfoFilePath)):
os.makedirs(os.path.dirname(runInfoFilePath))
with open(runInfoFilePath, "wb") as f:
yaml.dump(runInfo, f)
def writeCustomConfig(config, filename = "config.yaml"):
if "customConfig" in config:
if os.path.exists(filename):
# Append if it already exists
mode = "ab"
else:
mode = "wb"
# Write out configuration
with open(filename, mode) as f:
yaml.dump(config["customConfig"], f, default_flow_style = False)
def dump_yaml(data, file_handle):
"""Wrapper function to nicely dump dictionaries as yaml files.
"""
assert(yaml is not None), "\nError: ruamel yaml python package not found."
yaml.dump(data, file_handle, Dumper=yaml.RoundTripDumper)
def as_yaml(self):
"""
Render the YAML node and subnodes as string.
"""
dumped = dump(self.as_marked_up(), Dumper=StrictYAMLDumper, allow_unicode=True)
return dumped if sys.version_info[0] == 3 else dumped.decode('utf8')
def start_line(self, document):
slicedpart = self._slice_segment(self._indices, document, include_selected=False)
if slicedpart is None or slicedpart == {} or slicedpart == []:
return 1
else:
return len(dump(slicedpart, Dumper=RoundTripDumper).rstrip().split('\n')) + 1
def end_line(self, document):
slicedpart = self._slice_segment(self._indices, document, include_selected=True)
return len(dump(slicedpart, Dumper=RoundTripDumper).rstrip().split('\n'))
def lines(self, document):
return "\n".join(dump(document, Dumper=RoundTripDumper).split('\n')[
self.start_line(document) - 1:self.end_line(document)
])
