| | import os |
| | import random |
| | import numpy as np |
| | from tqdm import tqdm |
| | from Generate_caption import load_model |
| | from tensorflow.keras.preprocessing.image import load_img, img_to_array |
| | from tensorflow.keras.applications.vgg16 import preprocess_input |
| | from tensorflow.keras.preprocessing.text import Tokenizer |
| | from tensorflow.keras.utils import pad_sequences, to_categorical |
| | from tensorflow.keras.models import Model, load_model |
| | from tensorflow.keras.layers import Input, Dense, Dropout, BatchNormalization, Embedding, GRU, add, LayerNormalization |
| | from tensorflow.keras.optimizers import Adam |
| | from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau |
| |
|
| |
|
| | def extract_image_features(model, image_folder): |
| | features = {} |
| | directory = os.path( image_folder) |
| | for item in tqdm(os.listdir(directory), desc="Extracting Features"): |
| | item_path = os.path.join(directory, item) |
| | if os.path.isfile(item_path): |
| | try: |
| | image = load_img(item_path, target_size=(224, 224)) |
| | img_array = img_to_array(image) |
| | img_array = img_array.reshape((1, img_array.shape[0], img_array.shape[1], img_array.shape[2])) |
| | img_array = preprocess_input(img_array) |
| | feature = model.predict(img_array, verbose=0) |
| | image_id = item.split('.')[0] |
| | features[image_id] = feature |
| | except Exception as e: |
| | print(f"Error processing image {item_path}: {e}") |
| | return features |
| |
|
| |
|
| | def read_captions_file(file_path): |
| | try: |
| | with open(file_path, 'r') as file: |
| | next(file) |
| | captions = file.read() |
| | return captions |
| | except Exception as e: |
| | raise RuntimeError(f"Error reading the file: {e}") |
| |
|
| |
|
| | def create_image_caption_mapping(captions): |
| | mapping = {} |
| | for line in tqdm(captions.split('\n'), desc="Processing Captions"): |
| | tokens = line.split(',') |
| | if len(tokens) < 2: |
| | continue |
| | image_id, caption = tokens[0], tokens[1:] |
| | caption = " ".join(caption) |
| | if image_id not in mapping: |
| | mapping[image_id] = [] |
| | mapping[image_id].append(caption) |
| | return mapping |
| |
|
| |
|
| | def preprocess_text(mapping): |
| | for key, captions in mapping.items(): |
| | for i in range(len(captions)): |
| | caption = captions[i].lower() |
| | caption = caption.replace('[^A-Za-z]', ' ').replace('\s+', ' ') |
| | caption = 'startseq ' + " ".join([word for word in caption.split() if len(word) > 1]) + ' endseq' |
| | captions[i] = caption |
| |
|
| |
|
| | def extract_captions(mapping): |
| | captions_list = [] |
| | for key in mapping: |
| | captions_list.extend(mapping[key]) |
| | return captions_list |
| |
|
| |
|
| | def prepare_tokenizer(captions_list): |
| | tokenizer = Tokenizer() |
| | tokenizer.fit_on_texts(captions_list) |
| | vocab_size = len(tokenizer.word_index) + 1 |
| | return tokenizer, vocab_size |
| |
|
| |
|
| | def calculate_max_length(captions_list): |
| | return max(len(caption.split()) for caption in captions_list) |
| |
|
| |
|
| | def split(image_ids, train_ratio, val_ratio=None): |
| | random.shuffle(image_ids) |
| | total = len(image_ids) |
| | train_split = int(total * train_ratio) |
| | val_split = int(total * (train_ratio + val_ratio)) if val_ratio else train_split |
| | train_ids = image_ids[:train_split] |
| | val_ids = image_ids[train_split:val_split] if val_ratio else [] |
| | test_ids = image_ids[val_split:] |
| | return train_ids, val_ids, test_ids |
| |
|
| |
|
| | def data_generator(data_keys, mapping, features, tokenizer, max_length, vocab_size, batch_size): |
| | X1, X2, y = [], [], [] |
| | n = 0 |
| | while True: |
| | for key in data_keys: |
| | n += 1 |
| | captions = mapping[key] |
| | for caption in captions: |
| | seq = tokenizer.texts_to_sequences([caption])[0] |
| | for i in range(1, len(seq)): |
| | in_seq, out_seq = seq[:i], seq[i] |
| | in_seq = pad_sequences([in_seq], maxlen=max_length, padding='post')[0] |
| | out_seq = to_categorical([out_seq], num_classes=vocab_size)[0] |
| | X1.append(features[key][0]) |
| | X2.append(in_seq) |
| | y.append(out_seq) |
| | if n == batch_size: |
| | yield {"image": np.array(X1), "text": np.array(X2)}, np.array(y) |
| | X1, X2, y = [], [], [] |
| | n = 0 |
| |
|
| |
|
| | def build_model(vocab_size, max_length): |
| | inputs1 = Input(shape=(4096,), name="image") |
| | fe1 = Dropout(0.4)(inputs1) |
| | fe2 = Dense(256, activation='relu')(fe1) |
| | fe3 = BatchNormalization()(fe2) |
| |
|
| | inputs2 = Input(shape=(max_length,), name="text") |
| | se1 = Embedding(vocab_size, 256, mask_zero=True)(inputs2) |
| | se2 = Dropout(0.4)(se1) |
| | se3 = GRU(256, recurrent_dropout=0.3, return_sequences=False)(se2) |
| |
|
| | decoder1 = add([fe3, se3]) |
| | decoder2 = LayerNormalization()(decoder1) |
| | decoder3 = Dense(512, activation='relu')(decoder2) |
| | decoder4 = Dropout(0.3)(decoder3) |
| | outputs = Dense(vocab_size, activation='softmax')(decoder4) |
| |
|
| | model = Model(inputs=[inputs1, inputs2], outputs=outputs) |
| | optimizer = Adam(learning_rate=0.001) |
| | model.compile(loss='categorical_crossentropy', optimizer=optimizer) |
| |
|
| | return model |
| |
|
| |
|
| | def load_existing_or_new_model(vocab_size, max_length, model_path="seven_version_model.keras"): |
| | if os.path.exists(model_path): |
| | print("Loading existing model...") |
| | return load_model(model_path) |
| | else: |
| | print("No existing model found. Creating a new one...") |
| | return build_model(vocab_size, max_length) |
| |
|
| |
|
| | def continue_training(model, train, val, mapping, features, tokenizer, max_length, vocab_size, batch_size, epochs): |
| | early_stopping = EarlyStopping(monitor='val_loss', patience=3, restore_best_weights=True, verbose=1) |
| | lr_scheduler = ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=2, min_lr=1e-6, verbose=1) |
| |
|
| | steps = len(train) // batch_size |
| |
|
| | for i in range(epochs): |
| | print(f"Epoch {i + 1}/{epochs}") |
| | generator = data_generator(train, mapping, features, tokenizer, max_length, vocab_size, batch_size) |
| | validation_generator = data_generator(val, mapping, features, tokenizer, max_length, vocab_size, batch_size) |
| |
|
| | model.fit(generator, validation_data=validation_generator, epochs=1, steps_per_epoch=steps, |
| | validation_steps=len(val) // batch_size, verbose=1, callbacks=[early_stopping, lr_scheduler]) |
| |
|
| | model.save("seven_version_model.keras") |
| | print("Updated model saved successfully.") |
| |
|
| | |
| | |
| | |
| | |
| |
|
