Cleaned up imports and PVs

This commit is contained in:
2024-06-21 12:13:57 +02:00
parent 1a204693dc
commit 2ee2b25c21

View File

@@ -5,17 +5,15 @@ Created on Tue Jun 11 11:28:38 2024
@author: mohacsi_i
"""
from ophyd import Device, Component, PVPositioner, EpicsSignal, EpicsSignalRO, Kind, DerivedSignal
from ophyd.status import Status, SubscriptionStatus, StatusBase, DeviceStatus
from ophyd.flyers import FlyerInterface
from time import sleep
import warnings
import numpy as np
import time
from ophyd import Component, PVPositioner, EpicsSignal, EpicsSignalRO, Kind
from ophyd.status import Status
class A3200Axis(PVPositioner):
"""Ophyd proxy class for the Aerotech Automation 1's core controller functionality"""
"""Positioner wrapper for motors on the Aerotech A3200 controller. As the
IOC does not provide a motor record, this class simply wraps axes into
a standard Ophyd positioner for the BEC. It also has some additional
functionality for diagnostics."""
# Basic PV positioner interface
done = Component(EpicsSignalRO, "-DONE", auto_monitor=True, kind=Kind.config)
readback = Component(EpicsSignalRO, "-RBV", auto_monitor=True, kind=Kind.hinted)
@@ -38,14 +36,14 @@ class A3200Axis(PVPositioner):
_rock_dist = Component(EpicsSignal, "-RINCP", put_complete=True, kind=Kind.config)
_rock_velo = Component(EpicsSignal, "-RSETV", put_complete=True, kind=Kind.config)
_rock_count = Component(EpicsSignal, "-COUNT", put_complete=True, kind=Kind.config)
_rock_accel = Component(EpicsSignal, "-RRATE", put_complete=True, kind=Kind.config)
#_rock_accel = Component(EpicsSignal, "-RRATE", put_complete=True, kind=Kind.config)
def rmove(self, distance, wait=True, timeout=None, moved_cb=None) -> Status:
""" Native relative movement on the A3200 """
# Before moving, ensure we can stop (if a stop_signal is configured).
if self.stop_signal is not None:
self.stop_signal.wait_for_connection()
status = super(PVPositioner, self).move(position=distance, timeout=timeout, moved_cb=moved_cb)
status = super(self).move(position=distance, timeout=timeout, moved_cb=moved_cb)
has_done = self.done is not None
if not has_done:
@@ -64,18 +62,25 @@ class A3200Axis(PVPositioner):
return status
def rock(self, distance, counts: int, velocity=None, acceleration=None, wait=True):
def rock(self, distance, counts: int, velocity=None, acceleration=None, wait=True) -> Status:
""" Repeated single axis zigzag scan I guess PSO should be configured for this"""
self._rock_dist.put(distance)
self._rock_count.put(counts)
if velocity is not None:
self._rock_velo.put(velocity)
if acceleration is not None:
self._rock_accel.put(acceleration)
#if acceleration is not None:
# self._rock_accel.put(acceleration)
self._rock.put(1)
status = super(PVPositioner, self).move(position=distance)
if wait:
status.wait()
return status
return status
# Automatically start an axis if directly invoked
if __name__ == "__main__":
omega = A3200Axis(prefix="X06DA-ES-DF1:OMEGA", name="omega")
omega.wait_for_connection()