Merge branch 'edge' of https://github.com/Opentrons/opentrons into pd…

…-flex-pipette-selection

api/src/opentrons/hardware_control/api.py

@@ -969,20 +969,11 @@ async def blow_out(
  Force any remaining liquid to dispense. The liquid will be dispensed at
  the current location of pipette
  """
- blowout_spec = self.plan_check_blow_out(mount, volume)
-
- instrument = self.get_pipette(mount)
- max_blowout_pos = instrument.config.blow_out
- # start at the bottom position and move additional distance det. by plan_check_blow_out
- blowout_distance = instrument.config.bottom - blowout_spec.plunger_distance
-
- if blowout_distance < max_blowout_pos:
- raise ValueError("Blow out distance exceeds plunger position limit")
-
+ blowout_spec = self.plan_check_blow_out(mount)
  self._backend.set_active_current({blowout_spec.axis: blowout_spec.current})
  target_pos = target_position_from_plunger(
  mount,
- blowout_distance,
+ blowout_spec.plunger_distance,
  self._current_position,
  )
 

api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py

@@ -609,35 +609,26 @@ def plan_check_dispense( # type: ignore[no-untyped-def]
  )
 
  @overload
- def plan_check_blow_out(
- self, mount: top_types.Mount, volume: Optional[float] = None
- ) -> LiquidActionSpec[Axis]:
+ def plan_check_blow_out(self, mount: top_types.Mount) -> LiquidActionSpec[Axis]:
  ...
 
  @overload
- def plan_check_blow_out(
- self, mount: OT3Mount, volume: Optional[float] = None
- ) -> LiquidActionSpec[OT3Axis]:
+ def plan_check_blow_out(self, mount: OT3Mount) -> LiquidActionSpec[OT3Axis]:
  ...
 
- def plan_check_blow_out(self, mount, volume: Optional[float] = None): # type: ignore[no-untyped-def]
+ def plan_check_blow_out(self, mount): # type: ignore[no-untyped-def]
  """Check preconditions and calculate values for blowout."""
  instrument = self.get_pipette(mount)
  self.ready_for_tip_action(instrument, HardwareAction.BLOWOUT)
  speed = self.plunger_speed(
  instrument, instrument.blow_out_flow_rate, "dispense"
  )
- # TODO(cm, 04/23): change this to use schema v2 configs
- if volume is not None:
- distance_mm = instrument.ul_per_mm(volume, "dispense")
- else:
- distance_mm = instrument.config.blow_out
 
  if isinstance(mount, top_types.Mount):
  return LiquidActionSpec(
  axis=Axis.of_plunger(mount),
  volume=0,
- plunger_distance=distance_mm,
+ plunger_distance=instrument.config.blow_out,
  speed=speed,
  instr=instrument,
  current=instrument.config.plunger_current,
@@ -646,7 +637,7 @@ def plan_check_blow_out(self, mount, volume: Optional[float] = None): # type: i
  return LiquidActionSpec(
  axis=OT3Axis.of_main_tool_actuator(mount),
  volume=0,
- plunger_distance=distance_mm,
+ plunger_distance=instrument.config.blow_out,
  speed=speed,
  instr=instrument,
  current=instrument.config.plunger_current,

api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py

@@ -594,11 +594,7 @@ def plan_check_blow_out(
  """Check preconditions and calculate values for blowout."""
  instrument = self.get_pipette(mount)
  self.ready_for_tip_action(instrument, HardwareAction.BLOWOUT)
- speed = self.plunger_speed(
- instrument,
- instrument.blow_out_flow_rate,
- "blowout",
- )
+ speed = self.plunger_speed(instrument, instrument.blow_out_flow_rate, "blowout")
  if volume is None:
  ul = self.get_attached_instrument(mount)["default_blow_out_volume"]
  else:

api/src/opentrons/protocol_engine/commands/__init__.py

@@ -94,6 +94,14 @@
  DropTipCommandType,
 )
 
+from .drop_tip_in_place import (
+ DropTipInPlace,
+ DropTipInPlaceParams,
+ DropTipInPlaceCreate,
+ DropTipInPlaceResult,
+ DropTipInPlaceCommandType,
+)
+
 from .home import (
  Home,
  HomeParams,
@@ -290,6 +298,12 @@
  "DropTipParams",
  "DropTipResult",
  "DropTipCommandType",
+ # drop tip in place command models
+ "DropTipInPlace",
+ "DropTipInPlaceCreate",
+ "DropTipInPlaceParams",
+ "DropTipInPlaceResult",
+ "DropTipInPlaceCommandType",
  # home command models
  "Home",
  "HomeParams",

api/src/opentrons/protocol_engine/commands/command_unions.py

@@ -73,6 +73,14 @@
  DropTipCommandType,
 )
 
+from .drop_tip_in_place import (
+ DropTipInPlace,
+ DropTipInPlaceParams,
+ DropTipInPlaceCreate,
+ DropTipInPlaceResult,
+ DropTipInPlaceCommandType,
+)
+
 from .home import (
  Home,
  HomeParams,
@@ -211,6 +219,7 @@
  BlowOut,
  BlowOutInPlace,
  DropTip,
+ DropTipInPlace,
  Home,
  LoadLabware,
  LoadLiquid,
@@ -263,6 +272,7 @@
  BlowOutParams,
  BlowOutInPlaceParams,
  DropTipParams,
+ DropTipInPlaceParams,
  HomeParams,
  LoadLabwareParams,
  LoadLiquidParams,
@@ -316,6 +326,7 @@
  BlowOutCommandType,
  BlowOutInPlaceCommandType,
  DropTipCommandType,
+ DropTipInPlaceCommandType,
  HomeCommandType,
  LoadLabwareCommandType,
  LoadLiquidCommandType,
@@ -368,6 +379,7 @@
  BlowOutCreate,
  BlowOutInPlaceCreate,
  DropTipCreate,
+ DropTipInPlaceCreate,
  HomeCreate,
  LoadLabwareCreate,
  LoadLiquidCreate,
@@ -420,6 +432,7 @@
  BlowOutResult,
  BlowOutInPlaceResult,
  DropTipResult,
+ DropTipInPlaceResult,
  HomeResult,
  LoadLabwareResult,
  LoadLiquidResult,

api/src/opentrons/protocol_engine/commands/dispense_in_place.py

@@ -1,8 +1,5 @@
 """Dispense-in-place command request, result, and implementation models."""
 
-# TODO(mm, 2022-08-15): This command is not yet in the JSON protocol schema.
-# Before our production code emits this command, we must add it to the schema,
-# and probably bump the schema version.
 
 from __future__ import annotations
 from typing import TYPE_CHECKING, Optional, Type

api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py

@@ -0,0 +1,75 @@
+"""Drop tip in place command request, result, and implementation models."""
+from __future__ import annotations
+from pydantic import Field, BaseModel
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+
+from .pipetting_common import PipetteIdMixin
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
+
+if TYPE_CHECKING:
+ from ..execution import TipHandler
+
+
+DropTipInPlaceCommandType = Literal["dropTipInPlace"]
+
+
+class DropTipInPlaceParams(PipetteIdMixin):
+ """Payload required to drop a tip in place."""
+
+ homeAfter: Optional[bool] = Field(
+ None,
+ description=(
+ "Whether to home this pipette's plunger after dropping the tip."
+ " You should normally leave this unspecified to let the robot choose"
+ " a safe default depending on its hardware."
+ ),
+ )
+
+
+class DropTipInPlaceResult(BaseModel):
+ """Result data from the execution of a DropTipInPlace command."""
+
+ pass
+
+
+class DropTipInPlaceImplementation(
+ AbstractCommandImpl[DropTipInPlaceParams, DropTipInPlaceResult]
+):
+ """Drop tip in place command implementation."""
+
+ def __init__(
+ self,
+ tip_handler: TipHandler,
+ **kwargs: object,
+ ) -> None:
+ self._tip_handler = tip_handler
+
+ async def execute(self, params: DropTipInPlaceParams) -> DropTipInPlaceResult:
+ """Drop a tip using the requested pipette."""
+ await self._tip_handler.drop_tip(
+ pipette_id=params.pipetteId, home_after=params.homeAfter
+ )
+
+ return DropTipInPlaceResult()
+
+
+class DropTipInPlace(BaseCommand[DropTipInPlaceParams, DropTipInPlaceResult]):
+ """Drop tip in place command model."""
+
+ commandType: DropTipInPlaceCommandType = "dropTipInPlace"
+ params: DropTipInPlaceParams
+ result: Optional[DropTipInPlaceResult]
+
+ _ImplementationCls: Type[
+ DropTipInPlaceImplementation
+ ] = DropTipInPlaceImplementation
+
+
+class DropTipInPlaceCreate(BaseCommandCreate[DropTipInPlaceParams]):
+ """Drop tip in place command creation request model."""
+
+ commandType: DropTipInPlaceCommandType = "dropTipInPlace"
+ params: DropTipInPlaceParams
+
+ _CommandCls: Type[DropTipInPlace] = DropTipInPlace

api/src/opentrons/protocol_engine/state/pipettes.py

@@ -28,6 +28,7 @@
  MoveRelativeResult,
  PickUpTipResult,
  DropTipResult,
+ DropTipInPlaceResult,
  HomeResult,
  BlowOutResult,
  TouchTipResult,
@@ -167,7 +168,7 @@ def _handle_command(self, command: Command) -> None:
  self._state.attached_tip_by_id[pipette_id] = attached_tip
  self._state.aspirated_volume_by_id[pipette_id] = 0
 
- elif isinstance(command.result, DropTipResult):
+ elif isinstance(command.result, (DropTipResult, DropTipInPlaceResult)):
  pipette_id = command.params.pipetteId
  self._state.aspirated_volume_by_id[pipette_id] = None
  self._state.attached_tip_by_id[pipette_id] = None

api/src/opentrons/protocol_engine/state/tips.py

@@ -10,7 +10,13 @@
  ResetTipsAction,
  AddPipetteConfigAction,
 )
-from ..commands import Command, LoadLabwareResult, PickUpTipResult, DropTipResult
+from ..commands import (
+ Command,
+ LoadLabwareResult,
+ PickUpTipResult,
+ DropTipResult,
+ DropTipInPlaceResult,
+)
 
 
 class TipRackWellState(Enum):
@@ -91,7 +97,7 @@ def _handle_command(self, command: Command) -> None:
  )
  self._state.length_by_pipette_id[pipette_id] = length
 
- elif isinstance(command.result, DropTipResult):
+ elif isinstance(command.result, (DropTipResult, DropTipInPlaceResult)):
  pipette_id = command.params.pipetteId
  self._state.length_by_pipette_id.pop(pipette_id, None)
 

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -5,6 +5,7 @@
 from math import copysign
 import pytest
 from mock import AsyncMock, patch, Mock, call, PropertyMock
+from hypothesis import given, strategies, settings, HealthCheck, assume, example
 
 from opentrons.calibration_storage.types import CalibrationStatus, SourceType
 from opentrons.config.types import (
@@ -64,7 +65,6 @@
  PipetteChannelType,
  PipetteVersionType,
 )
-from hypothesis import given, strategies, settings, HealthCheck, assume, example
 
 
 @pytest.fixture
@@ -471,8 +471,14 @@ async def test_blow_out_position(
 
 
 @pytest.mark.parametrize("load_configs", load_blowout_configs)
-@given(blowout_volume=strategies.floats(min_value=0, max_value=200))
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], max_examples=10)
+@given(blowout_volume=strategies.floats(min_value=0, max_value=300))
+@settings(
+ suppress_health_check=[
+ HealthCheck.function_scoped_fixture,
+ HealthCheck.filter_too_much,
+ ],
+ max_examples=20,
+)
 async def test_blow_out_error(
  ot3_hardware: ThreadManager[OT3API],
  load_configs: List[Dict[str, Any]],

api/tests/opentrons/protocol_engine/commands/test_drop_tip.py

@@ -1,4 +1,4 @@
-"""Test pick up tip commands."""
+"""Test drop tip commands."""
 import pytest
 from decoy import Decoy
 

api/tests/opentrons/protocol_engine/commands/test_drop_tip_in_place.py

@@ -0,0 +1,36 @@
+"""Test drop tip in place commands."""
+import pytest
+from decoy import Decoy
+
+from opentrons.protocol_engine.execution import TipHandler
+
+from opentrons.protocol_engine.commands.drop_tip_in_place import (
+ DropTipInPlaceParams,
+ DropTipInPlaceResult,
+ DropTipInPlaceImplementation,
+)
+
+
+@pytest.fixture
+def mock_tip_handler(decoy: Decoy) -> TipHandler:
+ """Get a mock TipHandler."""
+ return decoy.mock(cls=TipHandler)
+
+
+async def test_drop_tip_implementation(
+ decoy: Decoy,
+ mock_tip_handler: TipHandler,
+) -> None:
+ """A DropTip command should have an execution implementation."""
+ subject = DropTipInPlaceImplementation(tip_handler=mock_tip_handler)
+
+ params = DropTipInPlaceParams(pipetteId="abc", homeAfter=False)
+
+ result = await subject.execute(params)
+
+ assert result == DropTipInPlaceResult()
+
+ decoy.verify(
+ await mock_tip_handler.drop_tip(pipette_id="abc", home_after=False),
+ times=1,
+ )

api/tests/opentrons/protocol_engine/state/command_fixtures.py

@@ -300,6 +300,24 @@ def create_drop_tip_command(
  )
 
 
+def create_drop_tip_in_place_command(
+ pipette_id: str,
+) -> cmd.DropTipInPlace:
+ """Get a completed DropTip command."""
+ params = cmd.DropTipInPlaceParams(pipetteId=pipette_id)
+
+ result = cmd.DropTipInPlaceResult()
+
+ return cmd.DropTipInPlace(
+ id="command-id",
+ key="command-key",
+ status=cmd.CommandStatus.SUCCEEDED,
+ createdAt=datetime.now(),
+ params=params,
+ result=result,
+ )
+
+
 def create_move_to_well_command(
  pipette_id: str,
  labware_id: str = "labware-id",