seiral_moniter.py

#!/usr/bin/env python
# coding: utf-8

# Serial 설정과 동작을 관리하는 모듈

import sys

from PyQt5.QtWidgets import QWidget
from PyQt5.QtWidgets import QBoxLayout
from PyQt5.QtWidgets import QGridLayout
from PyQt5.QtWidgets import QLabel
from PyQt5.QtWidgets import QComboBox
from PyQt5.QtWidgets import QGroupBox
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWidgets import QLineEdit
from PyQt5.QtCore import Qt
from PyQt5.QtCore import QThread
from PyQt5.QtSerialPort import QSerialPort
from PyQt5.QtSerialPort import QSerialPortInfo
from PyQt5.QtCore import QIODevice
from PyQt5.QtCore import QWaitCondition
from PyQt5.QtCore import QMutex
from PyQt5.QtCore import QByteArray
from PyQt5.QtCore import pyqtSlot
from PyQt5.QtCore import pyqtSignal

__author__ = "Deokyu Lim <hong18s@gmail.com>"
__platform__ = sys.platform







class SerialReadThread(QThread):
    """
    시리얼 연결이 성공하면 항상 데이터를 수신할 수 있어야 하므로
    스레드로 만들어야 한다.

    """
    # 사용자 정의 시그널 선언
    # 받은 데이터 그대로를 전달 해주기 위해 QByteArray 형태로 전달
    received_data = pyqtSignal(QByteArray, name="receivedData")

    def __init__(self, serial):
        QThread.__init__(self)
        self.cond = QWaitCondition()
        self._status = False
        self.mutex = QMutex()
        self.serial = serial

    def __del__(self):
        self.wait()

    def run(self):
        """
        들어온 데이터가 있다면 시그널을 발생
        :return:
        """
        while True:
            self.mutex.lock()
            if not self._status:
                self.cond.wait(self.mutex)

            buf = self.serial.readAll()
            if buf:
                self.received_data.emit(buf)
            self.usleep(1)
            self.mutex.unlock()

    def toggle_status(self):
        self._status = not self._status
        if self._status:
            self.cond.wakeAll()

    @pyqtSlot(bool, name='setStatus')
    def set_status(self, status):
        self._status = status
        if self._status:
            self.cond.wakeAll()


