-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathPath.py
More file actions
62 lines (50 loc) · 1.83 KB
/
Copy pathPath.py
File metadata and controls
62 lines (50 loc) · 1.83 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from Rule import Rule
Entity = str | int
class Path:
"""
Represents a specific instantiation of a rule in the Knowledge Graph.
"""
def __init__(self, head: tuple[Entity, Entity, Entity], body: set[tuple[Entity, Entity, Entity]]):
self.head = head
self.body = body
def copy(self):
return Path(self.head, self.body.copy())
def get_nodes(self) -> set[Entity]:
nodes = set()
for s, p, o in self.body:
nodes.add(s)
nodes.add(o)
return nodes
def frontiers_closed_rule(self) -> Entity:
h1 = self.head[0]
nodes = self.get_nodes()
if not self.body or (len(nodes) == 1 and h1 in nodes):
return h1
counts: dict[Entity, int] = {}
for s, p, o in self.body:
if s == o:
continue
counts[s] = counts.get(s, 0) + 1
counts[o] = counts.get(o, 0) + 1
for node, count in counts.items():
if node == h1:
continue
if count == 1:
return node
return None
def to_rule_closed_rule(self) -> Rule:
entity_to_var: dict[Entity, int] = {self.head[0]: 0, self.head[2]: 1}
next_var = 2
def get_var(entity: Entity) -> int:
nonlocal next_var
if entity not in entity_to_var:
entity_to_var[entity] = next_var
next_var += 1
return entity_to_var[entity]
body_vars: set[tuple[int, Entity, int]] = set()
for s, p, o in self.body:
body_vars.add((get_var(s), p, get_var(o)))
head_vars = (entity_to_var[self.head[0]], self.head[1], entity_to_var[self.head[2]])
return Rule(body_vars, head_vars)
def __repr__(self):
return f"Path(head={self.head}, body={self.body})"