File size: 8,414 Bytes
c996cf4 |
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 |
import streamlit as st
import json
from collections import Counter
import contractions
import csv
import altair as alt
from typing import Tuple, List, Optional
from my_model.dataset.dataset_processor import process_okvqa_dataset
from my_model.config import dataset_config as config
class OKVQADatasetAnalyzer:
"""
Provides tools for analyzing and visualizing distributions of question types within given question datasets.
It supports operations such as data loading, categorization of questions based on keywords, visualization of q
uestion distribution, and exporting data to CSV files.
Attributes:
train_file_path (str): Path to the training dataset file.
test_file_path (str): Path to the testing dataset file.
data_choice (str): Choice of dataset(s) to analyze; options include 'train', 'test', or 'train_test'.
questions (List[str]): List of questions aggregated based on the dataset choice.
question_types (Counter): Counter object tracking the frequency of each question type.
Qs (Dict[str, List[str]]): Dictionary mapping question types to lists of corresponding questions.
"""
def __init__(self, train_file_path: str, test_file_path: str, data_choice: str):
"""
Initializes the OKVQADatasetAnalyzer with paths to dataset files and a choice of which datasets to analyze.
Parameters:
train_file_path (str): Path to the training dataset JSON file. This file should contain a list of questions.
test_file_path (str): Path to the testing dataset JSON file. This file should also contain a list of
questions.
data_choice (str): Specifies which dataset(s) to load and analyze. Valid options are 'train', 'test', or
'train_test'indicating whether to load training data, testing data, or both.
The constructor initializes the paths, selects the dataset based on the choice, and loads the initial data by
calling the `load_data` method.
It also prepares structures for categorizing questions and storing the results.
"""
self.train_file_path = train_file_path
self.test_file_path = test_file_path
self.data_choice = data_choice
self.questions = []
self.question_types = Counter()
self.Qs = {keyword: [] for keyword in config.QUESTION_KEYWORDS}
self.load_data()
def load_data(self) -> None:
"""
Loads the dataset(s) from the specified JSON file(s) based on the user's choice of 'train', 'test', or
'train_test'.
This method updates the internal list of questions depending on the chosen dataset.
"""
if self.data_choice in ['train', 'train_test']:
with open(self.train_file_path, 'r') as file:
train_data = json.load(file)
self.questions += [q['question'] for q in train_data['questions']]
if self.data_choice in ['test', 'train_test']:
with open(self.test_file_path, 'r') as file:
test_data = json.load(file)
self.questions += [q['question'] for q in test_data['questions']]
def categorize_questions(self) -> None:
"""
Categorizes each question in the loaded data into predefined categories based on keywords.
This method updates the internal dictionary `self.Qs` and the Counter `self.question_types` with categorized
questions.
"""
question_keywords = config.QUESTION_KEYWORDS
for question in self.questions:
question = contractions.fix(question)
words = question.lower().split()
question_keyword = None
if words[:2] == ['name', 'the']:
question_keyword = 'name the'
else:
for word in words:
if word in question_keywords:
question_keyword = word
break
if question_keyword:
self.question_types[question_keyword] += 1
self.Qs[question_keyword].append(question)
else:
self.question_types["others"] += 1
self.Qs["others"].append(question)
def plot_question_distribution(self) -> None:
"""
Plots an interactive bar chart of question types using Altair and Streamlit, displaying the count and percentage
of each type.
The chart sorts question types by count in descending order and includes detailed tooltips for interaction.
This method is intended for visualization in a Streamlit application.
"""
# Prepare data
total_questions = sum(self.question_types.values())
items = [(key, value, (value / total_questions) * 100) for key, value in self.question_types.items()]
df = pd.DataFrame(items, columns=['Question Keyword', 'Count', 'Percentage'])
# Sort data and handle 'others' category specifically if present
df = df[df['Question Keyword'] != 'others'].sort_values('Count', ascending=False)
if 'others' in self.question_types:
others_df = pd.DataFrame([('others', self.question_types['others'],
(self.question_types['others'] / total_questions) * 100)],
columns=['Question Keyword', 'Count', 'Percentage'])
df = pd.concat([df, others_df], ignore_index=True)
# Explicitly set the order of the x-axis based on the sorted DataFrame
order = df['Question Keyword'].tolist()
# Create the bar chart
bars = alt.Chart(df).mark_bar().encode(
x=alt.X('Question Keyword:N', sort=order, title='Question Keyword', axis=alt.Axis(labelAngle=-45)),
y=alt.Y('Count:Q', title='Frequency'),
color=alt.Color('Question Keyword:N', scale=alt.Scale(scheme='category20'), legend=None),
tooltip=[alt.Tooltip('Question Keyword:N', title='Type'),
alt.Tooltip('Count:Q', title='Count'),
alt.Tooltip('Percentage:Q', title='Percentage', format='.1f')]
)
# Create text labels for the bars with count and percentage
text = bars.mark_text(
align='center',
baseline='bottom',
dy=-5 # Nudges text up so it appears above the bar
).encode(
text=alt.Text('PercentageText:N')
).transform_calculate(
PercentageText="datum.Count + ' (' + format(datum.Percentage, '.1f') + '%)'"
)
# Combine the bar and text layers
chart = (bars + text).properties(
width=700,
height=400,
title='Distribution of Question Keywords'
).configure_title(fontSize=20).configure_axis(
labelFontSize=12,
titleFontSize=14
)
# Display the chart in Streamlit
st.altair_chart(chart, use_container_width=True)
def export_to_csv(self, qs_filename: str, question_types_filename: str) -> None:
"""
Exports the categorized questions and their counts to two separate CSV files.
Parameters:
qs_filename (str): The filename or path for exporting the `self.Qs` dictionary data.
question_types_filename (str): The filename or path for exporting the `self.question_types` Counter data.
This method writes the contents of `self.Qs` and `self.question_types` to the specified files in CSV format.
Each CSV file includes headers for better understanding and use of the exported data.
"""
# Export self.Qs dictionary
with open(qs_filename, mode='w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
writer.writerow(['Question Type', 'Questions'])
for q_type, questions in self.Qs.items():
for question in questions:
writer.writerow([q_type, question])
# Export self.question_types Counter
with open(question_types_filename, mode='w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
writer.writerow(['Question Type', 'Count'])
for q_type, count in self.question_types.items():
writer.writerow([q_type, count]) |