-
Notifications
You must be signed in to change notification settings - Fork 7
/
structure_miner.py
174 lines (154 loc) · 5.33 KB
/
structure_miner.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
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
# -*- coding: utf-8 -*-
"""
Created on Fri Mar 6 20:47:09 2020
@author: Manuel Camargo
"""
import os
import platform as pl
import subprocess
import traceback
import readers.bpmn_reader as br
import readers.process_structure as gph
from support_modules.log_repairing import conformance_checking as chk
class StructureMiner():
"""
This class extracts all the BPS parameters
"""
class Decorators(object):
@classmethod
def safe_exec(cls, method):
"""
Decorator to safe execute methods and return the state
----------
method : Any method.
Returns
-------
dict : execution status
"""
def safety_check(*args, **kw):
is_safe = kw.get('is_safe', method.__name__.upper())
if is_safe:
try:
method(*args)
except Exception as e:
print(e)
traceback.print_exc()
is_safe = False
return is_safe
return safety_check
def __init__(self, settings, log):
"""constructor"""
self.log = log
self.is_safe = True
self.settings = settings
def execute_pipeline(self) -> None:
"""
Main method for mining structure
"""
self.is_safe = self._mining_structure(is_safe=self.is_safe)
self.is_safe = self._evaluate_alignment(is_safe=self.is_safe)
@Decorators.safe_exec
def _mining_structure(self, **kwargs) -> None:
miner = self._get_miner(self.settings['mining_alg'])
miner(self.settings)
def _get_miner(self, miner):
if miner == 'sm1':
return self._sm1_miner
elif miner == 'sm2':
return self._sm2_miner
elif miner == 'sm3':
return self._sm3_miner
else:
raise ValueError(miner)
@staticmethod
def _sm2_miner(settings):
"""
Executes splitminer for bpmn structure mining.
Returns
-------
None
DESCRIPTION.
"""
print(" -- Mining Process Structure --")
# Event log file_name
file_name = settings['file'].split('.')[0]
input_route = os.path.join(settings['output'], file_name+'.xes')
sep = ';' if pl.system().lower() == 'windows' else ':'
# Mining structure definition
args = ['java']
if not pl.system().lower() == 'windows':
args.append('-Xmx2G')
args.extend(['-cp',
(settings['sm2_path']+sep+os.path.join(
'external_tools','splitminer2','lib','*')),
'au.edu.unimelb.services.ServiceProvider',
'SM2',
input_route,
os.path.join(settings['output'], file_name),
str(settings['concurrency'])])
subprocess.call(args)
@staticmethod
def _sm1_miner(settings) -> None:
"""
Executes splitminer for bpmn structure mining.
Returns
-------
None
DESCRIPTION.
"""
print(" -- Mining Process Structure --")
# Event log file_name
file_name = settings['file'].split('.')[0]
input_route = os.path.join(settings['output'], file_name+'.xes')
# Mining structure definition
args = ['java', '-jar', settings['sm1_path'],
str(settings['epsilon']), str(settings['eta']),
input_route,
os.path.join(settings['output'], file_name)]
subprocess.call(args)
@staticmethod
def _sm3_miner(settings) -> None:
"""
Executes splitminer for bpmn structure mining.
Returns
-------
None
DESCRIPTION.
"""
print(" -- Mining Process Structure --")
# Event log file_name
file_name = settings['file'].split('.')[0]
input_route = os.path.join(settings['output'], file_name+'.xes')
sep = ';' if pl.system().lower() == 'windows' else ':'
# Mining structure definition
args = ['java']
if not pl.system().lower() == 'windows':
args.extend(['-Xmx2G', '-Xss8G'])
args.extend(['-cp',
(settings['sm3_path']+sep+os.path.join(
'external_tools','splitminer3','lib','*')),
'au.edu.unimelb.services.ServiceProvider',
'SMD',
str(settings['epsilon']), str(settings['eta']),
'false', 'false', 'false',
input_route,
os.path.join(settings['output'], file_name)])
subprocess.call(args)
@Decorators.safe_exec
def _evaluate_alignment(self, **kwargs) -> None:
"""
Evaluates alignment
Returns
-------
None
DESCRIPTION.
"""
# load bpmn model
self.bpmn = br.BpmnReader(os.path.join(
self.settings['output'],
self.settings['file'].split('.')[0]+'.bpmn'))
self.process_graph = gph.create_process_structure(self.bpmn)
# Evaluate alignment
chk.evaluate_alignment(self.process_graph,
self.log,
self.settings)