""" (c) 2014 miraculixx at gmx.ch """ from shrutil.dictobj import DictObject class RuleContext(DictObject): """ rule context to store values and attributes (or any object) """ def __init__(self): super(RuleContext, self).__init__() self._executed=[] def __setitem__(self, item, value): self.__setattr__(item, value) def __getattr__(self, item): if not item in self.__dict__ and not item in self._data: return None return super(RuleContext, self).__getattr__(item) @property def as_dict(self): return {'context' : self } class Rule(object): """ basic rule """ def should_trigger(self, context): pass def perform(self, context): pass def record(self, context, result): context._executed.append((self.ruleid, result)) @property def ruleid(self): return self.__class__.__name__.split('.')[-1] class TableRuleset(Rule): """ a table ruleset created from a list of dict objects of the following format [ { 'if' : ['condition1', ...], 'then' : ['action1', ...], 'target' : ['target1', ...] }, ... ] Each rule is only executed if all conditions are met. In conditions and actions, use context. to reference variables. targets implicitly reference context. (target 'xy' means 'context.xy'). The result of the nth 'then' action is stored in the nth context.variable as defined in target. """ def __init__(self, rules, translations=None): self.rules = rules or {} if translations: translator = Translator(translations) for rule in self.rules: for key in rule.keys(): rule[key] = [translator.replace(item) for item in rule[key]] def should_trigger(self, context): return True def perform(self, context): count = 0 for rule in self.rules: if all([eval(condition, context.as_dict) for condition in rule['if']]): count = count + 1 self._current_ruleid = rule.get('id', count) for action, target in zip(rule['then'], rule['target']): if context._translations: action = context._translations.replace(action) target = context._translations.replace(target) result = context[target.replace('context.', '').strip()] = eval(action, context.as_dict) self.record(context, result) else: break else: self._current_ruleid = None return True return False @property def ruleid(self): if self._current_ruleid: return "%s.%s" % (super(TableRuleset, self).ruleid, self._current_ruleid) return super(TableRuleset, self).ruleid class NaturalLanguageRule(TableRulset): def __init__(self, translations): translator = Translator(translations) from inspect import getdoc def should_trigger(self, context) class SequencedRuleset(Rule): """ guaranteed to run in sequence """ def __init__(self, rules): self.rules = rules or [] def should_trigger(self, context): return True def perform(self, context): for rule in self.rules: if rule.should_trigger(context): result = rule.perform(context) rule.record(context, result) return True class CalculateBasicFare(Rule): def should_trigger(self, context): return True def perform(self, context): context.fare = context.distance * 20 return context.fare class CalculateWeekendFare(Rule): def should_trigger(self, context): return context.weekend def perform(self, context): context.fare = context.fare * 1.2 return context.fare class RuleEngine(object): def execute(self, ruleset, context): for rule in ruleset: if rule.should_trigger(context): result = rule.perform(context) rule.record(context, result) return context context = RuleContext() context.distance = 10 context.weekend = 0 context.weather = 'nice' context.sunday = 1 class Translator(object): def __init__(self, translations): self.translations = translations def replace(self, input): input = " %s " % input for source, target in self.translations: input = input.replace(" %s" % source, " %s " % target) return input translations=[ ('das Wetter', 'context.weather'), ('ist', '=='), ('schön', '"nice"'), ('Fahrpreis', 'context.fare'), ('Wochenende', 'context.weekend'), ('nicht', 'not'), ('am Sonntag', 'context.sunday'), ] manyrules = TableRuleset([ { 'if': ['nicht Wochenende', 'das Wetter ist schön'], 'then' : ['Fahrpreis * 1'], 'target' : ['Fahrpreis'], }, { 'if': ['am Sonntag', 'das Wetter ist schön'], 'then' : ['Fahrpreis * 1.5'], 'target' : ['Fahrpreis'], } ], translations=translations) ruleset = (CalculateBasicFare(), CalculateWeekendFare(), manyrules, #SequencedRuleset([CalculateBasicFare(), CalculateWeekendFare()]), ) engine = RuleEngine() #%timeit engine.execute(ruleset, context) context = engine.execute(ruleset, context) print "fare", context.fare print "executed", context._executed