import tensorflow as tf
from tensorflow.keras.layers import Input, Conv2D, MaxPooling2D, UpSampling2D
from tensorflow.keras.models import Model
import matplotlib.pyplot as plt
from tensorflow.keras.preprocessing.image import ImageDataGenerator, load_img, img_to_array
from sklearn.metrics import confusion_matrix, classification_report
from sklearn.model_selection import train_test_split # Add this import
# Data directories
normal_dir = '/content/drive/MyDrive/9/Normal'
anomalous_dir = '/content/drive/MyDrive/9/Anomolous'
# Function to load and preprocess images
def load_and_preprocess_images(image_dir, target_size=(128, 128, 3)):
images = []
for filename in os.listdir(image_dir):
if filename.endswith('.jpeg'):
img = load_img(os.path.join(image_dir, filename), target_size=target_size)
img_array = img_to_array(img)
img_array /= 255.0 # Normalize pixel values
images.append(img_array)
return np.array(images)
# Load and preprocess data
normal_data = load_and_preprocess_images(normal_dir)
anomalous_data = load_and_preprocess_images(anomalous_dir)
'''
# Data augmentation using ImageDataGenerator
datagen = ImageDataGenerator(
rotation_range=20,
width_shift_range=0.1,
height_shift_range=0.1,
horizontal_flip=True,
vertical_flip=True,
zoom_range=0.1,
shear_range=0.1
)
# Augment the training data
batch_size = 20
augmented_normal_data = []
for batch in datagen.flow(normal_data, batch_size, shuffle=False):
augmented_normal_data.append(batch)
break
augmented_normal_data = np.array(augmented_normal_data)
# Split the augmented normal data into training and validation sets
train_normal_data, val_normal_data = train_test_split(augmented_normal_data, test_size=0.2, random_state=42)
'''
# Data augmentation using ImageDataGenerator with a larger batch size
datagen = ImageDataGenerator(
rotation_range=20,
width_shift_range=0.1,
height_shift_range=0.1,
horizontal_flip=True,
vertical_flip=True,
zoom_range=0.1,
shear_range=0.1
)
# Augment the training data with a larger batch size
augmented_normal_data = []
batch_size = 32 # You can adjust this batch size as needed
# Create an iterator for the data generator
data_generator = datagen.flow(normal_data, batch_size=batch_size, shuffle=False)
# Generate augmented samples
for i in range(len(normal_data) // batch_size):
batch = next(data_generator)
augmented_normal_data.extend(batch)
# If there are remaining samples, you can append them
if len(augmented_normal_data) < len(normal_data):
batch = next(data_generator)
augmented_normal_data.extend(batch[:len(normal_data) - len(augmented_normal_data)])
augmented_normal_data = np.array(augmented_normal_data)
# Split the augmented normal data into training and validation sets
train_normal_data, val_normal_data = train_test_split(augmented_normal_data, test_size=0.2, random_state=42)
# Define the autoencoder architecture
input_img = Input(shape=(128, 128, 3))
x = Conv2D(32, (3, 3), activation='relu', padding='same')(input_img)
x = MaxPooling2D((2, 2), padding='same')(x)
encoded = Conv2D(16, (3, 3), activation='relu', padding='same')(x)
x = Conv2D(16, (3, 3), activation='relu', padding='same')(encoded)
x = UpSampling2D((2, 2))(x)
decoded = Conv2D(3, (3, 3), activation='sigmoid', padding='same')(x)
autoencoder = Model(input_img, decoded)
autoencoder.compile(optimizer='adam', loss='mse')
# Train the autoencoder on augmented normal data
autoencoder.fit(train_normal_data, train_normal_data, epochs=50, batch_size=20, shuffle=True, validation_data=(val_normal_data, val_normal_data))
# Calculate reconstruction errors on anomalous data
reconstruction_errors = []
for sample in anomalous_data:
reconstructed_sample = autoencoder.predict(np.expand_dims(sample, axis=0))
error = np.mean(np.square(sample - reconstructed_sample))
reconstruction_errors.append(error)
# Set a threshold based on the validation data (adjust as needed)
validation_reconstruction_errors = autoencoder.evaluate(val_normal_data, val_normal_data, verbose=0)
threshold = 2.0 * validation_reconstruction_errors
# Identify anomalies
anomalous_indices = [i for i, error in enumerate(reconstruction_errors) if error > threshold]
# Visualize results
plt.figure(figsize=(12, 6))
plt.imshow(np.hstack([anomalous_data[0], autoencoder.predict(np.expand_dims(anomalous_data[0], axis=0))[0]]))
plt.title('Anomaly Detection Result')
plt.show()
print(f"Total anomalies detected: {len(anomalous_indices)}")
# Optionally, evaluate the model's performance
ground_truth = np.zeros(len(reconstruction_errors), dtype=int) # All anomalies are manually marked
predictions = np.array([1 if error > threshold else 0 for error in reconstruction_errors])
print("ground_truth",ground_truth)
print("predictions",predictions )
print("\nConfusion Matrix:\n", confusion_matrix(ground_truth, predictions))
print("\nClassification Report:\n", classification_report(ground_truth, predictions))
https://drive.google.com/drive/folders/1w7iQLa_eBAwDJLmcREQrnrLLR2LTRDRZ?usp=sharing
Comments
Post a Comment