updated unit test args for tip_action

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

@@ -40,7 +40,6 @@
  create_gripper_jaw_home_group,
  create_gripper_jaw_hold_group,
  create_tip_action_group,
- PipetteAction,
  create_gear_motor_home_group,
  motor_nodes,
  LIMIT_SWITCH_OVERTRAVEL_DISTANCE,
@@ -260,7 +259,7 @@ def __init__(
  self._estop_detector: Optional[EstopDetector] = None
  self._estop_state_machine = EstopStateMachine(detector=None)
  self._position = self._get_home_position()
- self._gear_motor_position: float
+ self._gear_motor_position: Dict[NodeId, float] = {}
  self._encoder_position = self._get_home_position()
  self._motor_status = {}
  self._check_updates = check_updates
@@ -333,7 +332,7 @@ def _build_system_hardware(
  )
 
  @property
- def gear_motor_position(self) -> float:
+ def gear_motor_position(self) -> Dict[NodeId, float]:
  return self._gear_motor_position
 
  def _motor_nodes(self) -> Set[NodeId]:
@@ -362,7 +361,14 @@ async def update_motor_status(self) -> None:
  response = await get_motor_position(self._messenger, motor_nodes)
  self._handle_motor_status_response(response)
 
- async def update_motor_estimation(self, axes: Sequence[Axis]) -> None:
+ async def update_gear_motor_position(self) -> None:
+ motor_response = await update_gear_motor_position_estimation(self._messenger)
+ updated_successfully = motor_response[1]
+ # update position only if updated successfully by the firmware
+ if updated_successfully:
+ self._gear_motor_position = {NodeId.pipette_left: motor_response[0]}
+
+ async def update_motor_estimation(self, axes: Sequence[OT3Axis]) -> None:
  """Update motor position estimation for commanded nodes, and update cache of data."""
  nodes = set([axis_to_node(a) for a in axes])
  response = await update_motor_position_estimation(self._messenger, nodes)
@@ -623,34 +629,31 @@ def _filter_move_group(self, move_group: MoveGroup) -> MoveGroup:
  )
  return new_group
 
- async def home_gear_motors(
- self,
- distance: float,
- velocity: float,
- ) -> None:
- move_group = create_gear_motor_home_group(distance, velocity)
- runner = MoveGroupRunner(
- move_groups=[move_group],
- ignore_stalls=True if not ff.stall_detection_enabled() else False,
- )
- positions = await runner.run(can_messenger=self._messenger)
- if NodeId.pipette_left in positions:
- self._gear_motor_position = positions[NodeId.pipette_left][0]
-
  async def tip_action(
  self,
- moves: List[Move[OT3Axis]],
+ moves: Optional[List[Move[OT3Axis]]] = None,
+ distance: Optional[float] = None,
+ velocity: Optional[float] = None,
  tip_action: str = "home",
  ) -> None:
- move_group = create_tip_action_group(moves, [NodeId.pipette_left], tip_action)
+ move_group = []
+ if tip_action == "clamp":
+ assert moves is not None
+ move_group = create_tip_action_group(
+ moves, [NodeId.pipette_left], tip_action
+ )
+ elif tip_action == "home":
+ assert distance is not None and velocity is not None
+ move_group = create_gear_motor_home_group(float(distance), float(velocity))
  runner = MoveGroupRunner(
  move_groups=[move_group],
  ignore_stalls=True if not ff.stall_detection_enabled() else False,
  )
  positions = await runner.run(can_messenger=self._messenger)
  if NodeId.pipette_left in positions:
