Textual에서 시간이 오래 걸리는 작업을 처리하면 UI의 반응성이 떨어지는 상황이 발생한다. 이 때 취할 수 있는 UI 적인 개선으로는 몇 가지 방법이 있는데, 그 중 가장 직관적인 것이 진행률을 보여주는 것이다. 진행률을 보여주는 프로그레스 바는 원래 Rich에도 포함되어 있는데, 그 중 기본적인 유형을 Textual에도 사용할 수 있게 되었다.
ProgressBar 위젯을 생성하여 화면에 추가한 다음, total 속성을 전체 일의 양으로 지정하고, 매 작업을 처리할 때마다 update() 나 advance()를 사용하여 진행된 분량을 늘려주면 프로그레스바는 그에 따라 UI 내용을 자동으로 업데이트한다. 이러한 사용방법은 Rich의 ProgressBar와 사실상 동일하다. (Rich의 경우에는 task를 내부적으로 여러 개 관리할 수 있기 때문에, 좀 더 단순화된 버전이라 할 수 있다.)
아래 예제는 간단하게 프로그레스바를 구현한 예를 보여준다. 여기서는 다른 작업을 하지 않고 시간에 따라 자동으로 업데이트 되도록 하기 위해서 타이머를 사용했는데, time.sleep()을 주기적으로 호출하는 루프를 사용해서 시간이 걸리는 작업을 흉내내는 식으로 만들어도 될 것이다.
from textual.app import App, ComposeResult
from textual.timer import Timer
from textual.widget import ProgressBar
class SampleApp(App):
progress_timer: Timer
CSS = """Screen { align: center middle; }"""
def compose(self) -> ComposeResult:
yield ProgressBar(total=100)
def on_mount(self) -> None:
self.progress_timer = self.set_interval(
1 / 60,
self.update_progress,
pause=True) # <- 타이머의 초기상태를 멈춤으로
def update_progress(self):
self.query_one(ProgressBar).advance(1)
async def key_space(self) -> None:
self.progress_timer.resume()
SampleApp().run()프로그레스바를 로딩바처럼 사용하기
Rich의 ProgressBar와 마찬가지로, ProgressBar는 total= 속성이 None인 경우에는 종료 시점이 결정되지 않은 ‘nondeterminated ‘한 프로그레스바가 된다. 이 때에는 진행률에 상관없이 표시되는, 로딩 인디케이터처럼 사용할 수 있다.
from textual.app import App, ComposeResult
from textual.widgets import ProgressBar
class MyApp(App):
CSS = "Screen { align: center middle; }"
def compose(self) -> ComposeResult:
yield ProgressBar(total=None,
show_percentage=False,
show_eta=False)
MyApp().run()참고로 대부분의 UI 위젯들은 .loading = True 로만 속성을 변경해주면 자동으로 로딩 인디케이터를 표시해주고 있으니 참고.
스피너
위와 같이 프로그레스 바를 로딩 인디케이터처럼 사용하는 것은 가능한데, Rich와 같이 스피너를 출력할 수는 없을까? Rich의 스피너는 각 프레임을 표현하는 유니코드 문자들의 리스트를 준비하고, 시간에 따라 스피너의 각 시간에서의 프레임을 출력한다. 이러한 매커니즘을 그대로 구현하면 되지 않을까? Textual의 공식 튜토리얼에는 스톱워치 만드는 예제가 있는데, 이 예제에서는 아래와 같이 타이머를 이용해서 애니메이션(디지털시계)을 구현한다. 이 코드에서도 시간의 경과에 따라 Static 위젯이 자신을 업데이트하게끔 하고 있다.
class TimeDisplay(Static):
start_time = reactive(monotonic)
time = reactive(0.0)
total = reactive(0.0)
def on_mount(self) -> None:
self.update_timer = self.set_interval( 1 / 60,
self.update_time,
pause=True)
def self.update_time(self):
self.time = self.total + (monotonic() - self.start_time)
def watch_time(self, time: float) -> None:
minutes, seconds = divmod(time, 60)
hours, minutes = divmod(minutes, 60)
self.update(f"{hours:02d}:{minutes:02d}:{second:05.3f}")
set_interval(), set_timer()는 모두 메시지 펌프(message pump)로부터 상속받는 기능이므로, 사실 상 모든 위젯들에서 사용 가능하다. 그렇다면 타이머를 내부적으로 가지고 있는 Static이라면 되지 않을까?
from time import monotonic
from rich.spinner import Spinner
from textual.timer import Timer
from textual.widgets import Static
from textual.app import App, ComposeResult
class MySpinner(Static):
DEFAULT_CSS = """
MySpinner {
width: 2;
height: 1;
content-align: center middle;
}"""
_spinner_timer: Timer
_spinner: Spinner
def __init__(self, /, **kwds):
super().__init__("", **kwds)
self._spinner = Spinner("moon")
def on_mount(self) -> None:
self._spinner_timer = self.set_interval( 1 / 8, self.update_spinner)
def update_spinner(self) -> None:
self.update(self._spinner.render(monotonic()))
class MyApp(App):
CSS = "Screen { align: center middle; }"
def componse(self) -> ComponseResult:
yield MySpinner()
MyApp().run()Spinner 내부에 자체적으로 특정 시간이 흐른 후 표시할 프레임을 결정하는 함수가 있기 때문에, 업데이트 주기를 아주 빠르게 한다고해서 애니메이션이 더 빨라지거나 부드러워지지는 않는다. 좀 더 프레임 수가 많은 경우에는 1/8초마다 프레임을 업데이트하는 주기를 변경해줄 필요가 있을 것 같다.
부록, 다운로드 진행상황
다운로드 진행 상황의 경우 rich.progress를 사용하면 진행률을 용량으로, 막대나 퍼센트로 표시할 수 있고, 전속속도와 경과시간, 남은 예상시간 같은 정보를 실시간으로 표현할 수 있다. async http 라이브라리인 httpx를 사용하여 다운로드를 진행하면서 진행 상황을 표시하는 예제를 소개하면서 글을 마무리 할까한다.
import httpx
from rich.progress import (DownloadColumn, Progress, BarColumn,
TransferSpeedColumn, TimeRemainingColumn)
url = "https:// .... "
async def download(url: str, filename: str) -> None:
async with httpx.AsyncClient() as client:
res = await client.get(url)
total = int(res.headers["Content-Length"])
if not total:
print("No data")
return
with open(filename, 'wb') as f:
with Progress(DownloadColumn(),
BarColumn(),
TransferSpeedColumn(),
TimeRemainingColumn()) as progress:
async for chunk in res.aiter_bytes(1024):
progress.update(task, advance=len(chunk))
f.write(chunk)
task = progress.add_task("download", total=total)