Categories: python

python rich with asyncio

內容目錄

python rich 畫圖很漂亮

console.print() 可以很容易地取代原本的 print() 印出很漂亮有顏色的字。而要做成全終端機視窗的 TUI 程式，也可以另外使用 rich.live.Live，再封裝原本的 renderable 物件會自行更新畫面。這個功能是 Live 另外開一個 thread 在後面跑的。

不過新的 python 就有了 asyncio，不管是拿來收 stdio 的輸入或是用網路連接獲取資料都相當方便，但翻遍了文件查詢了網頁似乎都沒有看到 asyncio 怎麼跟 rich 合作。

那麼能不能也基於 asyncio 的機制上用 rich.live.Live 呢？

自行更新 rich.live.Live

是可以的，但是首先要先停用 rich.live.Live 背後的 thread。稍微看了一下 source code，就會發現只要設定 auto_refresh=False 後面就不會自行更新，也不會有另一個 thread，不過就需要自行告知什麼時候 refresh。

async def rich_task(dictlayout: 'DictLayout'):
    ...
    with Live(dictlayout.layout, screen=True, auto_refresh=False,
              console=console, redirect_stderr=False) as liver:
        while True:
            update()
            liver.refresh()
            await asyncio.sleep(0.05)

如上面片段範例，就可以用 asyncio 的方把，做一個 fps 20 的畫面更新，dictlayout.layout 就是全部畫面的 Renderable，而 redirect_stderr=False 是為了 debug 訊息才開的。

不過呢，這段 code 在某些系統某些狀況下會出現 exception
BlockingIOError: [Errno 11] write could not complete without blocking
在 macos 下可能看到是 Errno 35

原因是因為 rich.console.Console 在寫 stdout 的時候，沒有處理 non-blocking I/O 的能力。

aioconsole 只用 stdin

在寫這個 TUI 程式需要接收使用者的輸入，既然都是用 asyncio 寫了，當然不可以另外開一個 thread 聽啊，一定也是用 asyncio 的方法，所以借用了 aioconsole 這個 package。

from aioconsole import get_standard_streams

async def main():
    reader, writer = await get_standard_streams()

這個 function 會全部處理 stdin, stdout, stderr 將他們都變成了 non-blocking 的模式，雖然目前我只需要 stdin 取得使用者輸入就好，推測這也就是出現 BlockingIOError 的原因。
所以就直接翻一下 source code，將 stdin 獨立開出來。

from aioconsole.stream import StandardStreamReader
from aioconsole.stream import StandardStreamReaderProtocol

async def open_standard_pipe_connection(pipe_in, *, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    # Reader
    in_reader = StandardStreamReader(loop=loop)
    protocol = StandardStreamReaderProtocol(in_reader, loop=loop)
    await loop.connect_read_pipe(lambda: protocol, pipe_in)
    # Return
    return in_reader

async def input_task(dictlayout: 'DictLayout'):
    reader = await open_standard_pipe_connection(sys.stdin)
    while True:
        res = await reader.read(100)
        if not res:
            break

這樣應該就不會影響到 stdout 了吧！

我錯了，看了一下還是會發生 BlockingIOError，而且 stdout 還是被設定為 non-blocking mode

stdin/stdout non-blocking 連動

這真的很神奇，其實 stdin 與 stdout 他們的 non-blocking 設定是連動的啊，這裡可以看到有人在 stackoverflow 上有發問過，另外也可以利用下面這段改一改檢查一下。

還真的是耶，改一個兩個都變啊，還是會遇到 BlockingIOError

只好來改 rich.console.Console

好吧，既然系統會把 stdout 改成 non-blocking，這邊可能改動不了，那改能夠修改的部分，也就是 rich.console.Console。
一個是把 Console 改成支援 non-blocking 的方法，也就是看到 BlockIOError 就相當於 再試一次 Try Again 的意思。不過這邊可能先簡單採用另一種想法，就是 Console 在要寫出到 stdout 之前，就設定成 blocking ，然後再設定回來，改動地程度比較少。

import os
from rich.console import Console

class BlockingContext:
    '''A context manager set console.file blocking'''

    def __init__(self, console: 'NonBlockingConsole') -> None:
        self.file = console.file

    def __enter__(self) -> 'BlockingContext':
        os.set_blocking(self.file.fileno(), True)
        return self

    def __exit__(self, *exc_details) -> None:
        os.set_blocking(self.file.fileno(), False)

class NonBlockingConsole(Console):
    '''Support if NonBlocking stdout rich.console.Console
    '''
    def _check_buffer(self) -> None:
        with BlockingContext(self):
            super()._check_buffer()

...
async def rich_task(dictlayout: 'DictLayout'):
    tty.setcbreak(sys.stdin)
    console = NonBlockingConsole()
    ...
    with Live(dictlayout.layout, screen=True, auto_refresh=False,
              console=console, redirect_stderr=False) as liver:
        ...

發生寫 stdout 的部分是在 _check_buffer() 裡面，所以用一個 BlockingContext 包裝起來，記得 Live 在用的時候，要指定用這個 NonBlockingConsole，如此一來就不再出現 BlockingIOError 了。