eep sort

eep_sort.py

from __future__ import annotations

import asyncio
import collections.abc
import io
import os
import random
import sys
import time
import typing

T = typing.TypeVar("T")
Number: typing.TypeAlias = float | int | bool
NumberT = typing.TypeVar("NumberT", bound=Number)


async def arange(count: int, /):
    for i in range(count):
        yield i
        await asyncio.sleep(0.0)


@typing.overload
async def eepy_sort(
    seq: collections.abc.Sequence[NumberT],
) -> list[NumberT]:
    ...


@typing.overload
async def eepy_sort(
    seq: collections.abc.Sequence[T],
    key: collections.abc.Callable[[T], Number],
) -> list[T]:
    ...


async def eepy_sort(
    seq: collections.abc.Sequence[T],
    key: collections.abc.Callable[[T], Number] | None = None,
) -> list[T]:
    tasks: list[asyncio.Task[T]] = []

    async for i in arange(len(seq)):
        if isinstance(el := seq[i], Number):
            delay = el
        else:
            if key is None:
                raise TypeError(
                    "key must be provided if seq is not a sequence of integers"
                )
            delay = key(el)

        if delay < 0:
            raise ValueError("cannot sort negative numbers (can't eep backwards)")

        tasks.append(asyncio.create_task(_sleep_n(delay, el)))

    return [await f for f in asyncio.as_completed(tasks)]


async def _sleep_n(n: Number, value: T, /) -> T:
    await asyncio.sleep(typing.cast(float, n))

    return value


def print_temp(message: str) -> None:
    sys.stdout.write(f"\r\x1b[2K{message}")


def sliced_repr(seq: collections.abc.Sequence[typing.Any]) -> str:
    if not seq:
        return "[]"

    buffer = io.StringIO()

    beg, end = seq[:2], seq[-2:]

    buffer.write("[")

    for el in beg:
        if el in end:
            break
        buffer.write(repr(el) + ", ")

    buffer.write("...")

    for el in end:
        if el in beg:
            continue
        buffer.write(", " + repr(el))

    buffer.write("]")

    return buffer.getvalue()


def slow_random() -> float:
    time.sleep(0.001)

    return int(random.random() * 10)


# clever algorithm that I made up x)
def is_sorted(seq: collections.abc.Sequence[typing.Any]) -> bool:
    if len(seq) <= 1:
        return True

    return seq[0] == min(seq) and seq[-1] == max(seq) and is_sorted(seq[1:-1])


async def main() -> int:
    print_temp("\x1b[35m○ \x1b[1mGenerating array...\x1b[22;39m")
    array = [slow_random() for _ in range(1_000)]
    print_temp("\x1b[32m● \x1b[1mArray generated!\x1b[22;39m")

    print("\nOutput:", sliced_repr(array))

    print_temp("\x1b[35m○ \x1b[1mApplying sorting algorithm...\x1b[22;39m")
    sorted_array = await eepy_sort(array)
    print_temp("\x1b[32m● \x1b[1mAlgorithm finished!\x1b[22;39m")

    print("\nOutput:", sliced_repr(sorted_array))

    print_temp("\x1b[35m○ \x1b[1mChecking if the array is sorted...\x1b[22;39m")
    if is_sorted(sorted_array):
        print_temp("\x1b[32m● \x1b[1mSuccess!\x1b[22;39m")
    else:
        print_temp("\x1b[31m● \x1b[1mFailure: the array is not sorted.\x1b[22;39m")

    print()

    return os.EX_OK


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))