试写功能代码 ymal to csv,csv to yaml
工具站https://tableconvert.com/试写功能代码ymal to csv#!/usr/bin/env python# coding=utf-8import yamlimport csvimport globimport reimport sysreload(sys)sys.setdefaultencoding('utf8')def add_info_in_rows(titles,
·
工具站
https://tableconvert.com/
试写功能代码
ymal to csv
#!/usr/bin/env python
# coding=utf-8
import yaml
import csv
import glob
import re
import sys
reload(sys)
sys.setdefaultencoding('utf8')
def add_info_in_rows(titles,instance,groupname):
rows=[]
for tag in instance["subs"]:
varr=[]
v=""
for t in titles:
if t in tag:
if tag[t]==None:
print t+":"+"None"
v="None"
else:
if type(tag[t]) in [str,unicode]:
print t+":"+tag[t]
v=tag[t]
else:
print str(type(tag[t]))+":::::"
print t+":"+str(tag[t])
v=str(tag[t])
else:
print t+":"+"None"
v="None"
if t=="groupname":
v=groupname
varr.append (v)
rows.append(varr)
return rows
def add_var_info(titles,tag,groupname,isgroup):
varr=[]
v=""
for t in titles:
if t in tag:
if tag[t]==None:
print t+":"+"None"
v="None"
else:
if type(tag[t]) in [str,unicode]:
print t+":"+tag[t]
v=tag[t]
else:
print str(type(tag[t]))+":::::"
print t+":"+str(tag[t])
v=str(tag[t])
else:
print t+":"+"None"
v="None"
if t=="groupname":
v=groupname
if t=="isgroup":
v=isgroup
varr.append (v)
return varr
def add_vars_info_in_rows(titles,instance,groupname,isgroup):
rows=[]
for tag in instance["vars"]:
rows.append(add_var_info(titles,tag,groupname,isgroup))
return rows
def save_csv(filename):
with open(filename, 'w') as out:
csv_writer = csv.writer(out)
csv_writer.writerow(usedtitle)
csv_writer.writerows(rows_to_write)
print("Output file {} created".format(filename))
def to_table(filename='test', rept=','):
'''将符号分割文本转化为markdown表格形式
parameter:需要转化的文件名(str),文本的分割符号(str)
'''
file_obj_r = open(filename, 'r')
file_obj_w = open(filename+"mk.txt", 'w')
count = 0 # 哨兵,实现文件的第二行另做处理
fd=file_obj_r.read()
fd=fd.replace("|","\|")#表格分隔符号 转义
p= re.compile('.+?"(.+?)"')#提取“”内部内容,可能存在','符号,提取出来不做处理
rplist=p.findall(fd)
rplistflag=[]
for i in range(len(rplist)):
rplistflag.append("$zzz"+str(i)+"zzz$")
fd=fd.replace(rplist[i],rplistflag[i])
dlist_file_obj_r=fd.split('\n')
for line in dlist_file_obj_r:
rec = line.split(rept)
new_rec = [i.strip() for i in rec]
new_rec = '|'+ '|'.join(new_rec) +'|'+'\n'
for i in range(len(rplist)):#效率差点,凑合
new_rec=new_rec.replace(rplistflag[i],rplist[i])
file_obj_w.write(new_rec)
if count == 0:
head_rec = '|'
for i in rec:
head_rec += '---|'
head_rec += '\n'
file_obj_w.write(head_rec)
count += 1
file_obj_r.close()
file_obj_w.close()
#yaml_file_names = glob.glob('./config/base_setup.yaml')
#yaml_file_names = glob.glob('./config/sensing.yaml')
yaml_file_names = glob.glob('./config/bas_config/*.yaml')
yaml_file_names = yaml_file_names +glob.glob('./config/ses_config/*.yaml')
yaml_file_names = yaml_file_names +glob.glob('./config/map_config/*.yaml')
yaml_file_names = yaml_file_names +glob.glob('./config/trk_config/*.yaml')
yaml_file_names = yaml_file_names +glob.glob('./config/nav_config/*.yaml')
rows_to_write = []
titles=["name","namecn","desc","probe","run","runat","param","gui"]
titles2=["name","namecn","desc","probe","run","runat","param","gui","groupname"]
usedtitle=titles
'''
for i, each_yaml_file in enumerate(yaml_file_names):
print("Processing file {} of {} file name: {}".format(
i+1, len(yaml_file_names),each_yaml_file))
rows_to_write = []
with open(each_yaml_file) as file:
data = yaml.safe_load(file)
for instance in data["subs"]:
print instance["name"]
print instance["desc"]
values=dict()
isnotgroup=1
for tag in instance["subs"]:
isnotgroup= "run" in tag
if isnotgroup:
rows_to_write.extend(add_info_in_rows(titles,instance,""))
break
else:
usedtitle=titles2
rows_to_write.extend(add_info_in_rows(titles2,tag,instance["name"]+">"+tag["name"]))
print usedtitle
print rows_to_write
save_csv(each_yaml_file+".csv")
to_table(each_yaml_file+".csv")
'''
varstitle=["name","desc","label","flags","kind","v","cmd_param","choices","descs","choices_type","min","max","checkselfparam","script","groupname","isgroup","topic","msg"]
'''
- name : waypoint_follower
topic : /config/waypoint_follower
msg : ConfigWaypointFollower
vars :
-name : port_name
desc : 设备名
label : '设备名:'
flags : [disable]
kind : str
v : /dev/dev_chassis
choices: [ 'false', 'true' ]
descs : [ 'false', 'true' ]
choices_type: str
cmd_param :
min : 0
max : 5.0
'''
allvars=[]
for i, each_yaml_file in enumerate(yaml_file_names):
print("Processing file {} of {} file name: {}".format(
i+1, len(yaml_file_names),each_yaml_file))
#rows_to_write = []
with open(each_yaml_file) as file:
data = yaml.safe_load(file)
addgroup=1
for instance in data["params"]:
print (instance["name"],"-------------------------")
groupname=instance["name"]
#去重复
idx_gn=varstitle.index("name")
idx_isg=varstitle.index("isgroup")
#print(varstitle.index("groupname"))
isin=0
for row in allvars:
if row[idx_gn]==groupname and row[idx_isg]:
isin=1
break
if isin:
print ("repeat!!!!-------------------------",idx_gn,idx_isg)
continue
varslist=add_var_info(varstitle,instance,"",True)
varslist=[varslist]+add_vars_info_in_rows(varstitle,instance,groupname,False)
allvars=allvars+varslist
print varslist
usedtitle=varstitle
rows_to_write=allvars
save_csv("./zzzallparamsvars.csv")
csv to yaml
varstitle=["name","desc","label","flags","kind","v","cmd_param","choices","descs","choices_type","min","max","groupname","isgroup","topic","msg"]
'''
- name : waypoint_follower
topic : /config/waypoint_follower
msg : ConfigWaypointFollower
vars :
-name : port_name
desc : 设备名
label : '设备名:'
flags : [disable]
kind : str
v : /dev/dev_chassis
choices: [ 'false', 'true' ]
descs : [ 'false', 'true' ]
choices_type: str
cmd_param :
min : 0
max : 5.0
'''
csv_file_names = glob.glob('./zzzallparamsvars.csv')
allvars=[]
for i, each_csv_file in enumerate(csv_file_names):
print("Processing file {} of {} file name: {}".format(
i+1, len(csv_file_names),each_csv_file))
#rows_to_write = []
with open(each_csv_file) as csv_file:
all_lines=csv.reader(csv_file)
list_file=[]
for one_line in all_lines:
list_file.append(one_line)
#print one_line
varstitle=list_file[0]
idx_gn=varstitle.index("name")
idx_isg=varstitle.index("isgroup")
idx_gn_p=varstitle.index("groupname")
idx_kind=varstitle.index("kind")
paramslist=[]
for i in range (1,len(list_file)):
row=list_file[i]
#判断分组行
if row[idx_isg]=='True':
#创建组,组信息赋值
obj_grp={"vars":[]}
for idx in range (len(varstitle)):
if varstitle[idx]=="groupname" or varstitle[idx]=="isgroup":
continue
if row[idx]!="None":
obj_grp[varstitle[idx]]=row[idx]
#存入参数表
paramslist.append(obj_grp)
else:
obj_var={}
groupname=row[idx_gn_p]
#在参数表查找groupname对应的obj_grp
obj_grp=None
for param in paramslist:
if param["name"]==groupname:
obj_grp=param
break
if obj_grp is None:
print "eeeeeeee----------"
continue
#var 赋值
for idx in range (len(varstitle)):
if varstitle[idx]=="groupname" or varstitle[idx]=="isgroup":
continue
if row[idx]!="None":
v=row[idx]
if varstitle[idx] in ["cmd_param","checkselfparam","choices","min","max","v"] :
if v !="map" and row[idx_kind]!="str" or varstitle[idx]=="cmd_param":
#map 是python内置对象, kind为str 跳过,cmd_param 要转换
try :
v=eval(v)
except:
pass
obj_var[varstitle[idx]]=v
#放入对应分组
obj_grp["vars"].append(obj_var)
print paramslist
paramObj={"params" : paramslist}
print yaml.dump(paramObj)
file = open("./zzzallparamsvars.csv.yaml", 'w')
# safe_dump中文字符不会输出!!python/str强制类型
yaml.safe_dump(paramObj, file, default_flow_style=False, encoding='utf-8', allow_unicode=True)
file.close()
更多推荐
已为社区贡献7条内容
所有评论(0)