diff --git a/.vscode/launch.json b/.vscode/launch.json index 54982527..351895cd 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -8,7 +8,15 @@ "name": "DLS PMAC Control", "type": "debugpy", "request": "launch", - "module": "dls_pmac_control" + "module": "dls_pmac_control", + "args": [ + "-o", + "tcpip", + "-s", + "172.23.171.103", + "-p", + "1025" + ], }, { "name": "Debug Unit Test", diff --git a/src/dls_pmac_control/__main__.py b/src/dls_pmac_control/__main__.py index 1f0e91db..c0c94668 100755 --- a/src/dls_pmac_control/__main__.py +++ b/src/dls_pmac_control/__main__.py @@ -36,6 +36,7 @@ from dls_pmac_control.login import Loginform from dls_pmac_control.ppmacgather import PpmacGatherform from dls_pmac_control.status import PpmacStatusform, Statusform +from dls_pmac_control.status_dataclasses import ControllerStatus from dls_pmac_control.ui_form_control import UiControlForm from dls_pmac_control.watches import Watchesform @@ -137,6 +138,8 @@ def __init__(self, options, parent=None): ) self.comms_thread.started.connect(self.comms_worker.start) + self.comms_worker.update_received.connect(self.start_updating_motors) + self.comms_worker.watches_ready.connect(self.update_watches) self.comms_worker.finished.connect(self.comms_thread.quit) self.comms_thread.finished.connect(self.comms_worker.deleteLater) @@ -669,195 +672,151 @@ def add_to_txt_shell(self, command, ret_str=None, chk_show_all=True): ret_str.rstrip("\x06").lstrip("\x07").replace("\r", " ") ) - # Called when an event comes out of the polling thread - # and the jog ribbon. - def update_motors(self): + def start_updating_motors(self, status: ControllerStatus): + print("start_updating_motors") + print(f"The data that's been passed is {status} \n") + print(f"The cs is {status.coordinate_systems} \n") + print(f"The motors are {status.motors} \n") + under_voltage = False over_voltage = False over_temperature = False - self.comms_worker.resultQueue.qsize() - for _que_item in range(0, self.comms_worker.resultQueue.qsize()): - try: - value = self.comms_worker.resultQueue.get(False) - except Empty: - return + try: + # if isinstance(self.pmac, PPmacSshInterface): + self.update_identity(status.identifier_i65) + self.PpmacGlobalStatusScreen.update_status( + int(status.coordinate_systems[0].global_status.strip("$"), 16) + ) + self.PpmacCSStatusScreen.update_status( + int(status.coordinate_systems[0].cs_status.strip("$"), 16) + ) + self.PpmacCSStatusScreen.update_feed( + int(round(float(status.coordinate_systems[0].feedrate))) + ) - try: - motor_row = value[6] - # check for special cases - if isinstance(motor_row, str): + i2t_fault = False + over_current = False + + for motor in status.motors: + print(f"motor: {motor}\n") + + if isinstance(self.pmac, PPmacSshInterface): + velocity = motor.velocity + else: + velocity = round(float(motor.velocity) * self.servoCycleTime, 1) + + self.__item(motor.number - 1, 0).setText(str(motor.position)) + self.__item(motor.number - 1, 1).setText(str(velocity)) + self.__item(motor.number - 1, 2).setText(str(motor.following_error)) + + if motor.number - 1 < 8: if isinstance(self.pmac, PPmacSshInterface): - if motor_row == "G": - self.PpmacGlobalStatusScreen.update_status( - int(value[0].strip("$"), 16) - ) - continue - if motor_row.startswith("CS"): - self.PpmacCSStatusScreen.update_status( - int(value[0].strip("$"), 16) - ) - continue - if motor_row.startswith("FEED"): - self.PpmacCSStatusScreen.update_feed( - int(round(float(value[0]))) - ) - continue - if motor_row == "IDENT": - self.update_identity(int(value[0])) - continue - if motor_row == "UVOL": - if int(value[0]) != 0: + if int(motor.i2t_fault_status) > 0: + i2t_fault = True + # if int(value[5]) > 0: + # over_current = True + elif isinstance(self.pmac, PmacEthernetInterface): + amp_status = (int(motor.i2t_fault_status) & 448) >> 6 + if amp_status == 5: + i2t_fault = True + elif amp_status == 6: + over_current = True + if motor.number - 1 < 4: + if amp_status == 2: under_voltage = True - continue - if motor_row == "OVOL": - if int(value[0]) != 0: - over_voltage = True - continue - if motor_row == "OTEMP": - if int(value[0]) != 0: + elif amp_status == 3: over_temperature = True - continue - else: - if motor_row == "G": - self.global_status_screen.update_status(int(value[0], 16)) - continue - if motor_row.startswith("CS"): - self.cs_status_screen.update_status(int(value[0], 16)) - continue - if motor_row.startswith("FEED"): - self.cs_status_screen.update_feed( - int(round(float(value[0]))) - ) - continue - if motor_row == "IDENT": - self.update_identity(int(value[0])) - continue + elif amp_status == 4: + over_voltage = True + + status_word = int(motor.motor_status.strip("$"), 16) + + # define high and low limits for power pmac + if isinstance(self.pmac, PPmacSshInterface): + lo_lim = bool(status_word & 0x2000000000000000) # MinusLimit + hi_lim = bool(status_word & 0x1000000000000000) # PlusLimit + lo_lim_soft = bool( + status_word & 0x0080000000000000 + ) # SoftMinusLimit + hi_lim_soft = bool( + status_word & 0x0040000000000000 + ) # SoftPlusLimit + # define high and low limits for pmac else: - position = str(round(float(value[1]), 1)) - if isinstance(self.pmac, PPmacSshInterface): - velocity = str(round(float(value[2]), 1)) - else: - # On Turbo PMAC velocity is returned in counts per servo cycle - # so you have to use the servo cycle time to convert it to cts/msec - velocity = str(round(float(value[2]) * self.servoCycleTime, 1)) - folerr = str(round(float(value[3]), 1)) - - i2t_fault = False - over_current = False - - if motor_row < 8: - if isinstance(self.pmac, PPmacSshInterface): - if int(value[4]) > 0: - i2t_fault = True - if int(value[5]) > 0: - over_current = True - elif isinstance(self.pmac, PmacEthernetInterface): - amp_status = (int(value[4]) & 448) >> 6 - if amp_status == 5: - i2t_fault = True - elif amp_status == 6: - over_current = True - if motor_row < 4: - if amp_status == 2: - under_voltage = True - elif amp_status == 3: - over_temperature = True - elif amp_status == 4: - over_voltage = True - - self.__item(motor_row, 0).setText(position) - self.__item(motor_row, 1).setText(velocity) - self.__item(motor_row, 2).setText(folerr) - - status_word = int(value[0].strip("$"), 16) - - # define high and low limits for power pmac - if isinstance(self.pmac, PPmacSshInterface): - lo_lim = bool(status_word & 0x2000000000000000) # MinusLimit - hi_lim = bool(status_word & 0x1000000000000000) # PlusLimit - lo_lim_soft = bool( - status_word & 0x0080000000000000 - ) # SoftMinusLimit - hi_lim_soft = bool( - status_word & 0x0040000000000000 - ) # SoftPlusLimit - - # define high and low limits for pmac - else: - lo_lim = bool( - status_word & 0x400000000000 - ) # negative end limit set - hi_lim = bool( - status_word & 0x200000000000 - ) # positive end limit set - lo_lim_soft = False - hi_lim_soft = False - - # set limit indicators in polling table + lo_lim = bool( + status_word & 0x400000000000 + ) # negative end limit set + hi_lim = bool( + status_word & 0x200000000000 + ) # positive end limit set + lo_lim_soft = False + hi_lim_soft = False + + # set limit indicators in polling table + if hi_lim: + self.__item(motor.number - 1, 3).setIcon(QIcon(self.redLedOn)) + elif hi_lim_soft: + self.__item(motor.number - 1, 3).setIcon(QIcon(self.amberLedOn)) + else: + self.__item(motor.number - 1, 3).setIcon(QIcon(self.redLedOff)) + if lo_lim: + self.__item(motor.number - 1, 4).setIcon(QIcon(self.redLedOn)) + elif lo_lim_soft: + self.__item(motor.number - 1, 4).setIcon(QIcon(self.amberLedOn)) + else: + self.__item(motor.number - 1, 4).setIcon(QIcon(self.redLedOff)) + + # set amplifier status indicators in polling table + if i2t_fault: + self.__item(motor.number - 1, 5).setIcon(QIcon(self.redLedOn)) + else: + self.__item(motor.number - 1, 5).setIcon(QIcon(self.redLedOff)) + if over_current: + self.__item(motor.number - 1, 6).setIcon(QIcon(self.redLedOn)) + else: + self.__item(motor.number - 1, 6).setIcon(QIcon(self.redLedOff)) + + # Update also the jog ribbon + if motor.number == self.currentMotor: + self.lblPosition.setText(str(motor.position)) + self.lblVelo.setText(str(motor.velocity)) + self.lblFolErr.setText(str(motor.following_error)) if hi_lim: - self.__item(motor_row, 3).setIcon(QIcon(self.redLedOn)) + self.pixHiLim.setPixmap(self.redLedOn) elif hi_lim_soft: - self.__item(motor_row, 3).setIcon(QIcon(self.amberLedOn)) + self.pixHiLim.setPixmap(self.amberLedOn) else: - self.__item(motor_row, 3).setIcon(QIcon(self.redLedOff)) + self.pixHiLim.setPixmap(self.redLedOff) if lo_lim: - self.__item(motor_row, 4).setIcon(QIcon(self.redLedOn)) + self.pixLoLim.setPixmap(self.redLedOn) elif lo_lim_soft: - self.__item(motor_row, 4).setIcon(QIcon(self.amberLedOn)) + self.pixLoLim.setPixmap(self.amberLedOn) else: - self.__item(motor_row, 4).setIcon(QIcon(self.redLedOff)) + self.pixLoLim.setPixmap(self.redLedOff) + self.status_screen.update_status(status_word) + self.ppmacstatusScreen.update_status(status_word) - # set amplifier status indicators in polling table - if i2t_fault: - self.__item(motor_row, 5).setIcon(QIcon(self.redLedOn)) - else: - self.__item(motor_row, 5).setIcon(QIcon(self.redLedOff)) - if over_current: - self.__item(motor_row, 6).setIcon(QIcon(self.redLedOn)) - else: - self.__item(motor_row, 6).setIcon(QIcon(self.redLedOff)) - - # Update also the jog ribbon - if motor_row + 1 == self.currentMotor: - self.lblPosition.setText(position) - self.lblVelo.setText(velocity) - self.lblFolErr.setText(folerr) - if hi_lim: - self.pixHiLim.setPixmap(self.redLedOn) - elif hi_lim_soft: - self.pixHiLim.setPixmap(self.amberLedOn) - else: - self.pixHiLim.setPixmap(self.redLedOff) - if lo_lim: - self.pixLoLim.setPixmap(self.redLedOn) - elif lo_lim_soft: - self.pixLoLim.setPixmap(self.amberLedOn) - else: - self.pixLoLim.setPixmap(self.redLedOff) - self.status_screen.update_status(status_word) - self.ppmacstatusScreen.update_status(status_word) - - # set controller status indicators on main window - if under_voltage: - self.pixUnderVoltage.setPixmap(self.redLedOn) - else: - self.pixUnderVoltage.setPixmap(self.redLedOff) - if over_voltage: - self.pixOverVoltage.setPixmap(self.redLedOn) - else: - self.pixOverVoltage.setPixmap(self.redLedOff) - if over_temperature: - self.pixOverTemperature.setPixmap(self.redLedOn) - else: - self.pixOverTemperature.setPixmap(self.redLedOff) + # set controller status indicators on main window + if under_voltage: + self.pixUnderVoltage.setPixmap(self.redLedOn) + else: + self.pixUnderVoltage.setPixmap(self.redLedOff) + if over_voltage: + self.pixOverVoltage.setPixmap(self.redLedOn) + else: + self.pixOverVoltage.setPixmap(self.redLedOff) + if over_temperature: + self.pixOverTemperature.setPixmap(self.redLedOn) + else: + self.pixOverTemperature.setPixmap(self.redLedOff) - except (ValueError, IndexError): - # Catch the exception and continue, since there may be other - # updates waiting in the queue. - if self.verboseMode: - print("Update request received invalid response: ", value) + except (ValueError, IndexError): + # Catch the exception and continue, since there may be other + # updates waiting in the queue. + if self.verboseMode: + print(f"Update request received invalid response: {status}") domain_names = [ "BL", @@ -938,9 +897,6 @@ def customEvent(self, E): elif E.type() == self.downloadDoneEventType: self.progressDialog.setValue(self.progressDialog.maximum()) self.txtShell.append(str(E.data())) - elif E.type() == self.updatesReadyEventType: - self.update_motors() - self.update_watches() def signal_handler(self, signum, frame): if signum == 2: # SIGINT diff --git a/src/dls_pmac_control/comms_thread.py b/src/dls_pmac_control/comms_thread.py index ffcf53f6..f6c3bf2e 100644 --- a/src/dls_pmac_control/comms_thread.py +++ b/src/dls_pmac_control/comms_thread.py @@ -10,6 +10,12 @@ ) from PyQt6.QtCore import QCoreApplication, QEvent, QObject, QTimer, pyqtSignal, pyqtSlot +from dls_pmac_control.status_dataclasses import ( + ControllerStatus, + CurrentCoordinateSystemStatus, + MotorStatus, +) + class CustomEvent(QEvent): _data = None @@ -24,6 +30,7 @@ def data(self): class CommsWorker(QObject): update_received = pyqtSignal(object) + watches_ready = pyqtSignal() finished = pyqtSignal() def __init__(self, parent): @@ -106,44 +113,51 @@ def send_complete(self, message): ev_done = CustomEvent(self.parent.downloadDoneEventType, message) QCoreApplication.postEvent(self.parent, ev_done) - # Thread that sends the PMAC command to retrieve status, position, - # velocity and following error for each motor. - - def update_func(self): - if self.parent.pmac is None or not self.parent.pmac.isConnectionOpen: - time.sleep(0.1) - return - if self.gen: - # should be downloading a text file - try: - ( - was_successful, - self.lineNumber, - command, - pmac_response_str, - ) = self.gen.__next__() - except StopIteration: - self.send_complete( - "Downloaded " + str(self.lineNumber) + " lines from pmc file." + ### NEW CODE - POLLING AS DATACLASSES + + def parsed_poll_response(self, response): + print("parsed_poll_response \n") + + response_str_list = str(response).rstrip("\x06\r").split("\r") + print("response_str_list: " + repr(response_str_list)) + + status = ControllerStatus(identifier_i65=int(response_str_list[0])) + + status.coordinate_systems.append( + CurrentCoordinateSystemStatus( + global_status=response_str_list[1], + cs_status=response_str_list[2], + feedrate=float(response_str_list[3]), + ) + ) + + response_motors_list = response_str_list[4:] + response_motors_list = [ + response_motors_list[i : i + 6] + for i in range(0, len(response_motors_list), 6) + ] + print(f"response_motors_list: {response_motors_list}") + + motor_no = 1 + for motor_response in response_motors_list: + print(f"Motor_response: {motor_response}") + status.motors.append( + MotorStatus( + number=motor_no, + motor_status=str(motor_response[0]), + position=float(motor_response[1]), + velocity=float(motor_response[2]), + following_error=float(motor_response[3]), + i2t_fault_status=float(motor_response[4]), ) - else: - err = "" - if not was_successful: - err = "{}: command '{}' generated '{}'".format( - self.lineNumber, - command, - pmac_response_str.replace("\r", " ").replace("\x07", ""), - ) - self.send_tick(self.lineNumber, err) - return - if self.disablePollingStatusValue: - time.sleep(0.1) - return + ) + motor_no += 1 - # Reduce poll rate for serial interface (ignores if poll rate set to zero) - if isinstance(self.parent.pmac, PmacSerialInterface) and self.max_pollrate: - if time.time() - self.parent.pmac.last_comm_time < 1.0 / self.max_pollrate: - return + print(f"status: {status}") + return status + + def generate_cmd(self): + print("generate_cmd \n") cmd = f"i65???&{self.CSNum}??%" # Send a different command for the Power PMAC @@ -154,6 +168,7 @@ def update_func(self): # Add the 7 segment display status query cmd = f"i65???&{self.CSNum}??%" axes = self.parent.pmac.getNumberOfAxes() + 1 + for motor_no in range(1, axes): cmd = cmd + "#" + str(motor_no) + "?PVF " # Amplifier status checks only apply to the first 8 axes @@ -175,12 +190,43 @@ def update_func(self): # Use two dummy requests to keep the request chunks the same length (p99 always returns zero) cmd = cmd + "p99 p99" - # send polling command - (ret_str, was_successful) = self.parent.pmac.sendCommand(cmd) + # print(f"cmd: {cmd}") + return cmd + + def poll_status(self) -> ControllerStatus | None: + print("poll_status \n") + if self.parent.pmac is None: + return None + + if not self.parent.pmac.isConnectionOpen: + return None + + cmd = self.generate_cmd() + (send_command_response, success) = self.parent.pmac.sendCommand(cmd) + parsed_poll_response_status = self.parsed_poll_response(send_command_response) + print(f"parsed poll response: {parsed_poll_response_status}\n") + return parsed_poll_response_status + + def update_func(self): + if self.parent.pmac is None or not self.parent.pmac.isConnectionOpen: + time.sleep(0.1) + return + + status = self.poll_status() + print(f"status: {status} \n") + self.update_received.emit(status) + + # # Reduce poll rate for serial interface (ignores if poll rate set to + # # zero) + if isinstance(self.parent.pmac, PmacSerialInterface) and self.max_pollrate: + if time.time() - self.parent.pmac.last_comm_time < 1.0 / self.max_pollrate: + return + with self.lock: # send watch window commands value_list_watch = [] for key in self._watch_window: + print(f"watch window key: {key}") (ret, success) = self.parent.pmac.sendCommand(key) ret = ret.rstrip("\x06\r") if "error" in ret or "ERR" in ret: @@ -188,52 +234,170 @@ def update_func(self): # update watches dict self._watch_window[key] = ret value_list_watch.append(ret) + print(f"value_list_watch: {value_list_watch}") self.watchesQueue.put(value_list_watch) - if was_successful: - value_list = ret_str.rstrip("\x06\r").split("\r") - # fourth is the PMAC identity - if value_list[0].startswith("\x07"): - # error, probably in buffer - print(f"i65 returned {value_list[0].__repr__()}, sending CLOSE command") - self.parent.pmac.sendCommand("CLOSE") - return + self.watches_ready.emit() - # If we got a malformed response, abort now before writing anything to the result queue. - if len(value_list) < 4: - if self.parent.verboseMode: - print("Received malformed response to poll request: ", value_list) - return + # if was_successful: - # Identifier i65 - self.resultQueue.put([value_list[0], 0, 0, 0, 0, 0, "IDENT"]) - # Global status - self.resultQueue.put([value_list[1], 0, 0, 0, 0, 0, "G"]) - # CS status - self.resultQueue.put([value_list[2], 0, 0, 0, 0, 0, f"CS{self.CSNum}"]) - # Feedrate - self.resultQueue.put([value_list[3], 0, 0, 0, 0, 0, f"FEED{self.CSNum}"]) - - if isinstance(self.parent.pmac, PPmacSshInterface): - # Brick Under Voltage Status - self.resultQueue.put([value_list[4], 0, 0, 0, 0, 0, "UVOL"]) - # Brick Over Voltage Status - self.resultQueue.put([value_list[5], 0, 0, 0, 0, 0, "OVOL"]) - # Brick Over Temperature Status - self.resultQueue.put([value_list[6], 0, 0, 0, 0, 0, "OTEMP"]) - value_list = value_list[7:] - else: - value_list = value_list[4:] - - # All request chunks contain 7 elements - cols = 6 - for motor_row, i in enumerate(range(0, len(value_list), cols)): - return_list = value_list[i : i + cols] - return_list.append(motor_row) - self.resultQueue.put(return_list, False) - - ev_updates_ready = CustomEvent(self.parent.updatesReadyEventType, None) - QCoreApplication.postEvent(self.parent, ev_updates_ready) - else: - print(f'WARNING: Could not poll PMAC for motor status ("{ret_str}")') + # else: + # print(f'WARNING: Could not poll PMAC for motor status ("{ret_str}")') time.sleep(0.1) + + ### OLD update_func BELOW FOR REFERENCE ### + + # def updateFunc(self): + # try: + # # see if the gui wants us to do anything + # cmd, data = self.inputQueue.get(block=False) + # except Empty: + # # nope, nothing to do + # pass + # else: + # # work out what it wants us to do + # if cmd == "die": + # return True + # elif cmd == "sendSeries": + # try: + # self.gen = self.parent.pmac.sendSeries(data) + # except Exception: + # self.sendComplete("Couldn't start download") + # traceback.print_exc() + # elif cmd == "disablePollingStatus": + # self.disablePollingStatus = data + # elif cmd == "cancelSendSeries": + # if self.gen: + # self.gen.close() + # self.sendComplete("Download cancelled by the user") + # else: + # print(f"WARNING: don't know what to do with cmd {cmd}") + # if self.parent.pmac is None or not self.parent.pmac.isConnectionOpen: + # time.sleep(0.1) + # return + # if self.gen: + # # should be downloading a text file + # try: + # ( + # wasSuccessful, + # self.lineNumber, + # command, + # pmacResponseStr, + # ) = self.gen.__next__() + # except StopIteration: + # self.sendComplete( + # "Downloaded " + str(self.lineNumber) + " lines from pmc file." + # ) + # else: + # err = "" + # if not wasSuccessful: + # err = "{}: command '{}' generated '{}'".format( + # self.lineNumber, + # command, + # pmacResponseStr.replace("\r", " ").replace("\x07", ""), + # ) + # self.sendTick(self.lineNumber, err) + # return + # if self.disablePollingStatus: + # time.sleep(0.1) + # return + + # # Reduce poll rate for serial interface (ignores if poll rate set to + # # zero) + # if isinstance(self.parent.pmac, PmacSerialInterface) and self.max_pollrate: + # if time.time() - self.parent.pmac.last_comm_time < 1.0 / self.max_pollrate: + # return + # cmd = f"i65???&{self.CSNum}??%" + # # Send a different command for the Power PMAC + # if isinstance(self.parent.pmac, PPmacSshInterface): + # # There has to be a space before the first BrickLV string to avoid its B being interpreted as a 'begin' command + # cmd = f"i65?&{self.CSNum}?% BrickLV.BusUnderVoltage BrickLV.BusOverVoltage BrickLV.OverTemp" + # elif isinstance(self.parent.pmac, PmacEthernetInterface): + # # Add the 7 segment display status query + # cmd = f"i65???&{self.CSNum}??%" + # axes = self.parent.pmac.getNumberOfAxes() + 1 + # for motorNo in range(1, axes): + # cmd = cmd + "#" + str(motorNo) + "?PVF " + # # Amplifier status checks only apply to the first 8 axes + # if motorNo < 9: + # if isinstance(self.parent.pmac, PPmacSshInterface): + # # PowerBrick channels are zero-indexed + # cmd = ( + # cmd + # + "BrickLV.Chan[" + # + str(motorNo - 1) + # + "].I2tFaultStatus BrickLV.Chan[" + # + str(motorNo - 1) + # + "].OverCurrent" + # ) + # else: + # # Add a dummy request to keep the request chunks + # # the same length (p99 always returns zero) + # cmd = cmd + "m" + str(motorNo) + "90 p99" + # else: + # # Use two dummy requests to keep the request chunks + # # the same length (p99 always returns zero) + # cmd = cmd + "p99 p99" + + # # send polling command + # (retStr, wasSuccessful) = self.parent.pmac.sendCommand(cmd) + # with self.lock: + # # send watch window commands + # valueListWatch = [] + # for key in self._watch_window: + # (ret, success) = self.parent.pmac.sendCommand(key) + # ret = ret.rstrip("\x06\r") + # if "error" in ret or "ERR" in ret: + # ret = "Error" + # # update watches dict + # self._watch_window[key] = ret + # valueListWatch.append(ret) + # self.watchesQueue.put(valueListWatch) + + # if wasSuccessful: + # valueList = retStr.rstrip("\x06\r").split("\r") + # # fourth is the PMAC identity + # if valueList[0].startswith("\x07"): + # # error, probably in buffer + # print(f"i65 returned {valueList[0].__repr__()}, sending CLOSE command") + # self.parent.pmac.sendCommand("CLOSE") + # return + + # # If we got a malformed response, abort now before writing anything + # # to the result queue. + # if len(valueList) < 4: + # if self.parent.verboseMode: + # print("Received malformed response to poll request: ", valueList) + # return + + # # Identifier i65 + # self.resultQueue.put([valueList[0], 0, 0, 0, 0, 0, "IDENT"]) + # # Global status + # self.resultQueue.put([valueList[1], 0, 0, 0, 0, 0, "G"]) + # # CS status + # self.resultQueue.put([valueList[2], 0, 0, 0, 0, 0, f"CS{self.CSNum}"]) + # # Fedrate + # self.resultQueue.put([valueList[3], 0, 0, 0, 0, 0, f"FEED{self.CSNum}"]) + + # if isinstance(self.parent.pmac, PPmacSshInterface): + # # Brick Under Voltage Status + # self.resultQueue.put([valueList[4], 0, 0, 0, 0, 0, "UVOL"]) + # # Brick Over Voltage Status + # self.resultQueue.put([valueList[5], 0, 0, 0, 0, 0, "OVOL"]) + # # Brick Over Temperature Status + # self.resultQueue.put([valueList[6], 0, 0, 0, 0, 0, "OTEMP"]) + # valueList = valueList[7:] + # else: + # valueList = valueList[4:] + # # All request chunks contain 7 elements + # cols = 6 + # for motorRow, i in enumerate(range(0, len(valueList), cols)): + # returnList = valueList[i : i + cols] + # returnList.append(motorRow) + # self.resultQueue.put(returnList, False) + + # evUpdatesReady = CustomEvent(self.parent.updatesReadyEventType, None) + # QCoreApplication.postEvent(self.parent, evUpdatesReady) + # else: + # print(f'WARNING: Could not poll PMAC for motor status ("{retStr}")') + # time.sleep(0.1) diff --git a/src/dls_pmac_control/status_dataclasses.py b/src/dls_pmac_control/status_dataclasses.py new file mode 100644 index 00000000..e90fa7f4 --- /dev/null +++ b/src/dls_pmac_control/status_dataclasses.py @@ -0,0 +1,31 @@ +from dataclasses import dataclass, field + + +@dataclass +class MotorStatus: + number: int + motor_status: str + position: float + velocity: float + following_error: float + i2t_fault_status: float + overcurrent: float | None = None + + +@dataclass +class CurrentCoordinateSystemStatus: + # number: int + # running: bool + # in_position: bool + global_status: str + cs_status: str + feedrate: float + + +@dataclass +class ControllerStatus: + identifier_i65: int + coordinate_systems: list[CurrentCoordinateSystemStatus] = field( + default_factory=list + ) + motors: list[MotorStatus] = field(default_factory=list)