Implement pipeline some algorithm

SD (stable diffusion) for data compression

pretrained model (fine tuning parameter)
SR (super resolution) from image (512x512)

pretrained model (fine tuning parameter)
Classification traffic network

DL, ML, Graph, diffusion, GAN,…
Prediction

thinking (how to prediction and how to work) for satellite channel
Anomaly traffic

thinking (how to prediction and how to work) for satellite channel
EfficientNet Model with Action Recognition

# -*- coding: utf-8 -*-
"""rp_ml3_.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1290I3Q-_mxVngo3JuFl8JFsfeH9z6et0

# **Prepare libraries**
"""

import tensorflow as tf
import numpy as np
import pandas as pd
import os
import cv2
import matplotlib.pyplot as plt
import itertools

from keras.models import Model, load_model
from keras.applications import EfficientNetV2B0
from keras.layers import *
from keras.losses import SparseCategoricalCrossentropy
from keras.optimizers import Adam
from keras.utils import *
from sklearn.metrics import confusion_matrix

"""# **Prepare data**"""

# Set random seed for reproducibility
np.random.seed(40)

num_train_clips_per_label = 50
selected_labels = [
    'ApplyEyeMakeup', 'ApplyLipstick', 'Archery', 'BlowDryHair', 'Bowling',
    'HorseRace', 'HorseRiding', 'IceDancing', 'Mixing', 'PullUps',
    'Punch', 'Rowing',  'SkyDiving', 'Swing', 'Typing'
]

# Load and preprocess training, test, and validation data
train_df = pd.read_csv('/content/drive/MyDrive/ucf101/train.csv', sep=',')
test_df = pd.read_csv('/content/drive/MyDrive/ucf101/test.csv', sep=',')
val_df = pd.read_csv('/content/drive/MyDrive/ucf101/val.csv', sep=',')

# Filter selected labels from DataFrame
filtered_train_df = train_df[train_df['label'].isin(selected_labels)]
filtered_test_df = test_df[test_df['label'].isin(selected_labels)]
filtered_val_df = val_df[val_df['label'].isin(selected_labels)]

filtered_train_data = train_df[train_df['label'].isin(selected_labels)].groupby('label').head(num_train_clips_per_label).to_numpy()[:750]

print(f"Number of clips to training: {len(filtered_train_data)}")
print(f"Number of clips to validate: {len(filtered_val_df)}")
print(f"Number of clips to testing: {len(filtered_test_df)}")

filtered_train_df.head()

filtered_test_df.head()

filtered_val_df.head()

"""# **Preprocess data**"""

train_data = filtered_train_data
test_data = filtered_test_df.to_numpy()
val_data = filtered_val_df.to_numpy()

# Shuffle the data
np.random.shuffle(train_data)
np.random.shuffle(test_data)
np.random.shuffle(val_data)

# Get file paths for the data
train_paths = '/content/drive/MyDrive/ucf101' + train_data[:, 1]
test_paths = '/content/drive/MyDrive/ucf101' + test_data[:, 1]
val_paths = '/content/drive/MyDrive/ucf101' + val_data[:, 1]

# Get labels for the data
train_labels = train_data[:, 2]
test_labels = test_data[:, 2]
val_labels = val_data[:, 2]

# Convert labels to unique integer values
unique_labels = np.unique(val_labels)
int_unique_labels = {label: i for i, label in enumerate(unique_labels)}
train_int_labels = tf.convert_to_tensor([int_unique_labels[label] for label in train_labels], tf.float32)
test_int_labels = tf.convert_to_tensor([int_unique_labels[label] for label in test_labels], tf.float32)
val_int_labels = tf.convert_to_tensor([int_unique_labels[label] for label in val_labels], tf.float32)

# Class containing hyperparameters
class C:
    frame_count = 10
    frame_size = (224, 224)
    frame_step = 5
    batch_size = 16
    dtype = tf.float32
    channels = 3

