Skip to content

Commit

Permalink
remove labware params from params and refactor
Browse files Browse the repository at this point in the history
  • Loading branch information
@sfoster1
sfoster1 committed Feb 24, 2025
1 parent 92bd4f6 commit b92b97f
Show file tree
Hide file tree
Showing 5 changed files with 238 additions and 200 deletions.
11 changes: 0 additions & 11 deletions api/src/opentrons/protocol_api/core/engine/module_core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -725,20 +725,9 @@ def retrieve(self) -> None:
raise CannotPerformModuleAction(
f"Flex Stacker {self.module_id} has no labware to retrieve"
)
custom_labware_params = (
self._engine_client.state.labware.find_custom_labware_load_params()
)
namespace, version = load_labware_params.resolve(
stacker.pool_primary_definition.parameters.loadName,
None,
None,
custom_labware_params,
)
self._engine_client.execute_command(
cmd.flex_stacker.RetrieveParams(
moduleId=self.module_id,
namespace=namespace,
version=version,
)
)

Expand Down
157 changes: 75 additions & 82 deletions api/src/opentrons/protocol_engine/commands/flex_stacker/retrieve.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

if TYPE_CHECKING:
from opentrons.protocol_engine.state.state import StateView
from opentrons.protocol_engine.state.module_substates import FlexStackerSubState
from opentrons.protocol_engine.execution import EquipmentHandler

