rebase

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

@@ -368,7 +368,7 @@ async def update_gear_motor_position(self) -> None:
  if updated_successfully:
  self._gear_motor_position = {NodeId.pipette_left: motor_response[0]}
 
- async def update_motor_estimation(self, axes: Sequence[OT3Axis]) -> None:
+ async def update_motor_estimation(self, axes: Sequence[Axis]) -> 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)
@@ -605,14 +605,14 @@ async def home(
  ]
  async with self._monitor_overpressure(moving_pipettes):
  positions = await asyncio.gather(*coros)
- if Axis.Q in checked_axes:
- await self.tip_action(
- [Axis.Q],
- self.axis_bounds[Axis.Q][1] - self.axis_bounds[Axis.Q][0],
- self._configuration.motion_settings.max_speed_discontinuity.high_throughput[
- Axis.to_kind(Axis.Q)
- ],
- )
+ # if Axis.Q in checked_axes:
+ # await self.tip_action(
+ #  [Axis.Q],
+ #  self.axis_bounds[Axis.Q][1] - self.axis_bounds[Axis.Q][0],
+ #  self._configuration.motion_settings.max_speed_discontinuity.high_throughput[
+ #  Axis.to_kind(Axis.Q)
+ #  ],
+ # )
  for position in positions:
  self._handle_motor_status_response(position)
  return axis_convert(self._position, 0.0)