def get_fft_img(img, img_size, mode, k):
    img = tf.image.resize_with_crop_or_pad(img, *img_size)
    trow, tcol = img_size[0] // 2, img_size[1] // 2
    channels = img.shape[2]

    if mode == "LPF":
        H = np.zeros(img_size)
        H[trow-k:trow+k-1, tcol-k:tcol+k-1] = 1
    elif mode == "HPF":
        H = np.ones(img_size)
        H[trow-k:trow+k-1, tcol-k:tcol+k-1] = 0
    else:
        print("Error mode")

    f = np.zeros_like(img, dtype=np.complex128)
    for i in range(channels):
        f[:, :, i] = np.fft.fftshift(np.fft.fft2(img[:, :, i]))
        f[:, :, i] = f[:, :, i] * H
        f[:, :, i] = np.fft.ifft2(np.fft.ifftshift(f[:, :, i]))

    g = np.abs(f).astype(np.uint8)

    return tf.image.convert_image_dtype(g, tf.float32)

# Function to extract a fixed number of frames from a video file
def get_frames(path, frame_count, frame_size, frame_step, channels):
    frames = []
    cap = cv2.VideoCapture(path)
    rlength = frame_count * frame_step
    clength = cap.get(cv2.CAP_PROP_FRAME_COUNT)
    if clength < rlength:
        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
    else:
        start = np.random.randint(0, clength - rlength + 1)
        cap.set(cv2.CAP_PROP_POS_FRAMES, start)

    # Extract frames from the video file and preprocess each frame
    for i in range(frame_count * frame_step):
        ret, frame = cap.read()
        if i % frame_step == 0:
            if ret:
                frame = get_fft_img(frame, frame_size, "LPF", 105)
                # frame = preprocess_frame(frame, frame_size)
                frames.append(frame)
            else:
                black_frame = np.zeros(frame_size + (channels,))
                frames.append(black_frame)

    cap.release()
    frames = np.array(frames)[..., [2, 1, 0]] # Convert from BGR to RGB format
    return frames

# Preprocessing function for each frame
def preprocess_frame(frame, frame_size):
    frame = tf.image.resize_with_crop_or_pad(frame, *frame_size)
    # Convert the resulting frame to a tensor of the appropriate datatype
    preprocessed_frame = tf.image.convert_image_dtype(frame, tf.float32)
    return preprocessed_frame

# Function to extract a fixed number of frames from a video file
def get_frames_0fft(path, frame_count, frame_size, frame_step, channels):
    frames = []
    cap = cv2.VideoCapture(path)
    rlength = frame_count * frame_step
    clength = cap.get(cv2.CAP_PROP_FRAME_COUNT)
    if clength < rlength:
        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
    else:
        start = np.random.randint(0, clength - rlength + 1)
        cap.set(cv2.CAP_PROP_POS_FRAMES, start)

    # Extract frames from the video file and preprocess each frame
    for i in range(frame_count * frame_step):
        ret, frame = cap.read()
        if i % frame_step == 0:
            if ret:
                frame = preprocess_frame(frame, frame_size)
                # frame = preprocess_frame(frame, frame_size)
                frames.append(frame)
            else:
                black_frame = np.zeros(frame_size + (channels,))
                frames.append(black_frame)

    cap.release()
    frames = np.array(frames)[..., [2, 1, 0]] # Convert from BGR to RGB format
    return frames

# Class to generate preprocessed data for training, validation, and testing
class DataGenerator:
    def __init__(self, paths, labels, frame_count, frame_size, frame_step, channels, fft_func=True):
        self.paths = paths
        self.labels = labels
        self.frame_count = frame_count
        self.frame_size = frame_size
        self.frame_step = frame_step
        self.channels = channels
        self.fft_func = fft_func

    def __call__(self):
        data = list(zip(self.paths, self.labels))

        for path, label in data:
            if self.fft_func:
                frame = get_frames(path, self.frame_count, self.frame_size, self.frame_step, self.channels)
                yield frame, label
            else:
                frame = get_frames_0fft(path, self.frame_count, self.frame_size, self.frame_step, self.channels)
                yield frame, label