class SerialController(QWidget):
    # 시리얼포트 상수 값
    BAUDRATES = (
        QSerialPort.BaudRate.Baud1200,
        QSerialPort.BaudRate.Baud2400,
        QSerialPort.BaudRate.Baud4800,
        QSerialPort.BaudRate.Baud9600,
        QSerialPort.BaudRate.Baud19200,
        QSerialPort.BaudRate.Baud38400,
        QSerialPort.BaudRate.Baud57600,
        QSerialPort.BaudRate.Baud115200,
    )

    DATABITS = (
        QSerialPort.DataBits.Data5,
        QSerialPort.DataBits.Data6,
        QSerialPort.DataBits.Data7,
        QSerialPort.DataBits.Data8,
    )

    FLOWCONTROL = (
        QSerialPort.FlowControl.NoFlowControl,
        QSerialPort.FlowControl.HardwareControl,
        QSerialPort.FlowControl.SoftwareControl,
    )

    PARITY = (
        QSerialPort.Parity.NoParity,
        QSerialPort.Parity.EvenParity,
        QSerialPort.Parity.OddParity,
        QSerialPort.Parity.SpaceParity,
        QSerialPort.Parity.MarkParity,
    )

    STOPBITS = (
        QSerialPort.StopBits.OneStop,
        QSerialPort.StopBits.OneAndHalfStop,
        QSerialPort.StopBits.TwoStop,
    )

    received_data = pyqtSignal(QByteArray, name="receivedData")
    sent_data = pyqtSignal(str, name="sentData")

    def __init__(self):
        QWidget.__init__(self, flags=Qt.Widget)
        # 위젯 선언
        self.gb = QGroupBox(self.tr("Serial"))
        self.cb_port = QComboBox()
        self.cb_baud_rate = QComboBox()
        self.cb_data_bits = QComboBox()
        self.cb_flow_control = QComboBox()
        self.cb_parity = QComboBox()
        self.cb_stop_bits = QComboBox()
        self.cb_commend = QComboBox()
        self.cb_commend_value = QLineEdit()
        self.cb_commend_send = QPushButton(self.tr("send"))
        self.commend_label = QLabel()

        # 시리얼 인스턴스 생성
        # 시리얼 스레드 설정 및 시작
        self.serial = QSerialPort()
        self.serial_info = QSerialPortInfo()
        self.serial_read_thread = SerialReadThread(self.serial)
        self.serial_read_thread.received_data.connect(lambda v: self.received_data.emit(v))
        self.serial_read_thread.start()

        self.init_widget()

    def init_widget(self):
        self.setWindowTitle("Serial Controller")
        layout = QBoxLayout(QBoxLayout.TopToBottom, parent=self)
        grid_box = QGridLayout()

        grid_box.addWidget(QLabel(self.tr("Port")), 0, 0)
        grid_box.addWidget(self.cb_port, 0, 1)

        grid_box.addWidget(QLabel(self.tr("Baud Rate")), 1, 0)
        grid_box.addWidget(self.cb_baud_rate, 1, 1)

        grid_box.addWidget(QLabel(self.tr("Data Bits")), 2, 0)
        grid_box.addWidget(self.cb_data_bits, 2, 1)

        grid_box.addWidget(QLabel(self.tr("Flow Control")), 3, 0)
        grid_box.addWidget(self.cb_flow_control, 3, 1)

        grid_box.addWidget(QLabel(self.tr("Parity")), 4, 0)
        grid_box.addWidget(self.cb_parity, 4, 1)

        grid_box.addWidget(QLabel(self.tr("Stop Bits")), 5, 0)
        grid_box.addWidget(self.cb_stop_bits, 5, 1)

        grid_box.addWidget(QLabel(self.tr("Commend")), 6, 0)
        grid_box.addWidget(self.cb_commend, 6, 1)
        grid_box.addWidget(self.cb_commend_value, 6, 2)
        grid_box.addWidget(self.cb_commend_send, 6, 3)
        grid_box.addWidget(self.commend_label, 7, 2)

        self.Changcb("AT+RFSET")

        self.cb_commend_send.clicked.connect(lambda: self.SendData(self.cb_commend.currentText() + self.cb_commend_value.text()))
        self.cb_commend.activated.connect(lambda: self.Changcb(self.cb_commend.currentText()))

        self._fill_serial_info()
        self.gb.setLayout(grid_box)
        layout.addWidget(self.gb)
        self.setLayout(layout)

    def _fill_serial_info(self):
        # 시리얼 상수 값들을 위젯에 채운다
        self.cb_port.insertItems(0, self._get_available_port())
        self.cb_baud_rate.insertItems(0, [str(x) for x in self.BAUDRATES])
        self.cb_data_bits.insertItems(0, [str(x) for x in self.DATABITS])
        flow_name = {0: "None", 1: "Hardware", 2: "Software"}
        self.cb_flow_control.insertItems(0, [flow_name[x] for x in self.FLOWCONTROL])
        parity_name = {0: "None", 2: "Even", 3: "Odd", 4: "Space", 5: "Mark"}
        self.cb_parity.insertItems(0, [parity_name[x] for x in self.PARITY])
        stop_bits_name = {1: "1", 3: "1.5", 2: "2"}
        self.cb_stop_bits.insertItems(0, [stop_bits_name[x] for x in self.STOPBITS])
        commend_name = {0: "AT+RFSET", 1: "AT+SENSDATA_START", 2: "AT+SENSDATA_END", 3: "AT+RFSEND", 4: "AT+UPDATE", 5: "AT+SETNUM", 6: "AT+NOWSTATE",
                        7: "AT+SETDEL_WSH", 8: "AT+SETDEL_DRY"}
        self.cb_commend.insertItems(0, [commend_name[x] for x in [0,1,2,3,4,5,6,7,8]])

    @staticmethod
    def get_port_path():
        """
        현재플래폼에 맞게 경로 또는 지정어를 반환
        :return:
        """
        return {"linux": '/dev/ttyS', "win32": 'COM'}[__platform__]

    def _get_available_port(self):
        """
        255개의 포트를 열고 닫으면서 사용가능한 포트를 찾아서 반환
        :return:
        """
        available_port = list()
        port_path = self.get_port_path()

        for number in range(255):
            port_name = port_path + str(number)
            if not self._open(port_name):
                continue
            available_port.append(port_name)
            self.serial.close()
        return available_port

    def _open(self, port_name, baudrate=QSerialPort.Baud9600, data_bits=QSerialPort.Data8,
              flow_control=QSerialPort.NoFlowControl, parity=QSerialPort.NoParity, stop_bits=QSerialPort.OneStop):
        """
        인자값으로 받은 시리얼 접속 정보를 이용하여 해당 포트를 연결한다.
        :param port_name:
        :param baudrate:
        :param data_bits:
        :param flow_control:
        :param parity:
        :param stop_bits:
        :return: bool
        """
        info = QSerialPortInfo(port_name)
        self.serial.setPort(info)
        self.serial.setBaudRate(baudrate)
        self.serial.setDataBits(data_bits)
        self.serial.setFlowControl(flow_control)
        self.serial.setParity(parity)
        self.serial.setStopBits(stop_bits)
        return self.serial.open(QIODevice.ReadWrite)

    def SendData(self, data):
        # data_lis = []
        # for i in data:
        #     data_lis.append(i)
        data_bytes = bytes(data, "utf-8")
        test = bytes([0x02]) + bytes("TEST DATA", "utf-8") + bytes([0x03])
        self.serial.writeData(data_bytes)
        print(data_bytes)

    def Changcb(self, commend):
        print(commend)
        commend_dict = {'AT+RFSET': '"XXXXX"', 'AT+SENSDATA_START': '', 'AT+SENSDATA_END': '', 'AT+RFSEND': '"원하는 문자열(영어)"',
                        'AT+SETNUM': '"X,X"', 'AT+NOWSTATE': '', 'AT+SETDEL_WSH': '"X"',
                        'AT+SETDEL_DRY': '"X"', 'AT+UPDATE': '"X,X"'}
        commend_label_dict = {'AT+RFSET': '5자리 RF 주소 설정', 'AT+SENSDATA_START': '센서 데이터 UART로 바로 전송',
                              'AT+SENSDATA_END': '센서 데이터 UART로 바로 전송 종료',
                              'AT+RFSEND': 'UART로 받은 값 RF통신으로 전송','AT+SETNUM': '채널별 장치 번호 설정',
                              'AT+NOWSTATE': '현재 RF 주소 및 채널별 장치 번호 출력',
                              'AT+SETDEL_WSH': '센서 반응 지연 시간 설정(10초간격)','AT+SETDEL_DRY': '센서 반응 지연 시간 설정(1초간격)',
                              'AT+UPDATE': '꺼짐 켜짐 RF통신으로 수동 전송(0/1,장치 번호)'}
        self.cb_commend_value.setText(commend_dict[commend])
        self.commend_label.setText(commend_label_dict[commend])

    def connect_serial(self):
        serial_info = {
            "port_name": self.cb_port.currentText(),
            "baudrate": self.BAUDRATES[self.cb_baud_rate.currentIndex()],
            "data_bits": self.DATABITS[self.cb_data_bits.currentIndex()],
            "flow_control": self.FLOWCONTROL[self.cb_flow_control.currentIndex()],
            "parity": self.PARITY[self.cb_parity.currentIndex()],
            "stop_bits": self.STOPBITS[self.cb_stop_bits.currentIndex()],
        }
        status = self._open(**serial_info)
        self.serial_read_thread.setStatus(status)
        return status

    def disconnect_serial(self):
        return self.serial.close()

    @pyqtSlot(bytes, name="writeData")
    def write_data(self, data):
        self.serial.writeData(data)


