bec/pxiii_bec
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

SmarGon reads

Browse Source
This commit is contained in:
gac-x06da
2025-01-15 17:44:46 +01:00
parent d3d016108e
commit 03a5850bbf
2 changed files with 139 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
pxiii_bec/devices/A3200.py
View File

@@ -100,11 +100,11 @@ class AerotechAbrStage(Device):
""" Standard PX stage on A3200 controller
This is the wrapper class for the standard rotation stage layout for the PX
beamlines at SLS. It wraps the main rotation axis OMEGA and the associated
motion axes GMX, GMY and GMZ. The ophyd class associates to the general PX
measurement procedure, which is that the actual scan script is running as
an AeroBasic program on the controller and we communicate to it via 10+1
global variables.
beamlines at SLS. It wraps the main rotation axis OMEGA (Aerotech ABR)and
the associated motion axes GMX, GMY and GMZ. The ophyd class associates to
the general PX measurement procedure, which is that the actual scan script
is running as an AeroBasic program on the controller and we communicate to
it via 10+1 global variables.
"""
USER_ACCESS = ['reset', 'kickoff', 'complete']
@@ -135,8 +135,7 @@ class AerotechAbrStage(Device):
gmy_done = Component(EpicsSignalRO, "-DF1:GMY-DONE", kind=Kind.normal)
gmz_done = Component(EpicsSignalRO, "-DF1:GMZ-DONE", kind=Kind.normal)
# For some reason the task interface is called PSO...
scan_command = Component(
EpicsSignal, "-PSO:CMD", put_complete=True, kind=Kind.omitted)
start_command = Component(
@@ -157,8 +156,7 @@ class AerotechAbrStage(Device):
_var_10 = Component(EpicsSignal, "-PSO:VAR-10", put_complete=True, kind=Kind.omitted)
# Task status PVs (programs always run on task 1)
task1 = Component(
EpicsSignalRO, "-AERO:TSK1-DONE", auto_monitor=True, kind=Kind.hinted)
task1 = Component(EpicsSignalRO, "-AERO:TSK1-DONE", auto_monitor=True)
task2 = Component(EpicsSignalRO, "-AERO:TSK2-DONE", auto_monitor=True)
task3 = Component(EpicsSignalRO, "-AERO:TSK3-DONE", auto_monitor=True)
task4 = Component(EpicsSignalRO, "-AERO:TSK4-DONE", auto_monitor=True)
@@ -168,8 +166,10 @@ class AerotechAbrStage(Device):
raster_num_rows = Component(EpicsSignal, "-GRD:ROW-DONE", kind=Kind.config)
def set_axis_mode(self, mode: str, settle_time=0.1) -> None:
""" Set acix mode to direct/measurement mode.
Measurement mode blocks axis commands from the IOC.
""" Set axis mode to direct/measurement mode.
Measurement ensures that the scrips run undisturbed by blocking axis
motion commands from the IOC (i.e. internal only).
Parameters:
-----------
@@ -250,9 +250,31 @@ class AerotechAbrStage(Device):
status = SubscriptionStatus(self.raster_scan_done, is_idle, timeout=timeout, settle_time=0.5)
return status
def reset(self, settle_time=0.1):
"""Attempts to reset the currently running measurement task on the A3200"""
def reset(self, settle_time=0.1, wait_after_reload=1) -> None:
""" Resets the Aerotech controller state
Attempts to reset the currently running measurement task on the A3200
by stopping current motions, reloading aerobasic programs and going
to DIRECT mode.
This will stop movements in both DIRECT and MEASURING modes. During the
stop the `status` temporarely goes to ERROR but reverts to OK after a
couple of seconds.
"""
# Disarm commands
self.scan_command.set(CMD_NONE, settle_time=settle_time).wait()
self.stop_command.set(1, settle_time=settle_time)
# Reload tasks
self.taskStop.set(1, settle_time=wait_after_reload).wait()
# Go to direct mode
self.set_axis_mode("direct", settle_time=settle_time)
def stop(self, settle_time=1.0) -> None:
""" Stops current motions
"""
# Disarm commands
self.scan_command.set(CMD_NONE, settle_time=settle_time).wait()
# Go to direct mode
self.set_axis_mode("direct", settle_time=settle_time)
def is_ioc_ok(self):
@@ -283,6 +305,10 @@ class AerotechAbrStage(Device):
def measurement_state(self, value):
self.osc.phase.set(value).wait()
@property
def axis_mode(self):
return self.axisAxesMode.get()
# @property
# def shutter(self):
# return self._shutter.get()
@@ -303,56 +329,16 @@ class AerotechAbrStage(Device):
# self._shutter.set(state).wait()
# return state == self._shutter.get()
@property
def axis_mode(self):
return self.axisAxesMode.get()
def get_ready(self, ostart=None, orange=None, etime=None, wait=True):
self.wait_for_movements()
if self.measurement_state == ABR_BUSY:
raise RuntimeError("ABR is not DONE!!!!")
if self.measurement_state == ABR_READY:
self.measurement_state = ABR_DONE
if ostart is not None:
self.osc.ostart.set(ostart).wait()
if orange is not None:
self.osc.orange.set(orange).wait()
if etime is not None:
self.osc.etime.set(etime).wait()
self.osc.get_ready.set(1, settle_time=0.1).wait()
if wait:
for _ in range(10):
try:
self.wait_status(ABR_READY, timeout=5).wait()
except RuntimeWarning as ex:
logger.warning(f"{ex} --- trying ready again.")
self.osc.get_ready.set(1).wait()
def wait_for_movements(self, timeout=60.0):
timeout = timeout + time.time()
timeisup = False
""" Waits for all motor movements"""
t_start = time.time()
t_elapsed = 0
try:
moving = self.is_moving()
except:
moving = True
while not timeisup and moving:
time.sleep(0.1)
try:
moving = self.is_moving()
except Exception as ex:
logger.error(f"Failed to retrieve moving state for axes: {ex}")
moving = True
timeisup = timeout < time.time()
if timeisup:
raise RuntimeWarning("timeout waiting for all axis to stop moving")
while self.is_moving() and t_elapsed < timeout:
t_elapsed = time.time() - t_start
if timeout is not None and t_elapsed > timeout:
raise TimeoutError("Timeout waiting for all axis to stop moving")
time.sleep(0.5)
def is_moving(self):
"""Chechs if all axes are DONE"""
@@ -360,20 +346,6 @@ class AerotechAbrStage(Device):
self.omega_done.get()
and self.gmx_done.get() and self.gmy_done.get() and self.gmz_done.get() )
def stop(self, wait_after_reload=1.0):
"""Stops current motions; reloads aerobasic programs; goes DIRECT
This will stop movements in both DIRECT and MEASURING modes. During the
stop the `status` temporarely goes to ERROR but reverts to OK after a
couple of seconds.
"""
self.taskStop.set(1, settle_time=wait_after_reload).wait()
self.reset()
time.sleep(0.1)
reload_programs = stop
def start_exposure(self):
"""Starts the previously configured exposure."""
self.wait_for_movements()

