【Python】PyQtとMultiProcessingでリアルタイムGUIを作成する方法

SHARE

目次

はじめに

Pythonは何かを作る上で簡単に実行ができるので、便利ですが、何かしらのアプリを作るにはリアルタイム性がうまく実現できないことが多々あります。

今回は、MultiProcessingというライブラリを用いて、GUIを更新する処理と、内部での計算処理、状態を保存する処理を別々のプロセスで分けることでアプリケーションを実現していきます。

今回紹介する考え方を用いれば簡易デモソフト等を作りたい場合にも有効に使えますので、ぜひ活用してもらえたらと思います！

ライブラリのインストール

まず初めにPython用のライブラリをインストールしていきます。

今回使用する外部ライブラリは「PyQt」というもののみですので、下記コマンドでライブラリのインストールを行います。

pip install PyQt5

PyQtには複数のバージョンがありますが、今回は5で作成していきますので、注意してインストールをしてください。

作成するアプリケーションについて

今回はデモということで、PC用カウントダウンアプリを作成していきます。

基本的な構成は「変数やイベントトリガ」を設計するクラスと「カウントダウン」を実装するクラス、「GUIの見た目」を実装するクラスの3つで構成していきます。

それぞれのクラスのつながりをMultiProcessによってマルチスレッド化して実装していきます。

コードの全体

コードの詳細を説明する前に、今回のアプリケーションを実装する上でのコード全体を紹介します。

下記のコードをコピペして実行してみてください。

import sys
import time
import multiprocessing as mp
from PyQt5.QtWidgets import QApplication,QGridLayout,QLabel,QPushButton,QMainWindow
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QWidget

TIMER_DUARATION = 5.0
COUNT_DOWN_WIDTH_S = 0.05

class StateManagerProcess():
    def __init__(self) -> None:
        self._all_exit_event = mp.Event()
        self._gui_update_event = mp.Event()
        self._count_down_exec_event = mp.Event()
        self.count_down_time = mp.Value("f",TIMER_DUARATION)

class CountDownProcess():
    def __init__(self, state_and_data:StateManagerProcess) -> None:
        self.state_and_data = state_and_data
        self._count_down_process = mp.Process(target=self.setup, name="CountDownProcess")
    
    def setup(self):
        while True:
            if self.state_and_data._count_down_exec_event.is_set():
                self.state_and_data._count_down_exec_event.clear()
                while True:
                    # カウントダウンの処理
                    self.state_and_data.count_down_time.value = self.state_and_data.count_down_time.value - COUNT_DOWN_WIDTH_S
                    if(self.state_and_data.count_down_time.value <= 0):
                        # 設定時間が経過したらループから抜ける
                        break
                    # GUI更新イベントの発火
                    self.state_and_data._gui_update_event.set()
                    time.sleep(COUNT_DOWN_WIDTH_S)
    def start(self):
        self._count_down_process.start()

class GUIProcess():
    def __init__(self,state_and_data:StateManagerProcess) -> None:
        self.state_and_data = state_and_data
        self._gui_process = mp.Process(target=self.setup,name="GUIProcess")

    def setup(self):
        self.app = QApplication(sys.argv)
        self.window = MainWindow(self.state_and_data)
        self.window.closeEvent = lambda _: self.close()
        self.window.show()
        self.app.processEvents()
        while True:
            if self.state_and_data._gui_update_event.is_set():
                self.window.update_timer()
                self.state_and_data._gui_update_event.clear
            self.app.processEvents()
        
    def start(self):
        self._gui_process.start()

    def close(self):
        self.state_and_data._all_exit_event.set()

class MainWindow(QMainWindow):
    def __init__(self,state_and_data:StateManagerProcess) -> None:
        super().__init__()
        self.state_and_data = state_and_data
        self.setup_gui()

    def setup_gui(self):
        # GUIのタイトル
        self.setWindowTitle("Qt CountDownTimer")

        central_widget = QWidget(self)
        self.setCentralWidget(central_widget)
        layout = QGridLayout(central_widget)

        self.time_label = QLabel(self.make_display_time())
        self.time_label.setAlignment(Qt.AlignCenter)  
        layout.addWidget(self.time_label,0,0)

        start_count_down_button_widget = QPushButton("Start",self)
        start_count_down_button_widget.clicked.connect(self.start_count_down)
        layout.addWidget(start_count_down_button_widget,1,0)

    def update_timer(self):
        self.time_label.setText(self.make_display_time())

    def start_count_down(self):
        self.state_and_data.count_down_time.value = TIMER_DUARATION
        self.state_and_data._count_down_exec_event.set()

    def make_display_time(self):
        return f'{self.state_and_data.count_down_time.value:>5.3f}'

