5.8 MiB
Лабораторная 12
Задача: Мультиклассовая классификация изображений на 5 категорий (daisy, dandelion, rose, sunflower, tulip)
Ссылка на датасет: https://www.kaggle.com/datasets/rahmasleam/flowers-dataset
import os
import cv2
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import keras
os.environ["KERAS_BACKEND"] = "torch"
print(keras.__version__)
def load_images_from_folder(folder, target_size=(512, 512)):
images = []
labels = []
for label in os.listdir(folder):
if label in ['daisy', 'dandelion', 'roses', 'sunflowers', 'tulips']:
label_folder = os.path.join(folder, label)
if os.path.isdir(label_folder):
for filename in os.listdir(label_folder):
img_path = os.path.join(label_folder, filename)
img = cv2.imread(img_path)
if img is not None:
img_resized = cv2.resize(img, target_size)
images.append(img_resized)
labels.append(label)
return images, labels
folder_path = "./static/csv/dataset_flower"
images, labels = load_images_from_folder(folder_path)
num_images_to_display = min(8, len(images))
def display_images(images, labels, max_images=10):
if not images:
print("Нет изображений для отображения.")
return
count = min(max_images, len(images))
cols = 4
rows = (count + cols - 1) // cols
plt.figure(figsize=(15, 5 * rows))
for i in range(count):
plt.subplot(rows, cols, i + 1)
plt.imshow(cv2.cvtColor(images[i], cv2.COLOR_BGR2RGB))
plt.title(labels[i])
plt.axis('off')
plt.tight_layout()
plt.show()
images, labels = load_images_from_folder(folder_path)
display_images(images, labels)
# Преобразование в массивы
images = np.array(images)
labels = np.array(labels)
Предобработка изображений
import cv2
import numpy as np
import matplotlib.pyplot as plt
def preprocess_images(images):
processed_images = []
for img in images:
img_resized = cv2.resize(img, (128, 128))
img_gray = cv2.cvtColor(img_resized, cv2.COLOR_BGR2GRAY)
img_eq = cv2.equalizeHist(img_gray)
img_bin = cv2.adaptiveThreshold(img_eq, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
cv2.THRESH_BINARY, 11, 2)
processed_images.append(img_bin)
return np.array(processed_images)
processed_images = preprocess_images(images)
def display_single_image(original, processed, index):
plt.figure(figsize=(10, 5))
plt.subplot(1, 2, 1)
plt.imshow(cv2.cvtColor(original[index], cv2.COLOR_BGR2RGB))
plt.title('Оригинальное изображение')
plt.axis('off')
plt.subplot(1, 2, 2)
plt.imshow(processed[index], cmap='gray')
plt.title('Обработанное изображение')
plt.axis('off')
plt.show()
index = 0
display_single_image(images, processed_images, index)
from sklearn.model_selection import train_test_split
images_resized = np.array([cv2.resize(img, (28, 28)) for img in processed_images])
if images_resized.shape[-1] == 3: #
images_resized = np.dot(images_resized[...,:3], [0.2989, 0.5870, 0.1140]) # Формула из OpenCV
images_resized = np.expand_dims(images_resized, axis=-1) # Добавляем размерность канала
images_resized = images_resized.astype("float32") / 255.0
X_train, X_valid, y_train, y_valid = train_test_split(
images_resized, labels, test_size=0.2, stratify=labels, random_state=42
)
from sklearn.preprocessing import LabelEncoder
import tensorflow as tf
n_classes = 5
le = LabelEncoder()
y_train = le.fit_transform(y_train)
y_valid = le.transform(y_valid)
X_train = X_train.reshape(-1, 28, 28, 1).astype("float32") / 255
X_valid = X_valid.reshape(-1, 28, 28, 1).astype("float32") / 255
y_train = keras.utils.to_categorical(y_train, n_classes)
y_valid = keras.utils.to_categorical(y_valid, n_classes)
display(X_train[0])
display(y_train[0])
Проектирование архитектуры LeNet-5
from keras.api.models import Sequential
from keras.api.layers import InputLayer, Conv2D, MaxPooling2D, Dropout, Flatten, Dense
lenet_model = Sequential()
# Входной слой
lenet_model.add(InputLayer(shape=(28, 28, 1)))
# Первый скрытый слой
lenet_model.add(Conv2D(32, kernel_size=(3, 3), activation="relu"))
# Второй скрытый слой
lenet_model.add(Conv2D(64, kernel_size=(3, 3), activation="relu"))
lenet_model.add(MaxPooling2D(pool_size=(2, 2)))
# Третий скрытый слой
lenet_model.add(Flatten())
lenet_model.add(Dense(128, activation="relu"))
lenet_model.add(Dropout(0.5))
# Выходной слой
lenet_model.add(Dense(n_classes, activation="softmax"))
lenet_model.summary()
Обучение глубокой модели
lenet_model.compile(
loss="categorical_crossentropy",
optimizer="adam",
metrics=["accuracy"],
)
lenet_model.fit(
X_train,
y_train,
batch_size=128,
epochs=10,
validation_data=(X_valid, y_valid),
)
Оценка качества модели
lenet_model.evaluate(X_valid, y_valid)
Классификация текстов
import pandas as pd
import os
import win32com.client as win32
from tqdm import tqdm
def read_doc(file_path):
try:
if not os.path.exists(file_path):
raise FileNotFoundError(f"Файл {file_path} не существует")
if not file_path.lower().endswith('.doc'):
raise ValueError(f"Неверное расширение файла: {file_path}")
word = win32.Dispatch("Word.Application")
word.Visible = False
try:
doc = word.Documents.Open(os.path.abspath(file_path))
text = doc.Content.Text
doc.Close(False)
return text.strip()
except Exception as e:
print(f"Ошибка при чтении {file_path}: {e}")
return None
finally:
word.Quit()
except Exception as e:
print(f"Критическая ошибка для {file_path}: {e}")
return None
def load_docs(dataset_path):
dataset_path = os.path.abspath(dataset_path)
if not os.path.exists(dataset_path):
raise FileNotFoundError(f"Директория {dataset_path} не существует")
df = pd.DataFrame(columns=["doc", "text"])
files = [f for f in os.listdir(dataset_path)
if f.lower().endswith('.doc') and not f.startswith('~$')]
for file_name in tqdm(files, desc="Обработка документов"):
full_path = os.path.join(dataset_path, file_name)
text = read_doc(full_path)
if text:
df.loc[len(df)] = {
"doc": file_name,
"text": text
}
return df
try:
base_path = ".//static//csv//wsw"
df = load_docs(base_path)
df["type"] = df["doc"].str.lower().str.startswith("tz_").astype(int)
df.sort_values("doc", inplace=True)
display(df.head())
print(f"Успешно обработано {len(df)} из {len(files)} файлов")
except Exception as e:
print(f"Ошибка в основном потоке: {e}")
Предобработка текста
from gensim.models.phrases import Phraser, Phrases
import spacy
sp = spacy.load("ru_core_news_lg")
def prep_text(text):
doc = sp(text)
lower_sents = []
for sent in doc.sents:
lower_sents.append([word.lemma_.lower() for word in sent if not word.is_punct and not word.is_stop and not word.is_space])
lower_bigram = Phraser(Phrases(lower_sents))
clean_sents = []
for sent in lower_sents:
clean_sents.append(lower_bigram[sent])
return clean_sents
df["prep_text"] = df.apply(lambda row: prep_text(row["text"]), axis=1)
df
Инициализация Keras
import os
os.environ["KERAS_BACKEND"] = "jax"
import keras
print(keras.__version__)
Загрузка данных для классификации с помощью глубоких сетей
from keras.api.datasets import imdb
import os
from sklearn.model_selection import train_test_split
from tensorflow.keras.preprocessing.text import Tokenizer # type: ignore
from keras.api.preprocessing.sequence import pad_sequences
import numpy as np
unique_words = 10000
max_length = 200
X_texts = [' '.join([word for sent in doc for word in sent]) for doc in df['prep_text']]
tokenizer = Tokenizer(num_words=unique_words)
tokenizer.fit_on_texts(X_texts)
sequences = tokenizer.texts_to_sequences(X_texts)
X_padded = pad_sequences(sequences, maxlen=max_length, padding='post', truncating='post')
output_dir = "tmp"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
X_train, X_valid, y_train, y_valid = train_test_split(X_padded, df["type"].values, test_size=0.2, random_state=42)
Приведение отзывов к длине max_length (100)`
from keras.api.preprocessing.sequence import pad_sequences
from scipy.sparse import issparse
X_train = pad_sequences(X_train, maxlen=max_length, padding='pre', truncating='pre', value=0)
X_valid = pad_sequences(X_valid, maxlen=max_length, padding="pre", truncating="pre", value=0)
Формирование архитектуры глубокой рекуррентной сети
from keras.api.models import Sequential
from keras.api.layers import InputLayer, Embedding, SpatialDropout1D, SimpleRNN, Dense
rnn_model = Sequential()
rnn_model.add(InputLayer(shape=(max_length,), dtype="float32"))
rnn_model.add(Embedding(unique_words, 64))
rnn_model.add(SpatialDropout1D(0.2))
rnn_model.add(SimpleRNN(256, dropout=0.2))
rnn_model.add(Dense(1, activation="sigmoid"))
rnn_model.summary()
Обучение модели
from keras.api.callbacks import ModelCheckpoint
rnn_model.compile(
loss="binary_crossentropy",
optimizer="adam",
metrics=["accuracy"],
)
rnn_model.fit(
X_train,
y_train,
batch_size=128,
epochs=16,
validation_data=(X_valid, y_valid),
callbacks=[ModelCheckpoint(filepath=output_dir + "/rnn_weights.{epoch:02d}.keras")],
)
rnn_model.load_weights(output_dir + "/rnn_weights.15.keras")
rnn_model.evaluate(X_valid, y_valid)
Визуализация распределения вероятностей результатов модели на валидационной выборке
import matplotlib.pyplot as plt
plt.hist(rnn_model.predict(X_valid))
_ = plt.axvline(x=0.5, color="red")
Формирование архитектуры глубокой сверточной сети
from keras.api.models import Sequential
from keras.api.layers import InputLayer, Embedding, SpatialDropout1D, Conv1D, GlobalMaxPooling1D, Dense, Dropout
conv_model = Sequential()
conv_model.add(InputLayer(shape=(max_length,), dtype="float32"))
conv_model.add(Embedding(unique_words, 64))
conv_model.add(SpatialDropout1D(0.2))
# сверточный слой
conv_model.add(Conv1D(256, 3, activation="relu"))
conv_model.add(GlobalMaxPooling1D())
# полносвязанный слой
conv_model.add(Dense(256, activation="relu"))
conv_model.add(Dropout(0.2))
# выходной слой
conv_model.add(Dense(1, activation="sigmoid"))
conv_model.summary()
Обучение модели
from keras.api.callbacks import ModelCheckpoint
conv_model.compile(
loss="binary_crossentropy",
optimizer="adam",
metrics=["accuracy"],
)
conv_model.fit(
X_train,
y_train,
batch_size=128,
epochs=10,
validation_data=(X_valid, y_valid),
callbacks=[ModelCheckpoint(filepath=output_dir + "/conv_weights.{epoch:02d}.keras")],
)
Загрузка лучшей модели и оценка ее качества
conv_model.load_weights(output_dir + "/conv_weights.10.keras")
conv_model.evaluate(X_valid, y_valid)
Визуализация распределения вероятностей результатов модели на валидационной выборке
import matplotlib.pyplot as plt
plt.hist(conv_model.predict(X_valid))
_ = plt.axvline(x=0.5, color="red")
Формирование архитектуры глубокой полносвязанной сети
from keras.api.models import Sequential
from keras.api.layers import Dense, Flatten, Dropout, Embedding, InputLayer
simple_model = Sequential()
simple_model.add(InputLayer(shape=(max_length,), dtype="float32"))
simple_model.add(Embedding(unique_words, 64))
simple_model.add(Flatten())
simple_model.add(Dense(64, activation="relu"))
simple_model.add(Dropout(0.5))
simple_model.add(Dense(1, activation="sigmoid"))
simple_model.summary()
Обучение модели
from keras.api.callbacks import ModelCheckpoint
simple_model.compile(
loss="binary_crossentropy",
optimizer="adam",
metrics=["accuracy"],
)
simple_model.fit(
X_train,
y_train,
batch_size=128,
epochs=10,
validation_data=(X_valid, y_valid),
callbacks=[ModelCheckpoint(filepath=output_dir + "/simple_weights.{epoch:02d}.keras")],
)
Загрузка лучшей модели и оценка ее качества
simple_model.load_weights(output_dir + "/simple_weights.10.keras")
simple_model.evaluate(X_valid, y_valid)
Визуализация распределения вероятностей результатов модели на валидационной выборке
import matplotlib.pyplot as plt
plt.hist(simple_model.predict(X_valid))
_ = plt.axvline(x=0.5, color="red")