hardware controller changes

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

@@ -638,13 +638,11 @@ async def tip_action(
  tip_action: str = "home",
  ) -> None:
  move_group = []
- if tip_action == "clamp":
- assert moves is not None
+ if 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
+ elif 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],

api/src/opentrons/hardware_control/ot3api.py

@@ -568,8 +568,6 @@ 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
@@ -732,35 +730,6 @@ async def home_z(
  axes = list(Axis.ot3_mount_axes())
  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_float = position_estimation_dict[Axis.P_L]
- gear_motor_axis_bound = self._backend.axis_bounds[Axis.Q][1]
- velocity = self._config.motion_settings.default_max_speed[
- GantryLoad.HIGH_THROUGHPUT
- ][OT3AxisKind.Q]
- if position_estimation_float is not None and position_estimation_float > 5.0:
- position_valid = abs(position_estimation_float) < gear_motor_axis_bound
- if position_valid:
- position_dict = {Axis.Q: position_estimation_float}
- safe_home_target = {Axis.Q: self._config.safe_home_distance}
- moves = self._build_moves(position_dict, safe_home_target)
-
- await self._backend.tip_action(moves=moves[0], tip_action="clamp")
-
- distance = axis_convert(self._backend.gear_motor_position, 0.0)[
- Axis.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
- await self._backend.tip_action(
- distance=gear_motor_axis_bound, velocity=velocity, tip_action="home"
- )
 
  async def home_gripper_jaw(self) -> None:
  """
@@ -843,6 +812,8 @@ async def refresh_positions(self) -> None:
  await self._backend.update_motor_status()
  await self._cache_current_position()
  await self._cache_encoder_position()
+ if self._gantry_load == GantryLoad.HIGH_THROUGHPUT:
+ await self._backend.update_gear_motor_position()
 
  async def _cache_current_position(self) -> Dict[Axis, float]:
  """Cache current position from backend and return in absolute deck coords."""
@@ -1711,16 +1682,25 @@ async def _motor_pick_up_tip(
  await self._move(target_down)
  # perform pick up tip
  await self._backend.update_gear_motor_position()
+ pipette_axis = Axis.of_main_tool_actuator(mount)
  gear_origin_float = axis_convert(self._backend.gear_motor_position, 0.0)[
- Axis.P_L
+ pipette_axis
  ]
  gear_origin_dict = {Axis.Q: gear_origin_float}
  clamp_move_target = pipette_spec.pick_up_distance
  gear_target_dict = {Axis.Q: clamp_move_target}
  clamp_moves = self._build_moves(gear_origin_dict, gear_target_dict)
  await self._backend.tip_action(moves=clamp_moves[0], tip_action="clamp")
 
- await self.home_gear_motors()
+ homing_velocity = self._config.motion_settings.default_max_speed[
+ GantryLoad.HIGH_THROUGHPUT
+ ][OT3AxisKind.Q]
+ q_axis_distance = axis_convert(self._backend.gear_motor_position, 0.0)[Axis.P_L]
+ await self._backend.tip_action(
+ distance=(q_axis_distance + pipette_spec.home_buffer),
+ velocity=homing_velocity,
+ tip_action="home",
+ )
 
  async def pick_up_tip(
  self,
@@ -1796,12 +1776,21 @@ async def drop_tip(
  gear_start_position = axis_convert(
  self._backend.gear_motor_position, 0.0
  )
- drop_target_dict = {Axis.Q: move.target_position}
- drop_moves = self._build_moves(gear_start_position, drop_target_dict)
+ pickup_target_dict = {Axis.Q: move.target_position}
+ drop_moves = self._build_moves(gear_start_position, pickup_target_dict)
 
  await self._backend.tip_action(moves=drop_moves[0], tip_action="clamp")
 
- await self.home_gear_motors()
+ homing_velocity = self._config.motion_settings.default_max_speed[
+ GantryLoad.HIGH_THROUGHPUT
+ ][OT3AxisKind.Q]
+ pipette_axis = Axis.of_main_tool_actuator(mount)
+ q_axis_distance = axis_convert(self._backend.gear_motor_position, 0.0)[pipette_axis]
+ await self._backend.tip_action(
+ distance=(q_axis_distance + move.home_buffer),
+ velocity=homing_velocity,
+ tip_action="home",
+ )
 
  else:
  target_pos = target_position_from_plunger(

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -1715,8 +1715,6 @@ async def test_tip_presence_disabled_ninety_six_channel(
  instr_data = AttachedPipette(config=pipette_config, id="fakepip")
  await ot3_hardware.cache_pipette(OT3Mount.LEFT, instr_data, None)
  await ot3_hardware._configure_instruments()
- await ot3_hardware.set_gantry_load(GantryLoad.HIGH_THROUGHPUT)
- # await ot3_hardware.cache_instruments()
  await ot3_hardware.pick_up_tip(OT3Mount.LEFT, 60)
 
  tip_present.assert_not_called()

hardware/opentrons_hardware/hardware_control/motor_position_status.py

@@ -184,25 +184,23 @@ def _listener_filter(arbitration_id: ArbitrationId) -> bool:
 
  data = []
 
- with WaitableCallback(can_messenger, _listener_filter) as reader:
+ with MultipleMessagesWaitableCallback(
+ can_messenger,
+ _listener_filter,
+ 2,
+ ) as reader:
  await can_messenger.send(
  node_id=NodeId.pipette_left,
  message=UpdateGearMotorPositionEstimationRequest(),
  )
  try:
- for i in range(2):
- response = await asyncio.wait_for(
- _parser_update_gear_motor_position_response(
- reader, NodeId.pipette_left
- ),
- timeout,
- )
- data.append(response)
- if not data[0][1] or not data[1][1]:
- # If the stepper_ok flag isn't set, that means the node didn't update position.
- raise RuntimeError(
- f"Failed to update motor position for node: {NodeId.pipette_left}"
- )
+ response = await asyncio.wait_for(
+ _parser_update_gear_motor_position_response(
+ reader, NodeId.pipette_left
+ ),
+ timeout,
+ )
+ data.append(response)
  except asyncio.TimeoutError:
  log.warning("Update motor position estimation timed out")
  return 0, False

shared-data/python/opentrons_shared_data/pipette/pipette_load_name_conversions.py

@@ -221,6 +221,7 @@ def convert_pipette_model(
  # to attached pipettes.
  # We need to figure out how to default the pipette model as well
  # rather than returning a p1000
+ breakpoint()
  if model and not provided_version:
  pipette_type, parsed_channels, parsed_version = model.split("_")
  channels = channels_from_string(parsed_channels)