Skip to content

Dataclasses #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
M-Chan wants to merge 45 commits into
base: new-main
Choose a base branch
from
Open

Dataclasses #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
45 commits
Select commit Hold shift + click to select a range
40cfcd2
Use QThread
M-Chan Jun 19, 2026
59cfaf2
using 'worker' instead of 'commsThread'
M-Chan Jun 22, 2026
6cc7d7e
ruff formatting
M-Chan Jun 23, 2026
e19d952
fixes disable/enable polling
M-Chan Jun 23, 2026
e4ffe1c
removed moved code
M-Chan Jun 23, 2026
01b390e
fixed sendSeries
M-Chan Jun 23, 2026
5c690a5
use 'worker'
M-Chan Jun 23, 2026
1341921
fix function naming for ruff
M-Chan Jun 23, 2026
86604f3
attempt to safely delete threads and stop timers
M-Chan Jun 25, 2026
314dde6
using 'worker'
M-Chan Jun 25, 2026
553e679
use pyqt slots and signals to manage threads and timers
M-Chan Jun 29, 2026
3ef5cdc
fix threading and remove obsolete code + comments
M-Chan Jun 30, 2026
f72946b
add dataclasses for polling statuses
M-Chan Jul 8, 2026
5fddac0
uses dataclasses for polling status
M-Chan Jul 8, 2026
20a4cea
uses dataclasses for polling status
M-Chan Jul 9, 2026
d3a8faa
uses dataclasses for polling status
M-Chan Jul 9, 2026
90a68d4
clean up imports
M-Chan Jul 9, 2026
09ca129
uses dataclasses for polling status
M-Chan Jul 9, 2026
51f95d3
uses dataclasses for polling status
M-Chan Jul 9, 2026
e7cdd51
add the CoordinateSystemStatus
M-Chan Jul 9, 2026
c960c23
pass the gathered status
M-Chan Jul 13, 2026
e35c1af
receive the status
M-Chan Jul 13, 2026
0d0da76
use ip of geobrick 3 in lab18
M-Chan Jul 13, 2026
969296c
add individual motor_status
M-Chan Jul 13, 2026
dc7edd4
use status dataclass values to update screen
M-Chan Jul 13, 2026
90e109b
comment out unused code
M-Chan Jul 13, 2026
1492d5e
add back watch window and queue
M-Chan Jul 14, 2026
0f558fe
delete unused code
M-Chan Jul 14, 2026
ae22bce
add commented-out old update_func function for reference
M-Chan Jul 22, 2026
28d9e01
Merge branch 'new-main' into dataclasses
M-Chan Aug 5, 2026
d02fc4e
tidy up comments and namings
M-Chan Aug 5, 2026
8d942d8
Delete repeated code
M-Chan Aug 6, 2026
b25430b
add test geobrick ip to launch.json
M-Chan Aug 11, 2026
5b672a6
restore default ip
M-Chan Aug 11, 2026
7c020e1
strip $ for ppmacs and make global_status of type str
M-Chan Aug 11, 2026
01aff5a
revert back to commsThread
M-Chan Aug 14, 2026
cf95da5
plural dataclasses and add both amplifier_statuses to account for ppmacs
M-Chan Aug 14, 2026
ef378d7
do velocity calculation beforehand and call setText once
M-Chan Aug 14, 2026
8a4253f
change motor_row to motor and call i2t_fault_status instead of amplif…
M-Chan Aug 14, 2026
bd85742
separate out sendCommand
M-Chan Aug 14, 2026
46e51a1
add back the pollrate calculation
M-Chan Aug 14, 2026
dee4846
fix typo and hence function
M-Chan Aug 19, 2026
3578028
move i65 to 'global' ControllerStatus
M-Chan Aug 19, 2026
332222d
do the success validation earlier at the sendCommand stage
M-Chan Aug 19, 2026
4fc20aa
keep velocity as a numerical value before it is set as a text
M-Chan Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion .vscode/launch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
304 changes: 130 additions & 174 deletions src/dls_pmac_control/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(

@JamesOHeaDLS JamesOHeaDLS Aug 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Need to be careful here as, looking at the original code, the PowerBrick responses include a $ on the hex value which is stripped off before being passed to update_status. The non-PowerBrick responses don't require this character stripping

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
Comment on lines +717 to +718

@JamesOHeaDLS JamesOHeaDLS Aug 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are 2 amplifier status checks which are performed for PowerBricks - I2tFaultStatus amd OverCurrent. So having a single amplifier_status field in the dataclass is not enough.

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",
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading