weaver.py

import inspect
import itertools
import networkx as nx
from networkx.drawing.nx_agraph import to_agraph


class Provided:
    pass


class Target:
    pass


def f0(_a: Provided, _b: Provided) -> str:
    return 'hello'


def f1(s: str) -> str:
    return s


def f2(s: str) -> int:
    return int(s)


def f3(_s: str, _i: int, _b: bool) -> float:
    return 1.1


def f4(b: bool) -> str:
    return str(b)


def f5(i: int) -> bool:
    return i > 0


def f6(_f: float, _b: bytes, _i: int) -> type:
    return str


def f7(_f: float) -> bool:
    return True


def f8(_s: str, _b: bool) -> bool:
    return True


def f9(_s: str, _c: bytes) -> int:
    return True


def f10(_i: int) -> bytes:
    return b''


def f11(_t: type, b:bytes) -> Target:
    return Target()


functions = {
    f0,
    f1,
    f2,
    f3,
    f4,
    f5,
    f6,
    f7,
    f8,
    f9,
    f10,
    f11
}


def return_type(f):
    return inspect.signature(f).return_annotation


def parameters(f):
    ps = inspect.signature(f).parameters.values()
    for p in ps:
        yield p.annotation


def graph(fs):
    g = nx.MultiDiGraph()
    g.add_node(Provided.__name__, color='Green')
    g.add_node(Target.__name__, color='Green')
    for f in fs:
        start = ','.join(sorted(p.__name__ for p in frozenset(parameters(f))))
        end = return_type(f)
        g.add_edge(start, end.__name__, label=f.__name__, color='blue', derived=False)
    depth = 1
    while not nx.has_path(g, Provided.__name__, Target.__name__):
        # try derived path
        recur = False
        ds = nx.descendants(g, Provided.__name__)
        ds = list(filter(lambda d: ',' not in d, ds))
        for i in range(len(ds)):
            for derived in itertools.combinations(ds, r=i+1):
                derived_node = ','.join(sorted(frozenset(derived)))
                if g.has_node(derived_node):
                    if not nx.has_path(g, Provided.__name__, derived_node):
                        for source in derived:
                            g.add_edge(source, derived_node, color='red', label=depth, derived=True)
                            recur = True
        if not recur:
            break
        depth += 1

    return g


def draw(g, name):
    a = to_agraph(g)
    a.layout('dot')
    a.draw(name + '.png')


def call_graph(g, s=Provided.__name__, t=Target.__name__):
    ng = nx.DiGraph()
    path = min(nx.all_simple_edge_paths(g, s, t), key=lambda p: len(p))
    eds = list({"source": s, "target": t, **G.get_edge_data(s, t, k)} for s, t, k in path)
    for ed in eds:
        print('handle edge', ed)
        if ed['derived']:
            print('expanding derived edge', ed)
            t = ed['target']
            for u1, v1 in g.in_edges(t):
                print('expanding in edge', u1, v1)
                for u2, v2 in call_graph(g, s, u1).edges:
                    ned = g.get_edge_data(u2, v2)[0]
                    ng.add_edge(u2, v2, label=ned['label'])
        else:
           ng.add_edge(ed['source'], ed['target'], label=ed['label'])
    return ng


if "__main__" == __name__:
    G = graph(functions)
    NG = call_graph(G)
    draw(NG, 'call')
    draw(G, 'all')