class Form(QWidget):
    """
    테스트용도의 단독 실행때만 사용하는 폼
    """
    def __init__(self):
        QWidget.__init__(self, flags=Qt.Widget)
        self.te = QTextEdit()
        self.pb = QPushButton("Connect")
        self.pb_send = QPushButton("Send")
        #self.pb_send = QPushButton("Send_a")
        self.serial = SerialController()
        self.init_widget()


    def init_widget(self):
        """
        현재 위젯의 모양등을 초기화
        """
        self.setWindowTitle("Debug Mode")
        form_lbx = QBoxLayout(QBoxLayout.TopToBottom, parent=self)
        self.setLayout(form_lbx)

        self.pb.clicked.connect(self.slot_clicked_connect_button)
        self.serial.received_data.connect(self.read_data)
        test_data = bytes([0x02]) + bytes("TEST DATA", "utf-8") + bytes([0x03])
        self.pb_send.clicked.connect(lambda: self.serial.writeData(test_data))
        form_lbx.addWidget(self.serial)
        form_lbx.addWidget(self.te)
        form_lbx.addWidget(self.pb_send)
        form_lbx.addWidget(self.pb)
        #form_lbx.addWidget(self.pb_a)
        # 많이 사용하는 옵션을 미리 지정해 둔다.
        # 9600 8N1
        self.serial.cb_baud_rate.setCurrentIndex(3)
        self.serial.cb_data_bits.setCurrentIndex(3)

    @pyqtSlot(QByteArray, name="readData")
    def read_data(self, rd):
        self.te.insertPlainText(str(rd, 'ascii', 'replace'))

    @pyqtSlot(name="clickedConnectButton")
    def slot_clicked_connect_button(self):
        if self.serial.serial.isOpen():
            self.serial.disconnect_serial()
        else:
            self.serial.connect_serial()
        self.pb.setText({False: 'Connect', True: 'Disconnect'}[self.serial.serial.isOpen()])


if __name__ == "__main__":
    from PyQt5.QtWidgets import QPushButton
    from PyQt5.QtWidgets import QTextEdit
    print("1")
    app = QApplication(sys.argv)
    print("2")
    excepthook = sys.excepthook
    print("3")
    sys.excepthook = lambda t, val, tb: excepthook(t, val, tb)
    print("4")
    form = Form()
    print("5")
    form.showMaximized()
    print("6")
    exit(app.exec_())