- self._gear_motor_position = positions[NodeId.pipette_left][0]
- print(f"gear position is now {self._gear_motor_position}")
+ self._gear_motor_position = {
+ NodeId.pipette_left: positions[NodeId.pipette_left][0]
+ }
 
  @requires_update
  async def gripper_grip_jaw(

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

@@ -29,8 +29,6 @@
  create_gripper_jaw_hold_group,
  create_gripper_jaw_grip_group,
  create_gripper_jaw_home_group,
- create_tip_action_group,
- PipetteAction,
  NODEID_SUBSYSTEM,
  motor_nodes,
  target_to_subsystem,
@@ -139,7 +137,7 @@ def __init__(
  self._initialized = False
  self._lights = {"button": False, "rails": False}
  self._estop_state_machine = EstopStateMachine(detector=None)
- self._gear_motor_position = 0
+ self._gear_motor_position: Dict[NodeId, float] = {}
 
  def _sanitize_attached_instrument(
  mount: OT3Mount, passed_ai: Optional[Dict[str, Optional[str]]] = None
@@ -252,7 +250,19 @@ async def update_motor_status(self) -> None:
  )
 
  @ensure_yield
- async def update_motor_estimation(self, axes: Sequence[Axis]) -> None:
+ async def update_gear_motor_position(self) -> None:
+ return None
+
+ @ensure_yield
+ async def home_gear_motors(
+ self,
+ distance: float,
+ velocity: float,
+ ) -> None:
+ return None
+
+ @ensure_yield
+ async def update_motor_estimation(self, axes: Sequence[OT3Axis]) -> None:
  """Update motor position estimation for commanded nodes, and update cache of data."""
  # Simulate conditions as if there are no stalls, aka do nothing
  return None
@@ -379,7 +389,9 @@ async def get_tip_present(self, mount: OT3Mount, tip_state: TipStateType) -> Non
 
  async def tip_action(
  self,
- moves: List[Move[OT3Axis]],
+ moves: Optional[List[Move[OT3Axis]]] = None,
+ distance: Optional[float] = None,
+ velocity: Optional[float] = None,
  tip_action: str = "home",
  ) -> None:
  pass

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

@@ -1,5 +1,5 @@
 """Shared utilities for ot3 hardware control."""
-from typing import Dict, Iterable, List, Set, Tuple, TypeVar, Sequence, cast
+from typing import Dict, Iterable, List, Set, Tuple, TypeVar
 from typing_extensions import Literal
 from logging import getLogger
 from opentrons.config.defaults_ot3 import DEFAULT_CALIBRATION_AXIS_MAX_SPEED
@@ -425,7 +425,7 @@ def create_gear_motor_home_group(
  velocity: float,
 ) -> MoveGroup:
  step = create_tip_action_step(
- velocity={NodeId.pipette_left: np.float64(velocity)},
+ velocity={NodeId.pipette_left: np.float64(-1 * velocity)},
  acceleration={NodeId.pipette_left: np.float64(0)},
  duration=np.float64(distance / velocity),
  present_nodes=[NodeId.pipette_left],

api/src/opentrons/hardware_control/ot3api.py

@@ -568,6 +568,8 @@ async def cache_instruments(
  )
  if not skip_configure:
  await self._configure_instruments()
+ if self._gantry_load == GantryLoad.HIGH_THROUGHPUT:
+ await self._backend.update_gear_motor_position()
 
  async def _cache_instruments( # noqa: C901
  self, require: Optional[Dict[top_types.Mount, PipetteName]] = None
@@ -731,28 +733,34 @@ async def home_z(
  await self.home(axes)
 
  async def home_gear_motors(self) -> None:
+ position_estimation_dict = axis_convert(self._backend.gear_motor_position, 0.0)
+ position_estimation = position_estimation_dict[OT3Axis.P_L]
  gear_motor_axis_bound = self._backend.axis_bounds[OT3Axis.Q][1]
- position_estimation = self._backend.gear_motor_position
  velocity = self._config.motion_settings.default_max_speed[
  GantryLoad.HIGH_THROUGHPUT
  ][OT3AxisKind.Q]
- if position_estimation is not None:
+ if position_estimation is not None and position_estimation > 5.0:
  position_valid = abs(position_estimation) < gear_motor_axis_bound
  if position_valid:
  position_dict = {OT3Axis.Q: position_estimation}
  safe_home_target = {OT3Axis.Q: self._config.safe_home_distance}
  moves = self._build_moves(position_dict, safe_home_target)
 
- await self._backend.tip_action(moves[0], "clamp")
- distance = self._backend.gear_motor_position
+ await self._backend.tip_action(moves=moves[0], tip_action="clamp")
 
- await self._backend.home_gear_motors(distance, velocity)
+ distance = axis_convert(self._backend.gear_motor_position, 0.0)[
+ OT3Axis.P_L
+ ]
+ await self._backend.tip_action(
+ distance=distance, velocity=velocity, tip_action="home"
+ )
  return
  # if gear motor position is not known:
  # start at the axis boundary, and move without acceleration
  # to avoid crashing if the gear motors' positions are not available
- axis_bound = self._backend.axis_bounds[OT3Axis.Q][1]
- await self._backend.home_gear_motors(axis_bound, velocity)
+ await self._backend.tip_action(
+ distance=gear_motor_axis_bound, velocity=velocity, tip_action="home"
+ )
 
  async def home_gripper_jaw(self) -> None:
  """
@@ -1272,7 +1280,6 @@ async def _home(self, axes: Sequence[Axis]) -> None:
  await self.home_gripper_jaw()
  elif axis == OT3Axis.Q:
  await self.home_gear_motors()
- # add homing with acceleration when position is known also
  else:
  await self._home_axis(axis)
  except ZeroLengthMoveError:
@@ -1704,13 +1711,13 @@ async def _motor_pick_up_tip(
  await self._move(target_down)
  # perform pick up tip
 
- gear_motor_origin = await self._backend.gear_motor_position_estimation()
- gear_origin_dict = {OT3Axis.Q: gear_motor_origin[0]}
+ await self._backend.update_gear_motor_position()
+ gear_origin_dict = axis_convert(self._backend.gear_motor_position, 0.0)
  clamp_move_target = pipette_spec.pick_up_distance
  gear_target_dict = {OT3Axis.Q: clamp_move_target}
  clamp_moves = self._build_moves(gear_origin_dict, gear_target_dict)
 
- await self._backend.tip_action(clamp_moves[0], "clamp")
+ await self._backend.tip_action(moves=clamp_moves[0], tip_action="clamp")
 
  await self.home_gear_motors()
 
@@ -1785,21 +1792,21 @@ async def drop_tip(
  # The speed check is needed because speed can sometimes be None.
  # Not sure why
 
- gear_start_position = (
- await self._backend.gear_motor_position_estimation()
+ await self._backend.update_gear_motor_position()
+ gear_start_position = axis_convert(
+ self._backend.gear_motor_position, 0.0
  )
- gear_start_pos_dict = {OT3Axis.Q: gear_start_position[0]}
  drop_target_dict = {OT3Axis.Q: move.target_position}
- drop_moves = self._build_moves(gear_start_pos_dict, drop_target_dict)
+ drop_moves = self._build_moves(gear_start_position, drop_target_dict)
 
- await self._backend.tip_action(drop_moves[0], "clamp")
+ await self._backend.tip_action(moves=drop_moves[0], tip_action="clamp")
 
- current_gear_pos = await self._backend.gear_motor_position_estimation()
- gear_pos_dict = {OT3Axis.Q: current_gear_pos[0]}
+ await self._backend.update_gear_motor_position()
+ current_gear_pos = axis_convert(self._backend.gear_motor_position, 0.0)
  home_target_dict = {OT3Axis.Q: 0.0}
- home_moves = self._build_moves(gear_pos_dict, home_target_dict)
+ home_moves = self._build_moves(current_gear_pos, home_target_dict)
 
- await self._backend.tip_action(home_moves[0], "home")
+ await self._backend.tip_action(moves=home_moves[0], tip_action="home")
 
  else:
  target_pos = target_position_from_plunger(

api/tests/opentrons/hardware_control/backends/test_ot3_controller.py

@@ -681,7 +681,7 @@ async def test_tip_action(
  controller: OT3Controller,
  mock_move_group_run: mock.AsyncMock,
 ) -> None:
- await controller.tip_action([Axis.P_L], 33, -5.5, tip_action="clamp")
+ await controller.tip_action(distance=33, velocity=-5.5, tip_action="clamp")
  for call in mock_move_group_run.call_args_list:
  move_group_runner = call[0][0]
  for move_group in move_group_runner._move_groups:
@@ -694,7 +694,7 @@ async def test_tip_action(
 
  mock_move_group_run.reset_mock()
 
- await controller.tip_action([Axis.P_L], 33, -5.5, tip_action="home")
+ await controller.tip_action(distance=33, velocity=-5.5, tip_action="home")
  for call in mock_move_group_run.call_args_list:
  move_group_runner = call[0][0]
  for move_group in move_group_runner._move_groups:

hardware/opentrons_hardware/firmware_bindings/messages/payloads.py

@@ -226,10 +226,14 @@ class MoveCompletedPayload(MoveGroupResponsePayload):
 
 
 @dataclass(eq=False)
-class GearMotorPositionResponse(EmptyPayload):
+class GearMotorPositionResponse(MoveGroupResponsePayload):
  """Read Gear Motor Position Estimation."""
 
- current_position: utils.UInt32Field
+ seq_id: utils.UInt8Field
+ current_position_um: utils.UInt32Field
+ encoder_position_um: utils.Int32Field
+ position_flags: MotorPositionFlagsField
+ ack_id: utils.UInt8Field
 
 
 @dataclass(eq=False)
@@ -568,12 +572,6 @@ class TipActionResponsePayload(MoveCompletedPayload):
  success: utils.UInt8Field
  gear_motor_id: GearMotorIdField
 
- seq_id: utils.UInt8Field
- current_position_um: utils.UInt32Field
- encoder_position_um: utils.Int32Field
- position_flags: MotorPositionFlagsField
- ack_id: utils.UInt8Field
-
 
 @dataclass(eq=False)
 class PeripheralStatusResponsePayload(SensorPayload):

hardware/opentrons_hardware/hardware_control/motion.py

@@ -1,5 +1,5 @@
 """A collection of motions that define a single move."""
-from typing import List, Dict, Iterable, Union, Optional, Tuple
+from typing import List, Dict, Iterable, Union
 from dataclasses import dataclass
 import numpy as np
 from logging import getLogger
@@ -191,10 +191,11 @@ def create_tip_action_step(
  action: PipetteTipActionType,
 ) -> MoveGroupStep:
  """Creates a step for tip handling actions that require motor movement."""
- if action == PipetteTipActionType.home:
- stop_condition = MoveStopCondition.limit_switch
- else:
- stop_condition = MoveStopCondition.none
+ stop_condition = (
+ MoveStopCondition.limit_switch
+ if action == PipetteTipActionType.home
+ else MoveStopCondition.none
+ )
  step: MoveGroupStep = {}
  for axis_node in present_nodes:
  step[axis_node] = MoveGroupTipActionStep(

hardware/opentrons_hardware/hardware_control/motor_position_status.py

@@ -19,7 +19,6 @@
  UpdateMotorPositionEstimationResponse,
  UpdateGearMotorPositionEstimationRequest,
  UpdateGearMotorPositionEstimationResponse,
- TipActionResponse,
 )
 from opentrons_hardware.firmware_bindings.arbitration_id import ArbitrationId
 from opentrons_hardware.firmware_bindings.constants import (
@@ -224,7 +223,7 @@ async def _parser_update_gear_motor_position_response(
  node = NodeId(arb_id.parts.originating_node_id)
  if node == expected:
  return (
- float(response.payload.current_position.value / 1000.0),
+ float(response.payload.current_position_um.value / 1000.0),
  bool(MotorPositionFlags.stepper_position_ok.value),
  )
  raise StopAsyncIteration

hardware/opentrons_hardware/hardware_control/move_group_runner.py

@@ -174,6 +174,8 @@ def _accumulate_move_completions(
  List[Tuple[Tuple[int, int], float, float, bool, bool]]
  ] = defaultdict(list)
  for arbid, completion in completions:
+ if isinstance(completion, TipActionResponse):
+ continue
  position[NodeId(arbid.parts.originating_node_id)].append(
  (
  (
@@ -405,15 +407,13 @@ def _remove_move_group(
 
  in_group = (node_id, seq_id) in self._moves[group_id]
  self._moves[group_id].remove((node_id, seq_id))
- print(f"removing {node_id}, {seq_id}")
  self._completion_queue.put_nowait((arbitration_id, message))
  log.debug(
  f"Received completion for {node_id} group {group_id} seq {seq_id}"
  f", which {'is' if in_group else 'isn''t'} in group"
  )
 
  if not self._moves[group_id]:
- print(f"\nsetting event: {node_id}\n")
  log.debug(f"Move group {group_id+self._start_at_index} has completed.")
  self._event.set()