if __name__ == "__main__":
    state_manager_process = StateManagerProcess()
    count_down_process = CountDownProcess(state_and_data=state_manager_process)
    gui_process = GUIProcess(state_and_data=state_manager_process)

    count_down_process.start()
    gui_process.start()

    while not state_manager_process._all_exit_event.is_set():
        pass

    count_down_process._count_down_process.terminate()
    count_down_process._count_down_process.join()
    gui_process._gui_process.terminate()
    gui_process._gui_process.join()

コードを実行すると下記画像のようなアプリケーションが立ち上がると思います。

「Start」と書いてあるボタンをクリックすると中央に表示されている数値が減少して0になると止まることが確認できると思います。

コードの解説（StateManagerProcess）

class StateManagerProcess():
    def __init__(self) -> None:
        self._all_exit_event = mp.Event()
        self._gui_update_event = mp.Event()
        self._count_down_exec_event = mp.Event()
        self.count_down_time = mp.Value("f",TIMER_DUARATION)

ここではMultiProcessingで立てられた複数のスレッド間でデータを共有する変数を定義しています。

mp.Event()は複数のスレッド間でのトリガーの受け渡しを安定して実現できるAPIを提供してくれています。

今回は全てのアプリケーション（処理）を削除するトリガー、GUIの表示を更新するトリガー、カウントダウンを開始するトリガーの3つ用に定義をしています。

mp.Value()ではプロセス間での共通メモリにデータを格納するためのAPIを提供します。基本的にプロセス間でのやり取りはこのAPIを経由してデータの受け渡しを行うことになります。

コードの解説（CountDownProcess）

class CountDownProcess():
    def __init__(self, state_and_data:StateManagerProcess) -> None:
        self.state_and_data = state_and_data
        self._count_down_process = mp.Process(target=self.setup, name="CountDownProcess")
    
    def setup(self):
        while True:
            if self.state_and_data._count_down_exec_event.is_set():
                self.state_and_data._count_down_exec_event.clear()
                while True:
                    # カウントダウンの処理
                    self.state_and_data.count_down_time.value = self.state_and_data.count_down_time.value - COUNT_DOWN_WIDTH_S
                    if(self.state_and_data.count_down_time.value <= 0):
                        # 設定時間が経過したらループから抜ける
                        break
                    # GUI更新イベントの発火
                    self.state_and_data._gui_update_event.set()
                    time.sleep(COUNT_DOWN_WIDTH_S)
    def start(self):
        self._count_down_process.start()

__init__ではStateManagerProcessで定義したデータをクラス変数にしてマルチプロセス用の関数の割り当てをしています。

mp.Process()でsetupという関数をマルチプロセスで動かす処理に割り当てています。

setup関数の中では、カウントダウン処理に該当するトリガーが発火するまでループによるトリガーの監視を行い、もしトリガーが発火していれば、カウントダウンを開始する処理を実装しています。

start関数では__init__関数にて定義したマルチプロセスをスタートする処理を記述しています。

コードの解説（GUIProcess）

class GUIProcess():
    def __init__(self,state_and_data:StateManagerProcess) -> None:
        self.state_and_data = state_and_data
        self._gui_process = mp.Process(target=self.setup,name="GUIProcess")

    def setup(self):
        self.app = QApplication(sys.argv)
        self.window = MainWindow(self.state_and_data)
        self.window.closeEvent = lambda _: self.close()
        self.window.show()
        self.app.processEvents()
        while True:
            if self.state_and_data._gui_update_event.is_set():
                self.window.update_timer()
                self.state_and_data._gui_update_event.clear
            self.app.processEvents()
        
    def start(self):
        self._gui_process.start()

    def close(self):
        self.state_and_data._all_exit_event.set()

