from pathlib import Path import time from typing import List, Union import math IntroSorts = 0 InsertionSorts = 0 HeapSorts = 0 # ------------------------------------------------------------------- # load_file # Purpose: # Read non-blank integer records from the named input file into the program collection or shared array. # Parameters: # fileSpec - input or output file path. # Returns: # The loaded integer sequence or the count/status provided by this implementation. # Side effects: # Opens and closes the input file, populates the collection or shared array, and may update its element count. # ------------------------------------------------------------------- def load_file(fileSpec: Union[str, Path]) -> List[int]: print("READING INPUT FILE...",end="") values: List[int] = [] with Path(fileSpec).open(encoding="ascii") as input_file: for line_number, line in enumerate(input_file, start=1): text = line.strip() if text: try: values.append(int(text)) except ValueError as error: raise ValueError( f"Invalid integer on line {line_number}: {text!r}" ) from error print("DONE.") return values # ------------------------------------------------------------------- # save_file # Purpose: # Write the sorted integer collection or shared array to the named output file, one value per line. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # fileSpec - input or output file path. # Returns: # No value; writes the output file. # Side effects: # Creates or replaces the output file and reports file progress. # ------------------------------------------------------------------- def save_file(values: List[int], fileSpec: Union[str, Path]) -> None: print("SAVING OUTPUT FILE...",end="") with Path(fileSpec).open("w", encoding="ascii") as output_file: for value in values: print(value, file=output_file) print("DONE.") # ------------------------------------------------------------------- # sift_down # Purpose: # Restore max-heap order below the specified root within the active heap range. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # root - current heap root index. # start - inclusive first index of the active range. # end - exclusive end index of the active range. # Returns: # No value; rearranges the range in place. # Side effects: # Mutates the sequence and may stop once the heap property is restored. # ------------------------------------------------------------------- def sift_down(values: List[int], root: int, start: int, end: int) -> None: while start + 2 * (root - start) + 1 < end: child = start + 2 * (root - start) + 1 if child + 1 < end and values[child] < values[child + 1]: child += 1 if values[root] >= values[child]: return values[root],values[child] = values[child],values[root] root = child # ------------------------------------------------------------------- # heap_sort # Purpose: # Sort the selected range in place with heapsort, the worst-case introsort fallback. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # start - inclusive first index of the active range. # end - exclusive end index of the active range. # Returns: # No value; sorts the selected range in place. # Side effects: # Mutates the range and increments the heapsort counter where present. # ------------------------------------------------------------------- def heap_sort(values: List[int], start: int, end: int) -> None: global HeapSorts HeapSorts += 1 for root in range(start + (end - start) // 2 - 1, start, -1): sift_down(values, root, start, end) for last in range(end - 1, start + 1, -1): values[start],values[last] = values[last],values[start] sift_down(values, start, start, last) # ------------------------------------------------------------------- # insertion_sort # Purpose: # Sort the selected range in place with insertion sort, which is efficient for short ranges. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # start - inclusive first index of the active range. # end - exclusive end index of the active range. # Returns: # No value; sorts the selected range in place. # Side effects: # Mutates the range and increments the insertion-sort counter where present. # ------------------------------------------------------------------- def insertion_sort(values: List[int], start: int, end: int) -> None: global InsertionSorts InsertionSorts += 1 for index in range(start + 1, end, +1): value = values[index] pos = index - 1 while pos >= start and values[pos] > value: values[pos + 1] = values[pos] pos -= 1 values[pos + 1] = value # ------------------------------------------------------------------- # partition # Purpose: # Partition the selected range around its middle-element pivot and return the first index of the right-hand range. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # start - inclusive first index of the active range. # end - exclusive end index of the active range. # Returns: # The first index of the right partition, which is also the left boundary of the recursive ranges. # Side effects: # Reorders the range in place while inward scans keep each access inside the active bounds. # ------------------------------------------------------------------- def partition(values: List[int], start: int, end: int) -> int: left = start right = end - 1 pivot = values[start + math.floor((end - start) / 2)] while left <= right: while values[left] < pivot: left += 1 while values[right] > pivot: right -= 1 if left <= right: values[left],values[right] = values[right],values[left] left += 1 right -= 1 return left # ------------------------------------------------------------------- # intro_sort # Purpose: # Sort the selected range in place: insertion sort for short ranges, heapsort when the depth budget expires, otherwise partition and recurse. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # start - inclusive first index of the active range. # end - exclusive end index of the active range. # maxdepth - remaining partition-depth budget before heapsort fallback. # Returns: # No value; sorts the selected range in place. # Side effects: # Mutates the range and updates the introsort counter; recursive calls reduce the depth budget. # ------------------------------------------------------------------- def intro_sort(values: List[int], start: int, end: int, maxdepth: int) -> None: global IntroSorts IntroSorts += 1 length = end - start if length <= 1: return if length < 16: insertion_sort(values, start, end) elif maxdepth <= 0: heap_sort(values, start, end) else: part = partition(values, start, end) intro_sort(values,start,part,maxdepth - 1) intro_sort(values,part,end,maxdepth - 1) # ------------------------------------------------------------------- # perform_sort # Purpose: # Choose the initial depth budget, time the top-level introsort, and report completion. # Parameters: # values - integer sequence to read, sort, or write; range operations modify it in place. # Returns: # Elapsed time measured by this implementation. # Side effects: # Starts and stops a timer, prints progress, and invokes introsort on the full sequence. # ------------------------------------------------------------------- def perform_sort(values: List[int]) -> float: print("SORTING ELEMENTS...",end="") startTime = time.perf_counter() maxdepth = 0 if len(values) > 1: maxdepth = log2(len(values)) * 2 intro_sort(values, 0, len(values), maxdepth) endTime = time.perf_counter() print("DONE.") return endTime - startTime # ------------------------------------------------------------------- # log2 # Purpose: # Compute the integer base-2 size estimate used to initialize introsort depth. # Parameters: # x - positive integer used to compute the base-2 size estimate. # Returns: # An integer base-2 size estimate. # Side effects: # Does not modify the input. # ------------------------------------------------------------------- def log2(x: int) -> int: n = 0 while x > 0: x /= 2 n += 1 return n # ------------------------------------------------------------------- # main # Purpose: # Program entry point: read the data, sort it, write the result, and report elapsed time and algorithm counts. # Parameters: # None; this routine uses shared state or its enclosing context. # Returns: # The runtime-specific exit status, or no value for a void entry point. # Side effects: # Performs file I/O, invokes sorting, and prints timing and counters. # ------------------------------------------------------------------- def main() -> None: print("========INTROSORT PYTHON========") values = load_file("D:/Sorting/Sorting_Input.txt") elapsed = perform_sort(values) save_file(values, "D:/Sorting/Output_Python.txt") print("==========RUN COMPLETE==========") print(f"INTROSORT TOOK {elapsed:.6f} SECONDS.") print("================================") print("SORT CALLED | TIMES") print("------------+-------------------") print(f"Intro | {IntroSorts}") print(f"Insert | {InsertionSorts}") print(f"Heap | {HeapSorts}") print("--------------------------------") if __name__ == "__main__": main()