Bases: Solver
A solver that uses Soufflé.
Soufflé is a logic programming language inspired by Datalog. For more details,
see The Souffle site.
Note that in order to use this integration, you need to install Souffle
and have it on your path.
>>> from typedlogic.integrations.frameworks.pydantic import FactBaseModel
>>> class AncestorOf(FactBaseModel):
... ancestor: str
... descendant: str
>>> solver = SouffleSolver()
>>> from typedlogic import SentenceGroup, PredicateDefinition
>>> solver.add_predicate_definition(PredicateDefinition(predicate="AncestorOf", arguments={'ancestor': str, 'descendant': str}))
>>> solver.add_fact(AncestorOf(ancestor='p1', descendant='p1a'))
>>> solver.add_fact(AncestorOf(ancestor='p1a', descendant='p1aa'))
>>> aa = SentenceGroup(name="transitivity-of-ancestor-of")
>>> solver.add_sentence_group(aa)
>>> soln = solver.check()
This solver does not implement the open-world assumption.
>>> from typedlogic.profiles import OpenWorld
>>> solver.profile.impl(OpenWorld)
False
Source code in src/typedlogic/integrations/solvers/souffle/souffle_solver.py
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118 | @dataclass
class SouffleSolver(Solver):
"""
A solver that uses Soufflé.
Soufflé is a logic programming language inspired by Datalog. For more details,
see [The Souffle site](https://souffle-lang.github.io/).
Note that in order to use this integration, you need to [install Souffle](https://souffle-lang.github.io/install)
and have it on your path.
>>> from typedlogic.integrations.frameworks.pydantic import FactBaseModel
>>> class AncestorOf(FactBaseModel):
... ancestor: str
... descendant: str
>>> solver = SouffleSolver()
>>> from typedlogic import SentenceGroup, PredicateDefinition
>>> solver.add_predicate_definition(PredicateDefinition(predicate="AncestorOf", arguments={'ancestor': str, 'descendant': str}))
>>> solver.add_fact(AncestorOf(ancestor='p1', descendant='p1a'))
>>> solver.add_fact(AncestorOf(ancestor='p1a', descendant='p1aa'))
>>> aa = SentenceGroup(name="transitivity-of-ancestor-of")
>>> solver.add_sentence_group(aa)
>>> soln = solver.check()
This solver does not implement the open-world assumption.
>>> from typedlogic.profiles import OpenWorld
>>> solver.profile.impl(OpenWorld)
False
"""
exec_name: str = field(default="souffle")
profile: ClassVar[Profile] = MixedProfile(
ClassicDatalog(), SortedLogic(), AllowsComparisonTerms(), SingleModelSemantics()
)
def models(self) -> Iterator[Model]:
compiler = SouffleCompiler()
program = compiler.compile(self.base_theory)
pdmap = {}
facts = []
with tempfile.TemporaryDirectory() as temp_dir:
# Create output directives for each predicate
output_files = {}
input_files = {}
for pd in self.base_theory.predicate_definitions:
pred = pd.predicate
pdmap[pred] = pd
output_file = Path(temp_dir) / f"{pred}.csv"
output_files[pred] = str(output_file)
program += f'\n.output {pred}(IO=file, filename="{output_file}")\n'
input_file = Path(temp_dir) / f"{pred}__in.csv"
input_files[pred] = str(input_file)
program += f'\n.input {pred}(IO=file, filename="{input_file}")\n'
with open(input_file, "w", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile, delimiter="\t")
for term in self.base_theory.ground_terms:
if term.predicate == pred:
writer.writerow(term.bindings.values())
with tempfile.NamedTemporaryFile(suffix=".dl", mode="w") as fp:
fp.write(program)
fp.flush()
res = subprocess.run([self.exec_name, fp.name], capture_output=True)
if res.stderr:
msg = res.stderr.decode()
import re
if re.match(r".*Variable (\S+) only occurs once.*", msg):
logger.info(msg)
else:
logger.error(msg)
for pred, filename in output_files.items():
if not Path(filename).exists():
continue
pd = pdmap[pred]
rows = []
with open(filename, "r") as csvfile:
reader = csv.reader(csvfile, delimiter="\t")
for row in reader:
rows.append(row)
facts.extend(make_terms(rows, pd))
model = Model(source_object=self, ground_terms=facts)
yield model
def check(self) -> Solution:
return Solution(satisfiable=None)
def dump(self) -> str:
compiler = SouffleCompiler()
return compiler.compile(self.base_theory)
|
