drepr/old_code/prototype/drepr/query/selection_interface.py
def align2graphs(g1: nx.DiGraph, g2: nx.DiGraph):
    """Get mapping between node in g2 -> g1"""
    data_node_lbl = '@data_node'

    lbl2ids = defaultdict(lambda: [])
    for n in g1.nodes:
        if g1.nodes[n]['is_class_node']:
            lbl = g1.nodes[n]['label']
        else:
            u, v, edata = next(iter(g1.in_edges(n, data=True)))
            lbl = f"{g1.nodes[u]['label']}--{edata['label']}"
        lbl2ids[lbl].append(n)

    gold_triples = set()
    for s, o in g1.edges:
        lbl = g1.edges[(s, o)]['label']
        gold_triples.add((s, lbl, o))

    pred_lbl2ids = defaultdict(lambda: [])
    for n in g2.nodes:
        if g2.nodes[n]['is_class_node']:
            lbl = g2.nodes[n]['label']
        else:
            u, v, edata = next(iter(g2.in_edges(n, data=True)))
            lbl = f"{g2.nodes[u]['label']}--{edata['label']}"
        pred_lbl2ids[lbl].append(n)

    # partition into which we know the mappings
    mapping = {}
    unsolved_mappings = []
    for lbl, nodes in pred_lbl2ids.items():
        if len(nodes) == 1 and len(lbl2ids[lbl]) == 1:
            mapping[nodes[0]] = lbl2ids[lbl][0]
        elif len(nodes) > 0 and len(lbl2ids[lbl]) > 0:
            unsolved_mappings.append(lbl)
        else:
            assert False, "Invalid selection graph"

    if len(unsolved_mappings) == 0:
        return [mapping]

    # generate possible mappings choices
    choices = []
    for lbl in unsolved_mappings:
        nodes = lbl2ids[lbl]
        pred_nodes = pred_lbl2ids[lbl]
        # don't care if two node map to one node because we assure that the accuracy is 1
        for y in pred_nodes:
            choices.append([(y, x) for x in nodes])

    iterator = RAIterator(MockRAReader(choices), [len(x) for x in choices], [0] * len(choices))
    # select best one
    best_hypo_mappings = []
    best_score = 0
    while True:
        res = iterator.next()
        if res is None:
            break
        hypo_mapping = dict(res[1])

        pred_triples = set()
        for sp, op in g2.edges:
            lbl = g2.edges[(sp, op)]['label']
            if sp in mapping:
                s = mapping[sp]
            else:
                s = hypo_mapping[sp]

            if op in mapping:
                o = mapping[op]
            else:
                o = hypo_mapping[op]
            pred_triples.add((s, lbl, o))

        acc = len(pred_triples.intersection(gold_triples))
        if acc > best_score:
            best_hypo_mappings = [hypo_mapping]
        elif acc == best_score:
            best_hypo_mappings.append(hypo_mapping)

    return [hypo_mapping.update(mapping) for hypo_mapping in best_hypo_mappings]
Get mapping between node in g2 -> g1
Get mapping between node in g2 -> g1
Get mapping between node in g2 - > g1
def align2graphs(g1: nx.DiGraph, g2: nx.DiGraph):
    data_node_lbl = '@data_node'
    lbl2ids = defaultdict(lambda: [])
    for n in g1.nodes:
        if g1.nodes[n]['is_class_node']:
            lbl = g1.nodes[n]['label']
        else:
            u, v, edata = next(iter(g1.in_edges(n, data=True)))
            lbl = f"{g1.nodes[u]['label']}--{edata['label']}"
        lbl2ids[lbl].append(n)
    gold_triples = set()
    for s, o in g1.edges:
        lbl = g1.edges[(s, o)]['label']
        gold_triples.add((s, lbl, o))
    pred_lbl2ids = defaultdict(lambda: [])
    for n in g2.nodes:
        if g2.nodes[n]['is_class_node']:
            lbl = g2.nodes[n]['label']
        else:
            u, v, edata = next(iter(g2.in_edges(n, data=True)))
            lbl = f"{g2.nodes[u]['label']}--{edata['label']}"
        pred_lbl2ids[lbl].append(n)
    mapping = {}
    unsolved_mappings = []
    for lbl, nodes in pred_lbl2ids.items():
        if len(nodes) == 1 and len(lbl2ids[lbl]) == 1:
            mapping[nodes[0]] = lbl2ids[lbl][0]
        elif len(nodes) > 0 and len(lbl2ids[lbl]) > 0:
            unsolved_mappings.append(lbl)
        else:
            assert False, "Invalid selection graph"
    if len(unsolved_mappings) == 0:
        return [mapping]
    choices = []
    for lbl in unsolved_mappings:
        nodes = lbl2ids[lbl]
        pred_nodes = pred_lbl2ids[lbl]
        for y in pred_nodes:
            choices.append([(y, x) for x in nodes])
    iterator = RAIterator(MockRAReader(choices), [len(x) for x in choices], [0] * len(choices))
    best_hypo_mappings = []
    best_score = 0
    while True:
        res = iterator.next()
        if res is None:
            break
        hypo_mapping = dict(res[1])
        pred_triples = set()
        for sp, op in g2.edges:
            lbl = g2.edges[(sp, op)]['label']
            if sp in mapping:
                s = mapping[sp]
            else:
                s = hypo_mapping[sp]
            if op in mapping:
                o = mapping[op]
            else:
                o = hypo_mapping[op]
            pred_triples.add((s, lbl, o))
        acc = len(pred_triples.intersection(gold_triples))
        if acc > best_score:
            best_hypo_mappings = [hypo_mapping]
        elif acc == best_score:
            best_hypo_mappings.append(hypo_mapping)
    return [hypo_mapping.update(mapping) for hypo_mapping in best_hypo_mappings]
