Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor(api): Delete dead PipetteStore code and type nozzle maps as non-Optional #16481

Merged
merged 10 commits into from
Oct 15, 2024
32 changes: 11 additions & 21 deletions api/src/opentrons/protocol_engine/state/pipettes.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,14 +96,17 @@ class StaticPipetteConfig:
nozzle_offset_z: float
pipette_bounding_box_offsets: PipetteBoundingBoxOffsets
bounding_nozzle_offsets: BoundingNozzlesOffsets
default_nozzle_map: NozzleMap
default_nozzle_map: NozzleMap # todo(mm, 2024-10-14): unused, remove?
lld_settings: Optional[Dict[str, Dict[str, float]]]


@dataclasses.dataclass
class PipetteState:
"""Basic pipette data state and getter methods."""

# todo(mm, 2024-10-14): It's getting difficult to ensure that all of these
# attributes are populated at the appropriate times. Refactor to a
# single dict-of-many-things instead of many dicts-of-single-things.
pipettes_by_id: Dict[str, LoadedPipette]
aspirated_volume_by_id: Dict[str, Optional[float]]
current_location: Optional[CurrentPipetteLocation]
Expand All @@ -112,7 +115,7 @@ class PipetteState:
movement_speed_by_id: Dict[str, Optional[float]]
static_config_by_id: Dict[str, StaticPipetteConfig]
flow_rates_by_id: Dict[str, FlowRates]
nozzle_configuration_by_id: Dict[str, Optional[NozzleMap]]
nozzle_configuration_by_id: Dict[str, NozzleMap]
liquid_presence_detection_by_id: Dict[str, bool]


Expand Down Expand Up @@ -167,11 +170,6 @@ def _set_load_pipette(self, state_update: update_types.StateUpdate) -> None:
self._state.aspirated_volume_by_id[pipette_id] = None
self._state.movement_speed_by_id[pipette_id] = None
self._state.attached_tip_by_id[pipette_id] = None
static_config = self._state.static_config_by_id.get(pipette_id)
if static_config:
self._state.nozzle_configuration_by_id[
pipette_id
] = static_config.default_nozzle_map
Comment on lines -170 to -174
Copy link
Contributor Author

@SyntaxColoring SyntaxColoring Oct 14, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because of the if static_config: line, this chunk of code in _set_load_pipette() depended on running after _update_pipette_config(). We do not obey that order, so it looked at first like this was broken.

However, it turns out _update_pipette_config() also sets this attribute, and _update_pipette_config() is always called in practice whenever _set_load_pipette() is called. So it's moot.

For clarity, therefore, delete this code. Each attribute of self._state is now affected either by _set_load_pipette() or _update_pipette_config(), never both.

Relying on _update_pipette_config() always being called in practice whenever _set_load_pipette() is called feels a little icky to me. I think we can upgrade that from being "in practice" to being "as guaranteed by the types" by rearranging StateUpdate a little bit, but I'm not doing that here.


def _update_tip_state(self, state_update: update_types.StateUpdate) -> None:
if state_update.pipette_tip_state != update_types.NO_CHANGE:
Expand Down Expand Up @@ -632,31 +630,23 @@ def get_plunger_axis(self, pipette_id: str) -> MotorAxis:

def get_nozzle_layout_type(self, pipette_id: str) -> NozzleConfigurationType:
"""Get the current set nozzle layout configuration."""
nozzle_map_for_pipette = self._state.nozzle_configuration_by_id.get(pipette_id)
if nozzle_map_for_pipette:
return nozzle_map_for_pipette.configuration
else:
return NozzleConfigurationType.FULL
nozzle_map_for_pipette = self._state.nozzle_configuration_by_id[pipette_id]
return nozzle_map_for_pipette.configuration

def get_is_partially_configured(self, pipette_id: str) -> bool:
"""Determine if the provided pipette is partially configured."""
return self.get_nozzle_layout_type(pipette_id) != NozzleConfigurationType.FULL

def get_primary_nozzle(self, pipette_id: str) -> Optional[str]:
def get_primary_nozzle(self, pipette_id: str) -> str:
"""Get the primary nozzle, if any, related to the given pipette's nozzle configuration."""
nozzle_map = self._state.nozzle_configuration_by_id.get(pipette_id)
return nozzle_map.starting_nozzle if nozzle_map else None
nozzle_map = self._state.nozzle_configuration_by_id[pipette_id]
return nozzle_map.starting_nozzle

