如何使用twisted处理多个R/W串口？

Question

浏览扭曲手指教程并查看 SO 问题：

• Question-1
• Question-2

但是，我（还）不能编写一个可以从多个串行端口读取和写入的扭曲程序，尤其是在协议涉及读取单行或多行并相应地写回设备的情况下。

我要做的是为 2 个调制解调器打开 2 对（即总共 4 个）串行端口。与调制解调器的通信使用 Hayes AT 命令集。虽然与调制解调器的大多数命令/响应交换是通过命令端口进行的，但对于每个调制解调器，只有很少的诊断信息只能通过诊断端口获得。诊断信息应导致状态机（设备状态、连接状态）被修改。

这是我理解为潜在方法的粗略骨架程序（基于单端口示例）：

class CommandProtocol(LineOnlyReceiver):
    def connectionMade(self):
        log.msg("Connected to command port")

    def lineReceived(self, line):
        print repr(line)
        processCommandLine(line)

class DiagnosticProtocol(LineOnlyReceiver):
    def connectionMade(self):
        log.msg("Connected to diag port")

    def lineReceived(self, line):
        print repr(line)
        processDiagnosticLine(line)

...

# modem1 ports
cmdPort[0] = SerialPort(CommandProtocol, "/dev/ttyUSB0", reactor, 115200)
diagPort[0] = SerialPort(DiagnosticProtocol, "/dev/ttyUSB1", reactor, 115200)
# modem2 ports
cmdPort[1] = SerialPort(CommandProtocol, "/dev/ttyUSB3", reactor, 115200)
diagPort[1] = SerialPort(DiagnosticProtocol, "/dev/ttyUSB4", reactor, 115200)

但是，我不知道如何执行以下操作：

• 如何/在哪里接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令？
• 将命令端口上接收到的信息关联起来，用于调制解调器 1 的 ttyUSB0 和 ttyUSB1 以及调制解调器 2 的另一对？请注意，每个调制解调器都有自己的状态机（设备状态和连接状态）
• twisted 是否提供任何机制来按应用程序管理多个状态机？
• 与调制解调器的 USB 串行连接可能由于调制解调器被拔出而被破坏，并在重新插入时重新建立。如何检测此类事件并将相应设备端口的监控添加到 reactor ？目前，我在主应用程序中进行静态处理。

Answer 1

注意您的示例代码

我没有看到您在将类注册到反应器之前实例化它们。我预计那会失败得很惨。这是我的运行代码的类似sn-p：

# stuff to process messages coming from the serial port
class SerialEater(basic.LineReceiver):
    statusCallback = None

    def __init__(self):
        self.keyinprocess = None

    def lineReceived(self, data):
      self.dealWithSerial(data)

    def connectionLost(self, reason):
      if(reactor.running):
        print "Serial lost but reactor still running! reason: " + str(reason) + " at time " + time.asctime()
    [...etc...]



# Register the serialport into twisted
serialhandler = SerialEater()                   # <------------- instantiate
SerialPort(serialhandler, '/dev/ttyUSB0', reactor, baudrate=115200)

---

我如何/在哪里接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令？

就像您如何将串行处理程序注册到 Twisted 中一样，您可以为标准 io 注册处理程序，例如：

# stuff to pull cbreak char input from stdin
class KeyEater(basic.LineReceiver):

    def __init__(self):
      self.setRawMode() # Switch from line mode to "however much I got" mode

    def connectionLost(self, reason):
      if(reactor.running):
        self.sendLine( "Keyboard lost but reactor still running! reason: " + str(reason) + " at time " + time.asctime())

    def rawDataReceived(self, data):
      key = str(data).lower()[0]
      try:
        if key == '?':
          key = "help"
     [...etc...]

# register the stdio handler into twisted
keyboardobj = KeyEater()
keyboardobj.serialobj = serialhandler
stdio.StandardIO(keyboardobj,sys.stdin.fileno())

---

将调制解调器 1 的 ttyUSB0 和 ttyUSB1 命令端口接收到的信息相关联，调制解调器2 的另一对也类似？请注意，每个调制解调器都有自己的状态机（设备状态和连接状态）

在正常使用中，每个连接实例都有自己的状态机（包装在与连接一起注册到反应器中的类的实例中）。

作为程序员，您可以选择如何连接类的状态，但通常是通过推送对合作伙伴类的引用。

下面，此答案包含可运行的代码，将说明数据如何在状态机/接口之间连接。此 SO 中也说明了这一点：Persistent connection in twisted

---

twisted 是否提供任何机制来管理应用程序的多个状态机？

