Python的擴展接口[1] -> 串口通信

串口通信 / Serial Communication

---

 

1 串口簡介 / Serial Introduction

串行接口(Serial Interface)簡稱串口，通常為COM接口，數據發送方式為一位一位地順序傳送，通信線路簡單，一對傳輸線即可實現雙向通信，適用於遠距離且傳輸速度慢的通信。其中最常用的也是標准接口為RS-232，除此之外還有RS-422，RS-485等。

 

2 pyserial環境搭建

Pyserial的安裝可以通過pip進行，直接在命令行輸入下列命令進行安裝，

pip install pyserial

 

3 pyserial基本用法

利用pyserial可以實現Python對串口的控制，其基本使用方式十分簡單，主要為一下幾步，

       1. 利用Serial類傳入配置參數，生成串口實例;

       2. 利用串口實例的write()/read()方法消息的發送與接收。

Note: 關於Serial類，主要參數為端口和波特率，更多參數可查看源代碼。

import serial

ser = serial.Serial(port='com1', baudrate=9600)
print(ser.portstr)

def serial_communi(ser, msg):
    # n is the length of msg sent
    n = ser.write(msg.encode())
    print(n)
    s = ser.read(n)
    print(s)

 

4 read_until 方法的實現與擴展
在pyserial中，read_until 方法只能處理一個終止符，而當需要處理多個終止標志的時候，則需要重新定義一個新的方法，參考 pyserial 源代碼中的 read_until 函數，其實現方式為：

1. 每次讀取一個字符，加入待返回字符串
2. 根據終止符長度n，獲取字符中后 n 位字符，檢測是否為終止符

下面的代碼供參考，

    def _read_until(self):
        line = ''
        while True:
            try:
                c = self.read(1)
            except (
                    AttributeError, TypeError,          # COM Exception
                    ValueError, EOFError,               # TELNET Exception
                    self._connection.OpenCloseException) as e:
                # Above Exception happens when communication manually closed,
                # if still should run, condition sleep here,
                # else return None to exit thread
                if self.should_run:
                    self._cond_wait(self._recv_cond)
                    continue
                else:
                    return None
            if c:
                line += c
                self._result.put(c)
                tokens = [self._term_token, self._abort_token, self._cr_token, self._lf_token]
                for token in tokens:
                    if not line[-len(token):] == token:
                        continue
                    if token == self._cr_token or token == self._lf_token:
                        return line
                    if token == self._term_token:
                        self.waiter_wake_up()
                        self.termi_signal(True)
                        self._result.save(strip=self._term_token)
                        self.logger.debug("TERMINATE %s" % self._term_token)
                        return line
                    if token == self._abort_token:
                        # TODO: handle this abort token encountered
                        self.logger.debug("ABORT %s" % self._abort_token)
                        pass
            # When timeout will enter this else
            # else:
            #     return line