__init__関数とstart関数ははCountDownProcessと同様の処理を行なっています。

setup関数の中ではPyQtを用いたGUIの構成と画面更新の処理を記述しています。

MainWindowというクラスも別途定義しており、その中で主な画面設定や画面更新用の関数を定義しているので、後ほど詳しく説明します。

setup関数の中のwhile TrueにてGUIを更新するトリガーを監視してトリガーが発火していれば画面更新の処理を実行する流れになっています。

close関数については画面が閉じられたことを検知したタイミングで全てのアプリケーションを終了させるトリガーを発火させる処理を記述しています。

コードの解説（MainWindow）

class MainWindow(QMainWindow):
    def __init__(self,state_and_data:StateManagerProcess) -> None:
        super().__init__()
        self.state_and_data = state_and_data
        self.setup_gui()

    def setup_gui(self):
        # GUIのタイトル
        self.setWindowTitle("Qt CountDownTimer")
        # 中央よせウィジェットの定義
        central_widget = QWidget(self)
        self.setCentralWidget(central_widget)
        # グリッドレイアウトを定義
        layout = QGridLayout(central_widget)
        # 時間を表示するラベルの定義とレイアウトにウィジェットを追加
        self.time_label = QLabel(self.make_display_time())
        self.time_label.setAlignment(Qt.AlignCenter)  
        layout.addWidget(self.time_label,0,0)
        # カウントダウンを開始するボタンの定義とボタンが押された際の処理、レイアウトにウィジェットを追加
        start_count_down_button_widget = QPushButton("Start",self)
        start_count_down_button_widget.clicked.connect(self.start_count_down)
        layout.addWidget(start_count_down_button_widget,1,0)

    def update_timer(self):
        self.time_label.setText(self.make_display_time())

    def start_count_down(self):
        self.state_and_data.count_down_time.value = TIMER_DUARATION
        self.state_and_data._count_down_exec_event.set()

    def make_display_time(self):
        return f'{self.state_and_data.count_down_time.value:>5.3f}'

このクラスでは QMainWindow というQtのクラスを継承しています。

setup_gui関数ではGUIの配置やレイアウト等を設定しています。詳しくはコード中にコメントを記載しているのでそちらを参照してください。

update_timer関数ではプロセス間共有変数に格納されている値を用いて画面を更新する処理を記述しています。

start_count_down関数はGUI上のボタンとリンクされており、ボタンが押された際にカウントダウンを開始するトリガーを発火させ、カウンドダウン用の変数を初期化する処理を実装しています。

make_display_time関数では表示する時間についてのフォーマットを指定しています。

コードの解説（Main処理）

if __name__ == "__main__":
    state_manager_process = StateManagerProcess()
    count_down_process = CountDownProcess(state_and_data=state_manager_process)
    gui_process = GUIProcess(state_and_data=state_manager_process)

    count_down_process.start()
    gui_process.start()

    while not state_manager_process._all_exit_event.is_set():
        pass

    count_down_process._count_down_process.terminate()
    count_down_process._count_down_process.join()
    gui_process._gui_process.terminate()
    gui_process._gui_process.join()

Main処理では各クラスをインスタンス化し、GUIを閉じられるまで待機する処理を実装しています。

GUIが閉じれられると_all_exit_eventが発火するのでそれを検知して待機状態を解除してマルチプロセスで実行している処理を terminate で停止しています。

終わりに

今回は「PyQtとMultiProcessingでリアルタイムGUIを作成する方法」と題してPythonのみでリアルタイムに動きがあるPC用GUIアプリケーションを実装してきました。

今回の例ではカウントダウンタイマーという簡単な例でしたが、複数のハードウェアやデバイスを連携させたシステム構築も行うことができます。

マルチプロセスで処理を並列化することで処理待ちが発生しにくくなりますので、デモ等を実装したい場合にぜひご活用頂けたらと思います。

最後まで読んでくれてありがとうございました！！

よかったらお気に入り登録をお願いします！！！！

コメントを残す

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

前の記事

【GAS】Googleフォームの自動返信メールをGASを使って…

次の記事

【GAS】GoogleフォームとGoogleカレンダーで予約シ…