问题描述
浏览扭曲手指教程并查看 SO 问题:
- 问题1
- 问题2
但是，我(还)不能编写一个可以读取 & 的扭曲程序.从多个串行端口写入，尤其是在协议涉及读取单行或多行并相应地写回设备的情况下.
我要做的是为 2 个调制解调器打开 2 对(即总共 4 个)串行端口.与调制解调器的通信使用 Hayes AT 命令集.虽然与调制解调器的大多数命令/响应交换是通过命令端口进行的，但对于每个调制解调器，只有很少的诊断信息只能通过诊断端口获得.诊断信息应导致状态机(设备状态、连接状态)被修改.
这是我理解为潜在方法的粗略骨架程序(基于单端口示例):
类 CommandProtocol(LineOnlyReceiver):def connectionMade(self):log.msg("连接到命令端口")def lineReceived(self, line):打印代表(行)进程命令行(行)类诊断协议(LineOnlyReceiver):def connectionMade(self):log.msg("连接到诊断端口")def lineReceived(self, line):打印代表(行)过程诊断线(线)...#modem1 端口cmdPort[0] = SerialPort(CommandProtocol, "/dev/ttyUSB0", reactor, 115200)diagPort[0] = SerialPort(诊断协议，/dev/ttyUSB1"，反应器，115200)#modem2 端口cmdPort[1] = SerialPort(CommandProtocol, "/dev/ttyUSB3", reactor, 115200)diagPort[1] = SerialPort(DiagnosticProtocol, "/dev/ttyUSB4", reactor, 115200)但是，我不知道如何执行以下操作:
- 如何/在哪里接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令?
- 关联在命令端口上接收到的信息，用于 ttyUSB0 和ttyUSB1 用于 modem1，同样用于 modem2 的另一对?请注意，每个调制解调器都有自己的状态机(设备状态和连接状态)
- twisted 是否提供任何机制来按应用管理多个状态机?
- 调制解调器的 USB 串行连接可能由于调制解调器被拔出而被破坏，并在重新插入时重新建立.如何检测此类事件并将相应设备端口的监控添加到 reactor ?目前，我在主应用程序中进行静态处理.
注意示例代码
我没有看到您在将类注册到反应器之前实例化它们.我预计那会失败得很惨.这是我的一个类似的运行代码片段:
# 处理来自串口的消息的东西类 SerialEater(basic.LineReceiver):statusCallback = 无def __init__(self):self.keyinprocess = 无def lineReceived(自我，数据):self.dealWithSerial(数据)def connectionLost(自我，原因):如果(反应器.运行):print "序列丢失但反应堆仍在运行！原因:" + str(reason) + " at time " + time.asctime()[...等等...]# 注册串口到twistedserialhandler = SerialEater() # <------------- 实例化串行端口(串行处理程序，'/dev/ttyUSB0'，反应器，波特率 = 115200)<小时><块引用>
我如何/在哪里接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令?
就像您如何将串行处理程序注册到 Twisted 中一样，您可以为标准 io 注册处理程序，例如:
# 从标准输入中提取 cbreak char 输入的东西类 KeyEater(basic.LineReceiver):def __init__(self):self.setRawMode() # 从线路模式切换到不管我得到多少"模式def connectionLost(自我，原因):如果(反应器.运行):self.sendLine("键盘丢失但反应堆仍在运行！原因:" + str(reason) + " at time " + time.asctime())def rawDataReceived(自我，数据):键 = str(data).lower()[0]尝试:如果键 == '?':键=帮助"[...等等...]# 将 stdio 处理程序注册到 twisted键盘obj = KeyEater()键盘obj.serialobj = serialhandlerstdio.StandardIO(keyboardobj,sys.stdin.fileno())<小时><块引用>
关联在命令端口上接收到的信息，用于 ttyUSB0 和ttyUSB1 用于 modem1，同样用于 modem2 的另一对?请注意，每个调制解调器都有自己的状态机(设备状态和连接状态)
在正常使用中，每个连接实例都有自己的状态机(包装在与连接一起注册到反应器中的类的实例中).
作为程序员，您可以选择如何连接类的状态，但通常是通过推送对伙伴类的引用.
下面，此答案包含可运行的代码，将说明数据如何在状态机/接口之间连接.此 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() 之前的注释以了解代码的更多背景:
#!/usr/bin/python## Frankenstein-esk 示例代码的混合物# 其中的关键来自 Twisted "Chat" 示例#(如:http://twistedmatrix.com/documents/12.0.0/core/examples/chatserver.py)import sys # 这样我就可以进入标准输入import os # for isattyimport termios, tty # 访问 posix IO 设置从随机导入随机来自twisted.internet 进口反应堆from twisted.internet import stdio # 相当于listenXXX的stdiofrom twisted.protocols import basic # for lineReceiver for keyboard从 twisted.internet.protocol 导入协议，ServerFactoryMyClientConnections 类(basic.LineReceiver):def __init__(self):self.storedState = "空闲"self.connectionpos = 无def connectionMade(self):self.factory.clients.append(self) # <--- 神奇之处:# 协议自动有一个到它的工厂类的链接，并且# 在这种情况下，它被用来推送每个新连接#(就是这个类的形式)放到一个列表中# factory 然后可以访问以获取每个连接self.connectionpos = str(self.factory.clients.index(self)) # 搞清楚# 我在连接数组中的位置print "有新客户了！(index:", self.connectionpos + ")"self.transport.write("---
你的连接:" + self.connectionpos + "
---
")def connectionLost(自我，原因):打印失去了一个客户！"self.factory.clients.remove(self)# 用来假装在 telnet 连接上输入了一些东西def fakeInput(自我，消息):self.transport.write("FAKING 输入:'" + message + "'
")self.lineReceived(消息)#this 仅在一个 def 中，所以我可以集中我的演示调用Laterdef stateUpdate(self, newState, delay):self.storedState = newState# 以下是对该界面中的虚假数据的破解reactor.callLater(延迟，self.fakeInput，newState +完成")定义进程输入(自我，新状态):# 这里所有的逻辑都是垃圾来做一个演示，真实的代码可能看起来像也可能不像# 这.这个垃圾逻辑虽然是一个示例状态机如果 self.storedState == "空闲":如果 newState == 开始":self.stateUpdate("状态 A", 1)# 向该连接发送消息self.transport.write("启动状态机
")# 向运行脚本的术语发送消息print "Connection [" + self.connectionpos + "] 启动状态机"elif self.storedState == "状态 A":如果 newState == "状态 A DONE":self.transport.write("开始状态 B
")self.stateUpdate("状态 B", 2)elif self.storedState == "状态 B":如果 newState == "状态 B 完成":self.transport.write("开始状态 C
")self.stateUpdate("状态 C", 2)elif self.storedState == "状态 C":如果 newState == "状态 C 完成":self.storedState = "空闲"# 向该连接发送消息self.transport.write("返回空闲状态
")# 向运行脚本的术语发送消息print "Connection[" + self.connectionpos + "] 返回空闲状态"def lineReceived(self, line):# print "received '" + line +"' from connection", self.factory.clients.index(self)self.processInput(行)MyServerFactory 类(ServerFactory):协议 = MyClientConnectionsdef __init__(self):self.clients = [] # 这是从上面的类中填充的def sendToAll(自我，消息):for c in self.clients: # 读取 MyClientConnections 类作为背景c.transport.write(消息)def randStart(自我，宽度):对于 self.clients 中的 c:开始延迟 = 随机() * 宽度打印启动客户端"+str(c.connectionpos)+in"+str(startDelay)+secs"reactor.callLater(startDelay，c.processInput，开始")# 将键盘设置为 cbreak 模式——只是因为我喜欢这样...Cbreaktty 类(对象):org_termio = 无my_termio = 无def __init__(self, ttyfd):如果(os.isatty(ttyfd)):self.org_termio = (ttyfd, termios.tcgetattr(ttyfd))tty.setcbreak(ttyfd)print '设置 cbreak 模式'self.my_termio = (ttyfd, termios.tcgetattr(ttyfd))别的:raise IOError #不是我可以设置 cbreak 的东西！def retToOrgState(self):(tty, org) = self.org_termioprint '恢复终端设置'termios.tcsetattr(tty, termios.TCSANOW, org)类 KeyEater(basic.LineReceiver):def __init__(self, factoryObj):self.setRawMode() # 从线路模式切换到不管我得到多少"模式# 以下是twisted中的关键连接思想之一，对象# 包含另一个状态机(实际上是所有的 tcp 状态机)# 已经通过它的 init 传递给这个类.self.factoryObj = factoryObjdef rawDataReceived(自我，数据):键 = str(data).lower()[0]如果键 == 's':# 以下行将调用(从工厂对象内)# 随机开始defself.factoryObj.randStart(5)elif 键 == 'd':打印连接状态转储"打印 " - - - - - - - - - - - - -"对于 self.factoryObj.clients 中的 c:打印 "#" + str(c.connectionpos) + " " + c.storedStateelif 键 == 'q':反应堆.stop()别的:打印 " - - - - - - - "print " 如果你还没有，通过 a 连接到这个脚本"print " 'telnet localhost 5000' 至少一个(多个连接)打印更好)"打印按:"print " s - 随机启动所有客户端"print " d - 转储所有已连接客户端的状态"打印"q - 干净地关闭"print " 注意:你可以在连接、事物中输入命令"print " 其中最有用的是 'start'"打印 " - - - - - - - -"# 为 SO:30397425 定制的示例## 这段代码是风格和技术的混搭.两者都提供了不同的例子来说明如何# 可以做点什么，因为我很懒.它是在 OSX 和 linux 上构建和测试的，# 它应该是可移植的(也许是termal cbreak 模式).如果你想问# 有关此代码的问题，请通过邮件直接联系我到 partialmesh.com 上的 mike## 虽然它不直接使用串口，但它使用的tcp连接很好# 平行线.## 应该通过运行脚本然后打开许多窗口 telnet'ing 来使用# 本地主机 5000.## 一旦运行，在运行脚本的窗口中按任意键，它会给出# 指示.# 正常的用例是输入s"来排队状态机# 启动，然后反复按 'd' 转储所有状态机的状态## 'start' 也可以输入到任何 telnet 连接中以手动启动它们.定义主():client_connection_factory = MyServerFactory()尝试:termstate = Cbreaktty(sys.stdin.fileno())除了 IOError:sys.stderr.write("错误:" + sys.argv[0] + " 仅用于交互式 ttys
")系统退出(1)键盘obj = KeyEater(client_connection_factory)stdio.StandardIO(keyboardobj,sys.stdin.fileno())reactor.listenTCP(5000, client_connection_factory)反应堆.run()termstate.retToOrgState()如果 __name__ == '__main__':主要的()<小时><块引用>
调制解调器的 USB 串行连接可能由于调制解调器被拔出而被破坏，并在重新插入时重新建立.如何检测此类事件并将相应设备端口的监控添加到 reactor ?目前，我在主应用程序中进行静态处理.
经过研究，我没有一个简单的答案.我仍然怀疑以下逻辑将接近解决方案，但我今天没有找到实现此功能的代码.
我的猜测是有一种合理的方法可以确定是否发生了 USB 事件，并确定是否添加了串行设备.但我怀疑是否有一种好方法可以确定它是否是您的串行设备之一——更不用说它是您的命令或诊断接口(除非您的构建硬件并且可以控制设备的 USB ID)
事件会因串行端口错误而触发(至少从我在 linux 上的经验来看)，但我不确定 USB 拔出将如何/在何处注册.
<小时>其他可能对您有用的链接
与 GSM 调制解调器通话的实用程序的扭曲实现USB 通过 AT 命令:https://github.com/smn/txgsm- 通过 USB 扭曲实现气象站:https://gist.github.com/claws/2464017
Going through the twisted finger tutorial and seen the SO questions:
- Question-1
- Question-2
However, I can't (yet) write a twisted program that can read & write from multiple serial ports, especially where the protocol involves reading single or multiple lines, and writing back to the device accordingly.
What I am trying to do is open 2 pairs (i.e. total of 4) serial ports, for 2 modems. Communication with modems is using Hayes AT command set. While most of the command/response exchanges with modem is via the command-port, there are few diagnostic information that are available only via the diagnostic-port, for each modem. The diagnostic information should lead to a state-machine (device-state, connection-state) to be modified.
Here is a rough skeletal program of what I understand as the potential approach (based on single port examples):
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)
However, I am at loss, as to how do I do the following:
- How/where do I accept CLI input from user, that then triggers sending a set of AT command to the modems ?
- Correlate the information received on command port for ttyUSB0 & ttyUSB1 for modem1, and similarly for the other pair for modem2 ? Note that each modem has it's own state-machine (device-state and connection-state)
- Does twisted provide any mechanism for management of multiple state-machines by application ?
- It is possible that the USB-serial connection to the modem is destroyed due to the modem being unplugged, and re-established on being plugged back-in. How can I detect such events and add the monitoring of the corresponding device-ports to the reactor ? Currently, I'm doing it statically in the main application.
Noteonyourexamplecode
I don't see you instantiating your classes before registering them to the reactor. I expect that will fail badly. Here is a similar snippet of running code of mine:
# 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)
How/where do I accept CLI input from user, that then triggers sending a set of AT command to the modems ?
Much like how you can register Serial handlers into Twisted, you can register handlers for standard io, for instance:
# 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())
Correlate the information received on command port for ttyUSB0 & ttyUSB1 for modem1, and similarly for the other pair for modem2 ? Note that each modem has it's own state-machine (device-state and connection-state)
In normal use, each connection-instance is going to have its own state machine (wrapped up in the instance of the class that you register into the reactor along with the connection).
You as the programmer choose how you want to connect the states of the classes, but often its via pushing reference to the partner classes.
Below, this answer contains runnable code that will illustrate how data is connected between state-machines/interface. This is also illustrated in this SO: Persistent connection in twisted
Does twisted provide any mechanism for management of multiple state-machines by application ?
If by "application" you mean "your twisted code" then then the answer is absolutely YES!
The typical Twisted app is an array of state-machines, all with some amazingly well defined interfaces. I started my Twisted adventure intending to write an app with two state-machines (a serial and keyboard), but when I became comfortable with twisted was doing I realized it was trivial to add on extra interfaces and state-machines (through all the wonder of the tx libraries). All in one afternoon I added on a rough web interface, a websocket interface, then laid SSL over both and even added on an SSH debug interface. Once you get a rolling, adding interfaces and state-machines become trivial.
In many (all?) cases, the twisted model is that a state-machine will reside in an instantiated class that is tied to a connection and that has been registered into the (one-and-only-one) main event-loop.
With connection types that spawn off new state-machines (think http connections) you register one factory-class/state-machine along with the listening connection which together enable the app of spawning off new classes/state-machines for each new connection. Twisted applications routinely 10s or even 100s of thousands of concurrent instances of state when run at scale.
Twisted is amazing if your trying to glue together different protocols and states (... with all of it being in a event loop of your choice (select/epoll/kqueue/etc))
The following is runnable sample code that should illustrate many of these points. Read the comments before def main() for more background on the code:
#!/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("---
Your connection: " + self.connectionpos + "
---
")
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 + "'
")
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
")
# 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
")
self.stateUpdate("State B", 2)
elif self.storedState == "State B":
if newState == "State B DONE":
self.transport.write("Beginning state C
")
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
")
# 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
")
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()
It is possible that the USB-serial connection to the modem is destroyed due to the modem being unplugged, and re-established on being plugged back-in. How can I detect such events and add the monitoring of the corresponding device-ports to the reactor ? Currently, I'm doing it statically in the main application.
After research I don't have an easy answer. I still suspect that the following logic will be close to a solution but I didn't have any luck finding code that implements this today.
My guess is there will be a reasonable way to figure out if a USB event has occurred, and work out if a serial device has been added. But I doubt there will a good way to figure out if it is one of your serial devices - much less if its your Command or Diagnostic interface (Unless your building hardware and can control the USB IDs of the devices)
Events are fired on errors with serial ports (at least from my experience on linux), but I'm unsure how/where a USB unplug would register.
Otherlinksthatmightbeofusetoyou
- Twisted implementation of
Utilities for talking to a GSM modem over USB via AT commands: https://github.com/smn/txgsm - Twisted implementation of weather station through USB: https://gist.github.com/claws/2464017
这篇关于如何使用twisted处理多个R/W串口?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！