def _get_critical_point_offset_without_tip(
self, pipette_id: str, critical_point: Optional[CriticalPoint]
) -> Point:
"""Get the offset of the specified critical point from pipette's mount position."""
nozzle_map = self._state.nozzle_configuration_by_id.get(pipette_id)
# Nozzle map is unavailable only when there's no pipette loaded
# so this is merely for satisfying the type checker
assert (
nozzle_map is not None
), "Error getting critical point offset. Nozzle map not found."
nozzle_map = self._state.nozzle_configuration_by_id[pipette_id]
match critical_point:
case CriticalPoint.INSTRUMENT_XY_CENTER:
return nozzle_map.instrument_xy_center_offset
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""Test for the ProtocolEngine-based instrument API core."""
from typing import cast, Optional, Union
from typing import cast, Optional

from opentrons_shared_data.errors.exceptions import PipetteLiquidNotFoundError
import pytest
Expand Down Expand Up @@ -1227,17 +1227,14 @@ def test_configure_nozzle_layout(
argnames=["pipette_channels", "nozzle_layout", "primary_nozzle", "expected_result"],
argvalues=[
(96, NozzleConfigurationType.FULL, "A1", True),
(96, NozzleConfigurationType.FULL, None, True),
(96, NozzleConfigurationType.ROW, "A1", True),
(96, NozzleConfigurationType.COLUMN, "A1", True),
(96, NozzleConfigurationType.COLUMN, "A12", True),
(96, NozzleConfigurationType.SINGLE, "H12", True),
(96, NozzleConfigurationType.SINGLE, "A1", True),
(8, NozzleConfigurationType.FULL, "A1", True),
(8, NozzleConfigurationType.FULL, None, True),
(8, NozzleConfigurationType.SINGLE, "H1", True),
(8, NozzleConfigurationType.SINGLE, "A1", True),
(1, NozzleConfigurationType.FULL, None, True),
],
)
def test_is_tip_tracking_available(
Expand All @@ -1246,7 +1243,7 @@ def test_is_tip_tracking_available(
subject: InstrumentCore,
pipette_channels: int,
nozzle_layout: NozzleConfigurationType,
primary_nozzle: Union[str, None],
primary_nozzle: str,
expected_result: bool,
) -> None:
"""It should return whether tip tracking is available based on nozzle configuration."""
Expand Down
45 changes: 36 additions & 9 deletions api/tests/opentrons/protocol_engine/state/test_pipette_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,25 +191,52 @@ def test_location_state_update(subject: PipetteStore) -> None:
)


def test_handles_load_pipette(subject: PipetteStore) -> None:
def test_handles_load_pipette(
subject: PipetteStore,
supported_tip_fixture: pipette_definition.SupportedTipsDefinition,
) -> None:
"""It should add the pipette data to the state."""
command = create_load_pipette_command(
dummy_command = create_succeeded_command()

load_pipette_update = update_types.LoadPipetteUpdate(
pipette_id="pipette-id",
pipette_name=PipetteNameType.P300_SINGLE,
mount=MountType.LEFT,
liquid_presence_detection=None,
)

config = LoadedStaticPipetteData(
model="pipette-model",
display_name="pipette name",
min_volume=1.23,
max_volume=4.56,
channels=7,
flow_rates=FlowRates(
default_aspirate={"a": 1},
default_dispense={"b": 2},
default_blow_out={"c": 3},
),
tip_configuration_lookup_table={4: supported_tip_fixture},
nominal_tip_overlap={"default": 5},
home_position=8.9,
nozzle_offset_z=10.11,
nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE),
back_left_corner_offset=Point(x=1, y=2, z=3),
front_right_corner_offset=Point(x=4, y=5, z=6),
pipette_lld_settings={},
)
config_update = update_types.PipetteConfigUpdate(
pipette_id="pipette-id",
config=config,
serial_number="pipette-serial",
)

subject.handle_action(
SucceedCommandAction(
private_result=None,
command=command,
command=dummy_command,
state_update=update_types.StateUpdate(
loaded_pipette=update_types.LoadPipetteUpdate(
pipette_id="pipette-id",
pipette_name=PipetteNameType.P300_SINGLE,
mount=MountType.LEFT,
liquid_presence_detection=None,
)
loaded_pipette=load_pipette_update, pipette_config=config_update
),
)
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ def get_pipette_view(
movement_speed_by_id: Optional[Dict[str, Optional[float]]] = None,
static_config_by_id: Optional[Dict[str, StaticPipetteConfig]] = None,
flow_rates_by_id: Optional[Dict[str, FlowRates]] = None,
nozzle_layout_by_id: Optional[Dict[str, Optional[NozzleMap]]] = None,
nozzle_layout_by_id: Optional[Dict[str, NozzleMap]] = None,
liquid_presence_detection_by_id: Optional[Dict[str, bool]] = None,
) -> PipetteView:
"""Get a pipette view test subject with the specified state."""
Expand Down
8 changes: 4 additions & 4 deletions api/tests/opentrons/protocol_engine/state/test_tip_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -1367,24 +1367,24 @@ def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleM
return configure_nozzle_private_result.nozzle_map

map = _reconfigure_nozzle_layout("A1", "A1", "A1")
for x in range(96):
for _ in range(96):
_get_next_and_pickup(map)
assert _get_next_and_pickup(map) is None

subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
map = _reconfigure_nozzle_layout("A12", "A12", "A12")
for x in range(96):
for _ in range(96):
_get_next_and_pickup(map)
assert _get_next_and_pickup(map) is None

subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
map = _reconfigure_nozzle_layout("H1", "H1", "H1")
for x in range(96):
for _ in range(96):
_get_next_and_pickup(map)
assert _get_next_and_pickup(map) is None

subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
map = _reconfigure_nozzle_layout("H12", "H12", "H12")
for x in range(96):
for _ in range(96):
_get_next_and_pickup(map)
assert _get_next_and_pickup(map) is None
Loading