File size: 10,181 Bytes
17c1e65 946c8a9 a55a660 17c1e65 97a9791 17c1e65 97a9791 a2b2a3a 755501b a2b2a3a 97a9791 a2b2a3a 97a9791 a2b2a3a 97a9791 a2b2a3a 97a9791 a2b2a3a 946c8a9 c316009 31229df 946c8a9 17c1e65 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 |
import pandas as pd
from collections import Counter
import json
import os
from PIL import Image
import numpy as np
import torch
import matplotlib.pyplot as plt
from IPython import get_ipython
import sys
import gc
import streamlit as st
class VQADataProcessor:
"""
A class to process OKVQA dataset.
Attributes:
questions_file_path (str): The file path for the questions JSON file.
annotations_file_path (str): The file path for the annotations JSON file.
questions (list): List of questions extracted from the JSON file.
annotations (list): List of annotations extracted from the JSON file.
df_questions (DataFrame): DataFrame created from the questions list.
df_answers (DataFrame): DataFrame created from the annotations list.
merged_df (DataFrame): DataFrame resulting from merging questions and answers.
"""
def __init__(self, questions_file_path, annotations_file_path):
"""
Initializes the VQADataProcessor with file paths for questions and annotations.
Parameters:
questions_file_path (str): The file path for the questions JSON file.
annotations_file_path (str): The file path for the annotations JSON file.
"""
self.questions_file_path = questions_file_path
self.annotations_file_path = annotations_file_path
self.questions, self.annotations = self.read_json_files()
self.df_questions = pd.DataFrame(self.questions)
self.df_answers = pd.DataFrame(self.annotations)
self.merged_df = None
def read_json_files(self):
"""
Reads the JSON files for questions and annotations.
Returns:
tuple: A tuple containing two lists: questions and annotations.
"""
with open(self.questions_file_path, 'r') as file:
data = json.load(file)
questions = data['questions']
with open(self.annotations_file_path, 'r') as file:
data = json.load(file)
annotations = data['annotations']
return questions, annotations
@staticmethod
def find_most_frequent(my_list):
"""
Finds the most frequent item in a list.
Parameters:
my_list (list): A list of items.
Returns:
The most frequent item in the list. Returns None if the list is empty.
"""
if not my_list:
return None
counter = Counter(my_list)
most_common = counter.most_common(1)
return most_common[0][0]
def merge_dataframes(self):
"""
Merges the questions and answers DataFrames on 'question_id' and 'image_id'.
"""
self.merged_df = pd.merge(self.df_questions, self.df_answers, on=['question_id', 'image_id'])
def join_words_with_hyphen(self, sentence):
return '-'.join(sentence.split())
def process_answers(self):
"""
Processes the answers by extracting raw and processed answers and finding the most frequent ones.
"""
if self.merged_df is not None:
self.merged_df['raw_answers'] = self.merged_df['answers'].apply(lambda x: [ans['raw_answer'] for ans in x])
self.merged_df['processed_answers'] = self.merged_df['answers'].apply(
lambda x: [ans['answer'] for ans in x])
self.merged_df['most_frequent_raw_answer'] = self.merged_df['raw_answers'].apply(self.find_most_frequent)
self.merged_df['most_frequent_processed_answer'] = self.merged_df['processed_answers'].apply(
self.find_most_frequent)
self.merged_df.drop(columns=['answers'], inplace=True)
else:
print("DataFrames have not been merged yet.")
# Apply the function to the 'most_frequent_processed_answer' column
self.merged_df['single_word_answers'] = self.merged_df['most_frequent_processed_answer'].apply(
self.join_words_with_hyphen)
def get_processed_data(self):
"""
Retrieves the processed DataFrame.
Returns:
DataFrame: The processed DataFrame. Returns None if the DataFrame is empty or not processed.
"""
if self.merged_df is not None:
return self.merged_df
else:
print("DataFrame is empty or not processed yet.")
return None
def save_to_csv(self, df, saved_file_name):
if saved_file_name is not None:
if ".csv" not in saved_file_name:
df.to_csv(os.path.join(saved_file_name, ".csv"), index=None)
else:
df.to_csv(saved_file_name, index=None)
else:
df.to_csv("data.csv", index=None)
def display_dataframe(self):
"""
Displays the processed DataFrame.
"""
if self.merged_df is not None:
print(self.merged_df)
else:
print("DataFrame is empty.")
def process_okvqa_dataset(questions_file_path, annotations_file_path, save_to_csv=False, saved_file_name=None):
"""
Processes the OK-VQA dataset given the file paths for questions and annotations.
Parameters:
questions_file_path (str): The file path for the questions JSON file.
annotations_file_path (str): The file path for the annotations JSON file.
Returns:
DataFrame: The processed DataFrame containing merged and processed VQA data.
"""
# Create an instance of the class
processor = VQADataProcessor(questions_file_path, annotations_file_path)
# Process the data
processor.merge_dataframes()
processor.process_answers()
# Retrieve the processed DataFrame
processed_data = processor.get_processed_data()
if save_to_csv:
processor.save_to_csv(processed_data, saved_file_name)
return processed_data
def show_image(image):
"""
Display an image in various environments (Jupyter, PyCharm, Hugging Face Spaces).
Handles different types of image inputs (file path, PIL Image, numpy array, OpenCV, PyTorch tensor).
Args:
image (str or PIL.Image or numpy.ndarray or torch.Tensor): The image to display.
"""
in_jupyter = is_jupyter_notebook()
in_colab = is_google_colab()
# Convert image to PIL Image if it's a file path, numpy array, or PyTorch tensor
if isinstance(image, str):
if os.path.isfile(image):
image = Image.open(image)
else:
raise ValueError("File path provided does not exist.")
elif isinstance(image, np.ndarray):
if image.ndim == 3 and image.shape[2] in [3, 4]:
image = Image.fromarray(image[..., ::-1] if image.shape[2] == 3 else image)
else:
image = Image.fromarray(image)
elif torch.is_tensor(image):
image = Image.fromarray(image.permute(1, 2, 0).numpy().astype(np.uint8))
# Display the image
if in_jupyter or in_colab:
from IPython.display import display
display(image)
else:
image.show()
def show_image_with_matplotlib(image):
if isinstance(image, str):
image = Image.open(image)
elif isinstance(image, np.ndarray):
image = Image.fromarray(image)
elif torch.is_tensor(image):
image = Image.fromarray(image.permute(1, 2, 0).numpy().astype(np.uint8))
plt.imshow(image)
plt.axis('off') # Turn off axis numbers
plt.show()
def is_jupyter_notebook():
"""
Check if the code is running in a Jupyter notebook.
Returns:
bool: True if running in a Jupyter notebook, False otherwise.
"""
try:
from IPython import get_ipython
if 'IPKernelApp' not in get_ipython().config:
return False
if 'ipykernel' in str(type(get_ipython())):
return True # Running in Jupyter Notebook
except (NameError, AttributeError):
return False # Not running in Jupyter Notebook
return False # Default to False if none of the above conditions are met
def is_pycharm():
return 'PYCHARM_HOSTED' in os.environ
def is_google_colab():
return 'COLAB_GPU' in os.environ or 'google.colab' in sys.modules
def get_image_path(name, path_type):
"""
Generates a path for models, images, or data based on the specified type.
Args:
name (str): The name of the model, image, or data folder/file.
path_type (str): The type of path needed ('models', 'images', or 'data').
Returns:
str: The full path to the specified resource.
"""
# Get the current working directory (assumed to be inside 'code' folder)
current_dir = os.getcwd()
# Get the directory one level up (the parent directory)
parent_dir = os.path.dirname(current_dir)
# Construct the path to the specified folder
folder_path = os.path.join(parent_dir, path_type)
# Construct the full path to the specific resource
full_path = os.path.join(folder_path, name)
return full_path
def get_model_path(model_name):
"""
Get the path to the specified model folder.
Args:
model_name (str): Name of the model folder.
Returns:
str: Absolute path to the specified model folder.
"""
# Directory of the current script
current_script_dir = os.path.dirname(os.path.abspath(__file__))
# Directory of the 'app' folder (parent of the 'my_model' folder)
app_dir = os.path.dirname(os.path.dirname(current_script_dir))
# Path to the 'models/{model_name}' folder
model_path = os.path.join(app_dir, "models", model_name)
return model_path
def free_gpu_resources():
"""
Clears GPU memory.
"""
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.empty_cache()
gc.collect()
gc.collect()
if __name__ == "__main__":
pass
#val_data = process_okvqa_dataset('OpenEnded_mscoco_val2014_questions.json', 'mscoco_val2014_annotations.json', save_to_csv=True, saved_file_name="okvqa_val.csv")
#train_data = process_okvqa_dataset('OpenEnded_mscoco_train2014_questions.json', 'mscoco_train2014_annotations.json', save_to_csv=True, saved_file_name="okvqa_train.csv")
|