@@ -0,0 +1,177 @@
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
miraculixx revised this gist