如果“应用程序”是指“你的扭曲代码”，那么答案是肯定的！

典型的 Twisted 应用程序是一组状态机，所有这些都具有一些非常明确的接口。我开始了我的 Twisted 冒险，打算用两个状态机（一个串行和键盘）编写一个应用程序，但是当我对 twisted 感到满意时，我意识到添加额外的接口和状态机是微不足道的（通过所有的奇迹的 tx 库）。在一个下午，我添加了一个粗略的 Web 界面，一个 websocket 界面，然后在两者上放置了 SSL，甚至还添加了一个 SSH 调试界面。一旦你得到一个滚动，添加接口和状态机就变得微不足道了。

在许多（所有？）情况下，扭曲模型是状态机将驻留在实例化类中，该类与连接绑定并已注册到（唯一的）主事件中-循环。

使用产生新状态机的连接类型（想想 http 连接），您注册一个工厂类/状态机以及侦听连接，它们共同使应用程序能够为每个新状态机产生新的类/状态机联系。 Twisted 应用程序在大规模运行时通常有 10 甚至 100 万个并发状态实例。

如果您尝试将不同的协议和状态粘合在一起，Twisted 会非常棒（......所有这些都在您选择的事件循环中（select/epoll/kqueue/etc））

以下是可运行的示例代码，可以说明其中的许多要点。阅读def main() 之前的 cmets 了解更多代码背景：

#!/usr/bin/python
#
# Frankenstein-esk amalgam of example code
#   Key of which comes from the Twisted "Chat" example
#   (such as: http://twistedmatrix.com/documents/12.0.0/core/examples/chatserver.py)

import sys # so I can get at stdin
import os # for isatty
import termios, tty # access to posix IO settings
from random import random
from twisted.internet import reactor
from twisted.internet import stdio # the stdio equiv of listenXXX
from twisted.protocols import basic # for lineReceiver for keyboard
from twisted.internet.protocol import Protocol, ServerFactory

class MyClientConnections(basic.LineReceiver):

    def __init__(self):
        self.storedState = "Idle"
        self.connectionpos = None

    def connectionMade(self):
        self.factory.clients.append(self) # <--- magic here :
            # protocol automagically has a link to its factory class, and
            # in this case that is being used to push each new connection
            # (which is in the form of this class) into a list that the
            # factory can then access to get at each of the connections
        self.connectionpos = str(self.factory.clients.index(self)) # figure out 
                                      # where I am in the connection array
        print "Got new client! (index:", self.connectionpos + ")"
        self.transport.write("---\nYour connection: " + self.connectionpos + "\n---\n")

    def connectionLost(self, reason):
        print "Lost a client!"
        self.factory.clients.remove(self)

    # used to pretend that something was typed on a telnet connection
    def fakeInput(self, message):
        self.transport.write("FAKING Input: '" + message + "'\n")
        self.lineReceived(message)

    #this is only in a def on its own so I can lump my demo callLater
    def stateUpdate(self, newState, delay):
        self.storedState = newState
        # the following is a hack to fake data coming in this interface
        reactor.callLater(delay, self.fakeInput, newState + " DONE")

    def processInput(self, newState):
        # all the logic in here is junk to make a demo, real code may or may-not look like
        # this.  This junk logic is an example statemachine though
        if self.storedState == "Idle":
            if newState == "start":
                self.stateUpdate("State A", 1)        
                # send a message to this connection
                self.transport.write("starting state machine\n")
                # send a message to the term in which the script it running
                print "Connection [" + self.connectionpos + "] starting state machine"
        elif self.storedState == "State A":
            if newState == "State A DONE":
                self.transport.write("Beginning state B\n")
                self.stateUpdate("State B", 2)
        elif self.storedState == "State B":
            if newState == "State B DONE":
                self.transport.write("Beginning state C\n")
                self.stateUpdate("State C", 2)
        elif self.storedState == "State C":
            if newState == "State C DONE":
                self.storedState = "Idle"
                # send a message to this connection
                self.transport.write("Returning to Idle state\n")
                # send a message to the term in which the script it running
                print "Connection [" + self.connectionpos + "] return to Idle state"

    def lineReceived(self, line):
        # print "received '" + line +"' from connection", self.factory.clients.index(self)
        self.processInput(line)

