はじめに

CLI（コマンドラインインターフェイス）上で動かしていた関数があって、これを直感的に使えるよう、GUI（グラフィカルインターフェイス）にし、exe化してみんなに配布したいと思った。
CLIでは、関数の進行状況をprint文で標準出力に表示していたが、GUIにしても関数をいじることなくウィンドウ内にそれを表示させたい。

問題

以下に示すような関数の様に、実行状況をprint文で表示させていたとする。
これを、fletのようなGUIを作成するフレームワークの中で呼んでも、当然のことながらただただコマンドライン上に表示されるだけで、GUI上のTextフィールドに出るわけではない。

def task():
    print("実行中")
    for i in range(10):
        time.sleep(1)
        print(f"Task {i}")
    print("終了")

でも、コードを変えると、CLI上で動かしたいときと泣き別れになってしまうのが嫌だった。

解決方法

contextlib.redirect_stdoutを使う。
標準出力をリダイレクトしてファイルとかに入れてくれるやつらしい。

ただ、変数などに入れてしまうと、すべての処理が終わってから表示することになってしまう。
やりたいのは、関数でprintされたらGUI上に表示させたいので、少し工夫を加えた。

具体的には、実行したい関数を別のスレッドで動かし、その出力をqueue に一時的に格納し、定期的にそれを確認し、GUI上に表示させるという感じだ。

具体的なコードを以下に示す。

import flet as ft
import time

import contextlib
import queue
import threading
import io


def task():
    print("実行中")
    for i in range(10):
        time.sleep(1)
        print(f"Task {i}")
    print("終了")


def main(page: ft.Page):
    page.title = "Print cmd on GUI"
    page.window.width = 350
    page.window.height = 350

    def button_clicked(e):

        class StreamToQueue(io.StringIO):
            def __init__(self, queue):
                super().__init__()
                self.queue = queue

            def write(self, message):
                super().write(message)
                self.queue.put(message)

        output_queue = queue.Queue()
        stream = StreamToQueue(output_queue)

        with contextlib.redirect_stdout(stream):
            thread = threading.Thread(target=task)
            thread.start()

            while thread.is_alive() or not output_queue.empty():
                thread.join(timeout=0.1)  # スレッドの状態を定期的にチェック
                while not output_queue.empty():  # キューの中身をチェック
                    text = output_queue.get().rstrip()  # キューから出力を取得
                    if text:
                        log.controls.append(ft.Text(text))  # GUI上に表示
                        page.update()  # GUIを更新

    button = ft.ElevatedButton(text="実行", on_click=button_clicked)

    log = ft.ListView(auto_scroll=True, expand=True)
    page.add(button, log)


ft.app(main)

別スレッドで関数を実行

関数を直接実行すると、メインスレッドがブロックされてしまい、GUIの更新が止まってしまうため、関数は別スレッドで実行する。これにより、メインスレッドであるGUIの動作を止めずに、バックグラウンドでタスクを並行して進めることが可能となる。

コード内では以下の部分でスレッドを作成している。

thread = threading.Thread(target=task)
thread.start()

ここで、task()関数が別スレッドで動作し、メインスレッドのGUIはそのまま操作を受け付け続けられる。

標準出力のリダイレクトとqueueへの格納

通常、print文は標準出力に結果を出力するが、今回はそれをGUI上に表示したいので、標準出力をリダイレクトする。Pythonのcontextlib.redirect_stdoutを使って、標準出力の内容を直接GUIへ渡せるようにする。

ここでは、標準出力をリダイレクトして、出力内容を キュー（queue） に格納する。キューはスレッド間で安全にデータをやり取りするために使用される構造。StreamToQueueクラスを使い、write()メソッドで標準出力の内容をキューに送信します。

class StreamToQueue(io.StringIO):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def write(self, message):
        super().write(message)
        self.queue.put(message)

print文が呼び出されるたびに、その出力がキューに入る仕組みです。

一定周期でキューをチェックしてGUIに表示

task()関数が別スレッドで動作している間、GUIは一定の間隔でキューの中身をチェックし、内容があればその文字列をGUI上のテキスト領域に追加する。この確認は、メインスレッドのwhileループの中で実施している。

while thread.is_alive() or not output_queue.empty():
    thread.join(timeout=0.1)  # スレッドの状態を定期的にチェック
    while not output_queue.empty():  # キューの中身をチェック
        text = output_queue.get().rstrip()  # キューから出力を取得
        if text:
            log.controls.append(ft.Text(text))  # GUI上に表示
            page.update()  # GUIを更新

ここでは、スレッドが終了するまで、またはキューに内容が残っている間、output_queueからデータを取り出し、GUI上に表示している。このループにより、タスクのprint出力がリアルタイムで反映される。

おわりに

ちょっとしたツールをGUI化してexeにして配布するというのは、平成かもと思いながらも、コードを渡して勝手に使って、というのよりは寄り添っていると思う今日この頃。令和なら、サーバ立ててリンクを送るのかな？