用于与 Zaber TLSR300B 通信的 Python、pyserial 程序

Question

我是新来的，所以如果我有任何错误或者我的问题不够具体，我深表歉意，请指正!我正在开发一个程序，用于通过串行连接在激光实验室中控制两个 Zaber TLSR300B 线性运动轨道。我使用 pyserial 与他们通信没有问题，我已经能够毫无问题地写入和读取数据。我的问题更多是关于如何构建我的程序以实现所需的功能(我几乎没有接受过正式的编程培训)。

我希望程序做的是提供一些方法，允许用户向轨道发送命令，然后返回轨道响应的内容。但是，我不希望程序在检查响应时挂起，所以我不能只用写入命令后跟读取命令来编写方法。对于某些命令，轨道会立即响应(返回 ID、返回当前位置等)，但对于其他命令，轨道会在执行请求的操作(移动到位置、移动回家等)后响应。例如，如果发送 move_absolute 命令，轨道将移动到所需位置，然后发送一个带有新位置的回复。此外，可以使用物理旋钮手动移动轨道，这会导致它们在移动时不断发送当前位置(这就是我需要连续读取串行数据的原因)。

我在下面附上了代码，我在其中实现了一个线程，当有数据要读取时从串行端口读取数据并将其放入队列中。然后另一个线程从队列中取出项目并处理它们，修改存储每个轨道当前属性的 ZaberTLSR300B 对象的适当值。底部的方法是我希望用户能够调用的一些方法。他们只需将命令写入串行端口，然后由持续运行的读取线程获取和处理响应。

我遇到的主要问题是这些方法不知道轨道响应什么，所以它们无法返回回复，我想不出解决这个问题的方法。所以我不能写这样的东西:

newPosition = TrackManager.move_absolute(track 1, 5000)

或

currentPosition = TrackManager.return_current_position(track 2)

这最终是我希望用户能够执行的操作(也是因为我希望将来在此基础上实现某种 GUI)。

无论如何，我的处理方式是否正确？还是有更简洁的方法来实现这种行为？如果需要，我愿意完全重写所有内容!

谢谢，如果有任何不清楚的地方，请告诉我。

Zaber TLSR300B 手册(如果需要):http://www.zaber.com/wiki/Manuals/T-LSR

代码:

轨道管理类

import serial
import threading
import struct
import time
from collections import deque
from zaberdevices import ZaberTLSR300B

class TrackManager:

def __init__(self):

    self.serial = serial.Serial("COM5",9600,8,'N',timeout=None)

    self.track1 = ZaberTLSR300B(1, self.serial)
    self.track2 = ZaberTLSR300B(2, self.serial)

    self.trackList = [self.track1, self.track2]

    self.serialQueue = deque()

    self.runThread1 = True
    self.thread1 = threading.Thread(target=self.workerThread1)
    self.thread1.start()

    self.runThread2 = True
    self.thread2 = threading.Thread(target=self.workerThread2)
    self.thread2.start()

def workerThread1(self):
    while self.runThread1 == True:
        while self.serial.inWaiting() != 0:
            bytes = self.serial.read(6)
            self.serialQueue.append(struct.unpack('<BBl', bytes))

def workerThread2(self):
    while self.runThread2 == True:
        try:
            reply = self.serialQueue.popleft()
            for track in self.trackList:
                if track.trackNumber == reply[0]:
                    self.handleReply(track, reply)
        except:
            continue

def handleReply(self, track, reply):
    if reply[1] == 10:
        track.update_position(reply[2])
    elif reply[1] == 16:
        track.storedPositions[address] = track.position
    elif reply[1] == 20:
        track.update_position(reply[2])
    elif reply[1] == 21:
        track.update_position(reply[2])
    elif reply[1] == 60:
        track.update_position(reply[2])

def move_absolute(self, trackNumber, position):
    packet = struct.pack("<BBl", trackNumber, 20, position)
    self.serial.write(packet)

def move_relative(self, trackNumber, distance):
    packet = struct.pack("<BBl", trackNumber, 21, distance)
    self.serial.write(packet)

def return_current_position(self, trackNumber):
    packet = struct.pack("<BBl", trackNumber, 60, 0)
    self.serial.write(packet)

def return_stored_position(self, trackNumber, address):
    packet = struct.pack("<BBl", trackNumber, 17, address)
    self.serial.write(packet)

def store_current_position(self, trackNumber, address):
    packet = struct.pack("<BBl", trackNumber, 16, address)
    self.serial.write(packet)

zaberdevices.py ZaberTLSR300B类

class ZaberTLSR300B:

def __init__(self, trackNumber, serial):
    self.trackNumber = trackNumber
    self.serial = serial
    self.position = None
    self.storedPositions = []

def update_position(self, position):
    self.position = position

Answer 1

Controller 上有一个选项可以禁用所有不立即跟随命令的回复。您可以通过启用 Device Mode setting 的位 0 和 5 来执行此操作.这些位对应于“禁用自动回复”和“禁用手动移动跟踪”。位 11 默认启用，因此启用这些位的组合值为 2081。

我的建议是禁用这些额外的响应，这样您就可以依赖可预测且可靠的命令->响应模型。例如，要移动设备，您可以发送移动命令，但永远不要寻找该命令的响应。要检查移动是否已完成，您可以使用 Return Current Position命令并读取位置响应，或使用返回状态(Cmd_54)命令检查设备是否忙(即移动)或空闲。