Skip to content

Commit

Permalink
fixes to tracking during automated motor moving and addition of a bea…
Browse files Browse the repository at this point in the history 
…m location for referencing during jet moves using image
  • Loading branch information
@amandashack
amandashack committed Jun 12, 2023
1 parent 40d860a commit 6c41b6f
Show file tree
Hide file tree
Showing 12 changed files with 350 additions and 120 deletions.
22 changes: 4 additions & 18 deletions jet_tracking/context.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,6 @@ class Context(object):
position_tolerance (float): Tolerance for motor position.
motor_averaging (int): Number of points to average for motor movement.
algorithm (str): Algorithm used for motor movement.
motor_running (bool): Flag indicating if the motor is currently running.
motor_mode (str): Mode of the motor.
live_motor (bool): Flag indicating if live motor data is enabled.
motor_position (float): Current motor position.
Expand Down Expand Up @@ -162,14 +161,14 @@ def __init__(self, signals):
self.high_limit = 0.1
self.low_limit = -0.1
self.step_size = 0.02
self.position_tolerance = 0.001
self.position_tolerance = 0.01
self.motor_averaging = 10
self.algorithm = 'Ternary Search'
self.motor_running = False
self.motor_mode = ''
self.motor_mode = 'sleep'
self.live_motor = True
self.motor_position = 0
self.read_motor_position = 0
self.motor_connected = False
self.live_data = True
self.display_flag = None

Expand Down Expand Up @@ -406,7 +405,7 @@ def calibrate_image(self):
"""
Initiate the image calibration process.
"""
if self.motor_running:
if self.motor_mode != "sleep":
self.signals.message.emit("the motor is currently running an algorithm. Try stopping motor first")
else:
self.update_motor_mode('calibrate')
Expand All @@ -420,10 +419,6 @@ def update_motor_mode(self, mode):
"""
self.motor_mode = mode
self.signals.motorMode.emit(self.motor_mode)
if self.motor_mode == 'run':
self.motor_running = True
else:
self.motor_running = False

def connect_motor(self):
"""
Expand Down Expand Up @@ -470,15 +465,6 @@ def update_image_calibration_steps(self, cs):
"""
self.image_calibration_steps = cs

def update_motor_running(self, running):
"""
Update the flag indicating if the motor is running.
Args:
running (bool): Flag indicating if the motor is running.
"""
self.motor_running = running

def update_peak_intensity(self, pi):
"""
Update the peak intensity.
Expand Down
62 changes: 34 additions & 28 deletions jet_tracking/datastream.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,11 +102,14 @@ def initialize_connections(self):
self.signal_ratio = EpicsSignal(ratio)
dropped = self.context.PV_DICT.get('dropped', None)
self.signal_dropped = EpicsSignal(dropped)
except NotImplementedError:
pass
except Exception as e:
self.live_initialized = False
self.connected = False
self.signals.message.emit(f"Could not connect to the PVs for the data stream, {type(e).__name__}")
else:
self.live_initialized = True
self.connected = True
self.live_initialized = False
self.connected = False
self.signals.message.emit("Could not connect to the PVs for the data stream")
elif not self.live_data:
self.simgen = SimulationGenerator(self.context, self.signals)
self.connected = True
Expand All @@ -121,7 +124,6 @@ def run_live_data(self, live):
self.live_data = live
if self.connected:
self.connected = False
self.initialize_connections()

def live_data_stream(self):
"""
Expand Down Expand Up @@ -425,6 +427,7 @@ def tracking(self, b):
A flag indicating whether tracking mode should be enabled.
"""
self.isTracking = b
print(self.isTracking)

def update_mode(self, mode):
"""
Expand Down Expand Up @@ -553,16 +556,16 @@ def check_status_update(self):
n_high = np.count_nonzero(high[~np.isnan(high)])
if n_miss > self.notification_tolerance:
self.signals.changeStatus.emit("Warning, missed shots", "red")
self.processor_worker.count_flags_and_execute('missed shot', 50,
self.processor_worker.count_flags_and_execute('missed shot', 20,
self.missed_shots)
elif n_drop > self.notification_tolerance:
self.signals.changeStatus.emit("Lots of dropped shots",
"yellow")
self.processor_worker.count_flags_and_execute('dropped shot', 50,
self.processor_worker.count_flags_and_execute('dropped shot', 20,
self.dropped_shots)
elif n_high > self.notification_tolerance:
self.signals.changeStatus.emit("High Intensity", "orange")
self.processor_worker.count_flags_and_execute('high intensity', 50,
self.processor_worker.count_flags_and_execute('high intensity', 20,
self.high_intensity)
else:
if not self.processor_worker.isCounting:
Expand Down Expand Up @@ -791,32 +794,28 @@ def missed_shots(self):
checks the bad_scan_counter and disables tracking if the limit is reached. If the motor is
not running and tracking is disabled, emits a message signal suggesting running a search.
"""
if self.context.motor_running:
print("processor worker called missed shots.. ", self.isTracking, self.context.motor_connected, self.context.motor_mode)
if self.context.motor_connected:
if self.status == "everything is good":
self.signals.notifyMotor.emit("missed shots")
elif self.status == "dropped shots":
self.signals.notifyMotor.emit("resume")
elif self.status == "high intensity":
self.signals.notifyMotor.emit("you downgraded")
if self.isTracking and not self.context.motor_running:
if self.isTracking and self.context.motor_mode == "sleep":
print("tracking and not motor running")
if self.bad_scan_counter < self.bad_scan_limit:
print(self.bad_scan_counter, self.bad_scan_limit)
self.signals.message.emit("lots of missed shots.. "
"starting motor")
self.context.update_motor_mode("run")
self.bad_scan_counter += 1
else:
self.signals.enableTracking.emit(False)
self.signals.trackingStatus.emit('disabled', "red")
elif not self.isTracking and not self.context.motor_running:
elif not self.isTracking and self.context.motor_mode == "sleep":
self.signals.message.emit("lots of missed shots.. consider running"
" a search")
if self.context.motor_running:
if self.status == "everything is good":
self.signals.notifyMotor.emit("missed shots")
elif self.status == "dropped shots":
self.signals.notifyMotor.emit("resume")
elif self.status == "high intensity":
self.signals.notifyMotor.emit("you downgraded")
self.status = "missed_shots"

def dropped_shots(self):
Expand All @@ -825,7 +824,9 @@ def dropped_shots(self):
Pauses the motor and emits a message signal indicating a high number of dropped shots.
"""
if self.context.motor_running:
print("processor worker called dropped shots.. ", self.isTracking, self.context.motor_connected,
self.context.motor_mode)
if self.context.motor_connected:
self.signals.message.emit("lots of dropped shots.. pausing motor")
self.signals.notifyMotor.emit("pause")
self.status = "dropped shots"
Expand All @@ -837,8 +838,10 @@ def high_intensity(self):
Resets the bad_scan_counter and emits appropriate notifyMotor signals based on the current
status. Sets the status to 'high intensity'.
"""
print("processor worker called high intensity.. ", self.isTracking, self.context.motor_connected,
self.context.motor_mode)
self.bad_scan_counter = 0
if self.context.motor_running:
if self.context.motor_connected:
if self.status == "dropped shots":
self.notifyMotor("resume")
elif self.status == "missed shots":
Expand All @@ -854,9 +857,11 @@ def everything_is_good(self):
Resets the bad_scan_counter and stops the processor_worker. Emits appropriate notifyMotor
signals based on the current status. Sets the status to 'everything is good'.
"""
print("processor worker called everything is good.. ", self.isTracking, self.context.motor_connected,
self.context.motor_mode)
self.bad_scan_counter = 0
self.processor_worker.stop_count()
if self.context.motor_running:
if self.context.motor_connected:
if self.status == "dropped shots":
self.signals.notifyMotor("resume")
elif self.status == "missed shots":
Expand Down Expand Up @@ -1559,7 +1564,6 @@ def live_motor(self, live):
self.live = live
if self.connected:
self.connected = False
self.connect_to_motor()

def connect_to_motor(self):
"""
Expand All @@ -1569,24 +1573,26 @@ def connect_to_motor(self):
"""
if self.live:
self.motor_name = self.context.PV_DICT.get('motor', None)
print("connecting to: ", self.motor_name)
self.signals.message.emit("Trying to connect to: "+ self.motor_name)
try:
self.motor = Motor(self.motor_name, name='jet_x')
time.sleep(1)
# for i in self.motor.component_names: ## this is a good diagnostic
# print(f"{i} {getattr(self.motor, i).connected}")
self.context.update_read_motor_position(self.motor.position)
except NameError or NotImplementedError:
except Exception as e:
self.connected = False
else: ## not sure what case this was added for so leaving it for now
self.context.update_motor_position(self.motor.position)
self.connected = True
self.wait = True
self.signals.message.emit(f"Could not connect to the PVs for the motor, {type(e).__name__}")
else:
self.connected = False
self.signals.message.emit(f"Could not connect to the PVs for the motor.")
elif not self.live:
self.motor = SimulatedMotor(self.context, self.signals)
self.context.update_motor_position(self.motor.position)
self.connected = True
self.wait = False
if self.connected:
self.context.motor_connected = True

def clear_values(self):
"""
Expand Down
24 changes: 1 addition & 23 deletions jet_tracking/gui/widgets/basicWidgets.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,33 +4,11 @@
from PyQt5.QtCore import (QAbstractAnimation, QParallelAnimationGroup,
QPropertyAnimation, QRect, Qt, pyqtSignal)
from PyQt5.QtGui import QBrush, QColor, QDoubleValidator, QPainter, QPalette
from PyQt5.QtWidgets import (QComboBox, QFrame, QGraphicsScene, QGraphicsView,
QLabel, QLineEdit, QMessageBox, QPushButton,
from PyQt5.QtWidgets import (QFrame, QLabel, QLineEdit, QMessageBox,
QScrollArea, QSizePolicy, QToolButton,
QVBoxLayout, QWidget)


class GraphicsView(QGraphicsView):
def __init__(self, parent=None):
super(GraphicsView, self).__init__(parent)
self.setMouseTracking(True)


class GraphicsScene(QGraphicsScene):
def __init__(self, parent=None):
super(GraphicsScene, self).__init__(parent)


class ComboBox(QComboBox):
def __init__(self, parent=None):
super(ComboBox, self).__init__(parent)


class PushButton(QPushButton):
def __init__(self, parent=None):
super(PushButton, self).__init__(parent)


class LineEdit(QLineEdit):
checkVal = pyqtSignal(float)

Expand Down
19 changes: 9 additions & 10 deletions jet_tracking/gui/widgets/controlWidget.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,11 +130,11 @@ def start_processes(self):
self._start_motor()
else:
self.signals.message.emit("Motor is not connected.\n"
"fix motor connection settings"
"and try again")
"Fix motor connection settings"
" and try again")
else:
self.signals.message.emit("The Value reader is not connecting"
"properly. Check settings.")
" properly. Check settings/PVs.")

def start_algorithm(self):
"""
Expand Down Expand Up @@ -179,7 +179,6 @@ def _start_motor(self):
"""
if not self.thread2.isRunning():
if self.thread1.isRunning():
self.context.update_motor_running(False)
self.context.update_motor_mode('sleep')
self.signals.startMotorThread.emit()
else:
Expand All @@ -195,7 +194,6 @@ def _stop_motor(self):
"""
if not self.worker_motor.paused:
self.context.update_motor_running(False)
self.signals.stopMotorThread.emit(False)
self.signals.message.emit("The motor was stopped"
"either press start or connect motor")
Expand Down Expand Up @@ -386,9 +384,8 @@ def popup_clicked(self, i):
self.context.update_live_graphing(False)
self.context.update_live_motor(False)
else:
print("here")
if self.bttngrp1.checkedId() == 1:
self.rdbttn_sim.setChecked(True)
self.rdbttn_all_sim.setChecked(True)
elif self.bttngrp1.checkedId() == 0:
self.rdbttn_live.setChecked(True)

Expand All @@ -400,20 +397,22 @@ def checkBttn(self, button):
button: The clicked button.
"""
bttn = button.text()
if bttn == "simulated data":
if bttn == "All simulated":
if not self.worker_motor.paused:
print("Motor running and trying to change to simulated")
self.signals.showMessageBox.emit("live", "simulated")
else:
self.context.update_live_graphing(False)
self.context.update_live_motor(False)
elif bttn == "live data":
elif bttn == "Live data":
if not self.worker_motor.paused:
print("Motor running and trying to change to live")
self.signals.showMessageBox.emit("simulated", "live")
else:
self.context.update_live_graphing(True)
self.context.update_live_motor(True)
elif bttn == "Sim data + Live motor/image":
self.context.update_live_graphing(False)
self.context.update_live_motor(True)
elif bttn == "manual \nmotor moving":
self.bttn_search.setEnabled(True)
self.bttn_tracking.setEnabled(False)
Expand Down
Loading

0 comments on commit 6c41b6f

Please sign in to comment.