【Streamlit】非同期処理の進捗をプログレスバーで表示

asyncioを使用した非同期処理の進捗をプログレスバーで表示する方法をメモしておきます。

同期処理のプログレスバー表示

同期処理で、進捗状況を段階的に表示するコードの簡単なサンプルです。

import streamlit as st
import time

# 進捗バーのプレースホルダを作成
progress_bar = st.progress(0)

# 進捗を0%から100%まで1秒ごとに更新
for percent_complete in range(100):
    time.sleep(0.1)
    progress_bar.progress(percent_complete + 1)

# 完了メッセージ
st.success('Task completed!')

非同期処理のプログレスバー表示

非同期処理の場合、別スレッドでの非同期処理実行とグローバル変数の使用により実現できました。

import streamlit as st
import asyncio
import time
from threading import Thread

# タスク完了数を追跡するための変数
completed_tasks = 0
# 合計タスク数
total_tasks = 5


async def my_task(seconds, task_id):
    global completed_tasks
    print(f'Task {task_id}: {seconds}秒待機開始')
    await asyncio.sleep(seconds)
    print(f'Task {task_id}: {seconds}秒待機終了')
    completed_tasks += 1  # タスクが完了したらカウントアップ


async def run_tasks():
    global completed_tasks

    tasks = []
    for i in range(total_tasks):
        tasks.append(my_task(i + 1, i + 1))

    await asyncio.gather(*tasks)
    completed_tasks = total_tasks  # 全タスク完了を保証


def start_async_tasks():
    global completed_tasks
    completed_tasks = 0  # タスクカウントをリセット
    asyncio.run(run_tasks())


if 'task_started' not in st.session_state:
    st.session_state['task_started'] = False

if st.button('Start Tasks') and not st.session_state['task_started']:
    st.session_state['task_started'] = True
    thread = Thread(target=start_async_tasks)
    thread.start()

progress_bar = st.empty()
status_text = st.empty()

if st.session_state['task_started']:
    while completed_tasks < total_tasks:
        # 完了したタスクの割合に基づいて進捗を更新
        progress = (completed_tasks / total_tasks) * 100
        progress_bar.progress(int(progress))
        status_text.text(f"Current Progress: {int(progress)}%")
        time.sleep(0.1)
    # すべてのタスクが完了したら進捗を100%に更新
    progress_bar.progress(100)
    status_text.text("All tasks completed!")
    st.session_state['task_started'] = False

completed_tasks 変数を使用して完了したタスクの数を追跡し、これを基に進捗バーを更新します。

全てのタスクが完了したら、進捗を明示的に100%に設定しています。

これにより、表示上の進捗が100にならない問題を解決し、タスク完了時に進捗バーが100%に更新されるようになります。