如何使用 twisted 处理 R/W 的多个串口？
How to deal with multiple serial ports for R/W using twisted?
完成扭曲手指教程并查看 SO 问题：
- Question-1
- Question-2
但是，我（还）不能编写一个可以从多个串行端口读取和写入的扭曲程序，尤其是在协议涉及读取单行或多行并相应地写回设备的情况下。
我想做的是为 2 个调制解调器打开 2 对（即总共 4 个）串口。与调制解调器的通信使用 Hayes AT 命令集。虽然大多数 command/response 与调制解调器的交换是通过命令端口进行的，但对于每个调制解调器，只有少数诊断信息只能通过诊断端口获得。诊断信息应该导致状态机（设备状态，连接状态）被修改。
这是我理解的潜在方法的粗略框架程序（基于单端口示例）：
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)
但是，我很茫然，我该怎么做：
- How/where 我是否接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令？
- 关联调制解调器 1 的 ttyUSB0 和 ttyUSB1 命令端口上收到的信息，调制解调器 2 的另一对接收到的信息也类似？请注意，每个调制解调器都有自己的状态机（设备状态和连接状态）
- twisted 是否提供任何由应用程序管理多个状态机的机制？
- 与调制解调器的 USB 串行连接可能由于拔下调制解调器而损坏，并在重新插入时重新建立。如何检测此类事件并将相应设备端口的监控添加到反应器中？目前，我正在主应用程序中静态地执行此操作。
注意你的示例代码
我没看到您在将 classes 注册到 reactor 之前对其进行实例化。我希望那会严重失败。这是我的 运行ning 代码的类似片段：
# 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 ?
就像将串行处理程序注册到 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())
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)
在正常使用中，每个连接实例都将拥有自己的状态机（包装在您与连接一起注册到反应器中的 class 实例中）。
作为程序员，您可以选择如何连接 classes 的状态，但通常是通过推送对伙伴 classes 的引用。
下面，此答案包含 运行 可用代码，将说明数据如何在 state-machines/interface 之间连接。此 SO 中也说明了这一点：Persistent connection in twisted
Does twisted provide any mechanism for management of multiple state-machines by application ?
如果 "application" 你的意思是 "your twisted code" 那么答案是肯定的！
典型的 Twisted 应用程序是一组状态机，所有这些都具有一些定义得非常好的接口。我开始了我的 Twisted 冒险，打算用两个状态机（串行和键盘）编写一个应用程序，但是当我对 Twisted 感到满意时，我意识到添加额外的接口和状态机是微不足道的（通过所有的奇迹tx 库）。在一个下午的时间里，我添加了一个粗糙的 Web 界面，一个 WebSocket 界面，然后在两者之上放置了 SSL，甚至添加了一个 SSH 调试界面。一旦开始滚动，添加接口和状态机就变得微不足道了。
在许多（所有？）情况下，扭曲的模型是状态机将驻留在实例化的 class 中，该实例化 class 绑定到连接并且已注册到（一和 -只有一个）主事件循环。
使用产生新状态机的连接类型（想想 http 连接），你注册一个 factory-class/state-machine 连同监听连接，它们一起使应用程序产生新的 classes/state-machines 每个新的联系。当 运行 大规模时，Twisted 应用程序通常有 10s 甚至 100s 的数千个并发状态实例。
如果你试图将不同的协议和状态粘合在一起（......所有这些都在你选择的事件循环中（select/epoll/kqueue/etc）），Twisted 是惊人的
以下是 运行可用的示例代码，应该可以说明其中的许多要点。阅读 def main() 之前的评论以了解有关代码的更多背景信息：
#!/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()
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.
经过研究，我没有一个简单的答案。我仍然怀疑以下逻辑将接近解决方案，但我今天没有运气找到实现它的代码。
我的猜测是会有一种合理的方法来确定是否发生了 USB 事件，并确定是否已添加串行设备。但我怀疑是否有一种好方法可以确定它是否是您的串行设备之一 - 如果它是您的命令或诊断接口则更不用说（除非您的构建硬件可以控制设备的 USB ID）
串口错误时会触发事件（至少根据我在 linux 上的经验），但我不确定 how/where 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
完成扭曲手指教程并查看 SO 问题：
- Question-1
- Question-2
但是，我（还）不能编写一个可以从多个串行端口读取和写入的扭曲程序，尤其是在协议涉及读取单行或多行并相应地写回设备的情况下。
我想做的是为 2 个调制解调器打开 2 对（即总共 4 个）串口。与调制解调器的通信使用 Hayes AT 命令集。虽然大多数 command/response 与调制解调器的交换是通过命令端口进行的，但对于每个调制解调器，只有少数诊断信息只能通过诊断端口获得。诊断信息应该导致状态机（设备状态，连接状态）被修改。
这是我理解的潜在方法的粗略框架程序（基于单端口示例）：
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)
但是，我很茫然，我该怎么做：
- How/where 我是否接受来自用户的 CLI 输入，然后触发向调制解调器发送一组 AT 命令？
- 关联调制解调器 1 的 ttyUSB0 和 ttyUSB1 命令端口上收到的信息，调制解调器 2 的另一对接收到的信息也类似？请注意，每个调制解调器都有自己的状态机（设备状态和连接状态）
- twisted 是否提供任何由应用程序管理多个状态机的机制？
- 与调制解调器的 USB 串行连接可能由于拔下调制解调器而损坏，并在重新插入时重新建立。如何检测此类事件并将相应设备端口的监控添加到反应器中？目前，我正在主应用程序中静态地执行此操作。
注意你的示例代码
我没看到您在将 classes 注册到 reactor 之前对其进行实例化。我希望那会严重失败。这是我的 运行ning 代码的类似片段：
# 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 ?
就像将串行处理程序注册到 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())
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)
在正常使用中，每个连接实例都将拥有自己的状态机（包装在您与连接一起注册到反应器中的 class 实例中）。
作为程序员，您可以选择如何连接 classes 的状态，但通常是通过推送对伙伴 classes 的引用。
下面，此答案包含 运行 可用代码，将说明数据如何在 state-machines/interface 之间连接。此 SO 中也说明了这一点：Persistent connection in twisted
Does twisted provide any mechanism for management of multiple state-machines by application ?
如果 "application" 你的意思是 "your twisted code" 那么答案是肯定的！
典型的 Twisted 应用程序是一组状态机，所有这些都具有一些定义得非常好的接口。我开始了我的 Twisted 冒险，打算用两个状态机（串行和键盘）编写一个应用程序，但是当我对 Twisted 感到满意时，我意识到添加额外的接口和状态机是微不足道的（通过所有的奇迹tx 库）。在一个下午的时间里，我添加了一个粗糙的 Web 界面，一个 WebSocket 界面，然后在两者之上放置了 SSL，甚至添加了一个 SSH 调试界面。一旦开始滚动，添加接口和状态机就变得微不足道了。
在许多（所有？）情况下，扭曲的模型是状态机将驻留在实例化的 class 中，该实例化 class 绑定到连接并且已注册到（一和 -只有一个）主事件循环。
使用产生新状态机的连接类型（想想 http 连接），你注册一个 factory-class/state-machine 连同监听连接，它们一起使应用程序产生新的 classes/state-machines 每个新的联系。当 运行 大规模时，Twisted 应用程序通常有 10s 甚至 100s 的数千个并发状态实例。
如果你试图将不同的协议和状态粘合在一起（......所有这些都在你选择的事件循环中（select/epoll/kqueue/etc）），Twisted 是惊人的
以下是 运行可用的示例代码，应该可以说明其中的许多要点。阅读 def main() 之前的评论以了解有关代码的更多背景信息：
#!/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()
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.
经过研究，我没有一个简单的答案。我仍然怀疑以下逻辑将接近解决方案，但我今天没有运气找到实现它的代码。
我的猜测是会有一种合理的方法来确定是否发生了 USB 事件，并确定是否已添加串行设备。但我怀疑是否有一种好方法可以确定它是否是您的串行设备之一 - 如果它是您的命令或诊断接口则更不用说（除非您的构建硬件可以控制设备的 USB ID）
串口错误时会触发事件（至少根据我在 linux 上的经验），但我不确定 how/where 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