def create_pepline(paths, labels, frame_count, frame_size, frame_step, channels, fft_func=True):
    data_generator = DataGenerator(paths, labels, frame_count, frame_size, frame_step, channels, fft_func)

    ds = tf.data.Dataset.from_generator(
        data_generator,
        output_types=(tf.float32, tf.float32),
        output_shapes=((frame_count, *frame_size, channels), ())
    )
    ds = ds.batch(batch_size=C.batch_size)
    return ds

train_ds = create_pepline(train_paths, train_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels)
test_ds = create_pepline(test_paths, test_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels)
val_ds = create_pepline(val_paths, val_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels)

train_ds_0fft = create_pepline(train_paths, train_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels, fft_func=False)
test_ds_0fft = create_pepline(test_paths, test_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels, fft_func=False)
val_ds_0fft = create_pepline(val_paths, val_int_labels, C.frame_count, C.frame_size, C.frame_step, C.channels, fft_func=False)



"""### **Show sample**"""

def display_images(images):
    fig = plt.figure(figsize=(12, 8))
    columns = 5
    rows = 2
    for i in range(1, min(len(images)+1, 11)):
        img = images[i-1]
        ax = fig.add_subplot(rows, columns, i)
        ax.imshow(img)

    plt.tight_layout()
    plt.show()

for x, y in train_ds.take(1):
    display_images(x[0])

for x, y in train_ds_0fft.take(1):
    display_images(x[0])

"""# **Create the model**"""

# Load the EfficientNetV2B0 model with pre-trained weights and freeze its layers
num_labels = len(selected_labels)

efficientnet_v2_b0 = EfficientNetV2B0(
    input_shape=(*C.frame_size, C.channels),
    include_top=False
)
efficientnet_v2_b0.trainable = False

# Input shape for the model
input_shape = (C.frame_count, *C.frame_size, C.channels)

# Define the input layer and add a rescaling layer
inputs = Input(shape=input_shape)
rescale = Rescaling(scale=255)(inputs)

# Add the TimeDistributed EfficientNetV2B0 layer and additional layers for classification
tdl = TimeDistributed(efficientnet_v2_b0)(rescale)
z = Dense(units=256, activation='relu')(tdl)
z = Dropout(0.3)(z)
# z = Dense(units=128, activation='relu')(z)
outputs = Dense(units=num_labels, activation='softmax')(z)
outputs = GlobalAveragePooling3D()(outputs)

# Compile the model with the specified loss function, optimizer, and metrics
model = Model(inputs, outputs)
model_0fft = Model(inputs, outputs)

model.compile(loss=SparseCategoricalCrossentropy(from_logits=True), optimizer=Adam(0.001), metrics=['accuracy'])
model_0fft.compile(loss=SparseCategoricalCrossentropy(from_logits=True), optimizer=Adam(0.001), metrics=['accuracy'])

model.summary()
print("--------------------------------------------------")
model_0fft.summary()

"""# **Train and save the model**"""

# Train the model
history = model.fit(train_ds, epochs=20, verbose=1, validation_data=val_ds)
model.save('model.h5')

history_0fft = model_0fft.fit(train_ds_0fft, epochs=20, verbose=1, validation_data=val_ds_0fft)
model_0fft.save('model_0fft.h5')

"""# **Evaluate**

### **Đánh giá 2 model với dữ liệu được xử lí với FFT**
"""

emodel = load_model('/content/drive/MyDrive/Report_ML2/model.h5')
emodel_0fft = load_model('/content/drive/MyDrive/Report_ML2/model_0fft.h5')

