|
7 | 7 | from bec_lib.endpoints import MessageEndpoints |
8 | 8 | from bec_lib.logger import bec_logger |
9 | 9 | from ophyd import Component as Cpt |
10 | | -from ophyd import Device, DeviceStatus, Kind, OphydObject, PositionerBase, Staged |
| 10 | +from ophyd import Device, DeviceStatus, Kind, OphydObject, PositionerBase, Signal, Staged |
11 | 11 |
|
| 12 | +from ophyd_devices.interfaces.base_classes.psi_pseudo_motor_base import PSIPseudoMotorBase |
12 | 13 | from ophyd_devices.sim.sim_camera import SimCamera |
13 | 14 | from ophyd_devices.sim.sim_positioner import SimPositioner |
14 | 15 | from ophyd_devices.sim.sim_signals import SetableSignal |
| 16 | +from ophyd_devices.utils.bec_processed_signal import BECProcessedSignal, ProcessedSignalModel |
15 | 17 | from ophyd_devices.utils.bec_signals import ( |
16 | 18 | AsyncSignal, |
17 | 19 | DynamicSignal, |
@@ -424,6 +426,49 @@ def complete_cam(): |
424 | 426 | return status |
425 | 427 |
|
426 | 428 |
|
| 429 | +class VirtualSlitCenter(PSIPseudoMotorBase): |
| 430 | + |
| 431 | + left_edge = Cpt(SimPositioner, name="left_edge", kind=Kind.normal) |
| 432 | + right_edge = Cpt(SimPositioner, name="right_edge", kind=Kind.normal) |
| 433 | + |
| 434 | + def __init__(self, name, device_manager, **kwargs): |
| 435 | + super().__init__(name, device_manager, **kwargs) |
| 436 | + positioners = {"left_edge": self.left_edge, "right_edge": self.right_edge} |
| 437 | + self.set_positioner_objects(positioners) |
| 438 | + |
| 439 | + def forward_calculation(self, left_edge: Signal, right_edge: Signal) -> float: |
| 440 | + return float((left_edge.get() + right_edge.get()) / 2) |
| 441 | + |
| 442 | + def inverse_calculation( |
| 443 | + self, position, left_edge: Signal, right_edge: Signal |
| 444 | + ) -> dict[str, float]: |
| 445 | + left_pos = left_edge.root.readback.get() |
| 446 | + right_pos = right_edge.root.readback.get() |
| 447 | + width = right_pos - left_pos |
| 448 | + new_right_pos = position + width / 2 |
| 449 | + new_left_pos = position - width / 2 |
| 450 | + return {"left_edge": new_left_pos, "right_edge": new_right_pos} |
| 451 | + |
| 452 | + def motors_are_moving(self, left_edge: Signal, right_edge: Signal) -> int: |
| 453 | + left_moving = left_edge.get() |
| 454 | + right_moving = right_edge.get() |
| 455 | + return int(left_moving or right_moving) |
| 456 | + |
| 457 | + |
| 458 | +class BECPseudoSignal(BECProcessedSignal): |
| 459 | + |
| 460 | + def __init__(self, name, signal_1: str, signal_2: str, device_manager=None, **kwargs): |
| 461 | + model = ProcessedSignalModel( |
| 462 | + devices={"signal_1": signal_1, "signal_2": signal_2}, |
| 463 | + compute_method=self.compute, |
| 464 | + return_type=float, |
| 465 | + ) |
| 466 | + super().__init__(name, model=model, device_manager=device_manager, **kwargs) |
| 467 | + |
| 468 | + def compute(self, signal_1: Signal, signal_2: Signal) -> float: |
| 469 | + return float(signal_1.get() + signal_2.get()) |
| 470 | + |
| 471 | + |
427 | 472 | if __name__ == "__main__": |
428 | 473 | cam = SimCameraWithPSIComponents(name="cam") |
429 | 474 | cam.read() |
0 commit comments