Merge pull request #182 from yannachen/robot_sample

Robot transfer sample

src/conftest.py

@@ -30,6 +30,7 @@
 from haven.instrument.dxp import DxpDetector
 from haven.instrument.dxp import add_mcas as add_dxp_mcas
 from haven.instrument.ion_chamber import IonChamber
+from haven.instrument.robot import Robot
 from haven.instrument.shutter import Shutter
 from haven.instrument.slits import ApertureSlits, BladeSlits
 from haven.instrument.xspress import Xspress3Detector
@@ -264,6 +265,13 @@ def aerotech():
     return stage
 
 
+@pytest.fixture()
+def robot():
+    RobotClass = make_fake_device(Robot)
+    robot = RobotClass(name="robotA", prefix="255idA:")
+    return robot
+
+
 @pytest.fixture()
 def aerotech_flyer(aerotech):
     flyer = aerotech.horiz

src/haven/__init__.py

@@ -39,6 +39,7 @@
 from .plans.mono_gap_calibration import calibrate_mono_gap  # noqa: F401
 from .plans.mono_ID_calibration import mono_ID_calibration  # noqa: F401
 from .plans.record_dark_current import record_dark_current  # noqa: F401
+from .plans.robot_transfer_sample import robot_transfer_sample  # noqa: F401
 from .plans.set_energy import set_energy  # noqa: F401
 from .plans.shutters import close_shutters, open_shutters  # noqa: F401
 from .plans.xafs_scan import xafs_scan  # noqa: F401

src/haven/instrument/aerotech.py

@@ -483,7 +483,8 @@ def send_command(self, cmd: str):
         return status
 
     def disable_pso(self):
