-
Notifications
You must be signed in to change notification settings - Fork 0
/
rules.py
122 lines (89 loc) · 3.3 KB
/
rules.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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
119
120
121
122
from enum import Enum
from utils.str2notes import str2notes
from .note import Note
import yaml
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
import os
class RuleType(Enum):
ONCE = 'once'
EVERYTIME = 'everytime'
class Rule(object):
def __init__(self, name, notes, trigger, rule_type, enabled=True):
self.set(name, notes, trigger, rule_type, enabled)
def set(self, name, notes, trigger, rule_type, enabled=True):
self.name = name
if type(notes) is str:
self.notes = set(str2notes(notes))
elif type(notes) is set:
self.notes = set(Note(note) if type(note) is not Note else note for note in notes)
self.trigger = trigger
self.type = RuleType(rule_type)
self.enabled = enabled
def __repr__(self):
return f'Rule(notes={self.notes}, trigger={self.trigger}, type={self.type}, enabled={self.enabled})'
def __str__(self):
return self.name
def match(self, context):
return self.notes.issubset(context.notes) and self.name not in context.triggered
@classmethod
def loads(cls, data, name_prefix=None, order=None):
# Deserialize data and return a rule instance
# Deal with the name
name = data.get('name', None)
if not name and order is not None:
name = f'#{order}'
if not name:
name = 'undefined'
if name_prefix:
name = f'{name_prefix}.{name}'
# Make the rule
r = Rule(name, data.get('notes'), data.get('trigger'), data.get('type'), data.get('enabled'))
return r
# TODO: serializing the name should be a problem. Needs refactoring here
# Maybe split "name" and "display"
# Need a way to store the original data that was loaded
def dumps(self):
out = {
'name': self.name,
'notes': ' '.join(str(note) for note in sorted(self.notes)),
'trigger': self.trigger,
'type': self.type.value,
'enabled': self.enabled,
}
return out
class RuleSet(object):
def __init__(self, name):
self.name = name
self.rules = []
def add(self, rule):
self.rules.append(rule)
def examine(self, context):
triggered = []
for rule in self.rules:
match = rule.match(context)
if match:
print(f'> {rule.trigger}')
triggered.append(rule.trigger)
context.register_trigger(rule.notes, rule.name)
return triggered
def __repr__(self):
return f'RuleSet({self.name})'
@classmethod
def load(cls, path):
rs = None
# Loads the yaml file
print(f'Loading rule set from file {path} with {cls}')
with open(path, 'r') as f:
rs = cls(os.path.basename(path))
document = yaml.load(f, Loader=Loader)
if type(document) is not list:
raise ValueError(f'{path} is not valid yaml file')
for index, entry in enumerate(document):
rule = Rule.loads(entry, name_prefix=rs.name, order=index)
rs.add(rule)
print(f'Rule {rule.name} loaded: {repr(rule)}')
# print(rule.dumps())
return rs