Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
EpicsMotor alarm handling
from ophyd.epics_motor import EpicsMotor, required_for_connection, motor_done_move, AlarmSeverity
PatchedEpicsMotor(EpicsMotor):
@required_for_connection
@motor_done_move.sub_value
def _move_changed(self, timestamp=None, value=None, sub_type=None,
**kwargs):
'''Callback from EPICS, indicating that movement status has changed'''
was_moving = self._moving
self._moving = (value != 1)
started = False
if not self._started_moving:
started = self._started_moving = (not was_moving and self._moving)
self.log.debug('[ts=%s] %s moving: %s (value=%s)', fmt_time(timestamp),
self, self._moving, value)
if started:
self._run_subs(sub_type=self.SUB_START, timestamp=timestamp,
value=value, **kwargs)
if was_moving and not self._moving:
success = True
# Check if we are moving towards the low limit switch
if self.direction_of_travel.get() == 0:
if self.low_limit_switch.get(use_monitor=False) == 1:
success = False
# No, we are going to the high limit switch
else:
if self.high_limit_switch.get(use_monitor=False) == 1:
success = False
# Check the severity of the alarm field after motion is complete.
# If there is any alarm at all warn the user, and if the alarm is
# greater than what is tolerated, mark the move as unsuccessful
severity = self.user_readback.alarm_severity
if severity != AlarmSeverity.NO_ALARM:
status = self.user_readback.alarm_status
if severity > self.tolerated_alarm:
self.log.error('Motion failed: %s is in an alarm state '
'status=%s severity=%s',
self.name, status, severity)
# *** TAKE CORRECTIVE ACTION HERE ***
success = False
else:
self.log.warning('Motor %s raised an alarm during motion '
'status=%s severity %s',
self.name, status, severity)
self._done_moving(success=success, timestamp=timestamp,
value=value)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment