import os
import random
import subprocess
import time
import tracemalloc
# Stop runtime if conda env is not active
import matplotlib
import numpy
import pandas
import tabulate
print(f"Matplotlib version: {matplotlib.__version__}")
print(f"NumPy version: {numpy.__version__}")
print(f"Pandas version: {pandas.__version__}")
print(f"Tabulate version: {tabulate.__version__}")
from sorting_algorithms import (
heap_sort,
merge_sort,
quick_sort,
quick_sort_deterministic,
)
# Constants
# Directories
PROJECT_ROOT = os.path.dirname(os.path.dirname(__file__))
RESULTS_DIR = os.path.join(PROJECT_ROOT, "results")
TEST_DATA_DIR = os.path.join(PROJECT_ROOT, "data", "size_10000")
TEST_DIR = os.path.join(PROJECT_ROOT, "tests")
os.makedirs(RESULTS_DIR, exist_ok=True)
os.makedirs(TEST_DIR, exist_ok=True)
# File Paths
PLOT_DATA_FILE = os.path.join(RESULTS_DIR, "plot_data.csv")
TABLE_DATA_FILE = os.path.join(RESULTS_DIR, "table_data.csv")
OUTPUT_FILE_PATH = os.path.join(RESULTS_DIR, "sorting_results.md")
# Script Paths
ALGORITHM_TESTS = os.path.join(PROJECT_ROOT, "src", "sorting_algorithms.py")
GENERATE_DATASETS_SCRIPT = os.path.join(PROJECT_ROOT, "src", "generate_datasets.py")
PLOTTING_SCRIPT = os.path.join(PROJECT_ROOT, "src", "plotting_script.py")
TABLE_SCRIPT = os.path.join(PROJECT_ROOT, "src", "table_script.py")
# Dataset Settings
SIZES = ["10000", "20000", "40000", "80000", "100000", "200000", "400000", "800000"]
NUM_DATASETS = 12 # Number of datasets per size
def datasets_exist(sizes, num_datasets):
"""Check if datasets are already generated for all sizes."""
for size in sizes:
for i in range(1, num_datasets + 1):
dataset_file = os.path.join(
PROJECT_ROOT, "data", f"size_{size}", f"input_{i}.txt"
)
if not os.path.exists(dataset_file):
return False
return True
def generate_datasets():
"""Generate datasets by running the generate_datasets.py script."""
print("Datasets missing. Generating datasets...")
subprocess.run(["python3", GENERATE_DATASETS_SCRIPT], check=True)
print("Datasets generated.")
def load_dataset(file_path):
"""Loads the dataset from the given file."""
with open(file_path, "r") as file:
dataset = [int(line.strip()) for line in file]
return dataset
def run_sorting_algorithm(algorithm, dataset, counter):
"""
Runs the sorting algorithm on the dataset, measures time, memory usage, and updates the counter.
:param algorithm: function, the sorting algorithm to run
:param dataset: list, the dataset to sort
:param counter: list, a list containing a single integer for counting comparisons
:return: tuple (time_taken, memory_usage)
"""
dataset_copy = dataset.copy()
tracemalloc.start()
start_time = time.perf_counter()
algorithm(dataset_copy, counter)
time_taken = time.perf_counter() - start_time
_, peak = tracemalloc.get_traced_memory()
memory_usage = peak / (1024 * 1024) # Convert to megabytes
tracemalloc.stop()
return time_taken, memory_usage
def quick_sort_wrapper(dataset, counter):
quick_sort(dataset, 0, len(dataset) - 1, counter)
def quick_sort_deterministic_wrapper(dataset, counter):
quick_sort_deterministic(dataset, 0, len(dataset) - 1, counter)
def save_sorted_data(sorted_data, algorithm_name, scenario):
"""Saves the sorted data to a file in the /tests/ directory."""
output_file = os.path.join(TEST_DIR, f"{algorithm_name}_{scenario}.txt")
with open(output_file, "w") as file:
for value in sorted_data:
file.write(f"{value}\n")
def sort_and_save(dataset, algorithm_name, algorithm, scenario):
"""Sorts the dataset using the specified algorithm and saves the result."""
dataset_copy = dataset.copy()
counter = [0]
# Capture the sorted array when using merge_sort
if algorithm_name == "Merge Sort":
sorted_data = algorithm(dataset_copy, counter)
else:
algorithm(dataset_copy, counter)
sorted_data = dataset_copy
save_sorted_data(sorted_data, algorithm_name, scenario)
def main():
# Check that algorithms work as expected
subprocess.run(["python3", ALGORITHM_TESTS], check=True)
# Check if datasets exist, if not, generate them
if not datasets_exist(SIZES, NUM_DATASETS):
generate_datasets()
# Define sorting algorithms
sorting_algorithms = {
"Quick Sort (Random Pivot)": quick_sort_wrapper,
"Quick Sort (Deterministic Pivot)": quick_sort_deterministic_wrapper,
"Merge Sort": merge_sort,
"Heap Sort": heap_sort,
}
algorithm_order = [
"Quick Sort (Random Pivot)",
"Merge Sort",
"Heap Sort",
"Quick Sort (Deterministic Pivot)",
]
# Load the dataset from input_1.txt for a test run
dataset_file = os.path.join(TEST_DATA_DIR, "input_1.txt")
dataset = load_dataset(dataset_file)
# Original dataset sorting (random order)
for algorithm_name, algorithm in sorting_algorithms.items():
sort_and_save(dataset, algorithm_name, algorithm, "random")
# Non-decreasing sorted dataset
sorted_dataset = sorted(dataset)
for algorithm_name, algorithm in sorting_algorithms.items():
sort_and_save(sorted_dataset, algorithm_name, algorithm, "non_decreasing")
# Non-increasing sorted dataset
reverse_sorted_dataset = sorted(dataset, reverse=True)
for algorithm_name, algorithm in sorting_algorithms.items():
sort_and_save(
reverse_sorted_dataset, algorithm_name, algorithm, "non_increasing"
)
with open(OUTPUT_FILE_PATH, "w") as result_file, open(
PLOT_DATA_FILE, "w"
) as plot_file, open(TABLE_DATA_FILE, "w") as table_file:
# Initialize markdown structure for results
result_file.write("# Sorting Algorithm Comparison Results\n")
result_file.write(
"This document contains the empirical results of Quick Sort (Random Pivot), Quick Sort (Deterministic Pivot), Merge Sort, and Heap Sort algorithms.\n\n"
)
result_file.write("## Table of Contents\n")
# Initialize plot and table files with headers
plot_headers = [
"Input Size",
"Quick Sort (Random Pivot)",
"Quick Sort (Deterministic Pivot)",
"Merge Sort",
"Heap Sort",
"QS_Random_Time",
"QS_Deterministic_Time",
"MS_Time",
"HS_Time",
"QS_Random_Memory",
"QS_Deterministic_Memory",
"MS_Memory",
"HS_Memory",
]
plot_file.write(",".join(plot_headers) + "\n")
table_headers = [
"Input Size",
"Quick Sort (Random) (Non-decreasing)",
"Merge Sort (Non-decreasing)",
"Heap Sort (Non-decreasing)",
"Quick Sort (Deterministic) (Non-decreasing)",
"Quick Sort (Random) (Non-increasing)",
"Merge Sort (Non-increasing)",
"Heap Sort (Non-increasing)",
"Quick Sort (Deterministic) (Non-increasing)",
]
table_file.write(",".join(table_headers) + "\n")
# Create a Table of Contents with links to input sizes
for size in SIZES:
result_file.write(f"- [Input Size: {size}](#input-size-{size})\n")
result_file.write("\n---\n")
total_start_time = time.perf_counter()
# Loop through each size
for size in SIZES:
print(f"\nRunning sorting algorithms for input size: {size}")
result_file.write(f"\n## Input Size: {size}\n")
result_file.write("### Random Order Datasets\n")
plot_file.write(f"{size},")
table_file.write(f"{size}")
# Initialize accumulators for averages
total_counters = {name: 0 for name in sorting_algorithms}
total_times = {name: 0.0 for name in sorting_algorithms}
total_memory = {name: 0.0 for name in sorting_algorithms}
dataset_start_time = time.perf_counter()
# Loop through the datasets for each size
for i in range(1, NUM_DATASETS + 1):
dataset_file = os.path.join(
PROJECT_ROOT, "data", f"size_{size}", f"input_{i}.txt"
)
print(f" Dataset {i} for size {size}...") # Progress message
# Open and read the dataset file
dataset = load_dataset(dataset_file)
result_file.write(f"\n#### Dataset {i}\n")
result_file.write(f"- **Input Size**: {size}\n")
# Set random seed
random.seed(20000629)
# Run each sorting algorithm
for name, algorithm in sorting_algorithms.items():
counter = [0]
time_taken, memory_usage = run_sorting_algorithm(
algorithm, dataset, counter
)
# Accumulate results
total_counters[name] += counter[0]
total_times[name] += time_taken
total_memory[name] += memory_usage
# Write individual dataset results
result_file.write(f"- {name} Comparisons: {counter[0]}\n")
result_file.write(f"- {name} Time: {time_taken:.4f} seconds\n")
result_file.write(f"- {name} Memory Usage: {memory_usage:.4f} MB\n")
# END OF DATASET LOOP
dataset_time = time.perf_counter() - dataset_start_time
print(f"Total Time for input size {size} -> {dataset_time:.2f} seconds")
# Calculate averages over the datasets
average_counters = {
name: total_counters[name] / NUM_DATASETS for name in sorting_algorithms
}
average_times = {
name: total_times[name] / NUM_DATASETS for name in sorting_algorithms
}
average_memory = {
name: total_memory[name] / NUM_DATASETS for name in sorting_algorithms
}
# Write averages to the plot data file
plot_data = [
int(average_counters["Quick Sort (Random Pivot)"]),
int(average_counters["Quick Sort (Deterministic Pivot)"]),
int(average_counters["Merge Sort"]),
int(average_counters["Heap Sort"]),
average_times["Quick Sort (Random Pivot)"],
average_times["Quick Sort (Deterministic Pivot)"],
average_times["Merge Sort"],
average_times["Heap Sort"],
average_memory["Quick Sort (Random Pivot)"],
average_memory["Quick Sort (Deterministic Pivot)"],
average_memory["Merge Sort"],
average_memory["Heap Sort"],
]
plot_file.write(",".join(map(str, plot_data)) + "\n")
# ----------------------------------------------------------------
# Non-decreasing sorted dataset ----------------------------------
print("\nSorting non-decreasing dataset...")
result_file.write(f"\n### Non-decreasing Sorted Dataset (Size: {size})\n\n")
sorted_data = sorted(dataset)
comparisons_non_decreasing = []
for name in algorithm_order:
algorithm = sorting_algorithms[name]
counter = [0]
algorithm(sorted_data.copy(), counter)
print(f" {name} (non-decreasing) comparisons: {counter[0]}")
result_file.write(
f"- {name} (Non-decreasing) Comparisons: {counter[0]}\n"
)
comparisons_non_decreasing.append(str(counter[0]))
table_file.write("," + ",".join(comparisons_non_decreasing))
# ----------------------------------------------------------------
# Non-increasing sorted dataset ----------------------------------
print("\nSorting non-increasing dataset...")
result_file.write(f"\n### Non-increasing Sorted Dataset (Size: {size})\n\n")
reverse_sorted_data = sorted(dataset, reverse=True)
comparisons_non_increasing = []
for name in algorithm_order:
algorithm = sorting_algorithms[name]
counter = [0]
algorithm(reverse_sorted_data.copy(), counter)
print(f" {name} (non-increasing) comparisons: {counter[0]}")
result_file.write(
f"- {name} (Non-increasing) Comparisons: {counter[0]}\n"
)
comparisons_non_increasing.append(str(counter[0]))
table_file.write("," + ",".join(comparisons_non_increasing) + "\n")
# END OF SIZE LOOP
total_time = time.perf_counter() - total_start_time
print(f"\nTotal Time -> {total_time:.2f} seconds\n")
result_file.write("---\n")
result_file.write("End of results.\n")
print(f"Results saved to {OUTPUT_FILE_PATH}")
print(f"Data written to {PLOT_DATA_FILE} and {TABLE_DATA_FILE}.")
# Run plotting and table generation scripts automatically
try:
# Run plotting script
print("\nRunning plotting script...")
subprocess.run(["python3", PLOTTING_SCRIPT], check=True)
# Run table generation script
print("\nRunning table generation script...")
subprocess.run(["python3", TABLE_SCRIPT], check=True)
except subprocess.CalledProcessError as e:
print(f"An error occurred while running the script: {e}")
if __name__ == "__main__":
main()