feat(api): Add RobotContext movement commands

api/src/opentrons/hardware_control/api.py

@@ -169,6 +169,16 @@ def _update_door_state(self, door_state: DoorState) -> None:
     def _reset_last_mount(self) -> None:
         self._last_moved_mount = None
 
+    def get_deck_from_machine(
+        self, machine_pos: Dict[Axis, float]
+    ) -> Dict[Axis, float]:
+        return deck_from_machine(
+            machine_pos=machine_pos,
+            attitude=self._robot_calibration.deck_calibration.attitude,
+            offset=top_types.Point(0, 0, 0),
+            robot_type=cast(RobotType, "OT-2 Standard"),
+        )
+
     @classmethod
     async def build_hardware_controller(  # noqa: C901
         cls,
@@ -657,11 +667,8 @@ async def home(self, axes: Optional[List[Axis]] = None) -> None:
         async with self._motion_lock:
             if smoothie_gantry:
                 smoothie_pos.update(await self._backend.home(smoothie_gantry))
-                self._current_position = deck_from_machine(
-                    machine_pos=self._axis_map_from_string_map(smoothie_pos),
-                    attitude=self._robot_calibration.deck_calibration.attitude,
-                    offset=top_types.Point(0, 0, 0),
-                    robot_type=cast(RobotType, "OT-2 Standard"),
+                self._current_position = self.get_deck_from_machine(
+                    self._axis_map_from_string_map(smoothie_pos)
                 )
             for plunger in plungers:
                 await self._do_plunger_home(axis=plunger, acquire_lock=False)
@@ -703,11 +710,8 @@ async def current_position(
         async with self._motion_lock:
             if refresh:
                 smoothie_pos = await self._backend.update_position()
-                self._current_position = deck_from_machine(
-                    machine_pos=self._axis_map_from_string_map(smoothie_pos),
-                    attitude=self._robot_calibration.deck_calibration.attitude,
-                    offset=top_types.Point(0, 0, 0),
-                    robot_type=cast(RobotType, "OT-2 Standard"),
+                self._current_position = self.get_deck_from_machine(
+                    self._axis_map_from_string_map(smoothie_pos)
                 )
             if mount == top_types.Mount.RIGHT:
                 offset = top_types.Point(0, 0, 0)
@@ -938,11 +942,8 @@ async def retract_axis(self, axis: Axis, margin: float = 10) -> None:
 
         async with self._motion_lock:
             smoothie_pos = await self._fast_home(smoothie_ax, margin)
-            self._current_position = deck_from_machine(
-                machine_pos=self._axis_map_from_string_map(smoothie_pos),
-                attitude=self._robot_calibration.deck_calibration.attitude,
-                offset=top_types.Point(0, 0, 0),
-                robot_type=cast(RobotType, "OT-2 Standard"),
+            self._current_position = self.get_deck_from_machine(
+                self._axis_map_from_string_map(smoothie_pos)
             )
 
     # Gantry/frame (i.e. not pipette) config API
@@ -1256,11 +1257,8 @@ async def tip_drop_moves(
                     axes=[ot2_axis_to_string(ax) for ax in move.home_axes],
                     margin=move.home_after_safety_margin,
                 )
-                self._current_position = deck_from_machine(
-                    machine_pos=self._axis_map_from_string_map(smoothie_pos),
-                    attitude=self._robot_calibration.deck_calibration.attitude,
-                    offset=top_types.Point(0, 0, 0),
-                    robot_type=cast(RobotType, "OT-2 Standard"),
+                self._current_position = self.get_deck_from_machine(
+                    self._axis_map_from_string_map(smoothie_pos)
                 )
 
         for shake in spec.shake_moves:

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -929,7 +929,6 @@ async def home_tip_motors(
     def _build_tip_action_group(
         self, origin: float, targets: List[Tuple[float, float]]
     ) -> MoveGroup:
-
         move_targets = [
             MoveTarget.build({Axis.Q: target_pos}, speed)
             for target_pos, speed in targets

api/src/opentrons/hardware_control/motion_utilities.py

@@ -1,4 +1,6 @@
 """Utilities for calculating motion correctly."""
+from logging import getLogger
+
 from functools import lru_cache
 from typing import Callable, Dict, Union, Optional, cast
 from collections import OrderedDict
@@ -11,6 +13,7 @@
 
 from .types import Axis, OT3Mount
 
+log = getLogger(__name__)
 
 # TODO: The offset_for_mount function should be defined with an overload
 # set, as with other functions in this module. Unfortunately, mypy < 0.920
@@ -36,6 +39,19 @@
 # ) -> Point:
 #     ...
 
+EMPTY_ORDERED_DICT = OrderedDict(
+    (
+        (Axis.X, 0.0),
+        (Axis.Y, 0.0),
+        (Axis.Z_L, 0.0),
+        (Axis.Z_R, 0.0),
+        (Axis.Z_G, 0.0),
+        (Axis.P_L, 0.0),
+        (Axis.P_R, 0.0),
+        (Axis.Q, 0.0),
+    )
+)
+
 
 @lru_cache(4)
 def offset_for_mount(
@@ -68,6 +84,7 @@ def target_position_from_absolute(
     )
     primary_cp = get_critical_point(mount)
     primary_z = Axis.by_mount(mount)
+
     target_position = OrderedDict(
         (
             (Axis.X, abs_position.x - offset.x - primary_cp.x),
@@ -97,6 +114,57 @@ def target_position_from_relative(
     return target_position
 
 
+def target_axis_map_from_absolute(
+    primary_mount: Union[OT3Mount, Mount],
+    axis_map: Dict[Axis, float],
+    get_critical_point: Callable[[Union[Mount, OT3Mount]], Point],
+    left_mount_offset: Point,
+    right_mount_offset: Point,
+    gripper_mount_offset: Optional[Point] = None,
+) -> "OrderedDict[Axis, float]":
+    """Create an absolute target position for all specified machine axes."""
+    keys_for_target_position = list(axis_map.keys())
+
+    offset = offset_for_mount(
+        primary_mount, left_mount_offset, right_mount_offset, gripper_mount_offset
+    )
+    primary_cp = get_critical_point(primary_mount)
+    primary_z = Axis.by_mount(primary_mount)
+    target_position = OrderedDict()
+
+    if Axis.X in keys_for_target_position:
+        target_position[Axis.X] = axis_map[Axis.X] - offset.x - primary_cp.x
+    if Axis.Y in keys_for_target_position:
+        target_position[Axis.Y] = axis_map[Axis.Y] - offset.y - primary_cp.y
+    if primary_z in keys_for_target_position:
+        # Since this function is intended to be used in conjunction with `API.move_axes`
+        # we must leave out the carriage offset subtraction from the target position as
+        # `move_axes` already does this calculation.
+        target_position[primary_z] = axis_map[primary_z] - primary_cp.z
+
+    target_position.update(
+        {ax: val for ax, val in axis_map.items() if ax not in Axis.gantry_axes()}
+    )
+    return target_position
+
+
+def target_axis_map_from_relative(
+    axis_map: Dict[Axis, float],
+    current_position: Dict[Axis, float],
+) -> "OrderedDict[Axis, float]":
+    """Create a target position for all specified machine axes."""
+    target_position = OrderedDict(
+        (
+            (ax, current_position[ax] + axis_map[ax])
+            for ax in EMPTY_ORDERED_DICT.keys()
+            if ax in axis_map.keys()
+        )
+    )
+    log.info(f"Current position {current_position} and axis map delta {axis_map}")
+    log.info(f"Relative move target {target_position}")
+    return target_position
+
+
 def target_position_from_plunger(
     mount: Union[Mount, OT3Mount],
     delta: float,

api/src/opentrons/hardware_control/ot3api.py

@@ -352,7 +352,9 @@ def _update_estop_state(self, event: HardwareEvent) -> "List[Future[None]]":
     def _reset_last_mount(self) -> None:
         self._last_moved_mount = None
 
-    def _deck_from_machine(self, machine_pos: Dict[Axis, float]) -> Dict[Axis, float]:
+    def get_deck_from_machine(
+        self, machine_pos: Dict[Axis, float]
+    ) -> Dict[Axis, float]:
         return deck_from_machine(
             machine_pos=machine_pos,
             attitude=self._robot_calibration.deck_calibration.attitude,
@@ -1020,14 +1022,14 @@ async def _refresh_jaw_state(self) -> None:
 
     async def _cache_current_position(self) -> Dict[Axis, float]:
         """Cache current position from backend and return in absolute deck coords."""
-        self._current_position = self._deck_from_machine(
+        self._current_position = self.get_deck_from_machine(
             await self._backend.update_position()
         )
         return self._current_position
 
     async def _cache_encoder_position(self) -> Dict[Axis, float]:
         """Cache encoder position from backend and return in absolute deck coords."""
-        self._encoder_position = self._deck_from_machine(
+        self._encoder_position = self.get_deck_from_machine(
             await self._backend.update_encoder_position()
         )
         if self.has_gripper():
@@ -1227,7 +1229,9 @@ async def move_axes(  # noqa: C901
                     message=f"{axis} is not present", detail={"axis": str(axis)}
                 )
 
+        self._log.info(f"Attempting to move {position} with speed {speed}.")
         if not self._backend.check_encoder_status(list(position.keys())):
+            self._log.info("Calling home in move_axes")
             await self.home()
         self._assert_motor_ok(list(position.keys()))
 
@@ -2560,7 +2564,7 @@ def get_instrument_max_height(
         mount: Union[top_types.Mount, OT3Mount],
         critical_point: Optional[CriticalPoint] = None,
     ) -> float:
-        carriage_pos = self._deck_from_machine(self._backend.home_position())
+        carriage_pos = self.get_deck_from_machine(self._backend.home_position())
         pos_at_home = self._effector_pos_from_carriage_pos(
             OT3Mount.from_mount(mount), carriage_pos, critical_point
         )
@@ -2662,7 +2666,7 @@ async def _liquid_probe_pass(
         )
         machine_pos = await self._backend.update_position()
         machine_pos[Axis.by_mount(mount)] = end_z
-        deck_end_z = self._deck_from_machine(machine_pos)[Axis.by_mount(mount)]
+        deck_end_z = self.get_deck_from_machine(machine_pos)[Axis.by_mount(mount)]
         offset = offset_for_mount(
             mount,
             top_types.Point(*self._config.left_mount_offset),

api/src/opentrons/hardware_control/protocols/liquid_handler.py

@@ -1,6 +1,8 @@
 from typing import Optional
 from typing_extensions import Protocol
 
+from opentrons.types import Point
+from opentrons.hardware_control.types import CriticalPoint
 from .types import MountArgType, CalibrationType, ConfigType
 
 from .instrument_configurer import InstrumentConfigurer
@@ -16,6 +18,22 @@ class LiquidHandler(
     Calibratable[CalibrationType],
     Protocol[CalibrationType, MountArgType, ConfigType],
 ):
+    def critical_point_for(
+        self,
+        mount: MountArgType,
+        cp_override: Optional[CriticalPoint] = None,
+    ) -> Point:
+        """
+        Determine the current critical point for the specified mount.
+
+        :param mount: A robot mount that the instrument is on.
+        :param cp_override: The critical point override to use.
+
+        If no critical point override is specified, the robot defaults to nozzle location `A1` or the mount critical point.
+        :return: Point.
+        """
+        ...
+
     async def update_nozzle_configuration_for_mount(
         self,
         mount: MountArgType,

api/src/opentrons/hardware_control/protocols/motion_controller.py

@@ -9,6 +9,12 @@
 class MotionController(Protocol[MountArgType]):
     """Protocol specifying fundamental motion controls."""
 
+    def get_deck_from_machine(
+        self, machine_pos: Dict[Axis, float]
+    ) -> Dict[Axis, float]:
+        """Convert machine coordinates to deck coordinates."""
+        ...
+
     async def halt(self, disengage_before_stopping: bool = False) -> None:
         """Immediately stop motion.
 

api/src/opentrons/protocol_api/core/common.py

@@ -14,6 +14,7 @@
 )
 from .protocol import AbstractProtocol
 from .well import AbstractWellCore
+from .robot import AbstractRobot
 
 
 WellCore = AbstractWellCore
@@ -26,4 +27,5 @@
 HeaterShakerCore = AbstractHeaterShakerCore
 MagneticBlockCore = AbstractMagneticBlockCore
 AbsorbanceReaderCore = AbstractAbsorbanceReaderCore
+RobotCore = AbstractRobot
 ProtocolCore = AbstractProtocol[InstrumentCore, LabwareCore, ModuleCore]

api/src/opentrons/protocol_api/core/engine/protocol.py

@@ -63,6 +63,7 @@
 from ..labware import LabwareLoadParams
 from .labware import LabwareCore
 from .instrument import InstrumentCore
+from .robot import RobotCore
 from .module_core import (
     ModuleCore,
     TemperatureModuleCore,
@@ -82,7 +83,9 @@
 
 class ProtocolCore(
     AbstractProtocol[
-        InstrumentCore, LabwareCore, Union[ModuleCore, NonConnectedModuleCore]
+        InstrumentCore,
+        LabwareCore,
+        Union[ModuleCore, NonConnectedModuleCore],
     ]
 ):
     """Protocol API core using a ProtocolEngine.
@@ -531,6 +534,12 @@ def _get_module_core(
                 load_module_result=load_module_result, model=model
             )
 
+    def load_robot(self) -> RobotCore:
+        """Load a robot core into the RobotContext."""
+        return RobotCore(
+            engine_client=self._engine_client, sync_hardware_api=self._sync_hardware
+        )
+
     def load_instrument(
         self,
         instrument_name: PipetteNameType,