def align2graphs ( g1 : nx . DiGraph , g2 : nx . DiGraph ) : data_node_lbl = '@data_node' lbl2ids = defaultdict ( lambda : [ ] ) for n in g1 . nodes : if g1 . nodes [ n ] [ 'is_class_node' ] : lbl = g1 . nodes [ n ] [ 'label' ] else : u , v , edata = next ( iter ( g1 . in_edges ( n , data = True ) ) ) lbl = f"{g1.nodes[u]['label']}--{edata['label']}" lbl2ids [ lbl ] . append ( n ) gold_triples = set ( ) for s , o in g1 . edges : lbl = g1 . edges [ ( s , o ) ] [ 'label' ] gold_triples . add ( ( s , lbl , o ) ) pred_lbl2ids = defaultdict ( lambda : [ ] ) for n in g2 . nodes : if g2 . nodes [ n ] [ 'is_class_node' ] : lbl = g2 . nodes [ n ] [ 'label' ] else : u , v , edata = next ( iter ( g2 . in_edges ( n , data = True ) ) ) lbl = f"{g2.nodes[u]['label']}--{edata['label']}" pred_lbl2ids [ lbl ] . append ( n ) mapping = { } unsolved_mappings = [ ] for lbl , nodes in pred_lbl2ids . items ( ) : if len ( nodes ) == 1 and len ( lbl2ids [ lbl ] ) == 1 : mapping [ nodes [ 0 ] ] = lbl2ids [ lbl ] [ 0 ] elif len ( nodes ) > 0 and len ( lbl2ids [ lbl ] ) > 0 : unsolved_mappings . append ( lbl ) else : assert False , "Invalid selection graph" if len ( unsolved_mappings ) == 0 : return [ mapping ] choices = [ ] for lbl in unsolved_mappings : nodes = lbl2ids [ lbl ] pred_nodes = pred_lbl2ids [ lbl ] for y in pred_nodes : choices . append ( [ ( y , x ) for x in nodes ] ) iterator = RAIterator ( MockRAReader ( choices ) , [ len ( x ) for x in choices ] , [ 0 ] * len ( choices ) ) best_hypo_mappings = [ ] best_score = 0 while True : res = iterator . next ( ) if res is None : break hypo_mapping = dict ( res [ 1 ] ) pred_triples = set ( ) for sp , op in g2 . edges : lbl = g2 . edges [ ( sp , op ) ] [ 'label' ] if sp in mapping : s = mapping [ sp ] else : s = hypo_mapping [ sp ] if op in mapping : o = mapping [ op ] else : o = hypo_mapping [ op ] pred_triples . add ( ( s , lbl , o ) ) acc = len ( pred_triples . intersection ( gold_triples ) ) if acc > best_score : best_hypo_mappings = [ hypo_mapping ] elif acc == best_score : best_hypo_mappings . append ( hypo_mapping ) return [ hypo_mapping . update ( mapping ) for hypo_mapping in best_hypo_mappings ]
Get mapping between node in g2 -> g1
Get mapping between node in g2 - > g1
"""Get mapping between node in g2 -> g1""" # partition into which we know the mappings # generate possible mappings choices # don't care if two node map to one node because we assure that the accuracy is 1 # select best one
g1
nx.DiGraph
g2
nx.DiGraph