class MyServerFactory(ServerFactory):
    protocol = MyClientConnections

    def __init__(self):
        self.clients = [] # this gets filled from the class above

    def sendToAll(self, message):
      for c in self.clients:  # Read MyClientConnections class for background
        c.transport.write(message)

    def randStart(self, width):
      for c in self.clients:
        startDelay = random() * width
        print "Starting client " + str(c.connectionpos) + " in " +str(startDelay) + " secs" 
        reactor.callLater(startDelay, c.processInput, "start")

# to set keyboard into cbreak mode -- just because I like it that way...
class Cbreaktty(object):
    org_termio = None
    my_termio = None

    def __init__(self, ttyfd):
        if(os.isatty(ttyfd)):
            self.org_termio = (ttyfd, termios.tcgetattr(ttyfd))
            tty.setcbreak(ttyfd)
            print '  Set cbreak mode'
            self.my_termio = (ttyfd, termios.tcgetattr(ttyfd))
        else:
          raise IOError #Not something I can set cbreak on!

    def retToOrgState(self):
        (tty, org) = self.org_termio
        print '  Restoring terminal settings'
        termios.tcsetattr(tty, termios.TCSANOW, org)


class KeyEater(basic.LineReceiver):

    def __init__(self, factoryObj):
        self.setRawMode() # Switch from line mode to "however much I got" mode
        # the following is one of the key connecting ideas in twisted, the object
        # that contains another state machine (really all of the tcp statemachines)
        # has been passed into this class via its init.
        self.factoryObj = factoryObj

    def rawDataReceived(self, data):
        key = str(data).lower()[0]
        if key == 's':
            # The following line is going to call (from within the factory object)
            # the random start def
            self.factoryObj.randStart(5)
        elif key == 'd':
            print "State Dump of connections"
            print "-------------------------"
            for c in self.factoryObj.clients:
                print "#" + str(c.connectionpos) + "      " + c.storedState
        elif key == 'q':
            reactor.stop()
        else:
            print "--------------"
            print "  If you haven't already, connect to this script via a"
            print "  'telnet localhost 5000' at least one (multiple connections"
            print "  are better)"
            print "Press:"
            print "      s  - randomly start all clients"
            print "      d  - dump the state of all connected clients"
            print "      q  - to cleanly shutdown"
            print " Note: you can type commands in the connections, things"
            print "       most useful of which is 'start'"
            print "---------------"

# Custom tailored example for SO:30397425
# 
# This code is a mishmash of styles and techniques. Both to provide different examples of how
# something can be done and because I'm lazy.  Its been built and tested on OSX and linux,
# it should be portable (other then perhaps termal cbreak mode).  If you want to ask
# questions about this code contact me directly via mail to mike at partialmesh.com
#
# While it isn't directly using serial ports, the tcp connections that its using are a good
# parallel.
#
# It should be used by running the script and then opening up many windows telnet'ing into
# localhost 5000.
#
# Once running press any key in the window where the script was run and it will give
# instructions.  
# The normal use case would be to type "s" to queue statemachine
# start-ups, then repeatedly press 'd' to dump the status of all the state machines
#
# 'start' can be typed into any of the telnet connections to start them by hand too.


def main():
    client_connection_factory = MyServerFactory()

    try:
      termstate = Cbreaktty(sys.stdin.fileno())
    except IOError:
      sys.stderr.write("Error: " + sys.argv[0] + " only for use on interactive ttys\n")
      sys.exit(1)

    keyboardobj = KeyEater(client_connection_factory)

    stdio.StandardIO(keyboardobj,sys.stdin.fileno())
    reactor.listenTCP(5000, client_connection_factory)
    reactor.run()
    termstate.retToOrgState()
if __name__ == '__main__':
  main()

---

调制解调器的 USB 串行连接可能由于调制解调器被拔出而被破坏，并在重新插入时重新建立。如何检测此类事件并将相应设备端口的监控添加到 reactor ？目前，我在主应用程序中进行静态处理。

经过研究，我没有一个简单的答案。我仍然怀疑以下逻辑将接近解决方案，但我今天没有找到实现此功能的代码。

我的猜测是有一种合理的方法可以确定是否发生了 USB 事件，并确定是否添加了串行设备。但我怀疑是否有一种好方法可以确定它是否是您的串行设备之一 - 如果它是您的命令或诊断接口则更不用说（除非您的构建硬件并且可以控制设备的 USB ID）

事件会因串行端口错误而触发（至少从我在 linux 上的经验来看），但我不确定 USB 拔出将如何/在何处注册。

---

其他可能对您有用的链接

• Utilities for talking to a GSM modem over USB via AT commands 的扭曲实现：https://github.com/smn/txgsm
• 通过 USB 扭曲实现气象站： https://gist.github.com/claws/2464017