从 pty master 端从 python 调用 libc.select

Question

我正在尝试创建一个模拟的 TTY 连接目标设备，我可以通过即 minicom 连接到该设备。我正在使用/dev/ptmx 创建一个 pty 并打印出要由即 minicom 打开的从属名称： Please connect to: /dev/pts/4。然后在 python 端，我使用 os.read 和 os.write 来做 io 并模拟我的目标：

import os, re, termios                                                                                                                                                                                                                                                         
from ctypes import *
class dev():
    def __init__(self):
        pass  
    def createpty(self):                                                                                                                                                                                                                                                       

        self.fd3 = os.open("/dev/ptmx", os.O_RDWR | os.O_NONBLOCK);                                                                                                                                                                                                            
        if self.fd3 < 0:                                                                                                                                                                                                                                                       
            print("Couldn't open output /dev/ptmx\n")                                                                                                                                                                                                                          
        libc = cdll.LoadLibrary("libc.so.6")                                                                                                                                                                                                                                   
        libc.grantpt(self.fd3);                                                                                                                                                                                                                                                
        libc.unlockpt(self.fd3);                                                                                                                                                                                                                                               
        libc.ptsname.restype = c_char_p                                                                                                                                                                                                                                        
        self.slave = libc.ptsname(self.fd3)                                                                                                                                                                                                                                    
        print("Please connect to:" + self.slave);                                                                                                                                                                                                                                                    

        self.old = termios.tcgetattr(self.fd3)                                                                                                                                                                                                                                 
        n = termios.tcgetattr(self.fd3)                                                                                                                                                                                                                                        

        n[3] = n[3] & ~(termios.ECHO|termios.ICANON) # c_lflag                                                                                                                                                                                                                 
        n[3] = n[3] & 0                                                                                                                                                                                                                                                        
        n[4+1] = n[4+1] & 0xffff0000;                                                                                                                                                                                                                                          

        termios.tcsetattr(self.fd3, termios.TCSANOW, n)           

整个过程相当繁琐。我想在 python 端使用select，这迫使我再次使用 ctypes。我无法使用 os.fdopen 将 self.fd3 包装到文件中，因为我需要防止在 self.fd3 上调用 close。

所以我有两个问题：

• 有谁知道在 python 中实现 pty master 端时处理 pty 创建和操作的现成 python 库？
• 如果没有：是否有示例说明如何通过 ctypes 调用 libc.select？

Answer 1

正如@CristiFati 所说，Python 的 select 可用于 FD，但是为了完整起见，下面是我使用 ctype 的 select 的肮脏技巧。这实现了一个 tty 回显服务器。连接打印出来的pty slave，即minicom -D /dev/pty/4

import os, re, termios
import threading, os, re, sys, select, math, binascii, errno, time
from ctypes import *
import errno

class pty():
    def __init__(self, n="undef"):
        self.cmds = {}
        self.createpty()
        self.name = n
        self.goon = 1
        self.input = ""
        self.output = []

    def createpty(self):

        self.fd3 = os.open("/dev/ptmx", os.O_RDWR | os.O_NONBLOCK);
        if self.fd3 < 0:
            print("Couldn't open output /dev/ptmx\n")
        libc = cdll.LoadLibrary(None)
        libc.grantpt(self.fd3);
    libc.unlockpt(self.fd3);
        libc.ptsname.restype = c_char_p
        self.slave = libc.ptsname(self.fd3)
        print("Use: " + self.slave);

        self.old = termios.tcgetattr(self.fd3)
        n = termios.tcgetattr(self.fd3)

        n[3] = n[3] & ~(termios.ECHO|termios.ICANON) # c_lflag
        n[3] = n[3] & 0

#define VTIME 5
#define VMIN 6
#struct termios
#  {
#    tcflag_t c_iflag;      /* input mode flags */
#    tcflag_t c_oflag;      /* output mode flags */
#    tcflag_t c_cflag;      /* control mode flags */
#    tcflag_t c_lflag;      /* local mode flags */
#    cc_t c_line;           /* line discipline */
#    cc_t c_cc[NCCS];       /* control characters */
#    speed_t c_ispeed;      /* input speed */
#    speed_t c_ospeed;      /* output speed */
##define _HAVE_STRUCT_TERMIOS_C_ISPEED 1
##define _HAVE_STRUCT_TERMIOS_C_OSPEED 1
#  };

        n[4+1] = n[4+1] & 0xffff0000;
    #ioc.c_cc[VMIN] = 0;  # byte 6 + 7
    #ioc.c_cc[VTIME] = 0;

        termios.tcsetattr(self.fd3, termios.TCSANOW, n)

        #self.io = os.fdopen(self.fd3, "r+");


#       fd3io = fdopen (fd3, "r+");
#       fflush (fd3io);
#       setbuf (fd3io, NULL);
#     }


#typedef long int __fd_mask;
#define __NFDBITS   (8 * (int) sizeof (__fd_mask))
#define __FD_ELT(d) ((d) / __NFDBITS)
#define __FD_MASK(d)    ((__fd_mask) (1UL << ((d) % __NFDBITS)))

__FD_SETSIZE=1024

def fdmask():
    l = __FD_SETSIZE/64
    r = (c_ulong*l)()
    for i in range(l):
        r[i] = 0;
    return r

def fdset(a,b):
    a[b>>6] = a[b>>6] | (1 << (b % 64))

def fdisset(a,b):
    return (a[b>>6] & (1 << (b % 64)))

def setset(b, a):
    for i in a:
        fdset(b, i)

def setget(b, a):
    return [ i for i in a if fdisset(b,i) ]

class timeval(Structure):
    _fields_ = [("tv_sec", c_long), ("tv_usec", c_long)]

def cselect(r,w,x,timeout):

    libc = cdll.LoadLibrary(None)

    ra = fdmask();
    wa = fdmask();
    xa = fdmask();
    m = max(r+w+x)
    setset(ra, r),
    setset(wa, w),
    setset(xa, x),
    tv = timeval()
    tv.tv_sec = int(timeout)
    tv.tv_usec = int(1000000.0 * (timeout%1.0))
    ret = libc.select(m+1, ra, wa, xa, pointer(tv));
    if (ret < 0):
        print ("Error %d" %(ret));
    elif (ret == 0):
        return ([],[],[])
    else:
        return (setget(ra,r),setget(wa,w),setget(xa,x))

def main():
    if sizeof(c_ulong) != 8:
        raise("64bit os expected with long int = 64bit")
    a = pty();
    while True:
        try:
            (r,w,x) = cselect([a.fd3],[],[a.fd3],1)
            if (len(x) > 0):
                time.sleep(0.1);
                print("x");
                continue
            if (len(r) > 0):
                # implement an echo-server:
                v = os.read(a.fd3, 1)
                print (".");
                os.write(a.fd3,v)
        except pty as e:
            print (str(e));
            raise(e)

if __name__ == "__main__":
    main()