我正在尝试在Linux上对串口进行多路复用。我正在使用嵌入式系统,该系统仅具有一个串口,因此最好有多个进程与之通信。
常见的用例是:
- 运行测试的主程序(发送命令并接收输出); - 另一个记录所有串口活动的程序; - 用户终端打开以发送其他命令和/或在测试期间发生错误后执行事后分析。
首先,我编写了一个简单的Python脚本来打开n个伪终端对(加上串口),并使用poll语句将输入/输出定向到正确的位置:
常见的用例是:
- 运行测试的主程序(发送命令并接收输出); - 另一个记录所有串口活动的程序; - 用户终端打开以发送其他命令和/或在测试期间发生错误后执行事后分析。
首先,我编写了一个简单的Python脚本来打开n个伪终端对(加上串口),并使用poll语句将输入/输出定向到正确的位置:
# Removed boiler plate and error checking for clarity
##### Serial port setup
ttyS = serial.Serial(device, baudrate, width, parity, stopbits, 1, xon, rtc)
ttyS.setTimeout(0) # Non-blocking
##### PTYs setup
pts = []
for n in range(number_of_slave_terminals):
master, slave = os.openpty()
# Print slave names so others know where to connect
print >>sys.stderr, 'MUX > fd: %d pty: %s' % (slave, os.ttyname(slave))
pts.append(master)
##### Poller setup
poller = select.poll()
poller.register(ttyS.fd, select.POLLIN | select.POLLPRI)
for pt in pts:
poller.register(pt, select.POLLIN | select.POLLPRI)
##### MAIN
while True:
events = poller.poll(500)
for fd, flag in events:
# fd has input
if flag & (select.POLLIN | select.POLLPRI):
# Data on serial
if fd == ttyS.fd:
data = ttyS.read(80)
for pt in pts:
os.write(pt, data)
# Data on other pty
else:
ttyS.write(os.read(fd, 80))
如果每个pty都已连接,这种方法非常有效。如果有一些未连接的pty,则最终其缓冲区会填满,并在写入时被阻塞。似乎我需要知道哪些从设备已连接或某种按需打开pty的方法。
我在这个问题上发现了一个巧妙的技巧,在该问题中,该人仅需要读取串行端口部分,因此我调整了我的脚本:
##### Serial port setup
ttyS = serial.Serial(device, baudrate, width, parity, stopbits, 1, xon, rtc)
ttyS.setTimeout(0) # Non-blocking
##### PTYs setup
pts = []
for n in range(number_of_slave_terminals):
master, slave = os.openpty()
# slaves
print >>sys.stderr, 'MUX > fd: %d pty: %s' % (slave, os.ttyname(slave))
os.close(slave) # POLLHUP trick
# masters
pts.append(master)
##### Poller setup
reader = select.poll()
writer = select.poll()
reader.register(ttyS, select.POLLIN | select.POLLPRI)
for pt in pts:
reader.register(pt, select.POLLIN | select.POLLPRI)
writer.register(pt, select.POLLIN | select.POLLPRI | select.POLLOUT)
def write_to_ptys(data):
events = writer.poll(500)
for fd, flag in events:
# There is someone on the other side...
if not (flag & select.POLLHUP):
os.write(fd, data)
##### MAIN
while True:
events = reader.poll(500)
for fd, flag in events:
if flag & (select.POLLIN | select.POLLPRI):
# Data on serial
if fd == ttyS.fd:
write_to_tty(ttyS.read(80))
# Data on other pty
else:
ttyS.write(os.read(fd, 80))
这个方法可以实现,但是会使用100%的CPU,因为读取器轮询被POLLHUP事件淹没了。
我想如果我使用TCP套接字而不是伪终端,我就可以得到我想要的结果。缺点是我必须修改所有已经使用终端的其他脚本来使用套接字(我知道我可以使用socat,但我只想要更简单的东西)。此外,还有所有的网络开销...
那么,有什么好的想法吗?
我不介意使用其他工具,只要设置简单即可。我也不介意使用其他语言,我只是最喜欢Python。
screen
或dtach
。 - twalbergminicom
或其他工具)在嵌入式系统上运行它,并且它允许你运行多个虚拟终端,你可以使用按键命令在它们之间切换(还可以创建/销毁虚拟屏幕和许多其他命令)。 - twalberg