martino/MARTeAutoDoc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: reduced GAMDataSources

Browse Source
This commit is contained in:
Ferrari Martino Giordano
2024-03-11 11:04:38 +01:00
parent f5e3792893
commit c44d484fe9
2 changed files with 103 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
autodoc/__pycache__/autodoc.cpython-311.pyc
BIN
View File
Binary file not shown.
autodoc/autodoc.py
+103 -38
View File
@@ -12,8 +12,8 @@ def __sig__(obj, label, type):
return (obj, label, type)
def __edge__(src, trg):
return (src, trg)
def __edge__(src, trg, thread):
return (src, trg, thread)
def __sig_name__(name):
@@ -71,7 +71,7 @@ def __sig_alias__(name, signal):
return name if "Alias" not in signal else signal["Alias"]
def __doc_common_gam__(name, obj, edges):
def __doc_common_gam__(name, obj, edges, thread):
inputs = ""
outputs = ""
if "InputSignals" in obj:
@@ -83,6 +83,7 @@ def __doc_common_gam__(name, obj, edges):
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(name, l, mtype),
thread,
)
)
if i != 0:
@@ -97,6 +98,7 @@ def __doc_common_gam__(name, obj, edges):
__edge__(
__sig__(name, l, mtype),
__sig__(datasrc, alias, mtype),
thread,
)
)
if i != 0:
@@ -105,7 +107,7 @@ def __doc_common_gam__(name, obj, edges):
return "{ {" + inputs + " } | { " + outputs + "} }", edges
def __doc_triggered_iogam__(name, obj, edges):
def __doc_triggered_iogam__(name, obj, edges, thread):
node = ""
if "InputSignals" in obj:
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
@@ -116,6 +118,7 @@ def __doc_triggered_iogam__(name, obj, edges):
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(name, l, mtype),
thread,
)
)
if i > 1:
@@ -136,6 +139,7 @@ def __doc_triggered_iogam__(name, obj, edges):
__edge__(
__sig__(name, l, mtype),
__sig__(datasrc, alias, mtype),
thread,
)
)
if i != 0:
@@ -145,7 +149,7 @@ def __doc_triggered_iogam__(name, obj, edges):
return node, edges
def __doc_iogam__(id, obj, edges):
def __doc_iogam__(id, obj, edges, thread):
node = ""
if "InputSignals" in obj and "OutputSignals" in obj:
inputs, outputs = obj["InputSignals"], obj["OutputSignals"]
@@ -173,6 +177,7 @@ def __doc_iogam__(id, obj, edges):
__edge__(
__sig__(datasrc, alias, mtype),
__sig__(id, in_label, mtype),
thread,
)
)
while osize < isize and j < len(outputs):
@@ -191,6 +196,7 @@ def __doc_iogam__(id, obj, edges):
__edge__(
__sig__(id, out_label, mtype),
__sig__(datasrc, alias, mtype),
thread,
)
)
if isize == osize and irow and orow:
@@ -206,7 +212,7 @@ def __doc_iogam__(id, obj, edges):
return node, edges
def __doc_constgam__(id, obj, edges):
def __doc_constgam__(id, obj, edges, thread):
node = ""
if "OutputSignals" in obj:
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
@@ -214,28 +220,30 @@ def __doc_constgam__(id, obj, edges):
alias = __sig_alias__(l, sig)
mtype = __sig_type__(sig)
edges.append(
__edge__(__sig__(id, l, mtype), __sig__(datasrc, alias, mtype))
__edge__(__sig__(id, l, mtype), __sig__(datasrc, alias, mtype), thread)
)
if i != 0:
node += " | "
node += f'<out_{__sig_name__(l)}> {sig["Default"]} '
node += (
f'<out_{__sig_name__(l)}> {__type_to_str__(mtype)}: {sig["Default"]}'
)
node += ""
return node, edges
def __doc_gam__(label, obj, edges):
def __doc_gam__(label, obj, edges, thread):
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)
sigs, datasources = __doc_constgam__(id, obj, edges, thread)
elif cls == "IOGAM":
sigs, datasources = __doc_iogam__(id, obj, edges)
sigs, datasources = __doc_iogam__(id, obj, edges, thread)
elif cls == "TriggeredIOGAM":
sigs, datasources = __doc_triggered_iogam__(id, obj, edges)
sigs, datasources = __doc_triggered_iogam__(id, obj, edges, thread)
else:
sigs, datasources = __doc_common_gam__(id, obj, edges)
sigs, datasources = __doc_common_gam__(id, obj, edges, thread)
if sigs:
node += sigs + '"];'
else:
@@ -243,48 +251,107 @@ def __doc_gam__(label, obj, edges):
return node, datasources
def __process_gamds__(name, signals):
def __process_gamds__(name, signals, edges):
node = ""
for i, sig_name in enumerate(signals):
ok = False
outputs = {}
inputs = {}
for sig_name in signals:
for edge in edges:
if edge[0][0] == name and edge[0][1] == sig_name:
if sig_name not in outputs:
outputs[sig_name] = []
outputs[sig_name].append(edge)
elif edge[1][0] == name and edge[1][1] == sig_name:
if sig_name not in inputs:
inputs[sig_name] = []
inputs[sig_name].append(edge)
to_reduce = []
to_keep = []
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:
edges.remove(edge)
for edge in outs:
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'{name}_{__sig_name__(sig_name)}[ label=" {name}::{sig_name} ({sig_type[0]}[{sig_type[1]}])", color=blue];\n'
return node
node += f"<{__sig_name__(sig_name)}> {sig_name} ({__type_to_str__(sig_type)})"
if node:
node = f'{name} [label="{name}&#92;nGAMDataSource | {node}", color=blue];\n'
return node, edges
def __process_ds__(name, obj, edges):
cls = obj["Class"]
sigs = {}
node = ""
thread = ""
for edge in edges:
if edge[0][0] == name:
sigs[edge[0][1]] = edge[0][2]
if not thread:
thread = edge[-1]
if edge[1][0] == name:
sigs[edge[1][1]] = edge[1][2]
if not thread:
thread = edge[-1]
if sigs:
if cls == "GAMDataSource":
node = __process_gamds__(name, sigs)
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}&#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]}])"
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'
return node
if cls != "RealTimeThreadAsyncBridge" and node:
node = f"subgraph cluster_{thread}" + "{" + node + "};"
return node, edges
def __type_to_str__(mtype):
if mtype[1] != 1:
return f"{mtype[0]}[{mtype[1]}]"
return mtype[0]
def __process_edges__(edges, datasources):
arrows = ""
for source, target in edges:
for source, target, _ in edges:
src_obj = source[0]
src_sig = __sig_name__(source[1])
trg_obj = target[0]
@@ -292,27 +359,25 @@ def __process_edges__(edges, datasources):
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}"
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}"
b += f":{trg_sig}"
else:
b += f":in_{trg_sig}"
arrows += f"{a} -> {b};\n"
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 += "grapn [ranksep=4];\n"
graph += "graph [ranksep=2];\n"
graph += "rankdir = LR;\n"
graph += "beautify = true;\n"
graph += "node [shape=record];\n"
@@ -324,20 +389,20 @@ def __doc_state__(label, state, app):
if "+Threads" in state:
for tname, obj in state["+Threads"].items():
if tname[0] == "+":
# graph += f"subgraph cluster_{tname[1:]}" + "{\n"
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)
node, edges = __doc_gam__(fn, obj, edges, tname[1:])
graph += f"{node}\n"
# graph += f' label = "{tname[1:]}";\n'
# graph += "}\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)
node, edges = __process_ds__(key[1:], obj, edges)
graph += node
arrows = __process_edges__(edges, datasources)
graph += "\n" + ext + "\n" + arrows