Step 5. QThread 비동기 수신

0. 순서

Step 1. GUI + LED 제어 맛보기

• 버튼 1개
• 포트·보드레이트 코드에 고정
• GUI → 시리얼 → 아두이노 → 로그 흐름 경험 중심

 

Step 2. 포트 선택 추가

• 포트 선택 콤보박스 추가
• “어떤 장치와 통신하는가” 개념 학습
• Step 2는 기능 확장이 아니라, 통신 대상 선택이라는 개념을 추가하는 단계입니다.

 

Step 3. 문자열 송신/수신 추가

• 자유 문자열 전송 입력창 추가
• TX / RX 개념 명확화
• LED 제어 + Echo 응답 확인

 

Step 4. QTimer 수신 처리 추가

• 주기적 수신 처리 구조 도입
• GUI 멈춤 현상 체험
• “왜 비동기가 필요한가” 문제 제기

 

👉 Step 5. QThread 비동기 수신 추가

• RX 전용 스레드 분리
• Signal / Slot 구조 학습
• 안정적인 GUI + 실시간 수신 구현

 

1. 목표

목표

• 이 단계의 목표는
  시리얼 수신(RX)을 GUI 흐름에서 완전히 분리하여,
  안정적인 화면 응답성과 실시간 수신을 동시에 달성하는 것입니다.
• Step 4에서 “QTimer로도 GUI가 불안정해질 수 있다”  였다면,
  Step 5에서는 “RX는 GUI 스레드에서 처리하면 안 된다” 로 확장합니다.

 

학습을 마치면 할 수 있는 것

• RX 처리를 **전용 스레드(QThread)**로 분리할 수 있다.
• GUI 스레드와 작업 스레드의 역할 차이를 이해한다.
• Signal / Slot을 사용해 스레드 간 안전한 통신을 구현할 수 있다.
• 수신 데이터가 많아도 GUI가 멈추지 않는 구조를 만들 수 있다.

 

2. 아두이노 스케치 구현 

: 교재 P.110 ~ 143

Step 1. 동일

 

3. PySide6 앱 1단계 구현

└─ 3.1 전체 소스 코드

핵심 변화 요약

구분	Step 4 (QTimer)	Step 5 (QThread)
RX 처리 위치	GUI 스레드	전용 스레드
GUI 반응성	불안정	안정적
구조	임시 해결	근본 해결
목적	문제 체험	문제 해결

# Step 5: QThread 기반 비동기 수신 처리
# 목적: RX 전용 스레드 분리 + 안정적인 GUI + 실시간 수신

import sys
import time
import serial
from serial.tools import list_ports

from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import (
    QApplication, QMainWindow, QPushButton,
    QTextEdit, QVBoxLayout, QWidget,
    QComboBox, QLabel, QLineEdit, QHBoxLayout
)


class SerialReceiver(QThread):
    received = Signal(str)

    def __init__(self, ser):
        super().__init__()
        self.ser = ser
        self.running = True

    def run(self):
        while self.running:
            if self.ser and self.ser.in_waiting:
                line = self.ser.readline().decode().strip()
                if line:
                    self.received.emit(line)
            time.sleep(0.01)

    def stop(self):
        self.running = False


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Serial LED Test - Step 5")
        self.resize(420, 420)

        self.ser = None
        self.rx_thread = None
        self.led_on = False

        # --- 포트 선택 ---
        self.combo_port = QComboBox()
        self.load_ports()

        # --- 송신 UI ---
        self.edit_send = QLineEdit()
        self.edit_send.setPlaceholderText("Send text (ex: hello)")
        self.btn_send = QPushButton("Send")

        # --- LED 버튼 ---
        self.btn_led = QPushButton("LED OFF")

        self.log = QTextEdit()
        self.log.setReadOnly(True)

        # --- Layout ---
        layout = QVBoxLayout()
        layout.addWidget(QLabel("Port"))
        layout.addWidget(self.combo_port)

        send_row = QHBoxLayout()
        send_row.addWidget(self.edit_send)
        send_row.addWidget(self.btn_send)

        layout.addLayout(send_row)
        layout.addWidget(self.btn_led)
        layout.addWidget(self.log)

        central = QWidget()
        central.setLayout(layout)
        self.setCentralWidget(central)

        # --- Signals ---
        self.btn_led.clicked.connect(self.toggle_led)
        self.btn_send.clicked.connect(self.send_text)
        self.edit_send.returnPressed.connect(self.send_text)

    def load_ports(self):
        for p in list_ports.comports():
            self.combo_port.addItem(p.device)

    def ensure_connection(self):
        if self.ser is None:
            port = self.combo_port.currentText()
            if not port:
                return False
            self.ser = serial.Serial(port, 9600, timeout=0.1)
            self.log.append(f"Connected to {port}")

            # --- RX 스레드 시작 ---
            self.rx_thread = SerialReceiver(self.ser)
            self.rx_thread.received.connect(self.on_received)
            self.rx_thread.start()

        return True

    def toggle_led(self):
        if not self.ensure_connection():
            return
        self.led_on = not self.led_on
        msg = "1" if self.led_on else "0"
        self.send(msg)
        self.btn_led.setText("LED ON" if self.led_on else "LED OFF")

    def send_text(self):
        if not self.ensure_connection():
            return
        text = self.edit_send.text().strip()
        if not text:
            return
        self.send(text)
        self.edit_send.clear()

    def send(self, text):
        self.ser.write((text + "\n").encode())
        self.log.append(f"TX: {text}")

    # --- [핵심] 스레드에서 받은 데이터 처리 ---
    def on_received(self, text):
        self.log.append(f"RX: {text}")

    def closeEvent(self, event):
        if self.rx_thread:
            self.rx_thread.stop()
            self.rx_thread.wait()
        super().closeEvent(event)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    w = MainWindow()
    w.show()
    sys.exit(app.exec())​

 

└─ 3.2 전체 동작 흐름

[ 문자열 전송 (TX) ]  ────────────────┐
                                    ↓
                           [ Arduino 처리 ]
                                    ↓
[ RX 전용 스레드 ]  ←──── 수신 대기
        ↓
[ Signal 발생 ]
        ↓
[ GUI 슬롯(on_received) ]
        ↓
[ 로그 창에 RX 표시 ]

 

└─ 3.3 RX 전용 스레드 분리

소스코드 

class SerialReceiver(QThread):
    received = Signal(str)

 

의미

• RX는 GUI와 완전히 분리된 실행 흐름
• GUI 이벤트 루프를 절대 점유하지 않음

└─ 3.4 Signal / Slot 구조

소스코드 (스레드 → GUI 전달)

self.rx_thread.received.connect(self.on_received)

def on_received(self, text):
    self.log.append(f"RX: {text}")

 

의미

• 스레드에서 UI 직접 접근 금지
• 데이터는 반드시 Signal로 전달
• Slot은 항상 GUI 스레드에서 실행

└─ 3.5 Step 5에서 해결된 문제

문제	Step 4	Step 5
GUI 멈춤	발생	해결
RX 지연	가능	없음
구조 안정성	낮음	높음
실무 적합성	❌	⭕️

4. 테스트

테스트 1: 기본 동작

• LED 제어 정상
• Echo 응답 정상

 

테스트 2: 연속 수신

• 문자열 빠르게 연속 전송
• 로그가 실시간으로 출력됨
• GUI 버튼 반응 유지

 

정상 동작 기준

• RX 수신 중에도 GUI가 멈추지 않는다.
• TX / RX 로그가 즉시 갱신된다.
• 프로그램 종료 시 오류가 발생하지 않는다.