In [1]:
#1 Libraries
from __future__ import absolute_import, division, print_function, unicode_literals
import sys
import glob
import imageio
import os
import time
import datetime
import pandas as pd
import numpy as np
from math import sqrt
import argparse
import matplotlib.pyplot as plt
from pathlib import Path
from tensorflow.keras import layers
import tensorflow as tf
from src.datagens import WSIDatagen, DataSource
from src.training.experiments import SimpleExperiment, SummaryWriter
from src.training.utils import ExperimentLevel
from src.training.models import NASNetModel, ContextNASNetModel
from src.training.metrics import F1
from src.config import SLIDE_DIR
from summary import Summary
from gan_utils import plot_loss, sample_images, set_up_checkpoint
import cv2
from gan_utils import resize_and_clip
In [2]:
# Configuration
noise_dim = 10
num_examples_to_generate = 16
BATCH_SIZE = 48
BATCH_SIZE_ANOMALIES = int(BATCH_SIZE /100 *5)
EPOCHS = 60
# the epoch with which should the training be continued when resuming the training, mainly for plot and log consistency
# for a new training = 0
CONTINUE_EPOCH = 0
STEPS_PER_EPOCH = 10
STEPS_PER_EPOCH_TEST = 300
STEPS_PER_EPOCH_TEST = STEPS_PER_EPOCH_TEST//2 * 2
print('STEPS_PER_EPOCH_TEST ', STEPS_PER_EPOCH_TEST)
IMG_SIZE, GENERATED_TILE_SIZE = 512, 512
numpy_noise_seed = 42
CHANNELS = 3
DATA_SHAPE = (BATCH_SIZE, IMG_SIZE, IMG_SIZE, CHANNELS)
# if False, model_root_name does not matter
should_generate_new_stamp = False
# model_root_name = 'AnoEAN_mnist_paper'
model_root_name = 'AnoEAN_cell_4perc-ano_48batch'
should_get_pathology_data = True
should_generate_and_save_MNIST = False
should_load_MNIST = False
DATASET = 'cell'
# DATASET = 'mnist'
DATA_RANGE = 'minus_one_one'
np.random.seed(numpy_noise_seed)
global SEED
SEED = np.random.normal(0,1,(BATCH_SIZE, noise_dim))
config = {
'sample_interval': 5,
'save_interval': 5,
'noise_dim': 100,
'num_examples_to_generate': 16,
'batch_size': 48,
'epochs': 100,
"""
STEPS_PER_EPOCH and NUMBER_OF_BATCHES means the same but:
STEPS_PER_EPOCH is used for training
NUMBER_OF_BATCHES is used as a dataget parameter, to generate NUMBER_OF_BATCHES * BATCH_SIZE number of data
"""
'steps_per_epoch': 64,
'numpy_noise_seed': numpy_noise_seed,
}
# Nicely formatted time string
def hms_string(sec_elapsed):
h = int(sec_elapsed / (60 * 60))
m = int((sec_elapsed % (60 * 60)) / 60)
s = sec_elapsed % 60
return "{}:{:>02}:{:>05.2f}".format(h, m, s)
In [3]:
# CREATE DATASOURCE
if (should_get_pathology_data):
from gan_utils import get_pathology_data
train_gen, test_gen = get_pathology_data(tile_size= GENERATED_TILE_SIZE, batch_size = BATCH_SIZE,
train_steps=STEPS_PER_EPOCH, test_steps=STEPS_PER_EPOCH_TEST)
In [6]:
if (should_get_pathology_data):
print(np.max(train_gen[-2][0][0]))
print(np.min(train_gen[-2][0][0]))
plt.imshow(np.clip(train_gen[-2][0][0], a_min=0, a_max=1))
plt.show()
In [7]:
if (should_generate_new_stamp):
model_time_stamp = model_root_name + datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
model_name = model_time_stamp
summary = Summary(model_name, config, old_model_id = model_time_stamp)
sw = summary.summary_writer
sw.update_log()
else:
# 'AnoEAN_mnist_tanh_norm_2020-12-10_13:45:41' - range (-1,1)
# 'AnoEAN_mnist_tanh_norm_2020-12-08_19:05:45' - range (0,1)
# AnoEAN_cell_paper2020-12-10_22:11:02
# AnoEAN_cell_paper2021-01-02_02:03:28
model_time_stamp = 'AnoEAN_cell_4perc-ano_48batch2021-01-05_19:30:34'
model_name = model_time_stamp
summary = Summary(model_name, config, old_model_id = model_time_stamp)
sw = summary.summary_writer
sw.update_log()
print(model_name)
In [38]:
import seaborn as sns
%matplotlib inline
def sample_encoding(model, epoch, images, num_examples_to_generate=16):
# x_min, x_max = -0.2, 1.2
x_min, x_max = None, None
if (model == 'gauss'):
print('gauss')
gen_imgs = np.random.normal(0,1/2,(num_examples_to_generate, noise_dim))
exp_dir = Path('')
else:
gen_imgs = model.encoder(images, training=False)
exp_dir = model.summary.experiment_result_dir
x_min, x_max = np.min(gen_imgs), np.max(gen_imgs)
print(x_min, x_max)
print('generated_image.shape ', gen_imgs.shape)
# img_max, img_min = gen_imgs[0].max(), gen_imgs[0].min()
samples_count = num_examples_to_generate if num_examples_to_generate < gen_imgs.shape[0] else gen_imgs.shape[0]
square_root = int(sqrt(samples_count))
fig, axs = plt.subplots(square_root, square_root, figsize=(100,30))
x, y = None, None
for i in range(samples_count):
x, y = i // square_root, i % square_root
# if(model != 'gauss'):
# because sigmoid activation in the final layer projects the result into <0,1> range
if (x_min and x_max):
axs[x, y].axis(xmin = x_min, xmax = x_max)
sns.histplot(gen_imgs[i], ax=axs[x, y])
# axs[x, y].plot(gen_imgs[i])
fig.savefig(exp_dir / "{}.png".format(epoch))
plt.show()
plt.close()
# plt.plot(gen_imgs[0])
sns.displot(gen_imgs[0])
if (x_min and x_max):
plt.xlim(x_min, x_max)
# plt.axis('off')
plt.savefig(exp_dir / "{}_large.png".format(epoch), \
bbox_inches='tight', transparent=True, pad_inches=0)
plt.show()
plt.close()
sample_encoding(model='gauss', epoch='99999', images=None, num_examples_to_generate=16)
In [39]:
if (DATASET == 'cell'):
from tensorflow.keras.applications import VGG16 as Net
from tensorflow.keras.layers import Input
# from tensorflow.keras.applications import EfficientNetB0 as Net
class AnoEAN():
def __init__(self, config, summary, load_model=False):
self.gauss_mean = 0
self.gauss_stddev = 1/2
self.init_kernel = tf.keras.initializers.GlorotUniform(seed = numpy_noise_seed)
self.enc_iter = 5
self.summary = summary
self.checkpoint_dir = self.summary.experiment_result_dir / 'checkpoints'
self.encoder, self.pretrained_model = self.make_encoder_model()
self.discriminator = self.make_discriminator_model()
noise = tf.random.normal([1, IMG_SIZE, IMG_SIZE, CHANNELS])
generated_encoding = self.encoder(noise, training=False)
#
print('generated_encoding.shape ', generated_encoding.shape)
print('generated_encoding_min ', np.min((generated_encoding[0])))
print('generated_encoding_max ', np.max((generated_encoding[0])))
# plt.plot(generated_encoding[0])
# plt.show()
#
decision = self.discriminator(generated_encoding)
print("decision:", decision)
# This method returns a helper function to compute cross entropy loss
self.cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)
self.encoder_optimizer = tf.keras.optimizers.Adam(learning_rate=0.0001)
self.discriminator_optimizer = tf.keras.optimizers.Adam(learning_rate=0.0001)
self.summary.summary_writer.set_value('encoder', value=self.encoder.to_json())
self.summary.summary_writer.update_log()
self.summary.summary_writer.set_value('discriminator', value=self.discriminator.to_json())
self.summary.summary_writer.update_log()
# sets up:
# model.checkpoint_dir - path to checkpoints dir
# model.checkpoint - tf.train.Checkpoint object
# model.checkpoint_manager - tf.train.CheckpointManager object
if (load_model == False):
set_up_checkpoint(self, model_type='AnoEAN', max_to_keep=50)
else:
# self = self.load_checkpoint()
self.load_checkpoint()
def load_checkpoint(self):
self.checkpoint = tf.train.Checkpoint(
step=tf.Variable(22),
encoder_optimizer = self.encoder_optimizer,
discriminator_optimizer = self.discriminator_optimizer,
encoder = self.encoder,
discriminator = self.discriminator,
)
latest = tf.train.latest_checkpoint(self.checkpoint_dir)
print(latest)
self.checkpoint.restore(tf.train.latest_checkpoint(self.checkpoint_dir))
# self.checkpoint.assert_consumed()
self.checkpoint_manager = tf.train.CheckpointManager(self.checkpoint, self.checkpoint_dir, max_to_keep=3)
return self.checkpoint
#3 Encoder
def make_encoder_model(self):
inp = Input(shape=(IMG_SIZE, IMG_SIZE, CHANNELS), batch_size=BATCH_SIZE, name='input')
pretrained_model = Net(include_top=False, input_tensor=inp, weights='imagenet', pooling='max')
pretrained_model.trainable = False
# Apply regularization on pretrained network ?
# Add regularization in general?
out = pretrained_model(inp)
out = layers.Dropout(0.3)(out)
out = layers.Dense(512, activation=tf.keras.layers.LeakyReLU(), kernel_initializer=self.init_kernel)(out)
out = layers.Dense(256, activation=tf.keras.layers.LeakyReLU(), kernel_initializer=self.init_kernel)(out)
out = layers.Dense(noise_dim, activation='linear', kernel_initializer=self.init_kernel)(out)
model = tf.keras.Model(inp, out)
return model, pretrained_model
#4 Discriminator
def make_discriminator_model(self):
model = tf.keras.Sequential()
model.add(layers.Dense(units=32, input_shape=([noise_dim]), kernel_initializer= self.init_kernel))
model.add(layers.LayerNormalization())
model.add(layers.LeakyReLU())
model.add(layers.Dropout(0.3))
model.add(layers.Dense(16,
kernel_initializer= self.init_kernel))
model.add(layers.LayerNormalization())
model.add(layers.LeakyReLU())
model.add(layers.Dropout(0.3))
model.add(layers.Flatten())
model.add(layers.Dense(1, activation='sigmoid',
kernel_initializer= self.init_kernel))
return model
@tf.function
def get_noise(self):
noise = tf.random.normal([BATCH_SIZE, noise_dim], mean=self.gauss_mean, stddev=self.gauss_stddev)
return noise
#5 Losses
def discriminator_loss(self, real_output, fake_output, anomaly_output):
real_loss = 2 * self.cross_entropy(tf.ones_like(real_output), real_output)
# real_loss = self.cross_entropy(tf.ones_like(real_output), real_output)
fake_loss = self.cross_entropy(tf.zeros_like(fake_output), fake_output)
anomaly_loss = self.cross_entropy(tf.zeros_like(anomaly_output), anomaly_output)
# should not be total_loss = (real_loss + fake_loss) * 0.5 ?
# total_loss = real_loss*1.4 + fake_loss*1.4 + anomaly_loss*0.2
total_loss = real_loss + fake_loss + anomaly_loss
# total_loss, Dz_loss, DEx_normal_loss, DEx_anomaly_loss
return total_loss, real_loss, fake_loss, anomaly_loss
def encoder_loss(self, fake_output, anomaly_output):
fake_loss = self.cross_entropy(tf.ones_like(fake_output), fake_output)
anomaly_loss = self.cross_entropy(tf.zeros_like(anomaly_output), anomaly_output)
# total_loss = fake_loss*1.8 + anomaly_loss*0.2
# total_loss = fake_loss + anomaly_loss*1.4
total_loss = fake_loss + anomaly_loss
return total_loss, fake_loss, anomaly_loss
#6 Train def
# @tf.function causes the function to be compiled
@tf.function
def train_step_encoder(self, images, anomaly_images):
print('images.shape: ', images.shape)
# this was used for previous model:
# noise = tf.random.normal([BATCH_SIZE, noise_dim], mean=0, stddev=1, dtype=tf.dtypes.float32)
noise = self.get_noise()
with tf.GradientTape() as enc_tape, tf.GradientTape() as disc_tape:
generated_encoding = self.encoder(images, training=True)
anomaly_encoding = self.encoder(anomaly_images, training=True)
fake_output = self.discriminator(generated_encoding, training=False)
anomaly_output = self.discriminator(anomaly_encoding, training=False)
enc_loss, fake_loss, anomaly_loss = self.encoder_loss(fake_output, anomaly_output)
gradients_of_encoder = enc_tape.gradient(enc_loss, self.encoder.trainable_variables)
self.encoder_optimizer.apply_gradients(zip(gradients_of_encoder, self.encoder.trainable_variables))
return enc_loss, fake_loss, anomaly_loss
# @tf.function causes the function to be compiled
@tf.function
def train_step_discriminator(self, images, anomaly_images):
print('images.shape: ', images.shape)
# this was used for previous model:
# noise = tf.random.normal([BATCH_SIZE, noise_dim], mean=0, stddev=1, dtype=tf.dtypes.float32)
noise = self.get_noise()
with tf.GradientTape() as enc_tape, tf.GradientTape() as disc_tape:
generated_encoding = self.encoder(images, training=False)
anomaly_encoding = self.encoder(anomaly_images, training=False)
# real_output = self.discriminator(images, training=True)
real_output = self.discriminator(noise, training=True)
fake_output = self.discriminator(generated_encoding, training=True)
anomaly_output = self.discriminator(anomaly_encoding, training=True)
# total_loss, Dz_loss, DEx_normal_loss, DEx_anomaly_loss
# total_loss, real_loss, fake_loss, anomaly_loss
disc_loss, Dz_loss, DEx_normal_loss, DEx_anomaly_loss = self.discriminator_loss(
real_output,
fake_output,
anomaly_output
)
gradients_of_discriminator = disc_tape.gradient(disc_loss, self.discriminator.trainable_variables)
self.discriminator_optimizer.apply_gradients(zip(gradients_of_discriminator, self.discriminator.trainable_variables))
return disc_loss, Dz_loss, DEx_normal_loss, DEx_anomaly_loss, fake_output, anomaly_output, real_output
def train(self, datagen, anomaly_datagen=None, epochs=0, should_train=True):
LABEL_INDEX = 1
DATA_INDEX = 0
start = time.time()
# statistics of loss functions
gen_loss_stat, disc_loss_stat = [], []
Dz_loss_stat, DEx_normal_loss_stat, DEx_anomaly_loss_stat = [], [], []
fake_output_stat, anomaly_output_stat, real_output_stat = [], [], []
"""check if the labels are correct"""
datagen.on_epoch_end(labels = [0])
datagen.on_epoch_end_anomalous()
print('normal labels: ', datagen[0][LABEL_INDEX])
print('anomaly labels: ', datagen.anomalous(0)[LABEL_INDEX])
if should_train:
for epoch in range(CONTINUE_EPOCH, CONTINUE_EPOCH + epochs):
print('pretrained_model.trainable: ', self.pretrained_model.trainable)
if(DATASET.lower() == 'cell'):
# __getitem__ is set up for normal data
# datagen.anomalous(index) is a "get item" for anomalous data
datagen.on_epoch_end(labels = [0])
datagen.on_epoch_end_anomalous()
epoch_start = time.time()
gen_loss, disc_loss = [], []
Dz_loss_batch, DEx_normal_loss_batch, DEx_anomaly_loss_batch = [], [], []
fake_output_batch, anomaly_output_batch, real_output_batch = [], [], []
gen_fake_loss, gen_anomaly_loss = [], []
for batch_num in range(STEPS_PER_EPOCH):
elif(DATASET.lower() == 'cell'):
batch = datagen[batch_num][DATA_INDEX]
anomaly_batch = datagen.anomalous(batch_num)[DATA_INDEX][0:BATCH_SIZE_ANOMALIES]
# disc_loss, Dz_loss, DEx_normal_loss, DEx_anomaly_loss, fake_output, anomaly_output, real_output
disc_loss_sample, Dz_loss_sample, DEx_normal_loss_sample, DEx_anomaly_loss_sample, fake_output_sample, anomaly_output_sample , real_output_sample \
= self.train_step_discriminator(
batch,
anomaly_batch
)
disc_loss.append(disc_loss_sample)
Dz_loss_batch.append(Dz_loss_sample)
DEx_normal_loss_batch.append(DEx_normal_loss_sample)
DEx_anomaly_loss_batch.append(DEx_anomaly_loss_sample)
fake_output_batch.append(fake_output_sample)
anomaly_output_batch.append(anomaly_output_sample)
real_output_batch.append(real_output_sample)
for i in range(self.enc_iter):
gen_loss_sample, gen_fake_loss_sample, gen_anomaly_loss_sample = \
self.train_step_encoder(batch, anomaly_batch)
gen_loss.append(gen_loss_sample)
gen_fake_loss.append(gen_fake_loss_sample.numpy())
gen_anomaly_loss.append(gen_anomaly_loss_sample.numpy())
# print('loss: disc: ', disc_loss_sample, 'gen: ', gen_loss_sample)
g_loss = sum(gen_loss) / len(gen_loss) if len(gen_loss)!=0 else None
d_loss = sum(disc_loss) / len(disc_loss)
Dz_loss_sum = sum(Dz_loss_batch) / len(Dz_loss_batch)
DEx_n_loss_sum = sum(DEx_normal_loss_batch) / len(DEx_normal_loss_batch)
DEx_a_loss_sum = sum(DEx_anomaly_loss_batch) / len(DEx_anomaly_loss_batch)
fake_output_sum = np.mean(fake_output_batch)
anomaly_output_sum = np.mean(anomaly_output_batch)
real_output_sum = np.mean(real_output_batch)
gen_fake_loss_sum = sum(gen_fake_loss) / len(gen_fake_loss)
gen_anomaly_loss_sum = sum(gen_anomaly_loss) / len(gen_anomaly_loss)
epoch_elapsed = time.time()-epoch_start
# print (
# 'Epoch {} loss, gen={}, disc={}, D(z)={}, D(E(x_n))={}, D(E(x_a))={}, {}'.format(
# epoch+1, g_loss, d_loss, Dz_loss_sum, DEx_n_loss_sum, DEx_a_loss_sum, hms_string(epoch_elapsed))
# )
# print('fake_output, anomaly_output, real_output: {}, {}, {}'.format(
# fake_output_sum, anomaly_output_sum, real_output_sum)
# )
gen_loss_stat.append(g_loss)
disc_loss_stat.append(d_loss)
Dz_loss_stat.append(Dz_loss_sum)
DEx_normal_loss_stat.append(DEx_n_loss_sum)
DEx_anomaly_loss_stat.append(DEx_a_loss_sum)
fake_output_stat.append(fake_output_sum)
anomaly_output_stat.append(anomaly_output_sum)
real_output_stat.append(real_output_sum)
# Log
# d_loss_list = list(map(lambda x: str(x.item()), d_loss))
if(g_loss != None):
self.summary.summary_writer.add_value('loss', value=[epoch, str(d_loss.numpy()), str(g_loss.numpy())])
self.summary.summary_writer.add_value('partial_loss', value=[epoch, str(Dz_loss_sum.numpy()), str(DEx_n_loss_sum.numpy()), str(DEx_a_loss_sum.numpy())])
self.summary.summary_writer.update_log()
# save images and checkpoints
if (epoch + 1) % config['sample_interval'] == 0:
# sample_images(self, epoch, SEED, num_examples_to_generate)
sample_encoding(self, epoch, SEED_IMAGES, num_examples_to_generate)
if (epoch +1) % config['sample_interval'] == 0:
plot_loss(self, [gen_loss_stat, disc_loss_stat], ["Encoder", "Discriminator"], epoch)
plot_loss(self, [Dz_loss_stat, DEx_normal_loss_stat, DEx_anomaly_loss_stat], ["D(z)", "D(E(x_n))", "D(E(x_a))"], epoch='{}_partial_loss'.format(epoch))
plot_loss(self, [fake_output_stat, anomaly_output_stat, real_output_stat], ["fake", "anomaly", "real"], epoch='{}_disc_decision'.format(epoch), labels = ['Epoch', 'Prediction'])
self.checkpoint.step.assign_add(1)
if (epoch + 1) % config['save_interval'] == 0:
save_path = self.checkpoint_manager.save()
print("Saved checkpoint for step {}: {}".format(int(self.checkpoint.step), save_path))
# self.checkpoint.save(file_prefix = self.checkpoint_prefix / 'ckpt_{}'.format(epoch + 1))
if(DATASET.lower() == 'mnist'):
np.random.shuffle(datagen)
# final images
sample_encoding(self, epochs, SEED_IMAGES, num_examples_to_generate)
return gen_loss_stat, disc_loss_stat
In [40]:
if('SEED_IMAGES' not in globals()):
global SEED_IMAGES
if(DATASET.lower() == 'mnist'):
SEED_IMAGES = train_dataset[0][0:num_examples_to_generate//2]
SEED_IMAGES = np.concatenate(
(SEED_IMAGES,
anomaly_train_dataset[0][0:num_examples_to_generate//2])
)
elif(DATASET.lower() == 'cell'):
train_gen.on_epoch_end(labels = [0])
train_gen.on_epoch_end_anomalous()
# SEED_IMAGES = train_gen[0][0][0:num_examples_to_generate//2]
# print(SEED_IMAGES.shape)
SEED_IMAGES = np.concatenate(
(train_gen[0][0][0:num_examples_to_generate//2],
train_gen.anomalous(0)[0][0:num_examples_to_generate//2])
)
print(num_examples_to_generate)
print(SEED_IMAGES.shape)
In [24]:
%matplotlib inline
gan=None
SHOULD_LOAD_MODEL = True
"""moved to run method"""
# SHOULD_TRAIN = True
print(tf.config.experimental.list_physical_devices())
def run_train():
SHOULD_TRAIN = True
""" Runs the traning process"""
gan = AnoEAN(config, summary, load_model=SHOULD_LOAD_MODEL)
if(DATASET.lower() == 'mnist'):
gen_loss_stat, disc_loss_stat = gan.train(train_dataset, anomaly_train_dataset, epochs=EPOCHS, should_train=True)
elif(DATASET.lower() == 'cell'):
gen_loss_stat, disc_loss_stat = gan.train(train_gen, epochs=EPOCHS, should_train=SHOULD_TRAIN)
print(gen_loss_stat, disc_loss_stat)
return gan
def run_test():
SHOULD_TRAIN = False
""" Runs the training process"""
gan = AnoEAN(config, summary, load_model=SHOULD_LOAD_MODEL)
if(DATASET.lower() == 'mnist'):
gen_loss_stat, disc_loss_stat = gan.train(train_dataset, anomaly_train_dataset, epochs=EPOCHS, should_train=SHOULD_TRAIN)
elif(DATASET.lower() == 'cell'):
gen_loss_stat, disc_loss_stat = gan.train(train_gen, epochs=EPOCHS, should_train=SHOULD_TRAIN)
print(gen_loss_stat, disc_loss_stat)
return gan
# gan = run_train()
gan = run_test()
In [25]:
from scipy.spatial.distance import mahalanobis
elif(DATASET.lower() == 'cell'):
"""generate encodings"""
test_gen.on_epoch_end(labels = [0])
test_gen.on_epoch_end_anomalous()
LABEL_INDEX = 1
DATA_INDEX = 0
TEST_BATCH_COUNT_normal = 3200 // BATCH_SIZE
TEST_BATCH_COUNT_anomaly = 800 // BATCH_SIZE
normal_test_set = test_gen[0][DATA_INDEX]
test_set_y = test_gen[0][LABEL_INDEX]
array_test = gan.encoder(normal_test_set, training=False)
for i in range(1, TEST_BATCH_COUNT_normal):
sample = gan.encoder(test_gen[i][DATA_INDEX], training=False)
array_test = np.concatenate((array_test, sample))
test_set_y = np.concatenate((test_set_y, test_gen[i][LABEL_INDEX]))
for i in range(TEST_BATCH_COUNT_anomaly):
sample = gan.encoder(test_gen.anomalous(i)[DATA_INDEX], training=False)
array_test = np.concatenate((array_test, sample))
test_set_y = np.concatenate((test_set_y, test_gen.anomalous(i)[LABEL_INDEX]))
print(test_set_y)
test_length = array_test.shape[0]
print(test_length)
array_seed = gan.encoder(SEED_IMAGES, training=False)
test_dataset_y = test_set_y
print(len(test_set_y))
V = np.cov(array_test[0:test_length//2].T)
# IV = np.linalg.inv(V)
mean = 0
distance_normal = []
for i in range(test_length//2):
distance_normal.append(mahalanobis(array_test[i], mean, V))
print('distance_normal ', np.mean(distance_normal))
distance_anomaly = []
for i in range(test_length//2, test_length):
distance_anomaly.append(mahalanobis(array_test[i], mean, V))
print('distance_anomaly ', np.mean(distance_anomaly))
distance_SEED = []
for i in range(array_seed.shape[0]):
distance_SEED.append(mahalanobis(array_seed[i], mean, V))
print('distance_SEED ', np.mean(distance_SEED))
# array_test = gan.encoder(test_set, training=False)
distance_test = []
for i in range(test_length):
distance_test.append(mahalanobis(array_test[i], mean, V))
print('distance_test ', np.mean(distance_test))
threshold = np.mean((distance_normal, distance_anomaly))
print('threshold ', threshold)
distance_test_result = []
for i in range(len(distance_test)):
if (distance_test[i] >= threshold):
distance_test_result.append(1)
else:
distance_test_result.append(0)
print(len(distance_test_result))
In [27]:
c = 20
top_percentile = np.percentile(np.array(distance_test), 100 - c)
threshold = top_percentile
print('threshold ', threshold)
distance_test_result = []
for i in range(len(distance_test)):
if (distance_test[i] >= threshold):
distance_test_result.append(1)
else:
distance_test_result.append(0)
from sklearn import metrics
# metrics.confusion_matrix(y_true, y_pred)
confusion_matrix = metrics.confusion_matrix(test_set_y, distance_test_result, normalize='all')
# https://scikit-learn.org/stable/modules/generated/sklearn.metrics.confusion_matrix.html
TN = confusion_matrix[0][0]
FN = confusion_matrix[1][0]
TP = confusion_matrix[1][1]
FP = confusion_matrix[0][1]
# Sensitivity, hit rate, recall, or true positive rate
TPR = TP/(TP+FN)
# Specificity or true negative rate
TNR = TN/(TN+FP)
# Precision or positive predictive value
PPV = TP/(TP+FP)
# Negative predictive value
NPV = TN/(TN+FN)
# Fall out or false positive rate
FPR = FP/(FP+TN)
# False negative rate
FNR = FN/(TP+FN)
# False discovery rate
FDR = FP/(TP+FP)
# Overall accuracy
ACC = (TP+TN)/(TP+FP+FN+TN)
# F1 score
# F1 = 2 * (precision * recall) / (precision + recall)
F1 = 2 * (PPV * TPR) / (PPV + TPR)
print('{:0.4f} FP'.format(FP))
print('{:0.4f} FN'.format(FN))
print('{:0.4f} TP'.format(TP))
print('{:0.4f} TN'.format(TN))
print('{:0.4f} TPR = TP/(TP+FN) Recall'.format(TPR))
print('{:0.4f} TNR = TN/(TN+FP) '.format(TNR))
print('{:0.4f} PPV = TP/(TP+FP) Precision'.format(PPV))
print('{:0.4f} NPV = TN/(TN+FN) '.format(NPV))
print('{:0.4f} FPR = FP/(FP+TN) '.format(FPR))
print('{:0.4f} FNR = FN/(TP+FN) '.format(FNR))
print('{:0.4f} FDR = FP/(TP+FP) '.format(FDR))
print('{:0.4f} ACC = (TP+TN)/(TP+FP+FN+TN) accuracy'.format(ACC))
print('{:0.4f} F1 = 2 * (PPV * TPR) / (PPV + TPR) F1-score'.format(F1))
cmd = metrics.ConfusionMatrixDisplay(confusion_matrix, display_labels=['normal','anomalous'])
cmd.plot(cmap='Blues')
plt.grid(False)
In [29]:
from evaluations import do_hist, do_roc, do_prc
iteration='4'
name= model_name + '_i{}_'.format(iteration)
dir_name='AnoEAN/AnoEAN_16perc_ano_in_training-20perc_in_test-continue/'
if not os.path.exists(dir_name):
os.makedirs(dir_name)
print(name)
print(dir_name)
cmd.plot(cmap='Blues')
plt.grid(False)
plt.savefig(dir_name + 'confusion_{}_{}.png'.format(DATASET, name),
transparent=True, bbox_inches='tight')
with open('{}results_{}.txt'.format(dir_name, name), 'w') as outfile:
print('{:0.4f} FP'.format(FP), file = outfile)
print('{:0.4f} FN'.format(FN), file = outfile)
print('{:0.4f} TP'.format(TP), file = outfile)
print('{:0.4f} TN'.format(TN), file = outfile)
print('{:0.4f} TPR = TP/(TP+FN) Recall'.format(TPR), file = outfile)
print('{:0.4f} TNR = TN/(TN+FP) '.format(TNR), file = outfile)
print('{:0.4f} PPV = TP/(TP+FP) Precision'.format(PPV), file = outfile)
print('{:0.4f} NPV = TN/(TN+FN) '.format(NPV), file = outfile)
print('{:0.4f} FPR = FP/(FP+TN) '.format(FPR), file = outfile)
print('{:0.4f} FNR = FN/(TP+FN) '.format(FNR), file = outfile)
print('{:0.4f} FDR = FP/(TP+FP) '.format(FDR), file = outfile)
print('{:0.4f} ACC = (TP+TN)/(TP+FP+FN+TN) accuracy'.format(ACC), file = outfile)
print('{:0.4f} F1 = 2 * (PPV * TPR) / (PPV + TPR) F1-score'.format(F1), file = outfile)
print('{} threshold'.format(threshold), file = outfile)
if(top_percentile == threshold):
print('{} %(percent) of highest scores'.format(c), file = outfile)
do_prc(scores=np.array(distance_test), true_labels=test_set_y, file_name=name, directory=dir_name)
do_roc(scores=np.array(distance_test), true_labels=test_set_y, file_name=name, directory=dir_name)
do_hist(scores=np.array(distance_test), true_labels=test_set_y, directory=dir_name, dataset=DATASET,
random_seed='', file_name=name)
