prompt-toolkit / python-prompt-toolkit

Library for building powerful interactive command line applications in Python
https://python-prompt-toolkit.readthedocs.io/
BSD 3-Clause "New" or "Revised" License
9.1k stars 717 forks source link

asyncio concepts and prompt toolkit for UI updates #1847

Open mm4nn opened 4 months ago

mm4nn commented 4 months ago

I am currently struggling with asyncio concepts and prompt toolkit. I want to be able to update the UI when I have long running tasks in the background. In a regular UI (non-terminal application) the task would be spun up in a separate thread and when there is data available it could send... say an event or message (depending on the framework) with the data to the UI thread for it to update itself.

Again, I am struggling to try to get anything similar to that pattern with asyncio and prompt toolkit. I included a sample application that has a "ListView." Pressing "i" would run a long-running task that populates the list view. If I remove all of the asyncio stuff and run everything on the main thread, then everything works fine, but when I run it as-is, it only prints the first item. It does nothing after the first asyncio.sleep() call.

Also, I realize that the UI updates are happening in the "background" here (or at least I think they are not--I don't understand asyncio enough yet to know if it's sharing the main event loop?). Do you have any suggestion on how I can send events or messages to the main thread so all the UI work happens on one thread? (sample code?) Or is that not required for prompt-toolkit?

Any and all help and suggestions would be appreciated.

Just fyi. This is obviously sample code but in my real application the list view would receive updates from http calls.

# -*- coding: utf-8 -*-
"""
main.py

Part of the listupdate Project

Created: 2024-02-06T16:03:56

"""

import asyncio
import os
import random
import threading

from prompt_toolkit import Application
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.layout import FormattedTextControl, Window, Layout
from prompt_toolkit.styles import Style

LIST_DATA_HEADER_STYLE: str = 'class:list-data-header'
LIST_DATA_DATA_STYLE: str = 'class:list-data-data'

TAB_DATA: str = '    '

HDR_SEP: str = '|'
COL_SEP: str = ' '  # ' ' if you do not want any or '|' if you want lines

GREEK_ALPHABET = [
    'Alpha',
    'Beta',
    'Gamma',
    'Delta',
    'Epsilon',
    'Zeta',
    'Eta',
    'Theta',
    'Iota',
    'Kappa',
    'Lambda',
    'Mu',
    'Nu',
    'Xi',
    'Omicron',
    'Pi',
    'Rho',
    'Sigma',
    'Tau',
    'Upsilon',
    'Phi',
    'Chi',
    'Psi',
    'Omega',
]

MILITARY_ALPHABET = [
    'Alpha',
    'Bravo',
    'Charlie',
    'Delta',
    'Echo',
    'Foxtrot',
    'Golf',
    'Hotel',
    'India',
    'Juliet',
    'Kilo',
    'Lima',
    'Mike',
    'November',
    'Oscar',
    'Papa',
    'Quebec',
    'Romeo',
    'Sierra',
    'Tango',
    'Uniform',
    'Victor',
    'Whiskey',
    'X-ray',
    'Yankee',
    'Zulu',
]

def get_max_item_size(items: list[str]) -> int:
    max_size = 0
    for i in items:
        s = len(i)
        if s > max_size:
            max_size = s
    return max_size

COL_1_SIZE = 3
COL_2_SIZE = get_max_item_size(GREEK_ALPHABET) + 1 + get_max_item_size(MILITARY_ALPHABET)
COL_3_SIZE = len('desc')

class ListUpdate:
    def __init__(self):
        self._list = FormattedTextControl()
        self._root_container = self._build_ui()
        self._layout = Layout(self._root_container)
        self._key_bindings = self._make_key_bindings()
        self._style = self._make_style()

        self._app = Application(
            layout=self._layout,
            key_bindings=self._key_bindings,
            mouse_support=True,
            style=self._style,
            full_screen=True
        )

    def run(self):
        self._app.run()

    def _build_ui(self):

        term_width = os.get_terminal_size().columns
        col_4_width = term_width - (COL_1_SIZE + COL_2_SIZE + COL_3_SIZE + 9)

        data = [
            (LIST_DATA_HEADER_STYLE, f' {"ID": >{COL_1_SIZE}} '),
            (LIST_DATA_HEADER_STYLE, HDR_SEP),
            (LIST_DATA_HEADER_STYLE, f' {"NAME": <{COL_2_SIZE}} '),
            (LIST_DATA_HEADER_STYLE, HDR_SEP),
            (LIST_DATA_HEADER_STYLE, f' {"DESCRIPTION": <{COL_3_SIZE}} '),
            (LIST_DATA_HEADER_STYLE, HDR_SEP),
            (LIST_DATA_HEADER_STYLE, ' ' * col_4_width),
            (LIST_DATA_HEADER_STYLE, '\n'),
        ]

        self._list.text = data

        return Window(content=self._list)

    def _make_key_bindings(self):

        kb = KeyBindings()

        @kb.add("i")
        def _(event):
            self.on_add_list_data(event)

        @kb.add("c-q", eager=True)
        def _(event):
            self.on_exit_app(event)

        return kb

    def _make_style(self):
        return Style.from_dict({
            'list-data-header': 'bg:#47ddaa #000000 bold',
            'list-data-data': ''
        })

    def add_list_item(self, id_, name, description):
        data: list = self._list.text

        data.append((LIST_DATA_DATA_STYLE, f' {id_: >{COL_1_SIZE}} '))
        data.append((LIST_DATA_DATA_STYLE, COL_SEP))
        data.append((LIST_DATA_DATA_STYLE, f' {name: <{COL_2_SIZE}} '))
        data.append((LIST_DATA_DATA_STYLE, COL_SEP))
        data.append((LIST_DATA_DATA_STYLE, f' {description: <{COL_3_SIZE}} '))
        data.append((LIST_DATA_DATA_STYLE, '\n'))

    async def long_running_method(self, event):

        all_names: list = []
        for idx in range(len(GREEK_ALPHABET)):
            for jdx in range(len(MILITARY_ALPHABET)):
                all_names.append((GREEK_ALPHABET[idx], MILITARY_ALPHABET[jdx]))

        random.shuffle(all_names)

        # just show the first 40 because we haven't handled
        # scrolling in our "ListView"
        for idx, names in enumerate(all_names[0:40]):
            self.add_list_item(idx + 1, f'{names[0]}-{names[1]}', 'desc')
            await asyncio.sleep(1)

    def on_add_list_data(self, event):
        asyncio.create_task(self.long_running_method(event))

    def on_exit_app(self, event):
        self._app.exit()

def main():
    lu: ListUpdate = ListUpdate()
    lu.run()

if __name__ == '__main__':
    main()
jonathanslenders commented 4 months ago

You can call self._app.invalidate() to schedule a repaint.

If you'd like to use asyncio, I also suggest to make the main() function async, and have async from the start. Use asyncio.run(main()) to start the application, and rewrite the run() like this:

    async def run(self):
        await self._app.run_async()

Instead of asyncio.create_task, use either a TaskGroup (native asyncio task groups or from the anyio library), or use self._app.create_background_task(...).

I'd also like to point out that it's worth checking out the Textual library which might or might not be better for these things, depending on your needs.