@@ -631,7 +631,7 @@ def _filter_move_group(self, move_group: MoveGroup) -> MoveGroup:
 
  async def tip_action(
  self,
- moves: Optional[List[Move[OT3Axis]]] = None,
+ moves: Optional[List[Move[Axis]]] = None,
  distance: Optional[float] = None,
  velocity: Optional[float] = None,
  tip_action: str = "home",

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

@@ -203,7 +203,7 @@ def eeprom_data(self) -> EEPROMData:
  return EEPROMData()
 
  @property
- def gear_motor_position(self) -> Optional[float]:
+ def gear_motor_position(self) -> Dict[NodeId, float]:
  return self._gear_motor_position
 
  @property
@@ -262,7 +262,7 @@ async def home_gear_motors(
  return None
 
  @ensure_yield
- async def update_motor_estimation(self, axes: Sequence[OT3Axis]) -> None:
+ async def update_motor_estimation(self, axes: Sequence[Axis]) -> 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
@@ -389,7 +389,7 @@ async def get_tip_present(self, mount: OT3Mount, tip_state: TipStateType) -> Non
 
  async def tip_action(
  self,
- moves: Optional[List[Move[OT3Axis]]] = None,
+ moves: Optional[List[Move[Axis]]] = None,
  distance: Optional[float] = None,
  velocity: Optional[float] = None,
  tip_action: str = "home",
@@ -540,6 +540,7 @@ def axis_bounds(self) -> OT3AxisMap[Tuple[float, float]]:
  Axis.Y: phony_bounds,
  Axis.X: phony_bounds,
  Axis.Z_G: phony_bounds,
+ Axis.Q: phony_bounds,
  }
 
  @property

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
+from typing import Dict, Iterable, List, Set, Tuple, TypeVar, cast
 from typing_extensions import Literal
 from logging import getLogger
 from opentrons.config.defaults_ot3 import DEFAULT_CALIBRATION_AXIS_MAX_SPEED
@@ -214,7 +214,7 @@ def get_current_settings(
  conf_by_pip = config.by_gantry_load(gantry_load)
  currents = {}
  for axis_kind in conf_by_pip["hold_current"].keys():
- for axis in OT3Axis.of_kind(axis_kind):
+ for axis in Axis.of_kind(axis_kind):
  currents[axis] = CurrentConfig(
  conf_by_pip["hold_current"][axis_kind],
  conf_by_pip["run_current"][axis_kind],
@@ -225,7 +225,7 @@ def get_current_settings(
 def get_system_constraints(
  config: OT3MotionSettings,
  gantry_load: GantryLoad,
-) -> "SystemConstraints[OT3Axis]":
+) -> "SystemConstraints[Axis]":
  conf_by_pip = config.by_gantry_load(gantry_load)
  constraints = {}
  axis_kind_list = [
@@ -238,7 +238,7 @@ def get_system_constraints(
  if gantry_load == GantryLoad.HIGH_THROUGHPUT:
  axis_kind_list.append(OT3AxisKind.Q)
  for axis_kind in axis_kind_list:
- for axis in OT3Axis.of_kind(axis_kind):
+ for axis in Axis.of_kind(axis_kind):
  constraints[axis] = AxisConstraints.build(
  conf_by_pip["acceleration"][axis_kind],
  conf_by_pip["max_speed_discontinuity"][axis_kind],
@@ -399,7 +399,7 @@ def create_tip_action_home_group(
 
 
 def create_tip_action_group(
- moves: List[Move[OT3Axis]],
+ moves: List[Move[Axis]],
  present_nodes: Iterable[NodeId],
  action: str,
 ) -> MoveGroup:

api/src/opentrons/hardware_control/ot3api.py

@@ -734,22 +734,22 @@ async def home_z(
 
  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_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 is not None and position_estimation > 5.0:
- position_valid = abs(position_estimation) < gear_motor_axis_bound
+ 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 = {OT3Axis.Q: position_estimation}
- safe_home_target = {OT3Axis.Q: self._config.safe_home_distance}
+ 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)[
- OT3Axis.P_L
+ Axis.P_L
  ]
  await self._backend.tip_action(
  distance=distance, velocity=velocity, tip_action="home"
@@ -1278,7 +1278,7 @@ async def _home(self, axes: Sequence[Axis]) -> None:
  try:
  if axis == Axis.G:
  await self.home_gripper_jaw()
- elif axis == OT3Axis.Q:
+ elif axis == Axis.Q:
  await self.home_gear_motors()
  else:
  await self._home_axis(axis)
@@ -1307,7 +1307,7 @@ async def home(self, axes: Optional[List[Axis]] = None) -> None:
  else:
  checked_axes = [ax for ax in Axis if ax != Axis.Q]
  if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
- checked_axes.append(OT3Axis.Q)
+ checked_axes.append(Axis.Q)
  self._log.info(f"Homing {axes}")
 
  home_seq = [
@@ -1712,11 +1712,13 @@ async def _motor_pick_up_tip(
  # perform pick up tip
 
  await self._backend.update_gear_motor_position()
- gear_origin_dict = axis_convert(self._backend.gear_motor_position, 0.0)
+ gear_origin_float = axis_convert(self._backend.gear_motor_position, 0.0)[
+ Axis.P_L
+ ]
+ gear_origin_dict = {Axis.Q: gear_origin_float}
  clamp_move_target = pipette_spec.pick_up_distance
- gear_target_dict = {OT3Axis.Q: clamp_move_target}
+ gear_target_dict = {AxisQ: 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()
@@ -1796,17 +1798,19 @@ async def drop_tip(
  gear_start_position = axis_convert(
  self._backend.gear_motor_position, 0.0
  )
- drop_target_dict = {OT3Axis.Q: move.target_position}
+ drop_target_dict = {Axis.Q: move.target_position}
  drop_moves = self._build_moves(gear_start_position, drop_target_dict)
 
  await self._backend.tip_action(moves=drop_moves[0], tip_action="clamp")
 
- 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(current_gear_pos, home_target_dict)
+ await self.home_gear_motors()
+ # await self._backend.update_gear_motor_position()
+ # gear_pos_float = axis_convert(self._backend.gear_motor_position, 0.0)[AxisP_L]
+ # gear_pos_dict = {AxisQ: gear_pos_float}
+ # home_target_dict = {AxisQ: 0.0}
+ # home_moves = self._build_moves(gear_pos_dict, home_target_dict)
 
- await self._backend.tip_action(moves=home_moves[0], tip_action="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(distance=33, velocity=-5.5, tip_action="clamp")
+ 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:
@@ -690,7 +690,7 @@ async def test_tip_action(
  # we should be sending this command to the pipette axes to process.
  assert list(move_group[0].keys()) == [NodeId.pipette_left]
  step = move_group[0][NodeId.pipette_left]
- assert step.stop_condition == MoveStopCondition.none
+ assert step.stop_condition == MoveStopCondition.limit_switch
 
  mock_move_group_run.reset_mock()
 

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -1,6 +1,6 @@
 """ Tests for behaviors specific to the OT3 hardware controller.
 """
-from typing import Iterator, Union, Dict, Tuple, List, Any, OrderedDict
+from typing import Iterator, Union, Dict, Tuple, List, Any, OrderedDict, Optional
 from typing_extensions import Literal
 from math import copysign
 import pytest
@@ -60,6 +60,7 @@
 from opentrons.types import Point, Mount
 
 from opentrons_hardware.hardware_control.motion import MoveStopCondition
+from opentrons_hardware.hardware_control.motion_planning.types import Move
 
 from opentrons.config import gripper_config as gc
 from opentrons_shared_data.gripper.gripper_definition import GripperModel
@@ -469,6 +470,8 @@ async def test_blow_out_position(
  blowout_volume: float,
 ) -> None:
  for mount, configs in load_configs.items():
+ if configs["channels"] == 96:
+ await ot3_hardware.set_gantry_load(GantryLoad.HIGH_THROUGHPUT)
  instr_data, ot3_hardware = await prepare_for_mock_blowout(
  ot3_hardware, mount, configs
  )
@@ -481,7 +484,6 @@ async def test_blow_out_position(
  max_allowed_input_distance * instr_data["config"].shaft_ul_per_mm
  )
  assume(blowout_volume < max_input_vol)
-
  await ot3_hardware.blow_out(mount, blowout_volume)
  pipette_axis = Axis.of_main_tool_actuator(mount)
  position_result = await ot3_hardware.current_position_ot3(mount)
@@ -514,6 +516,8 @@ async def test_blow_out_error(
  blowout_volume: float,
 ) -> None:
  for mount, configs in load_configs.items():
+ if configs["channels"] == 96:
+ await ot3_hardware.set_gantry_load(GantryLoad.HIGH_THROUGHPUT)
  instr_data, ot3_hardware = await prepare_for_mock_blowout(
  ot3_hardware, mount, configs
  )
@@ -1369,23 +1373,48 @@ def _fake_function():
  # Move onto the posts
  tiprack_down=Point(0, 0, 0),
  tiprack_up=Point(0, 0, 0),
- pick_up_distance=0,
+ pick_up_distance=10,
  speed=0,
  currents={Axis.Q: 0},
  ),
  ),
  _fake_function,
  )
+
+ def _update_gear_motor_pos(
+ moves: Optional[List[Move[OT3Axis]]] = None,
+ distance: Optional[float] = None,
+ velocity: Optional[float] = None,
+ tip_action: str = "home",
+ ) -> None:
+ if NodeId.pipette_left not in backend._gear_motor_position:
+ backend._gear_motor_position = {NodeId.pipette_left: 0.0}
+ if moves:
+ for move in moves:
+ for block in move.blocks:
+ backend._gear_motor_position[
+ NodeId.pipette_left
+ ] += block.distance
+ elif distance:
+ backend._gear_motor_position[NodeId.pipette_left] += distance
+
+ tip_action.side_effect = _update_gear_motor_pos
+ await ot3_hardware.set_gantry_load(GantryLoad.HIGH_THROUGHPUT)
  await ot3_hardware.pick_up_tip(Mount.LEFT, 40.0)
+ # backend._gear_motor_position = 10
  pipette_handler.plan_check_pick_up_tip.assert_called_once_with(
  OT3Mount.LEFT, 40.0, None, None
  )
- tip_action.assert_has_calls(
- calls=[
- call([Axis.P_L], 0, 0, "clamp"),
- call([Axis.P_L], 0, 0, "home"),
- ]
- )
+ # breakpoint()
+ # first call should be "clamp", moving down
+ assert tip_action.call_args_list[0][-1]["tip_action"] == "clamp"
+ assert tip_action.call_args_list[0][-1]["moves"][0].unit_vector == {AxisQ: 1}
+ # next call should be "clamp", moving back up
+ assert tip_action.call_args_list[1][-1]["tip_action"] == "clamp"
+ assert tip_action.call_args_list[1][-1]["moves"][0].unit_vector == {AxisQ: -1}
+ # last call should be "home"
+ assert tip_action.call_args_list[2][-1]["tip_action"] == "home"
+ assert len(tip_action.call_args_list) == 3
 
 
 async def test_drop_tip_full_tiprack(
@@ -1405,8 +1434,8 @@ def _fake_function():
  DropTipSpec(
  drop_moves=[
  DropTipMove(
- target_position=1,
- current={Axis.P_L: 1.0},
+ target_position=10,
+ current={AxisP_L: 1.0},
  speed=1,
  is_ht_tip_action=True,
  )
@@ -1416,14 +1445,39 @@ def _fake_function():
  ),
  _fake_function,
  )
+
+ def _update_gear_motor_pos(
+ moves: Optional[List[Move[OT3Axis]]] = None,
+ distance: Optional[float] = None,
+ velocity: Optional[float] = None,
+ tip_action: str = "home",
+ ) -> None:
+ if NodeId.pipette_left not in backend._gear_motor_position:
+ backend._gear_motor_position = {NodeId.pipette_left: 0.0}
+ if moves:
+ for move in moves:
+ for block in move.blocks:
+ backend._gear_motor_position[
+ NodeId.pipette_left
+ ] += block.distance
+ elif distance:
+ backend._gear_motor_position[NodeId.pipette_left] += distance
+
+ tip_action.side_effect = _update_gear_motor_pos
+
+ await ot3_hardware.set_gantry_load(GantryLoad.HIGH_THROUGHPUT)
  await ot3_hardware.drop_tip(Mount.LEFT, home_after=True)
  pipette_handler.plan_check_drop_tip.assert_called_once_with(OT3Mount.LEFT, True)
- tip_action.assert_has_calls(
- calls=[
- call([Axis.P_L], 1, 1, "clamp"),
- call([Axis.P_L], 1, 1, "home"),
- ]
- )
+
+ # first call should be "clamp", moving down
+ assert tip_action.call_args_list[0][-1]["tip_action"] == "clamp"
+ assert tip_action.call_args_list[0][-1]["moves"][0].unit_vector == {Axis.Q: 1}
+ # next call should be "clamp", moving back up
+ assert tip_action.call_args_list[1][-1]["tip_action"] == "clamp"
+ assert tip_action.call_args_list[1][-1]["moves"][0].unit_vector == {Axis.Q: -1}
+ # last call should be "home"
+ assert tip_action.call_args_list[2][-1]["tip_action"] == "home"
+ assert len(tip_action.call_args_list) == 3
 
 
 @pytest.mark.parametrize(
@@ -1663,6 +1717,8 @@ 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/move_group_runner.py

@@ -613,6 +613,7 @@ async def run(self, can_messenger: CanMessenger) -> _Completions:
  self._start_at_index, self._start_at_index + len(self._moves)
  ):
  await self._run_one_group(group_id, can_messenger)
+
  def _reify_queue_iter() -> Iterator[_CompletionPacket]:
  while not self._completion_queue.empty():
  yield self._completion_queue.get_nowait()