print("FFT")
loss, accuracy = emodel.evaluate(test_ds)
print(f"Loss: {loss}, Accuracy: {accuracy}\n")

print("No FFT")
loss, accuracy = emodel_0fft.evaluate(test_ds)
print(f"Loss: {loss}, Accuracy: {accuracy}")

"""### **Đánh giá 2 model với dữ liệu không dùng FFT**"""

print("FFT 2")
loss, accuracy = emodel.evaluate(test_ds_0fft)
print(f"Loss: {loss}, Accuracy: {accuracy}\n")

print("No FFT 2")
loss, accuracy = emodel_0fft.evaluate(test_ds_0fft)
print(f"Loss: {loss}, Accuracy: {accuracy}")

"""# **Result**

### **Accuracy**
"""

def plot_training_curves(history):
    loss = np.array(history.history['loss'])
    val_loss = np.array(history.history['val_loss'])

    accuracy = np.array(history.history['accuracy'])
    val_accuracy = np.array(history.history['val_accuracy'])

    epochs = range(len(history.history['loss']))

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(17, 10))

    # Plot loss
    ax1.plot(epochs, loss, label='training_loss', marker='o', color='tab:blue')
    ax1.plot(epochs, val_loss, label='val_loss', marker='o', color='tab:orange')

    ax1.fill_between(epochs, loss, val_loss, where=(loss > val_loss), color='tab:blue', alpha=0.3, interpolate=True)
    ax1.fill_between(epochs, loss, val_loss, where=(loss < val_loss), color='tab:orange', alpha=0.3, interpolate=True)

    ax1.set_title('Loss (Lower Means Better)', fontsize=16)
    ax1.set_xlabel('Epochs', fontsize=12)
    ax1.legend()

    # Plot accuracy
    ax2.plot(epochs, accuracy, label='training_accuracy', marker='o', color='tab:green')
    ax2.plot(epochs, val_accuracy, label='val_accuracy', marker='o', color='tab:red')

    ax2.fill_between(epochs, accuracy, val_accuracy, where=(accuracy > val_accuracy), color='tab:green', alpha=0.3, interpolate=True)
    ax2.fill_between(epochs, accuracy, val_accuracy, where=(accuracy < val_accuracy), color='tab:red', alpha=0.3, interpolate=True)

    ax2.set_title('Accuracy (Higher Means Better)', fontsize=16)
    ax2.set_xlabel('Epochs', fontsize=12)
    ax2.legend()

    plt.show()

# FFT
plot_training_curves(history)

# No FFT
plot_training_curves(history_0fft)

"""### **Confusion matrix**"""

ds = test_ds_0fft.take(len(test_data))
x_ds = []
y_true = []
y_pred = []
for x, y in ds:
    for frames, yt in zip(x, y):
        x_ds.append(frames)
        pred = np.argmax(emodel(np.array([frames])))
        y_pred.append(pred)
        y_true.append(yt.numpy())

ds_0fft = test_ds_0fft.take(len(test_data))
x_ds_0fft = []
y_true_0fft = []
y_pred_0fft = []
for x, y in ds_0fft:
    for frames, yt in zip(x, y):
        x_ds_0fft.append(frames)
        pred = np.argmax(emodel_0fft(np.array([frames])))
        y_pred_0fft.append(pred)
        y_true_0fft.append(yt.numpy())

y_true = np.array(y_true)
y_pred = np.array(y_pred)

y_true_0fft = np.array(y_true_0fft)
y_pred_0fft = np.array(y_pred_0fft)

print(len(y_true))
print(len(y_pred))
print(len(y_true_0fft))
print(len(y_pred_0fft))

def plot_confusion_matrix(y_true, y_pred, classes, normalize=False, title='Confusion matrix', cmap=plt.cm.Blues, figsize=(10, 8)):
    plt.figure(figsize=figsize)  # Thay đổi kích thước của hình ảnh

    cm = confusion_matrix(y_true, y_pred)

    if normalize:
        cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]

    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title)
    plt.colorbar()

    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=90)  # Xoay nhãn ở trục x
    plt.yticks(tick_marks, classes)

    fmt = '.2f' if normalize else 'd'
    thresh = cm.max() / 2.

    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        plt.text(j, i, format(cm[i, j], fmt),
                 horizontalalignment="center",
                 color="white" if cm[i, j] > thresh else "black")

    plt.tight_layout()
    plt.ylabel('True label')
    plt.xlabel('Predicted label')

classes = [
    'ApplyEyeMakeup', 'ApplyLipstick', 'Archery', 'BlowDryHair', 'Bowling',
    'HorseRace', 'HorseRiding', 'IceDancing', 'Mixing', 'PullUps',
    'Punch', 'Rowing',  'SkyDiving', 'Swing', 'Typing'
]

# FFT
plot_confusion_matrix(y_true, y_pred, classes, normalize=False)
plt.show()

# No FFT
plot_confusion_matrix(y_true_0fft, y_pred_0fft, classes, normalize=False)
plt.show()

"""### **Recall and precision**"""

def calculate_recall_precision(y_true, y_pred, num_labels):
    # Tính toán ma trận nhầm lẫn
    cm = confusion_matrix(y_true, y_pred)

    recalls = []
    precisions = []

    for label in range(num_labels):
        # Tính toán True Positive, False Positive, False Negative cho từng nhãn
        tp = cm[label, label]
        fp = np.sum(cm[:, label]) - tp
        fn = np.sum(cm[label, :]) - tp

        # Tính toán recall và precision
        recall = tp / (tp + fn)
        precision = tp / (tp + fp)

        recalls.append(recall)
        precisions.append(precision)

    return recalls, precisions

recalls, precisions = calculate_recall_precision(y_true, y_pred, num_labels)
recalls_0fft, precisions_0fft = calculate_recall_precision(y_true_0fft, y_pred_0fft, num_labels)

print("FFT")
for i, label in enumerate(unique_labels):
    print("Label {}: Recall = {:.2f}, Precision = {:.2f}".format(label, recalls[i], precisions[i]))

print(f"\nNo FFT")
for i, label in enumerate(unique_labels):
    print("Label {}: Recall = {:.2f}, Precision = {:.2f}".format(label, recalls_0fft[i], precisions_0fft[i]))

def plot_recall(recalls):
    num_labels = len(recalls)
    labels = unique_labels

    plt.figure(figsize=(8, 5))
    plt.bar(labels, recalls, width=0.4, label='Recall', color='b', alpha=0.8)
    plt.xlabel('Label')
    plt.ylabel('Recall')
    plt.title('Recall for each Label')
    plt.xticks(labels, rotation=90)  # Xoay chữ viết dọc trên trục x
    plt.ylim(0, 1.1)
    plt.show()

# FFT
plot_recall(recalls)

# No FFT
plot_recall(recalls_0fft)

def plot_precision(precisions):
    num_labels = len(precisions)
    labels = unique_labels

    plt.figure(figsize=(8, 5))
    plt.bar(labels, precisions, width=0.4, label='Precision', color='r', alpha=0.8)
    plt.xlabel('Label')
    plt.ylabel('Precision')
    plt.title('Precision for each Label')
    plt.xticks(labels, rotation=90)  # Xoay chữ viết dọc trên trục x
    plt.ylim(0, 1.1)
    plt.show()

# FFT
plot_precision(precisions)

# No FFT
plot_precision(precisions_0fft)
Tài liệu tham khảo

Internet
Hết.
Implement pipeline some algorithm

Implement pipeline some algorithm

SD (stable diffusion) for data compression

SR (super resolution) from image (512x512)

Classification traffic network

Prediction

Anomaly traffic

EfficientNet Model with Action Recognition

Tài liệu tham khảo

CATALOG

FEATURED TAGS

LINKS
