feat: working version

This commit is contained in:
Ferrari Martino Giordano
2024-03-08 17:18:38 +01:00
parent e964b7347f
commit 65e399fa70
6 changed files with 439 additions and 165 deletions
+1
View File
@@ -1 +1,2 @@
output/
__pycache__
+1
View File
@@ -0,0 +1 @@
from autodoc.autodoc import *
+59 -45
View File
@@ -4,62 +4,76 @@ Main entry point for the MARTe Autodoc project"
Author: Martino Ferrari
Email: martinogiordano.ferrari@iter.org
"""
from argparse import ArgumentParser
import json
import logging
import sys
import os
import pprint
import sys
from argparse import ArgumentParser
from MARTe.StandardParser import StandardParser
from MARTe.StandardParser import ParseException
import json
from MARTe.StandardParser import ParseException, StandardParser
import autodoc
__version__ = "0.0.0"
def __args__():
"""Parse arguments."""
parser = ArgumentParser(
prog="autodoc",
description="MARTe configuration auto documentation tool",
)
parser.add_argument("filename", help="Configuration file to document", type=str)
parser.add_argument("--type", choices=["json", "marte"], default="marte", help="Configuration file type (default: `marte`)")
parser.add_argument("--loglevel", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], default="ERROR", help="Logging level (default: `ERROR`)")
parser.add_argument("--version", action="version", version=__version__)
return parser.parse_args()
"""Parse arguments."""
parser = ArgumentParser(
prog="autodoc",
description="MARTe configuration auto documentation tool",
)
parser.add_argument("filename", help="Configuration file to document", type=str)
parser.add_argument(
"--type",
choices=["json", "marte"],
default="marte",
help="Configuration file type (default: `marte`)",
)
parser.add_argument(
"--loglevel",
choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
default="ERROR",
help="Logging level (default: `ERROR`)",
)
parser.add_argument("--version", action="version", version=__version__)
parser.add_argument(
"-d", "--destination", help="Destination folder", type=str, default="."
)
return parser.parse_args()
def main():
"""Main function."""
args = __args__()
logging.basicConfig(level=args.loglevel)
logging.info(f"AutoDoc v{__version__}")
if not os.path.exists(args.filename):
logging.error(f"Specified file `{args.filename}` does not exists")
sys.exit(1)
config = {}
if args.type == 'marte':
parser = StandardParser()
try:
config = parser.parse_file(args.filename)
except ParseException as e:
logging.error(f"Impossible to parse file `{args.filename}` at position ({e.line},{e.col}): {e.msg}`")
sys.exit(1)
except Exception as e:
logging.error(f"Sometihng went wrong with file `{args.filename}`")
sys.exit(1)
else:
with open(args.filename, "r") as file:
config = json.load(file)
if not config:
logging.error(f"Impossible to load configuration `{args.filename}`")
sys.exit(1)
print(f"# `{os.path.splitext(os.path.basename(args.filename))[0]}` Documentation")
for label, obj in config.items():
if 'Class' in obj:
autodoc.document(label, obj)
"""Main function."""
args = __args__()
logging.basicConfig(level=args.loglevel)
logging.info(f"AutoDoc v{__version__}")
if not os.path.exists(args.filename):
logging.error(f"Specified file `{args.filename}` does not exists")
sys.exit(1)
config = {}
if args.type == "marte":
parser = StandardParser()
try:
config = parser.parse_file(args.filename)
except ParseException as e:
logging.error(
f"Impossible to parse file `{args.filename}` at position ({e.line},{e.col}): {e.msg}`"
)
sys.exit(1)
except Exception as e:
logging.error(f"Sometihng went wrong with file `{args.filename}`")
sys.exit(1)
else:
with open(args.filename, "r") as file:
config = json.load(file)
if not config:
logging.error(f"Impossible to load configuration `{args.filename}`")
sys.exit(1)
filename = os.path.splitext(os.path.basename(args.filename))[0]
autodoc.document(filename, args, config)
if __name__ == '__main__':
main()
if __name__ == "__main__":
main()
Binary file not shown.
Binary file not shown.
+378 -120
View File
@@ -2,134 +2,392 @@
Autodoc functions
"""
import logging
import os
import sys
__PATH__ = "."
def __doc_app_state__(state, tab=''):
if '+Threads' in state:
for i, tname in enumerate(state['+Threads']):
if tname[0] == '+':
ch = '├─●' if i < len(state['+Threads']) - 1 else '╰─●'
print(f'{tab} {ch} {tname[1:]}')
def __sig__(obj, label, type):
return (obj, label, type)
def __doc_state__(label, state, app, full=False):
if '+Threads' in state:
for tname, obj in state['+Threads'].items():
if tname[0] == '+':
print()
print(f"#### Thread `{tname[1:]}`\n")
print(f'Visual representation of the thread `{tname[1:]}`:\n')
print('```graphviz')
print('digraph structs {')
print(' rankdir=LR;')
print(' node [shape=record];')
datasources = {}
if 'Functions' in obj:
fns = obj['Functions']
for fn in fns:
obj = app['+Functions'][f'+{fn}']
node = f' fn{fn} [label="{fn}&#92;n{obj["Class"]}'
if full:
node += ' | '
outputs = ''
inputs = ''
if "InputSignals" in obj:
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
if sig['DataSource'] not in datasources:
datasources[sig['DataSource']] = {}
sn = sig['Alias'] if 'Alias' in sig else l
if sn not in datasources[sig['DataSource']]:
datasources[sig['DataSource']][sn] = [f'fn{fn}:in{i}']
else:
datasources[sig['DataSource']][sn].append(f'fn{fn}:in{i}')
def __edge__(src, trg):
return (src, trg)
if i != 0:
inputs += ' | '
inputs += f'<in{i}> {l} ({sig["Type"]})'
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
if sig['DataSource'] not in datasources:
datasources[sig['DataSource']] = {}
if l not in datasources[sig['DataSource']]:
datasources[sig['DataSource']][l] = [f'fn{fn}:out{i}']
else:
datasources[sig['DataSource']][l].append(f'fn{fn}:out{i}')
if i != 0:
outputs += ' | '
outputs += f'<out{i}> {l} ({sig["Type"]})'
node += '{ { '+inputs+' } | { '+ outputs +' } }"];'
else:
node += '"];'
if "InputSignals" in obj:
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
if sig['DataSource'] not in datasources:
datasources[sig['DataSource']] = []
if (fn, 'in') not in datasources[sig['DataSource']]:
datasources[sig['DataSource']].append((fn, 'in'))
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
if sig['DataSource'] not in datasources:
datasources[sig['DataSource']] = []
if (fn, 'out') not in datasources[sig['DataSource']]:
datasources[sig['DataSource']].append((fn, 'out'))
print(node)
for l, obj in datasources.items():
logging.debug(f"datasource: {l}")
ds = app['+Data'][f'+{l}']
arrows = ''
node = f' ds{l} [label="{l}&#92;n{ds["Class"]}'
if full:
node += ' | '
for i, s in enumerate(obj):
if i > 0:
def __sig_name__(name):
return name.replace(".", "_")
def __type_bytes__(mtype):
if mtype.startswith("uint"):
return int(mtype[4:]) // 8
if mtype.startswith("int"):
return int(mtype[3:]) // 8
if mtype.startswith("float"):
return int(mtype[5:]) // 8
if mtype == "bool":
return 1
logging.debug(f"Type `{mtype}` unknow, size 1 byte")
return 1
def __type_size__(mtype):
return __type_bytes__(mtype[0]) * mtype[1]
def __num_bytes__(sig):
noe = 1 if "NumberOfElements" not in sig else int(sig["NumberOfElements"])
nod = 1 if "NumberOfDimensions" not in sig else int(sig["NumberOfDimensions"])
bytes = __type_bytes__(sig["Type"])
return noe * nod * bytes
def __doc_app_state__(state, tab=""):
res = ""
if "+Threads" in state:
for i, tname in enumerate(state["+Threads"]):
if tname[0] == "+":
ch = "├─●" if i < len(state["+Threads"]) - 1 else "╰─●"
res += f"{tab} {ch} {tname[1:]}\n"
return res
def __doc_obj__(label, obj, id):
node = f' {id} [label="{label}&#92;n{obj["Class"]} | '
return node
def __sig_type__(sig):
type_name = sig["Type"]
noe = sig["NumberOfElements"] if "NumberOfElements" in sig else 1
nod = sig["NumberOfDimensions"] if "NumberOfDimensions" in sig else 1
dim = int(noe) * int(nod)
return (type_name, dim)
def __sig_alias__(name, signal):
return name if "Alias" not in signal else signal["Alias"]
def __doc_common_gam__(name, obj, edges):
inputs = ""
outputs = ""
if "InputSignals" in obj:
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
datasrc = sig["DataSource"]
edges.append(
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(name, l, mtype),
)
)
if i != 0:
inputs += " | "
inputs += f"<in_{__sig_name__(l)}> {l}"
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
datasrc = sig["DataSource"]
edges.append(
__edge__(
__sig__(name, l, mtype),
__sig__(datasrc, alias, mtype),
)
)
if i != 0:
outputs += " | "
outputs += f"<out_{__sig_name__(l)}> {l}"
return "{ {" + inputs + " } | { " + outputs + "} }", edges
def __doc_triggered_iogam__(name, obj, edges):
node = ""
if "InputSignals" in obj:
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
datasrc = sig["DataSource"]
edges.append(
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(name, l, mtype),
)
)
if i > 1:
node += " | "
node += f'<sig{i}> {s}'
for c in obj[s]:
if c.split(':')[1].startswith('in'):
arrows += f' ds{l}:sig{i} -> {c};\n'
else:
arrows += f' {c} -> ds{l}:sig{i};\n'
else:
for (fn, dir) in obj:
if dir == 'in':
arrows += f' ds{l} -> fn{fn};\n'
else:
arrows += f' fn{fn} -> ds{l};\n'
node += '" color=blue];'
print(node)
print(arrows)
print('}\n```')
node += f"<in_{__sig_name__(l)}> "
if i == 0:
node += "Trigger | { { "
else:
node += f"{l}"
node += " } | { "
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
datasrc = sig["DataSource"]
edges.append(
__edge__(
__sig__(name, l, mtype),
__sig__(datasrc, alias, mtype),
)
)
if i != 0:
node += " | "
node += f"<out_{__sig_name__(l)}> {l} "
node += " } }"
return node, edges
def __doc_iogam__(id, obj, edges):
node = ""
if "InputSignals" in obj and "OutputSignals" in obj:
inputs, outputs = obj["InputSignals"], obj["OutputSignals"]
output_names = list(outputs.keys())
input_names = list(inputs.keys())
j = 0
isize = 0
osize = 0
i = 0
irow = ""
orow = ""
while i < len(inputs):
in_label = input_names[i]
input = inputs[in_label]
if irow:
irow += " | "
irow += f"<in_{__sig_name__(in_label)}> {in_label}"
isize += __num_bytes__(input)
i += 1
alias = __sig_alias__(in_label, input)
mtype = __sig_type__(input)
datasrc = input["DataSource"]
edges.append(
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(id, in_label, mtype),
)
)
while osize < isize and j < len(outputs):
out_label = output_names[j]
output = outputs[out_label]
osize += __num_bytes__(output)
if orow:
orow += " | "
orow += f"<out_{__sig_name__(out_label)}> {out_label}"
j += 1
alias = __sig_alias__(out_label, output)
mtype = __sig_type__(output)
datasrc = output["DataSource"]
edges.append(
__edge__(
__sig__(id, out_label, mtype),
__sig__(datasrc, alias, mtype),
)
)
if isize == osize and irow and orow:
if node:
node += " | "
node += "{ {" + irow + "} | {" + orow + "} }"
irow = ""
orow = ""
if irow or orow:
if node:
node += " | "
node += "{ {" + irow + "} | {" + orow + "} }"
return node, edges
def __doc_constgam__(id, obj, edges):
node = ""
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
datasrc = sig["DataSource"]
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
edges.append(
__edge__(__sig__(id, l, mtype), __sig__(datasrc, alias, mtype))
)
if i != 0:
node += " | "
node += f'<out_{__sig_name__(l)}> {sig["Default"]} '
node += ""
return node, edges
def __doc_gam__(label, obj, edges):
id = f"fn{label}"
cls = obj["Class"]
node = f' {id} [label="{label}&#92;n{obj["Class"]} | '
if cls == "ConstantGAM":
sigs, datasources = __doc_constgam__(id, obj, edges)
elif cls == "IOGAM":
sigs, datasources = __doc_iogam__(id, obj, edges)
elif cls == "TriggeredIOGAM":
sigs, datasources = __doc_triggered_iogam__(id, obj, edges)
else:
sigs, datasources = __doc_common_gam__(id, obj, edges)
if sigs:
node += sigs + '"];'
else:
node = ""
return node, datasources
def __process_gamds__(name, signals):
node = ""
for i, sig_name in enumerate(signals):
sig_type = signals[sig_name]
node += f'{name}_{__sig_name__(sig_name)}[ label=" {name}::{sig_name} ({sig_type[0]}[{sig_type[1]}])", color=blue];\n'
return node
def __process_ds__(name, obj, edges):
cls = obj["Class"]
sigs = {}
node = ""
for edge in edges:
if edge[0][0] == name:
sigs[edge[0][1]] = edge[0][2]
if edge[1][0] == name:
sigs[edge[1][1]] = edge[1][2]
if sigs:
if cls == "GAMDataSource":
node = __process_gamds__(name, sigs)
else:
order = []
if "Signals" in obj:
for sig in obj["Signals"]:
if sig in sigs:
order.append(sig)
else:
order = list(sigs.keys())
node = f'{name} [label="{name}&#92;n{cls} | '
for i, sig_name in enumerate(order):
sig_type = sigs[sig_name]
if i > 0:
node += " | "
node += f"<{__sig_name__(sig_name)}> {sig_name} ({sig_type[0]}[{sig_type[1]}])"
node += '", color=blue];\n'
return node
def __process_edges__(edges, datasources):
arrows = ""
for source, target in edges:
src_obj = source[0]
src_sig = __sig_name__(source[1])
trg_obj = target[0]
trg_sig = __sig_name__(target[1])
a = f"{src_obj}"
b = f"{trg_obj}"
if src_obj in datasources:
if datasources[src_obj] == "GAMDataSource":
a += f"_{src_sig}"
else:
a += f":{src_sig}"
else:
a += f":out_{src_sig}"
if trg_obj in datasources:
if datasources[trg_obj] == "GAMDataSource":
b += f"_{trg_sig}"
else:
b += f":{trg_sig}"
else:
b += f":in_{trg_sig}"
arrows += f"{a} -> {b};\n"
return arrows
def __doc_state__(label, state, app):
mdstate = ""
graph = f"digraph {label[1:]}" + "{\n"
graph += "grapn [ranksep=4];\n"
graph += "rankdir = LR;\n"
graph += "beautify = true;\n"
graph += "node [shape=record];\n"
graph += "\n"
arrows = ""
ext = ""
edges = []
datasources = {}
if "+Threads" in state:
for tname, obj in state["+Threads"].items():
if tname[0] == "+":
# graph += f"subgraph cluster_{tname[1:]}" + "{\n"
if "Functions" in obj:
fns = obj["Functions"]
for fn in fns:
obj = app["+Functions"][f"+{fn}"]
node, edges = __doc_gam__(fn, obj, edges)
graph += f"{node}\n"
# graph += f' label = "{tname[1:]}";\n'
# graph += "}\n"
if "+Data" in app:
for key, obj in app["+Data"].items():
if key[0] == "+":
datasources[key[1:]] = obj["Class"]
node = __process_ds__(key[1:], obj, edges)
graph += node
arrows = __process_edges__(edges, datasources)
graph += "\n" + ext + "\n" + arrows
graph += "}\n"
fname = f"{label[1:]}"
with open(os.path.join(__PATH__, f"{fname}.dot"), "w") as file:
file.write(graph)
os.system(
f"dot -Tsvg -o{os.path.join(__PATH__, fname)}.svg {os.path.join(__PATH__, fname)}.dot"
)
mdstate += f"![{fname}]({fname}.svg)\n"
return mdstate
def __doc_application__(label, app):
print()
print(f"## Real-time application: `{label[1:]}`")
print()
print(f"Below are descrbed the states and threads of the real-time application {label[1:]}.")
print("```")
print(f'{""*(len(label)-1)}')
print(f'{label[1:]}')
print(f'╰┬{""*(len(label)-2)}')
states = app['+States']
for i, (label, obj) in enumerate(states.items()):
ch = '├─▷' if i < len(states) - 1 else '╰─▷'
tab = '' if i < len(states) - 1 else ' '
if label[0] == '+':
print(f' {ch} {label[1:]}')
__doc_app_state__(obj, tab)
print("```")
for label, obj in states.items():
if label[0] == '+':
print()
print(f'### State `{label[1:]}`')
__doc_state__(label, obj, app)
mdapp = f'\n## Real-time application "{label[1:]}"\n\n'
mdapp += f"Below are descrbed the states and threads of the real-time application {label[1:]}.\n"
mdapp += "```\n"
mdapp += f'{""*(len(label)-1)}\n'
mdapp += f"{label[1:]}\n"
mdapp += f'╰┬{""*(len(label)-2)}\n'
states = app["+States"]
for i, (label, obj) in enumerate(states.items()):
ch = "├─▷" if i < len(states) - 1 else "╰─▷"
tab = "" if i < len(states) - 1 else " "
if label[0] == "+":
mdapp += f" {ch} {label[1:]}\n"
mdapp += __doc_app_state__(obj, tab)
mdapp += "```\n"
for label, obj in states.items():
if label[0] == "+":
mdapp += f'\n### State "{label[1:]}"\n'
mdapp += __doc_state__(label, obj, app)
return mdapp
def document(fname, args, config):
"""Document MARTe object."""
global __PATH__
__PATH__ = os.path.join(args.destination, fname)
if not os.path.isdir(__PATH__):
os.mkdir(__PATH__)
def document(label, obj):
"""Document MARTe object."""
if obj['Class'] == 'RealTimeApplication':
__doc_application__(label, obj)
if obj['Class'] == 'StateMachine':
print()
print(f"## State-Machine `{label[1:]}`")
mddoc = f"# {fname}\n"
for label, obj in config.items():
if "Class" in obj:
if obj["Class"] == "RealTimeApplication":
mddoc += __doc_application__(label, obj)
if obj["Class"] == "StateMachine":
mddoc += f'\n## State-Machine "{label[1:]}"'
with open(os.path.join(__PATH__, f"{fname}.md"), "w") as file:
file.write(mddoc)