-        self.send_command(f"PSOCONTROL {self.axis} OFF")
+        for axis in range(2):
+            self.send_command(f"PSOCONTROL @{axis} OFF")
 
     def check_flyscan_bounds(self):
         """Check that the fly-scan params are sane at the scan start and end.

src/haven/instrument/heater.py

@@ -4,7 +4,7 @@
 from apstools.devices import PTC10AioChannel as PTC10AioChannelBase
 from apstools.devices import PTC10PositionerMixin, PTC10TcChannel
 from ophyd import Component as Cpt
-from ophyd import EpicsSignalRO, EpicsSignalWithRBV, PVPositioner, EpicsSignal
+from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, PVPositioner
 
 from .._iconfig import load_config
 from .device import aload_devices, make_device

src/haven/instrument/robot.py

@@ -12,11 +12,55 @@
 log = logging.getLogger(__name__)
 
 
+LOAD_TIMEOUT = 40
+
+
 class Sample(Device):
+    """An individual robot sample that can be loaded.
+
+    Signals
+    =======
+    present
+      Whether or not a sample is physically present on the stage.
+    empty
+      Whether or not no sample is physically present on the stage.
+    load
+      Direct the robot to physically move this sample to the loading
+      position. Can be slow (~25-30 seconds).
+    unload
+      Direct the robot to physically remove this sample to the loading
+      position. Can be slow (~25-30 seconds).
+    x
+      The x position of the robot in Cartesian coordinates.
+    y
+      The y position of the robot in Cartesian coordinates.
+    z
+      The z position of the robot in Cartesian coordinates.
+    rx
+      The rx position of the robot in Cartesian coordinates.
+    ry
+      The ry position of the robot in Cartesian coordinates.
+    rz
+      The position of the robot in Cartesian coordinates.
+
+    """
+
     present = Cpt(EpicsSignalRO, ":present")
     empty = Cpt(EpicsSignalRO, ":empty")
-    load = Cpt(EpicsSignal, ":load", kind="omitted")
-    unload = Cpt(EpicsSignal, ":unload", kind="omitted")
+    load = Cpt(
+        EpicsSignal,
+        ":load",
+        kind="omitted",
+        write_timeout=LOAD_TIMEOUT,
+        put_complete=True,
+    )
+    unload = Cpt(
+        EpicsSignal,
+        ":unload",
+        kind="omitted",
+        write_timeout=LOAD_TIMEOUT,
+        put_complete=True,
+    )
     x = Cpt(EpicsSignalRO, ":x")
     y = Cpt(EpicsSignalRO, ":y")
     z = Cpt(EpicsSignalRO, ":z")
@@ -94,7 +138,13 @@ class Robot(Device):
 
     # sample transfer
     current_sample = Cpt(EpicsSignalRO, ":current_sample", kind="config")
-    unload_current_sample = Cpt(EpicsSignal, ":unload_current_sample", kind="omitted")
+    unload_current_sample = Cpt(
+        EpicsSignal,
+        ":unload_current_sample",
+        kind="omitted",
+        write_timeout=LOAD_TIMEOUT,
+        put_complete=True,
+    )
     current_sample_reset = Cpt(EpicsSignal, ":current_sample_reset", kind="omitted")
     home = Cpt(EpicsSignal, ":home", kind="config")
     cal_stage = Cpt(EpicsSignal, ":cal_stage", kind="config")

src/haven/plans/robot_transfer_sample.py

@@ -0,0 +1,105 @@
+import logging
+
+from bluesky import plan_stubs as bps
+
+from ..instrument.instrument_registry import registry
+from ..motor_position import rbv
+
+log = logging.getLogger(__name__)
+
+
+__all__ = ["robot_transfer_sample"]
+
+
+ON = 1
+
+
+def robot_transfer_sample(robot, sampleN, *args):
+    """
+    Use robot to load sampleN at a fixed Aerotech stage position or any
+    motors.
+
+    e.g.
+    Load sample:
+    robot_sample(robot, 9, motor1, 100, motor2, 200, motor3, 50)
+
+    Unload sample:
+    robot_sample(robot, None, motor1, 100, motor2, 200, motor3, 50)
+
+    Parameters:
+    - robot: robot device
+    - sampleN: Sample number
+    - *args: multiple pairs of motor and pos
+    """
+
+    robot = registry.find(robot)
+
+    # Check if power is on
+    # if robot.power_rbv.get() == Off:
+    #    raise ValueError("Robot is NOT powered. Please turn on the power.")
+
+    # Check if remote control is on
+    # if not robot.remote_control.get() == Off:
+    #    raise ValueError(
+    #        "Robot is NOT in reomete control mode. Please click on Pad for remote control."
+    #    )
+
+    # Check if gripper is on
+    # if robot.gripper_activated.get() == Off:
+    #    raise ValueError("Gripper is not activated. Please activate the gripper.")
+
+    # Check if robot is running
+    # if robot.grogram_running.get() == On:
+    #   raise ValueError("Robot is running now.")
+
+    # Find motors
+    motor_list = [registry.find(motor) for motor in args[::2]]
+    new_positions = [pos for pos in args[1::2]]
+    # Record the motor positions before load sampls
+    initial_positions = [rbv(motor) for motor in motor_list]
+
+    # Move the aerotech to the loading position
+    for motor, pos in zip(motor_list, new_positions):
+        yield from bps.mv(motor, pos)
+
+    if sampleN == None:
+        # Unload sample
+        yield from bps.mv(robot.unload_current_sample, ON)
+
+    else:
+        # Load sampleN after all the other motor arrive at (pos1, pos2, pos3...)
+        sample = getattr(
+            robot.samples, f"sample{sampleN}"
+        )  # Access the Sample device corresponding to sampleN
+        print(sample.load)
+        yield from bps.mv(sample.load, ON)  # Assuming '1' initiates the loading action
+
+    # Return to the initial position
+    for motor, pos in zip(motor_list[-1::-1], initial_positions[-1::-1]):
+        yield from bps.mv(motor, pos)
+
+
+# -----------------------------------------------------------------------------
+# :author:    Yanna Chen
+# :email:     [email protected]
+# :copyright: Copyright © 2024, UChicago Argonne, LLC
+#
+# Distributed under the terms of the 3-Clause BSD License
+#
+# The full license is in the file LICENSE, distributed with this software.
+#
+# DISCLAIMER
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+#
+# -----------------------------------------------------------------------------

src/haven/tests/test_robot_transfer_sample.py

@@ -0,0 +1,43 @@
+from ophyd.sim import motor1, motor2, motor3
+
+from haven import robot_transfer_sample
+
+
+def test_robot_sample(robot):
+
+    plan = robot_transfer_sample(robot, 9, motor1, 100, motor2, 200, motor3, 50)
+
+    msgs = list(plan)
+
+    assert robot.name == "robotA"
+    assert len(msgs) == 14
+
+    unload = robot_transfer_sample(robot, None, motor1, 100)
+    msgs = list(unload)
+    assert len(msgs) == 6
+
+
+# -----------------------------------------------------------------------------
+# :author:    Yanna Chen
+# :email:     [email protected]
+# :copyright: Copyright © 2024, UChicago Argonne, LLC
+#
+# Distributed under the terms of the 3-Clause BSD License
+#
+# The full license is in the file LICENSE, distributed with this software.
+#
+# DISCLAIMER
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+#
+# -----------------------------------------------------------------------------