Asynchronous polling of file progress in Python

loopy.py

import asyncio
from pathlib import Path

from tqdm import tqdm


REPO_ROOT = Path(__file__).parent
SCRIPT_PATH = REPO_ROOT / 'foo.bash'
OUTPUT_PATH = REPO_ROOT / 'wat'

MAX_VALUE = 99


def output(text):
    print(f'now - {text}')


async def run_external_script(script_path, output_path):
    process = await asyncio.create_subprocess_exec(
        'bash', str(script_path), str(output_path),
        stdout=asyncio.subprocess.DEVNULL)
    await process.wait()
    if process.returncode != 0:
        raise RuntimeError(f'Script exited {process.returncode}')


async def poll_output(output_path, progress_bar, *, once=False):
    if not once:
        await asyncio.sleep(5)
    while True:
        result = output_path.read_text().splitlines()[-1]
        progress_bar.update(int(result) - progress_bar.n)
        if once:
            break
        await asyncio.sleep(5)


async def main():
    print('starting')
    progress_bar = tqdm(total=MAX_VALUE)
    poll_task = asyncio.create_task(poll_output(OUTPUT_PATH, progress_bar))
    await run_external_script(SCRIPT_PATH, OUTPUT_PATH)
    await poll_output(OUTPUT_PATH, progress_bar, once=True)
    progress_bar.close()
    poll_task.cancel()
    print('done')


if __name__ == '__main__':
    asyncio.run(main())