Skip to content

Instantly share code, notes, and snippets.

@nilp0inter
Created August 19, 2019 19:41
Show Gist options
  • Save nilp0inter/3417f52b9d232fc3e6870db390ab3bd5 to your computer and use it in GitHub Desktop.
Save nilp0inter/3417f52b9d232fc3e6870db390ab3bd5 to your computer and use it in GitHub Desktop.
from experta import *
import types
class ReturningEngine(KnowledgeEngine):
def run(self, steps=float('inf'), generate=False):
if not generate:
to_return = list()
for rhs in self._run_activations(steps=steps):
res = rhs()
if res is not None:
if isinstance(res, types.GeneratorType):
to_return.extend(res)
else:
return res
return to_return
else:
def _run_as_generator():
for rhs in self._run_activations(steps=steps):
res = rhs()
if res is not None:
if isinstance(res, types.GeneratorType):
yield from res
else:
return res
return _run_as_generator()
def _run_activations(self, steps=float('inf')):
"""
Execute agenda activations
"""
self.running = True
activation = None
execution = 0
try:
while steps > 0 and self.running:
added, removed = self.get_activations()
self.strategy.update_agenda(self.agenda, added, removed)
if watchers.worth('AGENDA', 'DEBUG'): # pragma: no cover
for idx, act in enumerate(self.agenda.activations):
watchers.AGENDA.debug(
"%d: %r %r",
idx,
act.rule.__name__,
", ".join(str(f) for f in act.facts))
activation = self.agenda.get_next()
if activation is None:
break
else:
steps -= 1
execution += 1
watchers.RULES.info(
"FIRE %s %s: %s",
execution,
activation.rule.__name__,
", ".join(str(f) for f in activation.facts))
yield lambda: activation.rule(
self,
**{k: v
for k, v in activation.context.items()
if not k.startswith('__')})
except:
raise
finally:
self.running = False
class KE(ReturningEngine):
@Rule()
def one(self):
yield 'R1'
yield 'R2'
@Rule()
def two(self):
yield 'R3'
ke=KE()
ke.reset()
print("Non generator, but yielding", ke.run())
ke=KE()
ke.reset()
print("Generator, yielding", list(ke.run(generate=True)))
class KE2(ReturningEngine):
@Rule()
def zero(self):
return 'RESULT'
ke2=KE2()
ke2.reset()
print("Just returning", ke2.run())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment