|
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
from .SequenceTypes import *
from .Modules import OutputModule, EDProducer, EDFilter, EDAnalyzer, Service, ESProducer, ESSource, _Module
from .Mixins import _Labelable
# Use this on Tasks in the Schedule
class ScheduleTaskValidator(object):
def __init__(self):
pass
def enter(self,visitee):
if visitee.isLeaf():
if isinstance(visitee, _Labelable):
if not visitee.hasLabel_():
raise ValueError("A task associated with the Schedule contains a module of type '"+visitee.type_()+"'\nwhich has no assigned label.")
elif isinstance(visitee, Service):
if not visitee._inProcess:
raise ValueError("A task associated with the Schedule contains a service of type '"+visitee.type_()+"'\nwhich is not attached to the process.")
def leave(self,visitee):
pass
# Use this on Paths
class PathValidator(object):
def __init__(self):
self.__label = ''
def setLabel(self,label:str):
self.__label = "'"+label+"' "
def enter(self,visitee):
if isinstance(visitee,OutputModule):
raise ValueError("Path "+self.__label+"cannot contain an OutputModule, '"+visitee.type_()+"', with label '"+visitee.label_()+"'")
if visitee.isLeaf():
if isinstance(visitee, _Labelable):
if not visitee.hasLabel_():
raise ValueError("Path "+self.__label+"contains a module of type '"+visitee.type_()+"' which has no assigned label.")
elif isinstance(visitee, Service):
if not visitee._inProcess:
raise ValueError("Path "+self.__label+"contains a service of type '"+visitee.type_()+"' which is not attached to the process.\n")
def leave(self,visitee):
pass
# Use this on EndPaths
class EndPathValidator(object):
_presetFilters = ["TriggerResultsFilter", "HLTPrescaler"]
def __init__(self):
self.filtersOnEndpaths = []
self.__label = ''
self._levelInTasks = 0
def setLabel(self,label:str):
self.__label = "'"+label+"' "
def enter(self,visitee):
if visitee.isLeaf():
if isinstance(visitee, _Labelable):
if not visitee.hasLabel_():
raise ValueError("EndPath "+self.__label+"contains a module of type '"+visitee.type_()+"' which has\nno assigned label.")
elif isinstance(visitee, Service):
if not visitee._inProcess:
raise ValueError("EndPath "+self.__label+"contains a service of type '"+visitee.type_()+"' which is not attached to the process.\n")
if isinstance(visitee, Task):
self._levelInTasks += 1
if self._levelInTasks > 0:
return
if isinstance(visitee,EDFilter):
if (visitee.type_() in self._presetFilters):
if (visitee.type_() not in self.filtersOnEndpaths):
self.filtersOnEndpaths.append(visitee.type_())
def leave(self,visitee):
if self._levelInTasks > 0:
if isinstance(visitee, Task):
self._levelInTasks -= 1
class NodeVisitor(object):
"""Form sets of all modules, ESProducers, ESSources and Services in visited objects. Can be used
to visit Paths, EndPaths, Sequences or Tasks. Includes in sets objects on sub-Sequences and sub-Tasks"""
def __init__(self):
self.modules = set()
self.esProducers = set()
self.esSources = set()
self.services = set()
def enter(self,visitee):
if visitee.isLeaf():
if isinstance(visitee, _Module):
self.modules.add(visitee)
elif isinstance(visitee, ESProducer):
self.esProducers.add(visitee)
elif isinstance(visitee, ESSource):
self.esSources.add(visitee)
elif isinstance(visitee, Service):
self.services.add(visitee)
def leave(self,visitee):
pass
class CompositeVisitor(object):
""" Combines 3 different visitor classes in 1 so we only have to visit all the paths and endpaths once"""
def __init__(self, validator, node, decorated, optional=None):
self._validator = validator
self._node = node
self._decorated = decorated
self._optional = optional
def enter(self, visitee):
self._validator.enter(visitee)
self._node.enter(visitee)
self._decorated.enter(visitee)
if self._optional:
self._optional.enter(visitee)
def leave(self, visitee):
self._validator.leave(visitee)
# The node visitor leave function does nothing
#self._node.leave(visitee)
self._decorated.leave(visitee)
if self._optional:
self._optional.leave(visitee)
class ModuleNamesFromGlobalsVisitor(object):
"""Fill a list with the names of Event module types in a sequence. The names are determined
by using globals() to lookup the variable names assigned to the modules. This
allows the determination of the labels before the modules have been attached to a Process."""
def __init__(self,globals_,l):
self._moduleToName = { v[1]:v[0] for v in globals_.items() if isinstance(v[1],_Module) }
self._names =l
def enter(self,node):
if isinstance(node,_Module):
self._names.append(self._moduleToName[node])
def leave(self,node):
return
if __name__=="__main__":
import unittest
class TestModuleCommand(unittest.TestCase):
def setUp(self):
"""Nothing to do """
pass
def testValidators(self):
producer = EDProducer("Producer")
analyzer = EDAnalyzer("Analyzer")
output = OutputModule("Out")
filter = EDFilter("Filter")
unlabeled = EDAnalyzer("UnLabeled")
producer.setLabel("producer")
analyzer.setLabel("analyzer")
output.setLabel("output")
filter.setLabel("filter")
s1 = Sequence(analyzer*producer)
s2 = Sequence(output+filter)
p1 = Path(s1)
p2 = Path(s1*s2)
p3 = Path(s1+unlabeled)
ep1 = EndPath(producer+output+analyzer)
ep2 = EndPath(filter+output)
ep3 = EndPath(s2)
ep4 = EndPath(unlabeled)
pathValidator = PathValidator()
endpathValidator = EndPathValidator()
p1.visit(pathValidator)
self.assertRaises(ValueError, p2.visit, pathValidator)
self.assertRaises(ValueError, p3.visit, pathValidator)
ep1.visit(endpathValidator)
ep2.visit(endpathValidator)
ep3.visit(endpathValidator)
self.assertRaises(ValueError, ep4.visit, endpathValidator)
unittest.main()
|