104
pxiii_bec/devices/SmarGon.py
View File

@@ -15,22 +15,64 @@ except ModuleNotFoundError:
class SmarGonSignal(Signal):
"""Small helper class to read PVs that need to be processed first."""
def __init__(self, group, addr, *args, **kwargs):
def __init__(self, prefix, *args, **kwargs):
super().__init__(*args, **kwargs)
self.group = group
self.addr = addr
self.prefix = prefix
self.addr = self.parent.name
# self.get()
def put(self, value, *args, **kwargs):
r = self.parent._go_n_put(f"target{self.group}?{self.addr}={value}")
print(r)
return super().put(*args, **kwargs)
self._go_n_put(f"target{self.prefix}?{self.addr}={value}")
return super().put(value, *args, force=True, **kwargs)
def get(self, *args, **kwargs):
r = self.parent._go_n_get(f"target{self.group}?{self.addr}")
r = self._go_n_get(f"target{self.prefix}")
print(r)
self.value = r[self.addr.upper()]
return super().get(*args, **kwargs)
def _go_n_get(self, name, **kwargs):
cmd = f"{self.parent.sg_url.get()}/{name}"
r = requests.get(cmd, timeout=1)
if not r.ok:
raise RuntimeError(f"[self.name] Error getting {name}; server returned {r.status_code} => {r.reason}")
return r.json()
def _go_n_put(self, name, **kwargs):
cmd = f"{self.parent.sg_url.get()}/{name}"
r = requests.put(cmd, timeout=1)
if not r.ok:
raise RuntimeError(f"[self.name] Error putting {name}; server returned {r.status_code} => {r.reason}")
class SmarGonSignalRO(Signal):
"""Small helper class to read PVs that need to be processed first.
TODO: Add monitoring
"""
def __init__(self, prefix, *args, **kwargs):
super().__init__(*args, **kwargs)
self._metadata["write_access"] = False
self.prefix = prefix
self.addr = self.parent.name
def get(self, *args, **kwargs):
r = self._go_n_get(f"readback{self.prefix}")
print(r)
self.put(r[self.addr.upper()], force=True)
return super().get(*args, **kwargs)
def _go_n_get(self, name, **kwargs):
cmd = f"{self.parent.sg_url.get()}/{name}"
r = requests.get(cmd, timeout=1)
if not r.ok:
raise RuntimeError(f"[self.name] Error getting {name}; server returned {r.status_code} => {r.reason}")
return r.json()
class SmarGonClient(Device):
"""SmarGon client deice
@@ -44,11 +86,11 @@ class SmarGonClient(Device):
sg_url = Component(Signal, kind=Kind.config, metadata={'write_access': False})
# Axis parameters
shx = Component(SmarGonSignal, group="SCS", name="shx", kind=Kind.config)
# shy = Component(SmarGonSignal, group="SCS", name="shy", kind=Kind.config)
# shz = Component(SmarGonSignal, group="SCS", name="shz", kind=Kind.config)
# chi = Component(SmarGonSignal, group="SCS", name="chi", kind=Kind.config)
# phi = Component(SmarGonSignal, group="SCS", name="phi", kind=Kind.config)
shx = Component(SmarGonSignal, group="SCS", addr="shx", kind=Kind.config)
# shy = Component(SmarGonSignal, group="SCS", addr="shy", kind=Kind.config)
# shz = Component(SmarGonSignal, group="SCS", addr="shz", kind=Kind.config)
# chi = Component(SmarGonSignal, group="SCS", addr="chi", kind=Kind.config)
# phi = Component(SmarGonSignal, group="SCS", addr="phi", kind=Kind.config)
def __init__(
self,
@@ -84,9 +126,47 @@ class SmarGonClient(Device):
class SmarGonAxis(Device):
"""SmarGon client deice
This class controls the SmarGon goniometer via the REST interface.
"""
# Status attributes
sg_url = Component(Signal, kind=Kind.config, metadata={'write_access': False})
# Axis parameters
readback = Component(SmarGonSignalRO, kind=Kind.config)
setpoint = Component(SmarGonSignal, kind=Kind.config)
def __init__(
self,
prefix="SCS",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sg_url: str = "http://x06da-smargopolo.psi.ch:3000",
**kwargs,
) -> None:
self.__class__.__dict__['readback'].kwargs['prefix'] = prefix
self.__class__.__dict__['readback'].kwargs['name'] = name
self.__class__.__dict__['setpoint'].kwargs['prefix'] = prefix
self.__class__.__dict__['setpoint'].kwargs['name'] = name
# super().__init__(prefix=prefix, name=name, kind=kind, read_attrs=read_attrs, configuration_attrs=configuration_attrs, parent=parent, device_manager=device_manager, **kwargs)
super().__init__(prefix=prefix, name=name, kind=kind, read_attrs=read_attrs, configuration_attrs=configuration_attrs, parent=parent, **kwargs)
self.sg_url._metadata["write_access"] = False
self.sg_url.set(sg_url, force=True).wait()
if __name__ == "__main__":
shz = SmarGonAxis(prefix="SCS", name="shz", sg_url="http://x06da-smargopolo.psi.ch:3000")
sg = SmarGonClient(prefix="X06DA-ES", name="smargon")
sg.wait_for_connection()