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()