-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathmain.py
107 lines (88 loc) · 4.81 KB
/
main.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
import numpy as np
import tensorflow as tf
from sklearn import preprocessing
from tensorflow.keras.layers import Conv2D, Activation
from sklearn.metrics import classification_report
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Flatten, Dense
from tensorflow.keras.layers import MaxPooling2D
from tensorflow.keras.optimizers import Nadam
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.layers import BatchNormalization
from ImagePPMiner import ImagePPMiner
seed = 123
np.random.seed(seed)
tf.random.set_seed(seed)
import argparse
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='CNN for next activity prediction.')
parser.add_argument('-event_log', type=str, help="Event log name")
parser.add_argument('-n_layers', type=int, help="Number of convolutional layers")
args = parser.parse_args()
dataset = args.event_log
n_layer = args.n_layers
pm = ImagePPMiner(dataset)
log = pm.import_log()
max_trace, n_caseid, n_activity = pm.dataset_summary(log=log)
train_act, train_temp, test_act, test_temp = pm.generate_prefix_trace(log=log, n_caseid=n_caseid)
X_train = pm.generate_image(act_val=train_act, time_val=train_temp, max_trace=max_trace, n_activity=n_activity)
X_test = pm.generate_image(act_val=test_act, time_val=test_temp, max_trace=max_trace, n_activity=n_activity)
l_train = pm.get_label(train_act)
l_test = pm.get_label(test_act)
df_labels = np.unique(l_train + l_test)
le = preprocessing.LabelEncoder()
le.fit_transform(df_labels)
l_train = le.transform(l_train)
l_test = le.transform(l_test)
num_classes = le.classes_.size
X_train = np.asarray(X_train)
l_train = np.asarray(l_train)
X_test = np.asarray(X_test)
l_test = np.asarray(l_test)
train_Y_one_hot = tf.keras.utils.to_categorical(l_train, num_classes)
test_Y_one_hot = tf.keras.utils.to_categorical(l_test, num_classes)
##############neural network architecture##############
model = Sequential()
reg = 0.0001
input_shape = (max_trace, n_activity, 2)
if int(n_layer) == 1:
model.add(Conv2D(32, (2, 2), input_shape=input_shape, padding='same', kernel_initializer='glorot_uniform',
kernel_regularizer=tf.keras.regularizers.l2(reg)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2)))
elif int(n_layer) == 2:
model.add(Conv2D(32, (2, 2), input_shape=input_shape, padding='same', kernel_initializer='glorot_uniform',
kernel_regularizer=tf.keras.regularizers.l2(reg)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2), padding='same'))
model.add(Conv2D(64, (4, 4), padding='same', kernel_regularizer=tf.keras.regularizers.l2(reg), ))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2), padding='same'))
elif int(n_layer) == 3:
model.add(Conv2D(32, (2, 2), input_shape=input_shape, padding='same', kernel_initializer='glorot_uniform',
kernel_regularizer=tf.keras.regularizers.l2(reg)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2), padding='same'))
model.add(Conv2D(64, (4, 4), padding='same', kernel_regularizer=tf.keras.regularizers.l2(reg), kernel_initializer='glorot_uniform'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2), padding='same'))
model.add(Conv2D(128, (8, 8), padding='same', kernel_regularizer=tf.keras.regularizers.l2(reg), kernel_initializer='glorot_uniform'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=(2, 2), padding='same'))
model.add(Flatten())
model.add(Dense(num_classes, activation='softmax', name='act_output'))
print(model.summary())
opt = Nadam(learning_rate=0.0002, beta_1=0.9, beta_2=0.999, epsilon=1e-08, schedule_decay=0.004, clipvalue=3)
model.compile(loss={'act_output': 'categorical_crossentropy'}, optimizer=opt, metrics=['accuracy'])
early_stopping = EarlyStopping(monitor='val_loss', patience=6)
history = model.fit(X_train, {'act_output': train_Y_one_hot}, validation_split=0.2, verbose=2, callbacks=[early_stopping], batch_size=128, epochs=500)
y_pred_test = model.predict(X_test)
max_y_pred_test = np.argmax(y_pred_test, axis=1)
max_y_test = np.argmax(test_Y_one_hot, axis=1)
print(classification_report(max_y_test, max_y_pred_test, digits=3))