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

fix: fix ddg code

Browse Source
This commit is contained in:
appel_c 2023-09-04 20:52:05 +02:00
parent ba9cb77ed9
commit b3237ceda5
1 changed files with 20 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -61,9 +61,7 @@ class DummyPositioner(PVPositioner):
setpoint = Component(EpicsSignal, "DelayAO", put_complete=True, kind=Kind.config) setpoint = Component(EpicsSignal, "DelayAO", put_complete=True, kind=Kind.config)
readback = Component(EpicsSignalRO, "DelayAI", kind=Kind.config) readback = Component(EpicsSignalRO, "DelayAI", kind=Kind.config)
done = Component(Signal, value=1) done = Component(Signal, value=1)
reference = Component( reference = Component(EpicsSignal, "ReferenceMO", put_complete=True, kind=Kind.config)
EpicsSignal, "ReferenceMO", put_complete=True, kind=Kind.config
)
class DelayPair(PseudoPositioner): class DelayPair(PseudoPositioner):
@ -93,16 +91,12 @@ class DelayPair(PseudoPositioner):
@pseudo_position_argument @pseudo_position_argument
def forward(self, pseudo_pos): def forward(self, pseudo_pos):
"""Run a forward (pseudo -> real) calculation""" """Run a forward (pseudo -> real) calculation"""
return self.RealPosition( return self.RealPosition(ch1=pseudo_pos.delay, ch2=pseudo_pos.delay + pseudo_pos.width)
ch1=pseudo_pos.delay, ch2=pseudo_pos.delay + pseudo_pos.width
)
@real_position_argument @real_position_argument
def inverse(self, real_pos): def inverse(self, real_pos):
"""Run an inverse (real -> pseudo) calculation""" """Run an inverse (real -> pseudo) calculation"""
return self.PseudoPosition( return self.PseudoPosition(delay=real_pos.ch1, width=real_pos.ch2 - real_pos.ch1)
delay=real_pos.ch1, width=real_pos.ch2 - real_pos.ch1
)
class TriggerSource(int, enum.Enum): class TriggerSource(int, enum.Enum):
@ -147,12 +141,8 @@ class DelayGeneratorDG645(Device):
"reload_config", "reload_config",
] ]
trigger_burst_readout = Component( trigger_burst_readout = Component(EpicsSignal, "EventStatusLI.PROC", name="read_burst_state")
EpicsSignal, "EventStatusLI.PROC", name="read_burst_state" burst_cycle_finished = Component(EpicsSignalRO, "EventStatusMBBID.B3", name="read_burst_state")
)
burst_cycle_finished = Component(
EpicsSignalRO, "EventStatusMBBID.B3", name="read_burst_state"
)
status = Component(EpicsSignalRO, "StatusSI", name="status") status = Component(EpicsSignalRO, "StatusSI", name="status")
clear_error = Component(EpicsSignal, "StatusClearBO", name="clear_error") clear_error = Component(EpicsSignal, "StatusClearBO", name="clear_error")
@ -199,9 +189,7 @@ class DelayGeneratorDG645(Device):
name="trigger_rate", name="trigger_rate",
kind=Kind.config, kind=Kind.config,
) )
trigger_shot = Component( trigger_shot = Component(EpicsSignal, "TriggerDelayBO", name="trigger_shot", kind="config")
EpicsSignal, "TriggerDelayBO", name="trigger_shot", kind="config"
)
# Burst mode # Burst mode
burstMode = Component( burstMode = Component(
EpicsSignal, EpicsSignal,
@ -358,10 +346,7 @@ class DelayGeneratorDG645(Device):
f"{name}_set_trigger_source": "SINGLE_SHOT", f"{name}_set_trigger_source": "SINGLE_SHOT",
} }
if ddg_config is not None: if ddg_config is not None:
[ [self.ddg_config.update({f"{name}_{key}": value}) for key, value in ddg_config.items()]
self.ddg_config.update({f"{name}_{key}": value})
for key, value in ddg_config.items()
]
super().__init__( super().__init__(
prefix=prefix, prefix=prefix,
name=name, name=name,
@ -372,9 +357,7 @@ class DelayGeneratorDG645(Device):
**kwargs, **kwargs,
) )
if device_manager is None and not sim_mode: if device_manager is None and not sim_mode:
raise DDGError( raise DDGError("Add DeviceManager to initialization or init with sim_mode=True")
"Add DeviceManager to initialization or init with sim_mode=True"
)
self.device_manager = device_manager self.device_manager = device_manager
if not sim_mode: if not sim_mode:
self._producer = self.device_manager.producer self._producer = self.device_manager.producer
@ -447,10 +430,7 @@ class DelayGeneratorDG645(Device):
self.set_channels( self.set_channels(
"reference", "reference",
0, 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)): for ii in range(len(self._all_delay_pairs)):
self.set_channels( self.set_channels(
@ -463,15 +443,12 @@ class DelayGeneratorDG645(Device):
self.level.set(self.thres_trig_level.get()) self.level.set(self.thres_trig_level.get())
def _check_burst_cycle(self) -> None: def _check_burst_cycle(self) -> None:
"''Checks burst cycle of delay generator''" """Checks burst cycle of delay generator
Force readout, return value from end of burst cycle
"""
while True: while True:
self.trigger_burst_readout.set(1) self.trigger_burst_readout.set(1)
if ( if self.burst_cycle_finished.read()[self.burst_cycle_finished.name]["value"] == 1:
self.burst_cycle_finished.read()[self.burst_cycle_finished.name][
"value"
]
== 1
):
self._acquisition_done = True self._acquisition_done = True
return return
if self._stopped == True: if self._stopped == True:
@ -497,9 +474,7 @@ class DelayGeneratorDG645(Device):
) )
total_exposure = exp_time total_exposure = exp_time
delay_burst = self.delay_burst.get() delay_burst = self.delay_burst.get()
self.burst_enable( self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0) self.set_channels("delay", 0)
# Set burst length to half of the experimental time! # Set burst length to half of the experimental time!
self.set_channels("width", exp_time) self.set_channels("width", exp_time)
@ -507,13 +482,9 @@ class DelayGeneratorDG645(Device):
exp_time = self.delta_width.get() + self.scaninfo.exp_time exp_time = self.delta_width.get() + self.scaninfo.exp_time
total_exposure = exp_time + self.scaninfo.readout_time total_exposure = exp_time + self.scaninfo.readout_time
delay_burst = self.delay_burst.get() delay_burst = self.delay_burst.get()
num_burst_cycle = ( num_burst_cycle = self.scaninfo.num_frames + self.additional_triggers.get()
self.scaninfo.num_frames + self.additional_triggers.get()
)
# set parameters in DDG # set parameters in DDG
self.burst_enable( self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0) self.set_channels("delay", 0)
# Set burst length to half of the experimental time! # Set burst length to half of the experimental time!
self.set_channels("width", exp_time) self.set_channels("width", exp_time)
@ -532,9 +503,7 @@ class DelayGeneratorDG645(Device):
# self.additional_triggers should be 0 for self.set_high_on_exposure or remove here fully.. # self.additional_triggers should be 0 for self.set_high_on_exposure or remove here fully..
num_burst_cycle = 1 + self.additional_triggers.get() num_burst_cycle = 1 + self.additional_triggers.get()
# set parameters in DDG # set parameters in DDG
self.burst_enable( self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0.0) self.set_channels("delay", 0.0)
# Set burst length to half of the experimental time! # Set burst length to half of the experimental time!
self.set_channels("width", exp_time) self.set_channels("width", exp_time)
@ -544,13 +513,9 @@ class DelayGeneratorDG645(Device):
exp_time = self.delta_width.get() + self.scaninfo.exp_time exp_time = self.delta_width.get() + self.scaninfo.exp_time
total_exposure = exp_time + self.scaninfo.readout_time total_exposure = exp_time + self.scaninfo.readout_time
delay_burst = self.delay_burst.get() delay_burst = self.delay_burst.get()
num_burst_cycle = ( num_burst_cycle = self.scaninfo.num_frames + self.additional_triggers.get()
self.scaninfo.num_frames + self.additional_triggers.get()
)
# set parameters in DDG # set parameters in DDG
self.burst_enable( self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
num_burst_cycle, delay_burst, total_exposure, config="first"
)
self.set_channels("delay", 0.0) self.set_channels("delay", 0.0)
# Set burst length to half of the experimental time! # Set burst length to half of the experimental time!
self.set_channels("width", exp_time) self.set_channels("width", exp_time)
@ -575,9 +540,7 @@ class DelayGeneratorDG645(Device):
def trigger(self) -> None: def trigger(self) -> None:
# if self.scaninfo.scan_type == "step": # if self.scaninfo.scan_type == "step":
if self.source.read()[self.source.name]["value"] == int( if self.source.read()[self.source.name]["value"] == int(TriggerSource.SINGLE_SHOT):
TriggerSource.SINGLE_SHOT
):
self.trigger_shot.set(1).wait() self.trigger_shot.set(1).wait()
super().trigger() super().trigger()