RetrieveCommandType = Literal["flexStacker/retrieve"]
Expand All @@ -41,14 +42,6 @@ class RetrieveParams(BaseModel):
...,
description="Unique ID of the Flex Stacker.",
)
namespace: str = Field(
...,
description="The namespace the lid labware definition belongs to.",
)
version: int = Field(
...,
description="The lid labware definition version.",
)
labwareId: str | SkipJsonSchema[None] = Field(
None,
description="An optional ID to assign to this labware. If None, an ID "
Expand All @@ -59,9 +52,6 @@ class RetrieveParams(BaseModel):
None,
description="An optional user-specified display name "
"or label for this labware.",
# NOTE: v4/5 JSON protocols will always have a displayName which will be the
# user-specified label OR the displayName property of the labware's definition.
# TODO: Make sure v6 JSON protocols don't do that.
json_schema_extra=_remove_default,
)
adapterId: str | SkipJsonSchema[None] = Field(
Expand All @@ -70,15 +60,6 @@ class RetrieveParams(BaseModel):
"will be generated.",
json_schema_extra=_remove_default,
)
adapterDisplayName: str | SkipJsonSchema[None] = Field(
None,
description="An optional user-specified display name "
"or label for an adapter.",
# NOTE: v4/5 JSON protocols will always have a displayName which will be the
# user-specified label OR the displayName property of the labware's definition.
# TODO: Make sure v6 JSON protocols don't do that.
json_schema_extra=_remove_default,
)
lidId: str | SkipJsonSchema[None] = Field(
None,
description="An optional ID to assign to a lid. If None, an ID "
Expand Down Expand Up @@ -138,28 +119,9 @@ def __init__(
self._state_view = state_view
self._equipment = equipment

async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
"""Execute the labware retrieval command."""
stacker_state = self._state_view.modules.get_flex_stacker_substate(
params.moduleId
)

if stacker_state.in_static_mode:
raise CannotPerformModuleAction(
"Cannot retrieve labware from Flex Stacker while in static mode"
)

stacker_loc = ModuleLocation(moduleId=params.moduleId)
# Allow propagation of ModuleNotAttachedError.
stacker_hw = self._equipment.get_module_hardware_api(stacker_state.module_id)

try:
self._state_view.labware.raise_if_labware_in_location(stacker_loc)
except LocationIsOccupiedError:
raise CannotPerformModuleAction(
"Cannot retrieve a labware from Flex Stacker if the carriage is occupied"
)

async def _load_labware_from_pool(
self, params: RetrieveParams, stacker_state: FlexStackerSubState
) -> tuple[RetrieveResult, update_types.StateUpdate]:
state_update = update_types.StateUpdate()

# If there is an adapter load it
Expand All @@ -170,10 +132,8 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
display_names_by_id: Dict[str, str | None] = {}
new_locations_by_id: Dict[str, LabwareLocation] = {}
if stacker_state.pool_adapter_definition is not None:
adapter_lw = await self._equipment.load_labware(
load_name=stacker_state.pool_adapter_definition.parameters.loadName,
namespace=params.namespace,
version=params.version,
adapter_lw = await self._equipment.load_labware_from_definition(
definition=stacker_state.pool_adapter_definition,
location=ModuleLocation(moduleId=params.moduleId),
labware_id=params.adapterId,
)
Expand All @@ -190,13 +150,13 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
raise CannotPerformModuleAction(
f"Flex Stacker {params.moduleId} has no labware to retrieve"
)
loaded_labware = await self._equipment.load_labware(
load_name=stacker_state.pool_primary_definition.parameters.loadName,
namespace=params.namespace,
version=params.version,
location=ModuleLocation(moduleId=params.moduleId)
if adapter_lw is None
else OnLabwareLocation(labwareId=adapter_lw.labware_id),
loaded_labware = await self._equipment.load_labware_from_definition(
definition=stacker_state.pool_primary_definition,
location=(
ModuleLocation(moduleId=params.moduleId)
if adapter_lw is None
else OnLabwareLocation(labwareId=adapter_lw.labware_id)
),
labware_id=params.labwareId,
)
definitions_by_id[loaded_labware.labware_id] = loaded_labware.definition
Expand All @@ -211,10 +171,8 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
)
# If there is a lid load it
if stacker_state.pool_lid_definition is not None:
lid_lw = await self._equipment.load_labware(
load_name=stacker_state.pool_lid_definition.parameters.loadName,
namespace=params.namespace,
version=params.version,
lid_lw = await self._equipment.load_labware_from_definition(
definition=stacker_state.pool_lid_definition,
location=OnLabwareLocation(labwareId=loaded_labware.labware_id),
labware_id=params.lidId,
)
Expand Down Expand Up @@ -280,26 +238,6 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
else None
)

labware_height = self._state_view.geometry.get_height_of_labware_stack(
definitions=list(definitions_by_id.values())
)

if stacker_hw is not None:
await stacker_hw.dispense_labware(labware_height=labware_height)

# Update the state to reflect the labware is now in the Flex Stacker slot
# todo(chb, 2025-02-19): This ModuleLocation piece should probably instead be an AddressableAreaLocation
# but that has implications for where labware are set by things like module.load_labware(..) and what
# happens when we move labware.
stacker_area = (
self._state_view.modules.ensure_and_convert_module_fixture_location(
deck_slot=self._state_view.modules.get_location(
params.moduleId
).slotName,
model=self._state_view.modules.get(params.moduleId).model,
)
)
state_update.set_addressable_area_used(stacker_area)
state_update.set_batch_loaded_labware(
definitions_by_id=definitions_by_id,
display_names_by_id=display_names_by_id,
Expand All @@ -316,11 +254,8 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
lid_ids=[lid_lw.labware_id],
)

state_update.retrieve_flex_stacker_labware(
module_id=params.moduleId, labware_id=loaded_labware.labware_id
)
return SuccessData(
public=RetrieveResult(
return (
RetrieveResult(
labwareId=loaded_labware.labware_id,
adapterId=adapter_lw.labware_id if adapter_lw is not None else None,
lidId=lid_lw.labware_id if lid_lw is not None else None,
Expand All @@ -331,6 +266,64 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
adapterLabwareURI=adapter_uri,
lidLabwareURI=lid_uri,
),
state_update,
)

async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
"""Execute the labware retrieval command."""
stacker_state = self._state_view.modules.get_flex_stacker_substate(
params.moduleId
)

if stacker_state.in_static_mode:
raise CannotPerformModuleAction(
"Cannot retrieve labware from Flex Stacker while in static mode"
)

stacker_loc = ModuleLocation(moduleId=params.moduleId)
# Allow propagation of ModuleNotAttachedError.
stacker_hw = self._equipment.get_module_hardware_api(stacker_state.module_id)

try:
self._state_view.labware.raise_if_labware_in_location(stacker_loc)
except LocationIsOccupiedError:
raise CannotPerformModuleAction(
"Cannot retrieve a labware from Flex Stacker if the carriage is occupied"
)

retrieve_result, state_update = await self._load_labware_from_pool(
params, stacker_state
)

assert state_update.batch_loaded_labware != update_types.NO_CHANGE
labware_height = self._state_view.geometry.get_height_of_labware_stack(
definitions=list(
state_update.batch_loaded_labware.definitions_by_id.values()
)
)

if stacker_hw is not None:
await stacker_hw.dispense_labware(labware_height=labware_height)

# Update the state to reflect the labware is now in the Flex Stacker slot
# todo(chb, 2025-02-19): This ModuleLocation piece should probably instead be an AddressableAreaLocation
# but that has implications for where labware are set by things like module.load_labware(..) and what
# happens when we move labware.
stacker_area = (
self._state_view.modules.ensure_and_convert_module_fixture_location(
deck_slot=self._state_view.modules.get_location(
params.moduleId
).slotName,
model=self._state_view.modules.get(params.moduleId).model,
)
)
state_update.set_addressable_area_used(stacker_area)

state_update.retrieve_flex_stacker_labware(
module_id=params.moduleId, labware_id=retrieve_result.labwareId
)
return SuccessData(
public=retrieve_result,
state_update=state_update,
)

Expand Down
52 changes: 39 additions & 13 deletions api/src/opentrons/protocol_engine/execution/equipment.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,43 @@ async def load_definition_for_details(
)
return definition, definition_uri

async def load_labware_from_definition(
self,
definition: LabwareDefinition,
location: LabwareLocation,
labware_id: Optional[str],
) -> LoadedLabwareData:
"""Load labware from already-found definition."""
definition_uri = uri_from_details(
load_name=definition.parameters.loadName,
namespace=definition.namespace,
version=definition.version,
)
return await self._load_labware_from_def_and_uri(
definition, definition_uri, location, labware_id
)

async def _load_labware_from_def_and_uri(
self,
definition: LabwareDefinition,
definition_uri: str,
location: LabwareLocation,
labware_id: Optional[str],
) -> LoadedLabwareData:
labware_id = (
labware_id if labware_id is not None else self._model_utils.generate_id()
)

# Allow propagation of ModuleNotLoadedError.
offset_id = self.find_applicable_labware_offset_id(
labware_definition_uri=definition_uri,
labware_location=location,
)

return LoadedLabwareData(
labware_id=labware_id, definition=definition, offsetId=offset_id
)

async def load_labware(
self,
load_name: str,
Expand Down Expand Up @@ -193,19 +230,8 @@ async def load_labware(
definition, definition_uri = await self.load_definition_for_details(
load_name, namespace, version
)

labware_id = (
labware_id if labware_id is not None else self._model_utils.generate_id()
)

# Allow propagation of ModuleNotLoadedError.
offset_id = self.find_applicable_labware_offset_id(
labware_definition_uri=definition_uri,
labware_location=location,
)

return LoadedLabwareData(
labware_id=labware_id, definition=definition, offsetId=offset_id
return await self._load_labware_from_def_and_uri(
definition, definition_uri, location, labware_id
)

async def reload_labware(self, labware_id: str) -> ReloadedLabwareData:
Expand Down
Loading

0 comments on commit b92b97f

Please sign in to comment.