async typewriting

aprint.py

import argparse
import asyncio
import os
import sys
import typing


class APrintNamespace(typing.Protocol):
    message: str
    tasks: int
    typing_delay: float


async def typewrite(message: str, *, delay: float = 0.1, flush: bool = False) -> None:
    for char in message:
        await asyncio.sleep(delay)
        sys.stdout.write(char)

        if flush:
            sys.stdout.flush()


async def aprint(
    message: str,
    typing_delay: float = 0.1,
    *,
    nb_tasks: int = 1,
    end: str = "\n",
    flush: bool = False,
) -> None:
    tasks: list[asyncio.Task[None]] = []

    for i in range(nb_tasks):
        task = asyncio.create_task(
            typewrite(
                message[i::nb_tasks],
                delay=typing_delay,
                flush=flush,
            )
        )
        tasks.append(task)

    await asyncio.wait(tasks)
    sys.stdout.write(end)


def parse_args() -> APrintNamespace:
    parser = argparse.ArgumentParser()

    parser.add_argument("message", nargs="?", default="Hello World!")
    parser.add_argument("--tasks", "-t", type=int, default=1)
    parser.add_argument("--typing-delay", "-d", type=float, default=0.1)

    return typing.cast(APrintNamespace, parser.parse_args())


async def main() -> int:
    namespace = parse_args()

    await aprint(
        namespace.message,
        namespace.typing_delay,
        nb_tasks=namespace.tasks,
        flush=True,
    )

    return os.EX_OK


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