Martino Ferrari
528 lines
17 KiB
Python
528 lines
17 KiB
Python
"""
|
|
Autodoc functions
|
|
"""
|
|
import logging
|
|
import os
|
|
|
|
__PATH__ = "."
|
|
|
|
|
|
def __sig__(obj, label, type):
|
|
return (obj, label, type)
|
|
|
|
|
|
def __edge__(src, trg, thread):
|
|
return (src, trg, thread)
|
|
|
|
|
|
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"])
|
|
nod = nod if nod > 0 else 1
|
|
noe = noe if noe > 0 else 1
|
|
bytes = __type_bytes__(sig["Type"])
|
|
tot = noe * nod
|
|
|
|
if "Ranges" in sig:
|
|
tot = 0
|
|
for i0, ie in sig["Ranges"]:
|
|
tot += int(ie) - int(i0) + 1
|
|
|
|
return tot * 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}\n{obj["Class"]} | '
|
|
return node
|
|
|
|
|
|
def __sig_type__(sig):
|
|
type_name = sig["Type"]
|
|
noe = int(sig["NumberOfElements"]) if "NumberOfElements" in sig else 1
|
|
nod = int(sig["NumberOfDimensions"]) if "NumberOfDimensions" in sig else 1
|
|
nod = nod if nod > 0 else 1
|
|
dim = noe * nod
|
|
range = sig["Ranges"] if "Ranges" in sig else []
|
|
return (type_name, dim, range)
|
|
|
|
|
|
def __sig_alias__(name, signal):
|
|
return name if "Alias" not in signal else signal["Alias"]
|
|
|
|
|
|
def __doc_common_gam__(name, obj, edges, thread):
|
|
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),
|
|
thread,
|
|
)
|
|
)
|
|
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),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
outputs += " | "
|
|
outputs += f"<out_{__sig_name__(l)}> {l}"
|
|
return "{ {" + inputs + " } | { " + outputs + "} }", edges
|
|
|
|
|
|
def __doc_triggered_iogam__(name, obj, edges, thread):
|
|
signals = ""
|
|
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),
|
|
thread,
|
|
)
|
|
)
|
|
if i > 1:
|
|
signals += " | "
|
|
signals += f"<in_{__sig_name__(l)}> "
|
|
|
|
if i == 0:
|
|
signals += "Trigger | { { "
|
|
else:
|
|
signals += f"{l}"
|
|
signals += " } | { "
|
|
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),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
signals += " | "
|
|
signals += f"<out_{__sig_name__(l)}> {l} "
|
|
signals += " } }"
|
|
node = f'{name} [tooltip=" TriggeredIOGAM::{name} ", label="{name}\nTriggeredIOGAM | {signals}", style=filled, fillcolor="#eee"];\n'
|
|
return node, edges
|
|
|
|
|
|
def __doc_iogam__(id, obj, edges, thread):
|
|
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
|
|
input_size = 0
|
|
output_size = 0
|
|
i = 0
|
|
irow = ""
|
|
orow = ""
|
|
counter = 0
|
|
while i < len(inputs) and j < len(outputs):
|
|
irelcount = 0
|
|
orelcount = 0
|
|
bold = False
|
|
while input_size <= output_size and i < len(inputs):
|
|
in_label = input_names[i]
|
|
input = inputs[in_label]
|
|
if irow:
|
|
irow += " | "
|
|
input_size += __num_bytes__(input)
|
|
i += 1
|
|
alias = __sig_alias__(in_label, input)
|
|
mtype = __sig_type__(input)
|
|
if mtype[2]:
|
|
bold = True
|
|
irow += f"<in_{irelcount}> {__type_to_str__(mtype)}"
|
|
datasrc = input["DataSource"]
|
|
edges.append(
|
|
__edge__(
|
|
__sig__(datasrc, alias, mtype),
|
|
__sig__(f"{id}_{counter}", f"{irelcount}", mtype),
|
|
thread,
|
|
)
|
|
)
|
|
irelcount += 1
|
|
while output_size < input_size and j < len(outputs):
|
|
out_label = output_names[j]
|
|
output = outputs[out_label]
|
|
output_size += __num_bytes__(output)
|
|
if orow:
|
|
orow += " | "
|
|
j += 1
|
|
alias = __sig_alias__(out_label, output)
|
|
mtype = __sig_type__(output)
|
|
if mtype[2]:
|
|
bold = True
|
|
orow += f"<out_{orelcount}> {__type_to_str__(mtype)}"
|
|
datasrc = output["DataSource"]
|
|
edges.append(
|
|
__edge__(
|
|
__sig__(f"{id}_{counter}", f"{orelcount}", mtype),
|
|
__sig__(datasrc, alias, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
orelcount += 1
|
|
fill = "#fff"
|
|
fsize = 14.0
|
|
if bold or irelcount > 1 or orelcount > 1:
|
|
fill = "#333"
|
|
else:
|
|
irow = f"<in_{irelcount-1}> "
|
|
orow = f"<out_{orelcount-1}>"
|
|
fill = "#666"
|
|
fsize = 0.0
|
|
node += f'{id}_{counter} [tooltip=" IOGAM::{id} " label="'
|
|
node += "{ {" + irow + "} | {" + orow + "} }"
|
|
node += f'", fillcolor="{fill}", color="#666", style=filled,'
|
|
node += f' fontcolor="#fff", fontsize={fsize}, height=0, width=0];\n'
|
|
irow = ""
|
|
orow = ""
|
|
counter += 1
|
|
return node, edges
|
|
|
|
|
|
def __doc_constgam__(id, obj, edges, thread):
|
|
node = f' {id} [label="{id[2:]}\n{obj["Class"]} | '
|
|
values = ""
|
|
types = ""
|
|
|
|
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),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
values += " | "
|
|
types += " | "
|
|
values += f'<out_{__sig_name__(l)}> {sig["Default"]}'
|
|
types += __type_to_str__(mtype)
|
|
node += "{{ " + types + " } | {" + values + "}}"
|
|
node += '", style=filled, fillcolor="#dff"];'
|
|
return node, edges
|
|
|
|
|
|
def __doc_gam__(label, obj, edges, thread):
|
|
id = f"fn{label}"
|
|
cls = obj["Class"]
|
|
node = f' {id} [label="{label}\n{obj["Class"]} | '
|
|
if cls == "ConstantGAM":
|
|
return __doc_constgam__(id, obj, edges, thread)
|
|
elif cls == "IOGAM":
|
|
return __doc_iogam__(id, obj, edges, thread)
|
|
elif cls == "TriggeredIOGAM":
|
|
return __doc_triggered_iogam__(id, obj, edges, thread)
|
|
else:
|
|
sigs, datasources = __doc_common_gam__(id, obj, edges, thread)
|
|
if cls == "MessageGAM":
|
|
import pprint
|
|
|
|
pprint.pprint(obj["+Events"])
|
|
if sigs:
|
|
node += sigs + '"];'
|
|
else:
|
|
node = ""
|
|
return node, datasources
|
|
|
|
|
|
def __process_gamds__(name, signals, edges):
|
|
node = ""
|
|
outputs = {}
|
|
inputs = {}
|
|
done = False
|
|
to_keep = []
|
|
while not done:
|
|
signals = {}
|
|
for edge in edges:
|
|
if edge[0][0] == name:
|
|
sig_name = edge[0][1]
|
|
signals[sig_name] = edge[0][2]
|
|
if sig_name not in outputs:
|
|
outputs[sig_name] = []
|
|
outputs[sig_name].append(edge)
|
|
elif edge[1][0] == name:
|
|
sig_name = edge[1][1]
|
|
signals[sig_name] = edge[1][2]
|
|
if sig_name not in inputs:
|
|
inputs[sig_name] = []
|
|
inputs[sig_name].append(edge)
|
|
to_reduce = []
|
|
to_keep = []
|
|
done = True
|
|
for sig_name in signals:
|
|
if sig_name in inputs and sig_name in outputs:
|
|
to_reduce.append(sig_name)
|
|
else:
|
|
to_keep.append(sig_name)
|
|
|
|
for sig_name in to_reduce:
|
|
ins = inputs[sig_name]
|
|
outs = outputs[sig_name]
|
|
for edge in ins:
|
|
if edge in edges:
|
|
done = False
|
|
edges.remove(edge)
|
|
for edge in outs:
|
|
if edge in edges:
|
|
done = False
|
|
edges.remove(edge)
|
|
for in_edge in ins:
|
|
for out_edge in outs:
|
|
edge = (in_edge[0], out_edge[1], in_edge[-1])
|
|
edges.append(edge)
|
|
|
|
for i, sig_name in enumerate(to_keep):
|
|
if i > 0:
|
|
node += " | "
|
|
sig_type = signals[sig_name]
|
|
node += f"<{__sig_name__(sig_name)}> {sig_name} ({__type_to_str__(sig_type)})"
|
|
if node:
|
|
node = f'{name} [label="{name}\nGAMDataSource | {node}", color=blue];\n'
|
|
return node, edges
|
|
|
|
|
|
def __process_ds__(name, obj, edges):
|
|
cls = obj["Class"]
|
|
sigs = {}
|
|
node = ""
|
|
thread = ""
|
|
multi_thread = False
|
|
for edge in edges:
|
|
if edge[0][0] == name:
|
|
sigs[edge[0][1]] = edge[0][2]
|
|
if not thread:
|
|
thread = edge[-1]
|
|
elif thread != edge[-1]:
|
|
multi_thread = True
|
|
if edge[1][0] == name:
|
|
sigs[edge[1][1]] = edge[1][2]
|
|
if not thread:
|
|
thread = edge[-1]
|
|
elif thread != edge[-1]:
|
|
multi_thread = True
|
|
if sigs:
|
|
if cls == "GAMDataSource":
|
|
node, edges = __process_gamds__(name, sigs, edges)
|
|
else:
|
|
order = []
|
|
lost = []
|
|
if "Signals" in obj:
|
|
for sig in obj["Signals"]:
|
|
if sig in sigs:
|
|
order.append(sig)
|
|
else:
|
|
order = list(sigs.keys())
|
|
for sig in sigs:
|
|
if sig not in order:
|
|
lost.append(sig)
|
|
node = f'{name} [label="{name}\n{cls} | '
|
|
for i, sig_name in enumerate(order):
|
|
if i > 0:
|
|
node += " | "
|
|
sig_type = sigs[sig_name]
|
|
node += f"<{__sig_name__(sig_name)}> {sig_name} ({__type_to_str__(sig_type)})"
|
|
|
|
for sig_name in lost:
|
|
logging.warning(f"Lost signal {name}:{sig_name}")
|
|
if node:
|
|
node += " | "
|
|
sig_type = sigs[sig_name]
|
|
node += f"<{__sig_name__(sig_name)}> {sig_name} ({__type_to_str__(sig_type)})"
|
|
node += '", color=blue];\n'
|
|
if not multi_thread and node:
|
|
node = f"subgraph cluster_{thread}" + "{\n" + node + "};\n"
|
|
return node, edges
|
|
|
|
|
|
def __type_to_str__(mtype):
|
|
range = ""
|
|
if mtype[2]:
|
|
range = (
|
|
"("
|
|
+ ", ".join(
|
|
[
|
|
str(i0) if i0 == i1 else (str(i0) + ":" + str(int(i1) + 1))
|
|
for i0, i1 in mtype[2]
|
|
]
|
|
)
|
|
+ ")"
|
|
)
|
|
if mtype[1] != 1:
|
|
return f"{mtype[0]}[{mtype[1]}]{range}"
|
|
return mtype[0] + range
|
|
|
|
|
|
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:
|
|
a += f":{src_sig}"
|
|
else:
|
|
a += f":out_{src_sig}"
|
|
if trg_obj in datasources:
|
|
b += f":{trg_sig}"
|
|
else:
|
|
b += f":in_{trg_sig}"
|
|
|
|
in_type = __type_to_str__(source[2])
|
|
out_type = __type_to_str__(target[2])
|
|
mtype = in_type if in_type == out_type else f"{in_type} -> {out_type}"
|
|
arrows += f'{a} -> {b} [tooltip=" {mtype} ", labelfloat=false];\n'
|
|
return arrows
|
|
|
|
|
|
def __doc_state__(label, state, app):
|
|
mdstate = ""
|
|
graph = f"digraph {label[1:]}" + "{\n"
|
|
graph += "ranksep = 1;\n"
|
|
graph += "nodesep = 0.5;\n"
|
|
graph += "rankdir = LR;\n"
|
|
graph += "node [shape=Mrecord];\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, tname[1:])
|
|
graph += f"{node}\n"
|
|
graph += f"\n label = < <B>{tname[1:]}</B> >;\n"
|
|
graph += " fontsize = 24.0;\n"
|
|
graph += " penwidth = 3;\n"
|
|
graph += ' color = "#0A0";\n'
|
|
graph += "}\n"
|
|
if "+Data" in app:
|
|
for key, obj in app["+Data"].items():
|
|
if key[0] == "+":
|
|
datasources[key[1:]] = obj["Class"]
|
|
node, edges = __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"\n"
|
|
return mdstate
|
|
|
|
|
|
def __doc_application__(label, 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__)
|
|
|
|
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)
|