bec/ophyd_devices
1
0
Fork 0
Code Issues 9 Pull Requests 10 Actions Packages Projects Releases 126 Wiki Activity

fix: online changes to all devices in preparation for beamtime

Browse Source
This commit is contained in:
appel_c 2023-09-04 16:23:14 +02:00
parent b6101cced2
commit c0b3418366
5 changed files with 234 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

194
ophyd_devices/epics/devices/DelayGeneratorDG645.py
View File

@ -42,10 +42,18 @@ class DelayStatic(Device):
kind=Kind.config,
)
amplitude = Component(
EpicsSignal, "OutputAmpAI", write_pv="OutputAmpAO", name="amplitude", kind=Kind.config
EpicsSignal,
"OutputAmpAI",
write_pv="OutputAmpAO",
name="amplitude",
kind=Kind.config,
)
offset = Component(
EpicsSignal, "OutputOffsetAI", write_pv="OutputOffsetAO", name="offset", kind=Kind.config
EpicsSignal,
"OutputOffsetAI",
write_pv="OutputOffsetAO",
name="offset",
kind=Kind.config,
)
@ -53,7 +61,9 @@ class DummyPositioner(PVPositioner):
setpoint = Component(EpicsSignal, "DelayAO", put_complete=True, kind=Kind.config)
readback = Component(EpicsSignalRO, "DelayAI", kind=Kind.config)
done = Component(Signal, value=1)
reference = Component(EpicsSignal, "ReferenceMO", put_complete=True, kind=Kind.config)
reference = Component(
EpicsSignal, "ReferenceMO", put_complete=True, kind=Kind.config
)
class DelayPair(PseudoPositioner):
@ -83,12 +93,16 @@ class DelayPair(PseudoPositioner):
@pseudo_position_argument
def forward(self, pseudo_pos):
"""Run a forward (pseudo -> real) calculation"""
return self.RealPosition(ch1=pseudo_pos.delay, ch2=pseudo_pos.delay + pseudo_pos.width)
return self.RealPosition(
ch1=pseudo_pos.delay, ch2=pseudo_pos.delay + pseudo_pos.width
)
@real_position_argument
def inverse(self, real_pos):
"""Run an inverse (real -> pseudo) calculation"""
return self.PseudoPosition(delay=real_pos.ch1, width=real_pos.ch2 - real_pos.ch1)
return self.PseudoPosition(
delay=real_pos.ch1, width=real_pos.ch2 - real_pos.ch1
)
class TriggerSource(int, enum.Enum):
@ -125,6 +139,14 @@ class DelayGeneratorDG645(Device):
current device
"""
USER_ACCESS = [
"set_channels",
"_set_trigger",
"burst_enable",
"burst_disable",
"reload_config",
]
state = Component(EpicsSignalRO, "EventStatusLI", name="status_register")
status = Component(EpicsSignalRO, "StatusSI", name="status")
clear_error = Component(EpicsSignal, "StatusClearBO", name="clear_error")
@ -172,63 +194,107 @@ class DelayGeneratorDG645(Device):
name="trigger_rate",
kind=Kind.config,
)
trigger_shot = Component(EpicsSignal, "TriggerDelayBO", name="trigger_shot", kind="config")
trigger_shot = Component(
EpicsSignal, "TriggerDelayBO", name="trigger_shot", kind="config"
)
# Burst mode
burstMode = Component(
EpicsSignal, "BurstModeBI", write_pv="BurstModeBO", name="burstmode", kind=Kind.config
EpicsSignal,
"BurstModeBI",
write_pv="BurstModeBO",
name="burstmode",
kind=Kind.config,
)
burstConfig = Component(
EpicsSignal, "BurstConfigBI", write_pv="BurstConfigBO", name="burstconfig", kind=Kind.config
EpicsSignal,
"BurstConfigBI",
write_pv="BurstConfigBO",
name="burstconfig",
kind=Kind.config,
)
burstCount = Component(
EpicsSignal, "BurstCountLI", write_pv="BurstCountLO", name="burstcount", kind=Kind.config
EpicsSignal,
"BurstCountLI",
write_pv="BurstCountLO",
name="burstcount",
kind=Kind.config,
)
burstDelay = Component(
EpicsSignal, "BurstDelayAI", write_pv="BurstDelayAO", name="burstdelay", kind=Kind.config
EpicsSignal,
"BurstDelayAI",
write_pv="BurstDelayAO",
name="burstdelay",
kind=Kind.config,
)
burstPeriod = Component(
EpicsSignal, "BurstPeriodAI", write_pv="BurstPeriodAO", name="burstperiod", kind=Kind.config
EpicsSignal,
"BurstPeriodAI",
write_pv="BurstPeriodAO",
name="burstperiod",
kind=Kind.config,
)
# bec_utils device ConfigSignal
delay_burst = Component(
bec_utils.ConfigSignal, name="delay_burst", kind="config", config_storage_name="ddg_configs"
bec_utils.ConfigSignal,
name="delay_burst",
kind="config",
config_storage_name="ddg_config",
)
delta_width = Component(
bec_utils.ConfigSignal, name="delta_width", kind="config", config_storage_name="ddg_configs"
bec_utils.ConfigSignal,
name="delta_width",
kind="config",
config_storage_name="ddg_config",
)
additional_triggers = Component(
bec_utils.ConfigSignal,
name="additional_triggers",
kind="config",
config_storage_name="ddg_configs",
config_storage_name="ddg_config",
)
polarity = Component(
bec_utils.ConfigSignal, name="polarity", kind="config", config_storage_name="ddg_configs"
bec_utils.ConfigSignal,
name="polarity",
kind="config",
config_storage_name="ddg_config",
)
amplitude = Component(
bec_utils.ConfigSignal, name="amplitude", kind="config", config_storage_name="ddg_configs"
bec_utils.ConfigSignal,
name="amplitude",
kind="config",
config_storage_name="ddg_config",
)
offset = Component(
bec_utils.ConfigSignal, name="offset", kind="config", config_storage_name="ddg_configs"
bec_utils.ConfigSignal,
name="offset",
kind="config",
config_storage_name="ddg_config",
)
thres_trig_level = Component(
bec_utils.ConfigSignal,
name="thres_trig_level",
kind="config",
config_storage_name="ddg_configs",
config_storage_name="ddg_config",
)
set_high_on_exposure = Component(
bec_utils.ConfigSignal,
name="set_high_on_exposure",
kind="config",
config_storage_name="ddg_configs",
config_storage_name="ddg_config",
)
set_high_on_stage = Component(
bec_utils.ConfigSignal,
name="set_high_on_stage",
kind="config",
config_storage_name="ddg_configs",
config_storage_name="ddg_config",
)
def __init__(
@ -242,6 +308,7 @@ class DelayGeneratorDG645(Device):
parent=None,
device_manager=None,
sim_mode=False,
ddg_config = None,
**kwargs,
):
"""_summary_
@ -259,11 +326,11 @@ class DelayGeneratorDG645(Device):
amplitude (_type_, optional): _description_. Defaults to None.
offset (_type_, optional): _description_. Defaults to None.
thres_trig_level (_type_, optional): _description_. Defaults to None.
delta_delay (_type_, float): Add delay for triggering in software trigger mode to allow fast shutter to open. Defaults to 0.
delay_burst (_type_, float): Add delay for triggering in software trigger mode to allow fast shutter to open. Defaults to 0.
delta_width (_type_, float): Add width to fast shutter signal to make sure its open during acquisition. Defaults to 0.
delta_triggers (_type_, int): Add additional triggers to burst mode (mcs card needs +1 triggers per line). Defaults to 0.
"""
self.ddg_configs = {
self.ddg_config = {
f"{name}_delay_burst": 0,
f"{name}_delta_width": 0,
f"{name}_additional_triggers": 0,
@ -274,6 +341,8 @@ class DelayGeneratorDG645(Device):
f"{name}_set_high_on_exposure": False,
f"{name}_set_high_on_stage": False,
}
if ddg_config is not None:
[self.ddg_config.update({f'{name}_{key}' : value}) for key, value in ddg_config.items()]
super().__init__(
prefix=prefix,
name=name,
@ -284,7 +353,9 @@ class DelayGeneratorDG645(Device):
**kwargs,
)
if device_manager is None and not sim_mode:
raise DDGError("Add DeviceManager to initialization or init with sim_mode=True")
raise DDGError(
"Add DeviceManager to initialization or init with sim_mode=True"
)
self.device_manager = device_manager
if not sim_mode:
self._producer = self.device_manager.producer
@ -292,10 +363,17 @@ class DelayGeneratorDG645(Device):
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self._all_channels = ["channelT0", "channelAB", "channelCD", "channelEF", "channelGH"]
self._all_channels = [
"channelT0",
"channelAB",
"channelCD",
"channelEF",
"channelGH",
]
self._all_delay_pairs = ["AB", "CD", "EF", "GH"]
self.wait_for_connection() # Make sure to be connected before talking to PVs
self._init_ddg()
logger.info(f'Current polarity value {self.polarity.get()}')
self.reload_config()
self._ddg_is_okay()
def _set_trigger(self, trigger_source: TriggerSource) -> None:
@ -322,7 +400,7 @@ class DelayGeneratorDG645(Device):
elif status != "STATUS OK":
raise DDGError(f"DDG failed to start with status: {status}")
def _set_channels(self, signal: str, value: Any, channels: List = None) -> None:
def set_channels(self, signal: str, value: Any, channels: List = None) -> None:
if not channels:
channels = self._all_channels
for chname in channels:
@ -338,26 +416,29 @@ class DelayGeneratorDG645(Device):
def _cleanup_ddg(self) -> None:
self._set_trigger(TriggerSource.SINGLE_SHOT)
def _init_ddg(self) -> None:
self._set_channels(
def reload_config(self) -> None:
self.set_channels(
"polarity",
self.polarity.get(),
channels=["channelT0", "channelCD", "channelEF", "channelGH"],
channels=["channelT0", "channelAB", "channelCD", "channelEF", "channelGH"],
)
# Set polarity for eiger inverted!
self._set_channels("polarity", 0, channels=["channelAB"])
self._set_channels("amplitude", self.amplitude.get())
self._set_channels("offset", self.offset.get())
# self.set_channels("polarity", 0, channels=["channelAB"])
self.set_channels("amplitude", self.amplitude.get())
self.set_channels("offset", self.offset.get())
# Setup reference
self._set_channels(
self.set_channels(
"reference",
0,
[f"channel{self._all_delay_pairs[ii]}.ch1" for ii in range(len(self._all_delay_pairs))],
[
f"channel{self._all_delay_pairs[ii]}.ch1"
for ii in range(len(self._all_delay_pairs))
],
)
for ii in range(len(self._all_delay_pairs)):
self._set_channels(
self.set_channels(
"reference",
2 * ii + 1,
0,
[f"channel{self._all_delay_pairs[ii]}.ch2"],
)
self._set_trigger(TriggerSource.SINGLE_SHOT)
@ -374,10 +455,10 @@ class DelayGeneratorDG645(Device):
delay_burst = self.delay_burst.get()
num_burst_cycle = 1 + self.additional_triggers.get()
# set parameters in DDG
self.burstEnable(num_burst_cycle, delay_burst, exp_time, config="first")
self._set_channels("delay", 0)
self.burst_enable(num_burst_cycle, delay_burst, exp_time, config="first")
self.set_channels("delay", 0)
# Set burst length to half of the experimental time!
self._set_channels("width", exp_time)
self.set_channels("width", exp_time)
elif self.scaninfo.scan_type == "fly":
# Prepare FSH DDG
if self.set_high_on_exposure.get():
@ -393,22 +474,28 @@ class DelayGeneratorDG645(Device):
# self.additional_triggers should be 0 for self.set_high_on_exposure or remove here fully..
num_burst_cycle = 1 + self.additional_triggers.get()
# set parameters in DDG
self.burstEnable(num_burst_cycle, delay_burst, total_exposure, config="first")
self._set_channels("delay", 0)
self.burst_enable(
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0)
# Set burst length to half of the experimental time!
self._set_channels("width", exp_time)
self.set_channels("width", exp_time)
else:
# define parameters
self._set_trigger(TriggerSource.SINGLE_SHOT)
exp_time = self.delta_width.get() + self.scaninfo.exp_time
total_exposure = exp_time + self.scaninfo.readout_time
delay_burst = self.delay_burst.get()
num_burst_cycle = self.scaninfo.num_frames + self.additional_triggers.get()
num_burst_cycle = (
self.scaninfo.num_frames + self.additional_triggers.get()
)
# set parameters in DDG
self.burstEnable(num_burst_cycle, delay_burst, total_exposure, config="first")
self._set_channels("delay", 0)
self.burst_enable(
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0)
# Set burst length to half of the experimental time!
self._set_channels("width", exp_time)
self.set_channels("width", exp_time)
else:
raise DDGError(f"Unknown scan type {self.scaninfo.scan_type}")
@ -427,18 +514,23 @@ class DelayGeneratorDG645(Device):
def trigger(self) -> None:
# if self.scaninfo.scan_type == "step":
if self.source.read()[self.source.name]["value"] == int(TriggerSource.SINGLE_SHOT):
if self.source.read()[self.source.name]["value"] == int(
TriggerSource.SINGLE_SHOT
):
self.trigger_shot.set(1).wait()
super().trigger()
def burstEnable(self, count, delay, period, config="all"):
def burst_enable(self, count, delay, period, config="all"):
"""Enable the burst mode"""
# Validate inputs
count = int(count)
assert count > 0, "Number of bursts must be positive"
assert delay >= 0, "Burst delay must be larger than 0"
assert period > 0, "Burst period must be positive"
assert config in ["all", "first"], "Supported bust configs are 'all' and 'first'"
assert config in [
"all",
"first",
], "Supported bust configs are 'all' and 'first'"
self.burstMode.set(1).wait()
self.burstCount.set(count).wait()
@ -450,7 +542,7 @@ class DelayGeneratorDG645(Device):
elif config == "first":
self.burstConfig.set(1).wait()
def burstDisable(self):
def burst_disable(self):
"""Disable the burst mode"""
self.burstMode.set(0).wait()

1
ophyd_devices/epics/devices/__init__.py
View File

@ -27,3 +27,4 @@ from .mcs_csaxs import McsCsaxs
from .eiger9m_csaxs import Eiger9mCsaxs
from .pilatus_csaxs import PilatusCsaxs
from .falcon_csaxs import FalconCsaxs
from .DelayGeneratorDG645 import DelayGeneratorDG645

25
ophyd_devices/epics/devices/eiger9m_csaxs.py
View File

@ -126,7 +126,9 @@ class Eiger9mCsaxs(DetectorBase):
**kwargs,
)
if device_manager is None and not sim_mode:
raise EigerError("Add DeviceManager to initialization or init with sim_mode=True")
raise EigerError(
"Add DeviceManager to initialization or init with sim_mode=True"
)
self.name = name
self.wait_for_connection() # Make sure to be connected before talking to PVs
@ -140,7 +142,9 @@ class Eiger9mCsaxs(DetectorBase):
# TODO
self.filepath = ""
self.scaninfo.username = "e21206"
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
self.service_cfg = {
"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"
}
self.filewriter = FileWriterMixin(self.service_cfg)
self.reduce_readout = 1e-3 # 3 ms
self.triggermode = 0 # 0 : internal, scan must set this if hardware triggered
@ -168,7 +172,9 @@ class Eiger9mCsaxs(DetectorBase):
f"Type of new value {type(value)}:{value} does not match old value {type(old_value)}:{old_value}"
)
cfg.update({cfg_key: value})
logger.info(f"Updated std_daq config for key {cfg_key} from {old_value} to {value}")
logger.info(
f"Updated std_daq config for key {cfg_key} from {old_value} to {value}"
)
def _init_standard_daq(self) -> None:
self.std_rest_server_url = "http://xbl-daq-29:5000"
@ -198,7 +204,9 @@ class Eiger9mCsaxs(DetectorBase):
energy = self.cam.beam_energy.read()[self.cam.beam_energy.name]["value"]
if setp_energy != energy:
self.cam.beam_energy.set(setp_energy) # .wait()
threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name]["value"]
threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name][
"value"
]
if not np.isclose(setp_energy / 2, threshold, rtol=0.05):
self.cam.threshold_energy.set(setp_energy / 2) # .wait()
@ -259,7 +267,9 @@ class Eiger9mCsaxs(DetectorBase):
self.arm_acquisition()
logger.info("Waiting for detector to be armed")
while True:
det_ctrl = self.cam.detector_state.read()[self.cam.detector_state.name]["value"]
det_ctrl = self.cam.detector_state.read()[self.cam.detector_state.name][
"value"
]
if det_ctrl == int(DetectorState.RUNNING):
break
time.sleep(0.005)
@ -279,13 +289,16 @@ class Eiger9mCsaxs(DetectorBase):
logger.info("Waiting for std daq to receive images")
while True:
det_ctrl = self.std_client.get_status()["acquisition"]["state"]
# TODO if no writing was performed before
if det_ctrl == "FINISHED":
break
time.sleep(0.005)
# Message to BEC
state = True
msg = BECMessage.FileMessage(file_path=self.filepath, done=True, successful=state)
msg = BECMessage.FileMessage(
file_path=self.filepath, done=True, successful=state
)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
msg.dumps(),

14
ophyd_devices/epics/devices/mcs_csaxs.py
View File

@ -131,7 +131,7 @@ class McsCsaxs(SIS38XX):
bec_utils.ConfigSignal,
name="num_lines",
kind="config",
config_storage_name="mcs_configs",
config_storage_name="mcs_config",
)
def __init__(
@ -145,11 +145,15 @@ class McsCsaxs(SIS38XX):
parent=None,
device_manager=None,
sim_mode=False,
mcs_config = None,
**kwargs,
):
self.mcs_configs = {
self.mcs_config = {
f"{name}_num_lines": 1,
}
if mcs_config is not None:
[self.mcs_config.update({f'{name}_{key}' : value}) for key, value in mcs_config.items()]
super().__init__(
prefix=prefix,
@ -231,6 +235,9 @@ class McsCsaxs(SIS38XX):
self._acquisition_done = True
self._send_data_to_bec()
self.stop_all.put(1, use_complete=False)
self._send_data_to_bec()
self.erase_all.set(1)
return
self.erase_start.set(1)
self._send_data_to_bec()
self.mca_data = defaultdict(lambda: [])
@ -328,7 +335,7 @@ class McsCsaxs(SIS38XX):
break
time.sleep(0.005)
logger.info("mcs is ready and running")
time.sleep(5)
return super().stage()
def unstage(self) -> List[object]:
@ -361,6 +368,7 @@ class McsCsaxs(SIS38XX):
self.stop_all.set(1)
# self.erase_all.set(1)
self._stopped = True
self._acquisition_done = True
super().stop(success=success)

78
ophyd_devices/galil/sgalil_ophyd.py
View File

@ -49,6 +49,9 @@ class GalilController(Controller):
"galil_show_all",
"socket_put_and_receive",
"socket_put_confirmed",
"sgalil_reference",
"fly_grid_scan",
"read_encoder_position",
]
def __init__(
@ -152,7 +155,7 @@ class GalilController(Controller):
def stop_all_axes(self) -> str:
# return self.socket_put_and_receive(f"XQ#STOP,1")
# Command stops all threads and motors!
return self.socket_put_and_receive(f"AB")
return self.socket_put_and_receive(f"ST")
def axis_is_referenced(self) -> bool:
return bool(float(self.socket_put_and_receive(f"MG allaxref").strip()))
@ -268,14 +271,23 @@ class GalilController(Controller):
"""
#
axes_referenced = self.controller.axis_is_referenced()
axes_referenced = self.axis_is_referenced()
sign_y = self._axis[ord("c") - 97].sign
sign_x = self._axis[ord("e") - 97].sign
# Check limits
# TODO check sign of stage, or not necessary
check_values = [start_y, end_y, start_x, end_x]
for val in check_values:
self.check_value(val)
speed = np.abs(end_y - start_y) / ((interval_y) * exp_time + (interval_y - 1) * readtime)
start_x *= sign_x
end_x *= sign_x
start_y *= sign_y
end_y *= sign_y
speed = np.abs(end_y - start_y) / (
(interval_y) * exp_time + (interval_y - 1) * readtime
)
if speed > 2.00 or speed < 0.02:
raise LimitError(
f"Speed of {speed:.03f}mm/s is outside of acceptable range of 0.02 to 2 mm/s"
@ -287,7 +299,9 @@ class GalilController(Controller):
# Hard coded to maximum offset of 0.1mm to avoid long motions.
self.socket_put_and_receive(f"off={(0):f}")
self.socket_put_and_receive(f"a_start={start_y:.04f};a_end={end_y:.04f};speed={speed:.04f}")
self.socket_put_and_receive(
f"a_start={start_y:.04f};a_end={end_y:.04f};speed={speed:.04f}"
)
self.socket_put_and_receive(
f"b_start={start_x:.04f};gridmax={gridmax:d};b_step={step_grid:.04f}"
)
@ -307,7 +321,9 @@ class GalilController(Controller):
val_axis4 = [] # x axis
while self.is_thread_active(thread_id):
posct = int(self.socket_put_and_receive(f"MGposct").strip().split(".")[0])
logger.info(f"SGalil is scanning - latest enconder position {posct+1} from {n_samples}")
logger.info(
f"SGalil is scanning - latest enconder position {posct+1} from {n_samples}"
)
time.sleep(1)
if posct > last_readout:
positions = self.read_encoder_position(last_readout, posct)
@ -318,7 +334,9 @@ class GalilController(Controller):
time.sleep(1)
# Readout of last positions after scan finished
posct = int(self.socket_put_and_receive(f"MGposct").strip().split(".")[0])
logger.info(f"SGalil is scanning - latest enconder position {posct} from {n_samples}")
logger.info(
f"SGalil is scanning - latest enconder position {posct} from {n_samples}"
)
if posct > last_readout:
positions = self.read_encoder_position(last_readout, posct)
val_axis4.extend(positions[0])
@ -330,7 +348,9 @@ class GalilController(Controller):
val_axis2 = [] # y axis
val_axis4 = [] # x axis
for ii in range(fromval, toval + 1):
rts = self.socket_put_and_receive(f"MGaposavg[{ii%2000}]*10,cposavg[{ii%2000}]*10")
rts = self.socket_put_and_receive(
f"MGaposavg[{ii%2000}]*10,cposavg[{ii%2000}]*10"
)
if rts == ":":
val_axis4.append(rts)
val_axis2.append(rts)
@ -369,11 +389,15 @@ class GalilReadbackSignal(GalilSignalRO):
"""
if self.parent.axis_Id_numeric == 2:
current_pos = float(
self.controller.socket_put_and_receive(f"MG _TP{self.parent.axis_Id}/mm")
self.controller.socket_put_and_receive(
f"MG _TP{self.parent.axis_Id}/mm"
)
)
elif self.parent.axis_Id_numeric == 4:
# hardware controller readback from axis 4 is on axis 0, A instead of E
current_pos = float(self.controller.socket_put_and_receive(f"MG _TP{'A'}/mm"))
current_pos = float(
self.controller.socket_put_and_receive(f"MG _TP{'A'}/mm")
)
current_pos *= self.parent.sign
return current_pos
@ -419,11 +443,17 @@ class GalilSetpointSignal(GalilSignalBase):
time.sleep(0.1)
if self.parent.axis_Id_numeric == 2:
self.controller.socket_put_confirmed(f"PA{self.parent.axis_Id}={target_val:.4f}*mm")
self.controller.socket_put_confirmed(
f"PA{self.parent.axis_Id}={target_val:.4f}*mm"
)
self.controller.socket_put_and_receive(f"BG{self.parent.axis_Id}")
elif self.parent.axis_Id_numeric == 4:
self.controller.socket_put_confirmed(f"targ{self.parent.axis_Id}={target_val:.4f}")
self.controller.socket_put_and_receive(f"XQ#POSE,{self.parent.axis_Id_numeric}")
self.controller.socket_put_confirmed(
f"targ{self.parent.axis_Id}={target_val:.4f}"
)
self.controller.socket_put_and_receive(
f"XQ#POSE,{self.parent.axis_Id_numeric}"
)
while self.controller.is_thread_active(0):
time.sleep(0.005)
@ -432,7 +462,9 @@ class GalilMotorIsMoving(GalilSignalRO):
@threadlocked
def _socket_get(self):
if self.parent.axis_Id_numeric == 2:
ret = self.controller.is_axis_moving(self.parent.axis_Id, self.parent.axis_Id_numeric)
ret = self.controller.is_axis_moving(
self.parent.axis_Id, self.parent.axis_Id_numeric
)
return ret
if self.parent.axis_Id_numeric == 4:
# Motion signal from axis 4 is mapped to axis 5
@ -470,8 +502,12 @@ class SGalilMotor(Device, PositionerBase):
kind="hinted",
)
user_setpoint = Cpt(GalilSetpointSignal, signal_name="setpoint")
motor_is_moving = Cpt(GalilMotorIsMoving, signal_name="motor_is_moving", kind="normal")
all_axes_referenced = Cpt(GalilAxesReferenced, signal_name="all_axes_referenced", kind="config")
motor_is_moving = Cpt(
GalilMotorIsMoving, signal_name="motor_is_moving", kind="normal"
)
all_axes_referenced = Cpt(
GalilAxesReferenced, signal_name="all_axes_referenced", kind="config"
)
high_limit_travel = Cpt(Signal, value=0, kind="omitted")
low_limit_travel = Cpt(Signal, value=0, kind="omitted")
@ -652,7 +688,9 @@ class SGalilMotor(Device, PositionerBase):
def axis_Id_numeric(self, val):
if isinstance(val, int):
if val not in [2, 4]:
raise ValueError(f"Numeric value {val} is not supported, it must be either 2 or 4.")
raise ValueError(
f"Numeric value {val} is not supported, it must be either 2 or 4."
)
self._axis_Id_alpha = val
self._axis_Id_numeric = (chr(val + 97)).capitalize()
else:
@ -676,7 +714,11 @@ if __name__ == "__main__":
else:
from ophyd_devices.utils.socket import SocketMock
samx = SGalilMotor("E", name="samx", host="129.129.122.26", port=23, socket_cls=SocketMock)
samy = SGalilMotor("C", name="samy", host="129.129.122.26", port=23, socket_cls=SocketMock)
samx = SGalilMotor(
"E", name="samx", host="129.129.122.26", port=23, socket_cls=SocketMock
)
samy = SGalilMotor(
"C", name="samy", host="129.129.122.26", port=23, socket_cls=SocketMock
)
samx.controller.galil_show_all()