a simple python based rule engine

rules.py

"""
(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