[Python] PyQt – QThread (OTP Program)

GUI를 구현할 때 thread를 사용해 producer와 consumer를 각각 구현한다. PyQt에선 좀더 쉽게 사용할 수 있도록 QThread를 제공한다. 아래 프로그램은 하나의 thread가 tic을 튀기면서 signal을 또 다른 thread에게 전달해주어 또 다른 thread가 화면에 뿌리도록 한다.

import sys
import time

from PyQt5.QtWidgets import QWidget
from PyQt5.QtWidgets import QTextEdit
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWidgets import QBoxLayout
from PyQt5.QtCore import Qt
from PyQt5.QtCore import QThread
from PyQt5.QtCore import pyqtSignal

class TicGenerator(QThread):
    tic = pyqtSignal(name="Tic")

    def __init__(self):
        QThread.__init__(self)

    def __del__(self):
        self.wait()

    def run(self):
        while True:
            t = int(time.time())
            if not t % 5 == 0:
                self.usleep(1)
                continue
            self.Tic.emit()
            self.msleep(1000)


class Form(QWidget):
    def __init__(self):
        QWidget.__init__(self, flags=Qt.Widget)
        self.te = QTextEdit()
        self.tic_gen = TicGenerator()
        self.init_widget()
        self.tic_gen.start()

    def init_widget(self):
        self.setWindowTitle("Custom Signal")
        form_lbx = QBoxLayout(QBoxLayout.TopToBottom, parent=self)
        self.setLayout(form_lbx)

        self.tic_gen.Tic.connect(
            lambda: self.te.insertPlainText(time.strftime("[%H:%M:%S] Tic!\n"))
        )

        form_lbx.addWidget(self.te)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    form = Form()
    form.show()
    exit(app.exec_())

Tic을 튀기는 thread의 동작은 TickGenerator(QThread)의 run() 함수에 기술된다. 내부 코드를 보면 현재 시간 초를 int로 받아와 5로 나눠질 때, 즉 5초마다 pyqtSignal().emit()을 수행시켜준다.

위 signal에 대한 연결은 실제 화면에 뿌려주는 main thread의 self.tic_gen.Tic.connect([함수 객체]) 내부에 선언한 함수의 동작을 수행한다. 실제 함수 객체 내부엔 현재 시간의 텍스트를 뿌려주도록 하고 있다.

Example

아래 코드는 금융에서 사용하는 OTP card의 예시다. 자세한 코드 리뷰는 코드의 주석으로 넣어서 설명한다.

import sys

from PyQt5.QtWidgets import QWidget
from PyQt5.QtWidgets import QLabel
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWidgets import QBoxLayout
from PyQt5.QtCore import Qt
from PyQt5.QtCore import QThread
from PyQt5.QtCore import pyqtSignal
import string
import time
import random

class OtpTokenGenerator(QThread):
    value_changed = pyqtSignal(str, name="ValueChanged")
    expires_in = pyqtSignal(int, name="ExpiresIn")

    EXPIRE_TIME = 5

    def __init__(self):
        QThread.__init__(self)
        # Get uppercase
        self.characters = list(string.ascii_uppercase)
        self.token = self.generate()

    def __del__(self):
        self.wait()

    def generate(self):
        # Get random string
        random.shuffle(self.characters)
        # Parse 5 bytes only
        return ''.join(self.characters[0:5])

    def run(self):
        self.value_changed.emit(self.token)  
        while True:
            # 매 초마다 갱신, 5 - 4 - 3 - 2 - 1 순으로 감소됨
            t = int(time.time()) % self.EXPIRE_TIME
            self.expires_in.emit(self.EXPIRE_TIME - t)  
            if t != 0:
                self.usleep(1)
                continue
            # 5초에 한 번씩 수행
            self.token = self.generate()
            self.value_changed.emit(self.token)
            self.msleep(1000)


class Form(QWidget):
    def __init__(self):
        QWidget.__init__(self, flags=Qt.Widget)
        self.lb_token = QLabel()
        self.lb_expire_time = QLabel()
        self.otp_gen = OtpTokenGenerator()
        self.init_widget()
        # Thread gets started
        self.otp_gen.start()

    def init_widget(self):
        self.setWindowTitle("Custom Signal")
        form_lbx = QBoxLayout(QBoxLayout.TopToBottom, parent=self)
        self.setLayout(form_lbx)

        # token connection
        self.otp_gen.ValueChanged.connect(self.lb_token.setText)
        # count connection, the value should be changed to string type
        self.otp_gen.ExpiresIn.connect(lambda v: self.lb_expire_time.setText(str(v)))

        form_lbx.addWidget(self.lb_token)
        form_lbx.addWidget(self.lb_expire_time)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    form = Form()
    form.show()
    exit(app.exec_())