Ferrari Martino Giordano
659 lines
23 KiB
Python
659 lines
23 KiB
Python
import logging
|
|
import os
|
|
|
|
import common as doc
|
|
|
|
|
|
def __doc_obj__(label, obj, id):
|
|
node = f' {id} [label="{label}\n{obj["Class"]} | '
|
|
return node
|
|
|
|
|
|
def __doc_common_gam__(name, obj, edges, thread):
|
|
inputs = ""
|
|
outputs = ""
|
|
if "InputSignals" in obj:
|
|
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
|
|
alias = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
datasrc = sig["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(datasrc, alias, mtype),
|
|
doc.signal(name, l, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
inputs += " | "
|
|
inputs += f"<in_{doc.format_name(l)}> {l}"
|
|
if "OutputSignals" in obj:
|
|
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
|
|
alias = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
datasrc = sig["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(name, l, mtype),
|
|
doc.signal(datasrc, alias, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
outputs += " | "
|
|
outputs += f"<out_{doc.format_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 = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
datasrc = sig["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(datasrc, alias, mtype),
|
|
doc.signal(name, l, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
if i > 1:
|
|
signals += " | "
|
|
signals += f"<in_{doc.format_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 = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
datasrc = sig["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(name, l, mtype),
|
|
doc.signal(datasrc, alias, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
signals += " | "
|
|
signals += f"<out_{doc.format_name(l)}> {l} "
|
|
signals += " } }"
|
|
node = f'{name} [tooltip=" TriggeredIOGAM::{name} ", label="{name}\nTriggeredIOGAM | {signals}", style=filled, fillcolor="#eee"];\n'
|
|
return node, edges
|
|
|
|
|
|
__set_keys__ = (
|
|
"Alias",
|
|
"Type",
|
|
"DataSource",
|
|
"NumberOfElements",
|
|
"NumberOfDimensions",
|
|
"Ranges",
|
|
)
|
|
|
|
|
|
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())
|
|
input_size = 0
|
|
output_size = 0
|
|
out_i = 0
|
|
in_i = 0
|
|
counter = 0
|
|
while in_i < len(inputs) and out_i < len(outputs):
|
|
irows = []
|
|
orows = []
|
|
bold = False
|
|
while input_size <= output_size and in_i < len(inputs):
|
|
in_label = input_names[in_i]
|
|
input = inputs[in_label]
|
|
input_size += doc.signal_byte_size(input)
|
|
in_i += 1
|
|
alias = doc.signal_alias(in_label, input)
|
|
mtype = doc.signal_type(input)
|
|
if mtype[2]:
|
|
bold = True
|
|
irow = f"{doc.type_to_string(mtype)}"
|
|
for key, val in input.items():
|
|
if key not in __set_keys__:
|
|
bold = True
|
|
irow += f"<br/>{key}: {val}"
|
|
datasrc = input["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(datasrc, alias, mtype),
|
|
doc.signal(f"{id}_{counter}", f"{len(irows)}", mtype),
|
|
thread,
|
|
)
|
|
)
|
|
irows.append(irow)
|
|
while output_size < input_size and out_i < len(outputs):
|
|
out_label = output_names[out_i]
|
|
output = outputs[out_label]
|
|
output_size += doc.signal_byte_size(output)
|
|
out_i += 1
|
|
alias = doc.signal_alias(out_label, output)
|
|
mtype = doc.signal_type(output)
|
|
if mtype[2]:
|
|
bold = True
|
|
orow = f"{doc.type_to_string(mtype)}"
|
|
for key, val in output.items():
|
|
if key not in __set_keys__:
|
|
bold = True
|
|
orow += f"<br/>{key}: {val}"
|
|
datasrc = output["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(f"{id}_{counter}", f"{len(orows)}", mtype),
|
|
doc.signal(datasrc, alias, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
orows.append(orow)
|
|
fill = "#eee"
|
|
fcolor = "#000"
|
|
total = max(len(orows), len(irows))
|
|
if bold or total > 1:
|
|
fill = "#eee"
|
|
bold = True
|
|
else:
|
|
fill = "#eee"
|
|
|
|
inode = f"{id}_{counter} [shape=plaintext,"
|
|
inode += f'tooltip=" IOGAM::{id} ", label=<'
|
|
inode += "<table cellspacing='0' cellborder='1'"
|
|
inode += f" cellpadding='4' border='0' bgcolor='{fill}'>"
|
|
|
|
if bold:
|
|
for i in range(total):
|
|
inode += "<tr>"
|
|
if i < len(irows):
|
|
inode += f'<td port="in_{i}" bgcolor="{fill}"'
|
|
if len(orows) > len(irows):
|
|
inode += f' rowspan="{len(orows)}"'
|
|
inode += ">"
|
|
inode += f'<font color="{fcolor}">'
|
|
inode += irows[i]
|
|
inode += "</font></td>"
|
|
if i < len(orows):
|
|
inode += f'<td port="out_{i}" bgcolor="{fill}"'
|
|
if len(orows) < len(irows):
|
|
inode += f' rowspan="{len(irows)}"'
|
|
inode += ">"
|
|
inode += f'<font color="{fcolor}">'
|
|
inode += orows[i]
|
|
inode += "</font></td>"
|
|
inode += "</tr>"
|
|
else:
|
|
inode += "<tr><td port='in_0'></td>"
|
|
inode += "<td port='out_0'></td></tr>"
|
|
inode += "</table>>"
|
|
inode += "];\n"
|
|
if not bold:
|
|
e_in, e_out = edges[-2:]
|
|
edges = edges[:-2]
|
|
edges.append((e_in[0], e_out[1], thread))
|
|
inode = ""
|
|
node += inode
|
|
irows = []
|
|
orows = []
|
|
counter += 1
|
|
return node, edges
|
|
|
|
|
|
def __doc_constgam__(id, obj, edges, thread):
|
|
node = f" {id} [shape=plaintext, label=<<table cellspacing='0' cellborder='1' cellpadding='4' border='0'>"
|
|
node += f"<tr><td colspan='2'><b>{id[2:]}</b><br/>{obj['Class']}</td></tr>"
|
|
|
|
if "OutputSignals" in obj:
|
|
for i, (l, sig) in enumerate(obj["OutputSignals"].items()):
|
|
datasrc = sig["DataSource"]
|
|
alias = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(id, l, mtype),
|
|
doc.signal(datasrc, alias, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
node += f'<tr><td bgcolor="#eee">{doc.type_to_string(mtype)}<br/><font point-size="8">{l}</font></td>'
|
|
node += f'<td port="out_{doc.format_name(l)}">'
|
|
node += f'{sig["Default"]}</td></tr>'
|
|
node += "</table>>];\n"
|
|
return node, edges
|
|
|
|
|
|
def __doc_message_gam__(id, obj, edges, thread):
|
|
label = id[2:]
|
|
node = f' {id} [label="{label}\n{obj["Class"]} | '
|
|
inputs = ""
|
|
if "InputSignals" in obj:
|
|
for i, (l, sig) in enumerate(obj["InputSignals"].items()):
|
|
alias = doc.signal_alias(l, sig)
|
|
mtype = doc.signal_type(sig)
|
|
datasrc = sig["DataSource"]
|
|
edges.append(
|
|
doc.edge(
|
|
doc.signal(datasrc, alias, mtype),
|
|
doc.signal(id, l, mtype),
|
|
thread,
|
|
)
|
|
)
|
|
if i != 0:
|
|
inputs += " | "
|
|
inputs += f"<in_{doc.format_name(l)}> {l}"
|
|
if inputs:
|
|
node += inputs + '", style=filled, fillcolor=lightyellow];'
|
|
else:
|
|
node = ""
|
|
|
|
if node:
|
|
add, edges = __doc_events__(id, obj, edges, thread)
|
|
node += add
|
|
return node, edges
|
|
|
|
|
|
def __doc_events__(id, obj, edges, thread):
|
|
nodes = ""
|
|
bgcolor = "lightyellow"
|
|
for event_id, event in obj["+Events"].items():
|
|
if event_id[0] == "+":
|
|
event_id = event_id[1:]
|
|
trigger = event["EventTrigger"]
|
|
message_ids = [id[1:] for id in event if id[0] == "+"]
|
|
for msg_id in message_ids:
|
|
nodes += f" {id}_{event_id}_{msg_id} [shape=plaintext, label=<"
|
|
nodes += '<table cellspacing="0" cellborder="1" border="0" cellpadding="4"><tr>'
|
|
nodes += f"<td bgcolor='{bgcolor}' align='left'><b>When</b></td><td balign='left'>"
|
|
for i, (sig, val) in enumerate(trigger.items()):
|
|
if i > 0:
|
|
nodes += "<br/>"
|
|
nodes += f"{sig} == {val}"
|
|
|
|
nodes += f"</td></tr><tr><td bgcolor='{bgcolor}' align='left'><b>Destination</b></td>"
|
|
msg = event[f"+{msg_id}"]
|
|
fn = msg["Function"]
|
|
nodes += f"<td align='left'>{msg['Destination']}</td></tr>"
|
|
nodes += f"<tr><td bgcolor='{bgcolor}' align='left'><b>Message</b></td><td align='left'>"
|
|
if fn == "SetOutput":
|
|
params = msg["+Parameters"]
|
|
nodes += f"<i>set</i> {params['SignalName']} <i>to</i> {params['SignalValue']}"
|
|
else:
|
|
nodes += f"{fn} "
|
|
nodes += "</td></tr></table>>];\n"
|
|
nodes += f" {id}:e -> {id}_{event_id}_{msg_id}:w"
|
|
nodes += " [arrowhead=box, constraint=true];\n"
|
|
return nodes, edges
|
|
|
|
|
|
def __sanitize__(expression):
|
|
if expression[0] == "\n":
|
|
expression = expression[1:]
|
|
return expression.replace(">", ">").replace("<", "<").replace("\n", "\n")
|
|
|
|
|
|
def __doc_mathgam__(id, obj, edges, thread):
|
|
label = id[2:]
|
|
node = f' {id} [label="{label}\n{obj["Class"]} | '
|
|
node += __sanitize__(obj["Expression"]) + " | "
|
|
sigs, edges = __doc_common_gam__(id, obj, edges, thread)
|
|
if sigs:
|
|
node += sigs + '"];'
|
|
else:
|
|
node = ""
|
|
return node, edges
|
|
|
|
|
|
__gams_fns__ = {
|
|
"IOGAM": __doc_iogam__,
|
|
"ConstantGAM": __doc_constgam__,
|
|
"TriggeredIOGAM": __doc_triggered_iogam__,
|
|
"MathExpressionGAM": __doc_mathgam__,
|
|
"MessageGAM": __doc_message_gam__,
|
|
}
|
|
|
|
|
|
def __doc_gam__(label, obj, edges, thread, classfilter=True):
|
|
id = f"fn{label}"
|
|
cls = obj["Class"]
|
|
if classfilter and cls in __gams_fns__:
|
|
return __gams_fns__[cls](id, obj, edges, thread)
|
|
node = f' {id} [label="{label}\n{obj["Class"]} | '
|
|
sigs, edges = __doc_common_gam__(id, obj, edges, thread)
|
|
if sigs:
|
|
node += sigs + '"];'
|
|
else:
|
|
node = ""
|
|
return node, edges
|
|
|
|
|
|
def __process_gamds__(name, cls, 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'<tr><td port="{doc.format_name(sig_name)}">'
|
|
f"{sig_name} ({doc.type_to_string(sig_type)})"
|
|
"</td></tr>"
|
|
)
|
|
if node:
|
|
label = f"""<table border="0" cellpadding="4" cellborder="1" cellspacing="0" bgcolor="#eee">
|
|
<tr><td><b>{name}</b><br/>{cls}</td></tr>
|
|
{node}
|
|
</table>
|
|
"""
|
|
node = f'{name} [shape=plaintext, label=<{label}>, color="#0af"];\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" or cls == "RealTimeThreadAsyncBridge":
|
|
node, edges = __process_gamds__(name, cls, 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} [shape=plaintext; label=<"
|
|
node += '<table border="0" cellpadding="4" cellborder="1" cellspacing="0" bgcolor="lightblue">\n'
|
|
node += f"<tr><td><b>{name}</b><br/>{cls}</td></tr>\n"
|
|
for i, sig_name in enumerate(order):
|
|
node += f'<tr><td port="{doc.format_name(sig_name)}">'
|
|
sig_type = sigs[sig_name]
|
|
node += f"{sig_name} ({doc.type_to_string(sig_type)})"
|
|
node += "</td></tr>\n"
|
|
for sig_name in lost:
|
|
logging.warning(f"Lost signal {name}:{sig_name}")
|
|
node += f'<tr><td port="{doc.format_name(sig_name)}">'
|
|
sig_type = sigs[sig_name]
|
|
node += f"{sig_name} ({doc.type_to_string(sig_type)})"
|
|
node += "</td></tr>\n"
|
|
node += '</table>>, color="#0af"];\n'
|
|
if not multi_thread and node:
|
|
node = f"subgraph cluster_{thread}" + "{\n" + node + "};\n"
|
|
return node, edges
|
|
|
|
|
|
def __process_edges__(edges, datasources):
|
|
# remove duplicated edges
|
|
clean_edges = []
|
|
for edge in edges:
|
|
if edge not in clean_edges:
|
|
clean_edges.append(edge)
|
|
edges = clean_edges
|
|
# generate arrows
|
|
arrows = ""
|
|
for source, target, _ in edges:
|
|
src_obj = source[0]
|
|
src_sig = doc.format_name(source[1])
|
|
trg_obj = target[0]
|
|
trg_sig = doc.format_name(target[1])
|
|
a = f"{src_obj}"
|
|
b = f"{trg_obj}"
|
|
ranked = True
|
|
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 = doc.type_to_string(source[2])
|
|
out_type = doc.type_to_string(target[2])
|
|
mtype = in_type if in_type == out_type else f"{in_type} -> {out_type}"
|
|
arrows += f'{a}:e -> {b}:w [tooltip=" {mtype} ", labelfloat=false'
|
|
arrows += f", constraint={ranked}];\n"
|
|
return arrows
|
|
|
|
|
|
def __state_full_graph__(label, state, app):
|
|
graph = f"digraph {label[1:]}" + "{\n"
|
|
graph += "ranksep = 2;\n"
|
|
graph += "nodesep = 0.05;\n"
|
|
graph += "rankdir = LR;\n"
|
|
graph += "node [shape=Mrecord];\n"
|
|
graph += f"label=<State: <B>{label[1:]}</B>>;\n"
|
|
graph += "fontsize=40;\n"
|
|
graph += "\n"
|
|
arrows = ""
|
|
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 = <Thread: <B>{tname[1:]}</B> >;\n"
|
|
graph += " fontsize = 24.0;\n"
|
|
graph += " penwidth = 3;\n"
|
|
graph += ' color = "#1A1A1A";\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\n" + arrows
|
|
graph += "}\n"
|
|
fname = f"{label[1:]}"
|
|
|
|
gpath = os.path.join(doc.path(), f"{fname}_full.dot")
|
|
with open(gpath, "w") as file:
|
|
file.write(graph)
|
|
return gpath
|
|
|
|
|
|
def __simple_gam__(label, obj):
|
|
cls = obj["Class"]
|
|
id = f"fn{doc.format_name(label)}"
|
|
style = "style=rounded"
|
|
if cls == "IOGAM":
|
|
style = 'style="rounded,filled", fillcolor="#eee"'
|
|
elif cls == "MessageGAM":
|
|
style = 'style="rounded,filled", fillcolor=lightyellow'
|
|
if "InputSignals" in obj:
|
|
for _, signal in obj["InputSignals"].items():
|
|
ds = f'ds{doc.format_name(signal["DataSource"])}'
|
|
if "OutputSignals" in obj:
|
|
for _, signal in obj["OutputSignals"].items():
|
|
ds = f'ds{doc.format_name(signal["DataSource"])}'
|
|
return f"{id}[label=<<B>{label}</B><BR/>{cls}>, {style}]\n"
|
|
|
|
|
|
def __simple_ds__(label, obj, edges):
|
|
cls = obj["Class"]
|
|
thread = ""
|
|
multi_thread = False
|
|
sources = []
|
|
targets = []
|
|
id = f"ds{doc.format_name(label)}"
|
|
for edge in edges:
|
|
if edge[0][0] == label:
|
|
if not thread:
|
|
thread = edge[-1]
|
|
elif thread != edge[-1]:
|
|
multi_thread = True
|
|
if edge[1][0] == label:
|
|
if not thread:
|
|
thread = edge[-1]
|
|
elif thread != edge[-1]:
|
|
multi_thread = True
|
|
|
|
fillcolor = "lightblue"
|
|
bordercolor = "#0af"
|
|
style = "penwidth=2"
|
|
if cls == "GAMDataSource":
|
|
fillcolor = "#eee"
|
|
node = f"{id}[label=<<B>{label}</B><BR/>{cls}>, "
|
|
node += f'{style}, fillcolor="{fillcolor}", color="{bordercolor}"]\n'
|
|
if thread and not multi_thread:
|
|
node = f"subgraph cluster_{thread}" + "{\n " + node + "}"
|
|
return node
|
|
|
|
|
|
def __simple_id__(name, datasources):
|
|
if name in datasources:
|
|
return f"ds{doc.format_name(name)}"
|
|
return name
|
|
|
|
|
|
def __simple_edges__(edges, datasources):
|
|
arrows = ""
|
|
done = []
|
|
|
|
for source, target, _ in edges:
|
|
src_obj = __simple_id__(source[0], datasources)
|
|
trg_obj = __simple_id__(target[0], datasources)
|
|
if (src_obj, trg_obj) not in done:
|
|
arrows += f"{src_obj}:e -> {trg_obj}:w\n"
|
|
done.append((src_obj, trg_obj))
|
|
return arrows
|
|
|
|
|
|
def __state_simple_graph__(label, state, app):
|
|
graph = f"digraph {label[1:]}" + "{\n"
|
|
graph += "rankdir = LR;\n"
|
|
graph += 'node [shape=rect, style="filled", fillcolor=white];\n'
|
|
graph += f"label=<State: <B>{label[1:]}</B>>;\n"
|
|
graph += "fontsize=40;\n"
|
|
graph += "\n"
|
|
arrows = ""
|
|
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}"]
|
|
_, edges = __doc_gam__(
|
|
fn, obj, edges, tname[1:], classfilter=False
|
|
)
|
|
graph += f" {__simple_gam__(fn, obj)}"
|
|
graph += f"\n label = <Thread: <B>{tname[1:]}</B> >;\n"
|
|
graph += " fontsize = 24.0;\n"
|
|
graph += " penwidth = 3;\n"
|
|
graph += ' color = "#1A1A1A";\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)
|
|
if node:
|
|
graph += __simple_ds__(key[1:], obj, edges)
|
|
graph += __simple_edges__(edges, datasources)
|
|
# arrows = __process_edges__(edges, datasources)
|
|
# graph += "\n" + ext + "\n" + arrows
|
|
graph += "}\n"
|
|
fname = f"{label[1:]}"
|
|
|
|
gpath = os.path.join(doc.path(), f"{fname}_simple.dot")
|
|
with open(gpath, "w") as file:
|
|
file.write(graph)
|
|
return gpath
|
|
|
|
|
|
def doc_state(label, state, app):
|
|
fname = f"{label[1:]}"
|
|
if doc.build():
|
|
logging.info(f"Building svg grpah {fname}")
|
|
src_path = __state_full_graph__(label, state, app)
|
|
os.system(f"dot -Tsvg -o{os.path.join(doc.path(), fname)}_full.svg {src_path}")
|
|
src_path = __state_simple_graph__(label, state, app)
|
|
os.system(f"dot -Tsvg -o{os.path.join(doc.path(), fname)}.svg {src_path}")
|
|
return f"" + "{width=100% height=90%}\n"
|