如何用Python代码替代socat实现虚拟串口通信？

你可以用Python标准库的pty模块结合线程实现虚拟串口对功能，完全不需要调用外部socat命令，代码示例如下：

import pty
import os
import threading

class VirtualSerialPair:
    def __init__(self):
        self.master1 = None
        self.master2 = None
        self.slave1 = None
        self.slave2 = None
        self.threads = []

    def _forward_data(self, src_fd, dest_fd):
        """从源文件描述符读取数据并写入目标描述符"""
        try:
            while True:
                data = os.read(src_fd, 1024)
                if not data:
                    break
                os.write(dest_fd, data)
        except Exception:
            pass
        finally:
            os.close(src_fd)
            os.close(dest_fd)

    def start(self):
        # 创建两个伪终端对
        self.master1, self.slave1 = pty.openpty()
        self.master2, self.slave2 = pty.openpty()

        # 获取伪终端的设备路径（比如/dev/pts/xx）
        self.port1 = os.ttyname(self.slave1)
        self.port2 = os.ttyname(self.slave2)

        # 启动双向数据转发线程
        thread1 = threading.Thread(target=self._forward_data, args=(self.master1, self.master2))
        thread2 = threading.Thread(target=self._forward_data, args=(self.master2, self.master1))
        
        thread1.daemon = True
        thread2.daemon = True
        
        thread1.start()
        thread2.start()
        
        self.threads.extend([thread1, thread2])
        print(f"虚拟串口对创建完成：{self.port1} <-> {self.port2}")

    def stop(self):
        # 关闭文件描述符触发线程退出
        if self.master1:
            os.close(self.master1)
        if self.master2:
            os.close(self.master2)
        # 等待线程结束
        for thread in self.threads:
            if thread.is_alive():
                thread.join(timeout=1)

使用方式 #

替代原有的socat_runner方法，直接实例化类并调用start即可：

# 示例调用
serial_pair = VirtualSerialPair()
serial_pair.start()
# 此时serial_pair.port1和serial_pair.port2就是可用的虚拟串口路径
# 不需要时调用stop()清理资源
# serial_pair.stop()

核心说明 #

• pty.openpty()创建伪终端对，slave端就是对外暴露的串口设备路径，格式和socat生成的/dev/pts/x完全一致。
• 两个线程分别负责不同方向的数据转发，实现串口间的双向通信。
• 守护线程设置确保主进程退出时自动清理线程资源。

内容的提问来源于stack exchange，提问作者Amit Chen