使用 serial_asyncio 的 Python SLIP 解码器

pip install -U pyserial-asyncio

pip install -U ansicolors

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
__author__ = "Uli Köhler"
__license__ = "CC0 1.0 Universal"

import asyncio
from colors import red
import serial_asyncio

SLIP_END = 0o300
SLIP_ESC = 0o333
SLIP_ESCEND = 0o334
SLIP_ESCESC = 0o335

def handle_slip_message(msg):
    print(f"Received message of length", len(msg))

class SLIPProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.msg = bytes() # 消息缓冲区
        self.transport = transport
        print('port opened', transport)
        transport.serial.rts = False  # You can manipulate Serial object via transport
        # 发送"回车"以提示输出
        self.buf = b''

    def check_for_slip_message(self):
        # 识别数据中的消息结尾
        decoded = []
        last_char_is_esc = False
        for i in range(len(self.buf)):
            c = self.buf[i]
            if last_char_is_esc:
                # 此字符必须是
                # SLIP_ESCEND 或 SLIP_ESCESC
                if c == SLIP_ESCEND: # 字面 END 字符
                    decoded.append(SLIP_END)
                elif c == SLIP_ESCESC: # 字面 ESC 字符
                    decoded.append(SLIP_ESC)
                else:
                    print(red("遇到无效的 SLIP 转义序列。忽略..."))
                    # 忽略消息的错误部分
                    self.buf = self.buf[i+1:]
                    break
                last_char_is_esc = False # 重置状态
            else: # 上一个字符不是 ESC
                if c == 192: # 消息结尾
                    # 从缓冲区移除当前消息
                    self.buf = self.buf[i+1:]
                    # 发出消息
                    return bytes(decoded)
                elif c == SLIP_ESC:
                    # 接下来处理转义字符
                    last_char_is_esc = True
                else: # 任何其他字符
                    decoded.append(c)
        # 缓冲区中没有更多字节 => 没有更多消息
        return None

    def data_received(self, data):
        # 将新数据追加到缓冲区
        self.buf += data
        while True:
            msg = self.check_for_slip_message()
            if msg is None:
                break # 需要等待更多数据
            else: # msg 不是 None
                handle_slip_message(msg)

    def connection_lost(self, exc):
        print('port closed')
        self.transport.loop.stop()

    def pause_writing(self):
        print('pause writing')
        print(self.transport.get_write_buffer_size())

    def resume_writing(self):
        print(self.transport.get_write_buffer_size())
        print('resume writing')

loop = asyncio.get_event_loop()
coro = serial_asyncio.create_serial_connection(loop, SLIPProtocol, '/dev/ttyACM0', baudrate=115200)
transport, protocol = loop.run_until_complete(coro)
loop.run_forever()
loop.close()