From f3a8400cc8b42753c6e5a4fb711d673349a9088d Mon Sep 17 00:00:00 2001 From: tamarzanzouri Date: Fri, 28 Jul 2023 17:06:16 -0400 Subject: [PATCH] WIP - use str/int as slots --- .../motion_planning/deck_conflict.py | 30 +++++++++---------- .../protocol_api/core/engine/deck_conflict.py | 14 ++++----- .../protocol_api/core/legacy/deck.py | 2 +- 3 files changed, 22 insertions(+), 24 deletions(-) diff --git a/api/src/opentrons/motion_planning/deck_conflict.py b/api/src/opentrons/motion_planning/deck_conflict.py index 00a48242aa4..94267589c6b 100644 --- a/api/src/opentrons/motion_planning/deck_conflict.py +++ b/api/src/opentrons/motion_planning/deck_conflict.py @@ -99,9 +99,9 @@ class OtherModule(_Module): class _NothingAllowed(NamedTuple): """Nothing is allowed in this slot.""" - location: int + location: Union[int, str] source_item: DeckItem - source_location: int + source_location: Union[int, str] def is_allowed(self, item: DeckItem) -> bool: return False @@ -129,9 +129,9 @@ def is_allowed(self, item: DeckItem) -> bool: class _NoModule(NamedTuple): """No module of any kind is allowed in this slot.""" - location: int + location: Union[int, str] source_item: DeckItem - source_location: int + source_location: Union[int, str] def is_allowed(self, item: DeckItem) -> bool: return not isinstance(item, _Module) @@ -175,9 +175,9 @@ class DeckConflictError(ValueError): # things that don't fit into a single deck slot, like the Thermocycler. # Refactor this interface to take a more symbolic location. def check( - existing_items: Mapping[int, DeckItem], + existing_items: Mapping[Union[int, str], DeckItem], new_item: DeckItem, - new_location: int, + new_location: Union[int, str], robot_type: str = "OT-2 Standard", ) -> None: """Check a deck layout for conflicts. @@ -190,7 +190,6 @@ def check( Raises: DeckConflictError: Adding this item should not be allowed. """ - print(f"robot_type: {robot_type}") restrictions: List[_DeckRestriction] = [_FixedTrashOnly()] # build restrictions driven by existing items @@ -284,7 +283,7 @@ def _create_ot2_restrictions(item: DeckItem, location: int) -> List[_DeckRestric return restrictions -def _create_flex_restrictions(item: DeckItem, location: int) -> List[_DeckRestriction]: +def _create_flex_restrictions(item: DeckItem, location: str) -> List[_DeckRestriction]: restrictions: List[_DeckRestriction] = [] if isinstance(item, ThermocyclerModule): @@ -323,12 +322,13 @@ def _create_flex_restrictions(item: DeckItem, location: int) -> List[_DeckRestri def _create_restrictions( - item: DeckItem, location: int, robot_type: str + item: DeckItem, location: Union[int, str], robot_type: str ) -> List[_DeckRestriction]: if robot_type == "OT-2 Standard": - return _create_ot2_restrictions(item, location) + return _create_ot2_restrictions(item, int(location)) else: + assert isinstance(location, str) return _create_flex_restrictions(item, location) @@ -373,16 +373,16 @@ def _flex_slots_covered_by_thermocycler() -> Set[int]: return {7, 10} -def _flex_right_left_hand_slots() -> List[int]: +def _flex_right_left_hand_slots() -> List[str]: return _flex_left_hand_slots() + _flex_left_hand_slots() -def _flex_left_hand_slots() -> List[int]: - return [1, 4, 7, 10] +def _flex_left_hand_slots() -> List[str]: + return ["A1", "B1", "C1", "D1"] -def _flex_right_hand_slots() -> List[int]: - return [3, 6, 9, 12] +def _flex_right_hand_slots() -> List[str]: + return ["A3", "B3", "C3", "D3"] def _is_fixed_trash(item: DeckItem) -> bool: diff --git a/api/src/opentrons/protocol_api/core/engine/deck_conflict.py b/api/src/opentrons/protocol_api/core/engine/deck_conflict.py index db51536e8ea..04b911d3ad1 100644 --- a/api/src/opentrons/protocol_api/core/engine/deck_conflict.py +++ b/api/src/opentrons/protocol_api/core/engine/deck_conflict.py @@ -96,7 +96,7 @@ def check( ) mapped_existing_modules = (m for m in all_existing_modules if m is not None) - existing_items: Dict[int, wrapped_deck_conflict.DeckItem] = {} + existing_items: Dict[str, wrapped_deck_conflict.DeckItem] = {} for existing_location, existing_item in itertools.chain( mapped_existing_labware, mapped_existing_modules ): @@ -104,7 +104,7 @@ def check( existing_items[existing_location] = existing_item wrapped_deck_conflict.check( - existing_items=existing_items, + existing_items=existing_items, # type: ignore[arg-type] new_item=new_item, new_location=new_location, robot_type=engine_state.config.robot_type, @@ -114,14 +114,14 @@ def check( def _map_labware( engine_state: StateView, labware_id: str, -) -> Optional[Tuple[int, wrapped_deck_conflict.DeckItem]]: +) -> Optional[Tuple[str, wrapped_deck_conflict.DeckItem]]: location_from_engine = engine_state.labware.get_location(labware_id=labware_id) if isinstance(location_from_engine, DeckSlotLocation): # This labware is loaded directly into a deck slot. # Map it to a wrapped_deck_conflict.Labware. return ( - _deck_slot_to_int(location_from_engine), + location_from_engine.slotName.id, wrapped_deck_conflict.Labware( name_for_errors=engine_state.labware.get_load_name( labware_id=labware_id @@ -154,12 +154,10 @@ def _map_labware( def _map_module( engine_state: StateView, module_id: str, -) -> Optional[Tuple[int, wrapped_deck_conflict.DeckItem]]: +) -> Optional[Tuple[str, wrapped_deck_conflict.DeckItem]]: module_model = engine_state.modules.get_connected_model(module_id=module_id) module_type = module_model.as_type() - mapped_location = _deck_slot_to_int( - engine_state.modules.get_location(module_id=module_id) - ) + mapped_location = engine_state.modules.get_location(module_id=module_id).slotName.id # Use the module model (e.g. "temperatureModuleV1") as the name for error messages # because it's convenient for us. Unfortunately, this won't necessarily match diff --git a/api/src/opentrons/protocol_api/core/legacy/deck.py b/api/src/opentrons/protocol_api/core/legacy/deck.py index 54bca33742b..7fb10022c17 100644 --- a/api/src/opentrons/protocol_api/core/legacy/deck.py +++ b/api/src/opentrons/protocol_api/core/legacy/deck.py @@ -174,7 +174,7 @@ def __setitem__(self, key: DeckLocation, val: DeckItem) -> None: # will raise DeckConflictError if items conflict deck_conflict.check( - existing_items=existing_items, + existing_items=existing_items, # type: ignore[arg-type] new_location=slot_key_int, new_item=self._map_to_conflict_checker_item(val), )