MARTeAutoDoc/autodoc/autodoc.py

"""
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}&#92;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):
    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),
                    thread,
                )
            )
            if i > 1:
                node += " | "
            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),
                    thread,
                )
            )
            if i != 0:
                node += " | "
            node += f"<out_{__sig_name__(l)}> {l} "
    node += " } }"
    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 = "#999"
            fsize = 14.0
            if bold or irelcount > 1 or orelcount > 1:
                fill = "#333"
            else:
                fsize = 8.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];\n'
            irow = ""
            orow = ""
            counter += 1
    return node, edges


def __doc_constgam__(id, obj, edges, thread):
    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),
                    thread,
                )
            )
            if i != 0:
                node += " | "
            node += (
                f'<out_{__sig_name__(l)}> {__type_to_str__(mtype)}: {sig["Default"]}'
            )

    node += ""
    return node, 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, thread)
    elif cls == "IOGAM":
        return __doc_iogam__(id, obj, edges, thread)
    elif cls == "TriggeredIOGAM":
        sigs, datasources = __doc_triggered_iogam__(id, obj, edges, thread)
    else:
        sigs, datasources = __doc_common_gam__(id, obj, edges, thread)
    if sigs:
        node += sigs + '"];'
    else:
        node = ""
    return node, datasources


def __process_gamds__(name, signals, edges):
    node = ""
    outputs = {}
    inputs = {}
    done = False
    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}&#92;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}&#92;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 += "graph [ranksep=2];\n"
    graph += "rankdir = LR;\n"
    graph += "beautify = true;\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"![{fname}]({fname}.svg)\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)