diff --git a/.gitignore b/.gitignore index ea1472e..77d26fc 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ output/ +__pycache__ diff --git a/autodoc/__init__.py b/autodoc/__init__.py index e69de29..18299c9 100644 --- a/autodoc/__init__.py +++ b/autodoc/__init__.py @@ -0,0 +1 @@ +from autodoc.autodoc import * diff --git a/autodoc/__main__.py b/autodoc/__main__.py index 79fa729..24bbe14 100644 --- a/autodoc/__main__.py +++ b/autodoc/__main__.py @@ -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() diff --git a/autodoc/__pycache__/__main__.cpython-311.pyc b/autodoc/__pycache__/__main__.cpython-311.pyc index d7eacc9..34b8f58 100644 Binary files a/autodoc/__pycache__/__main__.cpython-311.pyc and b/autodoc/__pycache__/__main__.cpython-311.pyc differ diff --git a/autodoc/__pycache__/autodoc.cpython-311.pyc b/autodoc/__pycache__/autodoc.cpython-311.pyc index bb6e121..a12babd 100644 Binary files a/autodoc/__pycache__/autodoc.cpython-311.pyc and b/autodoc/__pycache__/autodoc.cpython-311.pyc differ diff --git a/autodoc/autodoc.py b/autodoc/autodoc.py index 78c0d1f..1d3ee1c 100644 --- a/autodoc/autodoc.py +++ b/autodoc/autodoc.py @@ -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}\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' {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' {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}\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}\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" {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" {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' {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" " + + 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" {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_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_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' {sig["Default"]} ' + + node += "" + return node, edges + + +def __doc_gam__(label, obj, edges): + id = f"fn{label}" + cls = obj["Class"] + node = f' {id} [label="{label}\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}\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)