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.

Sign up for GitHub

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

feat(api): store labware objects in the stacker #18010

Open
wants to merge 2 commits into
base: edge
Choose a base branch
from
Open

feat(api): store labware objects in the stacker #18010

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fd648bc
update stacker stuff
sfoster1 Apr 1, 2025
d116172
a couple slight changes
sfoster1 Apr 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
823 changes: 821 additions & 2 deletions api/src/opentrons/protocol_engine/commands/flex_stacker/common.py
View file
Open in desktop

Large diffs are not rendered by default.

166 changes: 141 additions & 25 deletions api/src/opentrons/protocol_engine/commands/flex_stacker/empty.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

from __future__ import annotations

from __future__ import annotations
from typing import Optional, Literal, TYPE_CHECKING, Annotated
from typing_extensions import Type

Expand All @@ -15,8 +14,23 @@
)
from ...errors.exceptions import FlexStackerLabwarePoolNotYetDefinedError
from ...state import update_types
from ...types import StackerFillEmptyStrategy
from opentrons.calibration_storage.helpers import uri_from_details
from ...types import (
StackerFillEmptyStrategy,
StackerStoredLabwareGroup,
NotOnDeckLocationSequenceComponent,
OFF_DECK_LOCATION,
InStackerHopperLocation,
LabwareLocationSequence,
OnLabwareLocation,
LabwareLocation,
)
from .common import (
labware_locations_for_group,
primary_location_sequences,
adapter_location_sequences_with_default,
lid_location_sequences_with_default,
)


if TYPE_CHECKING:
from ...state.state import StateView
Expand Down Expand Up @@ -74,6 +88,49 @@ class EmptyResult(BaseModel):
None,
description="The labware definition URI of the lid labware.",
)
storedLabware: list[StackerStoredLabwareGroup] | SkipJsonSchema[None] = Field(
..., description="The primary labware loaded into the stacker labware pool."
)
removedLabware: list[StackerStoredLabwareGroup] | SkipJsonSchema[None] = Field(
...,
description="The labware objects that have just been removed from the stacker labware pool.",
)
originalPrimaryLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The previous position of each newly-removed primary labware, in the same order as removedLabware.",
)
originalAdapterLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The previous position of each newly-removed adapter labware, in the same order as removedLabware. None if the pool does not specify an adapter.",
)
originalLidLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The previous position of each newly-removed lid labware, in the same order as removedLabware. None if the pool does not specify a lid.",
)
newPrimaryLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The new position of each newly-removed primary labware, in the same order as removedLabware.",
)
newAdapterLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The new position of each newly-removed adapter labware, in the same order as removedLabware. None if the pool does not specify an adapter.",
)
newLidLabwareLocationSequences: (
list[LabwareLocationSequence] | SkipJsonSchema[None]
) = Field(
None,
description="The new position of each newly-removed lid labware, in the same order as removedLabware. None if the pool does not specify a lid labware.",
)


class EmptyImpl(AbstractCommandImpl[EmptyParams, SuccessData[EmptyResult]]):
Expand All @@ -99,11 +156,42 @@ async def execute(self, params: EmptyParams) -> SuccessData[EmptyResult]:

count = params.count if params.count is not None else 0

new_count = min(stacker_state.pool_count, count)
new_count = min(len(stacker_state.contained_labware_bottom_first), count)

new_stored_labware = stacker_state.contained_labware_bottom_first[:new_count]
removed_labware = stacker_state.contained_labware_bottom_first[new_count:]
new_locations_by_id: dict[str, LabwareLocation] = {}
new_offset_ids_by_id: dict[str, str | None] = {}

def _add_to_dicts(labware_group: StackerStoredLabwareGroup) -> None:
if labware_group.adapterLabwareId:
new_locations_by_id[labware_group.primaryLabwareId] = OnLabwareLocation(
labwareId=labware_group.adapterLabwareId
)
new_locations_by_id[labware_group.adapterLabwareId] = OFF_DECK_LOCATION
new_offset_ids_by_id[labware_group.primaryLabwareId] = None
new_offset_ids_by_id[labware_group.adapterLabwareId] = None
else:
new_locations_by_id[labware_group.primaryLabwareId] = OFF_DECK_LOCATION
new_offset_ids_by_id[labware_group.primaryLabwareId] = None
if labware_group.lidLabwareId:
new_locations_by_id[labware_group.lidLabwareId] = OnLabwareLocation(
labwareId=labware_group.primaryLabwareId
)
new_offset_ids_by_id[labware_group.lidLabwareId] = None

for group in removed_labware:
_add_to_dicts(group)

state_update = (
update_types.StateUpdate().update_flex_stacker_labware_pool_count(
params.moduleId, new_count
update_types.StateUpdate()
.update_flex_stacker_contained_labware(
module_id=params.moduleId,
contained_labware_bottom_first=new_stored_labware,
)
.set_batch_labware_location(
new_locations_by_id=new_locations_by_id,
new_offset_ids_by_id=new_offset_ids_by_id,
)
)

Expand All @@ -115,28 +203,56 @@ async def execute(self, params: EmptyParams) -> SuccessData[EmptyResult]:
"The Primary Labware must be defined in the stacker pool."
)

original_locations = [
labware_locations_for_group(
group, [InStackerHopperLocation(moduleId=params.moduleId)]
)
for group in removed_labware
]
new_locations = [
labware_locations_for_group(
group,
[
NotOnDeckLocationSequenceComponent(
logicalLocationName=OFF_DECK_LOCATION
)
],
)
for group in removed_labware
]

return SuccessData(
public=EmptyResult(
public=EmptyResult.model_construct(
count=new_count,
primaryLabwareURI=uri_from_details(
stacker_state.pool_primary_definition.namespace,
stacker_state.pool_primary_definition.parameters.loadName,
stacker_state.pool_primary_definition.version,
primaryLabwareURI=self._state_view.labware.get_uri_from_definition(
stacker_state.pool_primary_definition
),
adapterLabwareURI=self._state_view.labware.get_uri_from_definition_unless_none(
stacker_state.pool_adapter_definition
),
lidLabwareURI=self._state_view.labware.get_uri_from_definition_unless_none(
stacker_state.pool_lid_definition
),
storedLabware=new_stored_labware,
removedLabware=removed_labware,
originalPrimaryLabwareLocationSequences=primary_location_sequences(
original_locations
),
originalAdapterLabwareLocationSequences=adapter_location_sequences_with_default(
original_locations, stacker_state.pool_adapter_definition
),
originalLidLabwareLocationSequences=lid_location_sequences_with_default(
original_locations, stacker_state.pool_lid_definition
),
newPrimaryLabwareLocationSequences=primary_location_sequences(
new_locations
),
newAdapterLabwareLocationSequences=adapter_location_sequences_with_default(
new_locations, stacker_state.pool_adapter_definition
),
newLidLabwareLocationSequences=lid_location_sequences_with_default(
new_locations, stacker_state.pool_lid_definition
),
adapterLabwareURI=uri_from_details(
stacker_state.pool_adapter_definition.namespace,
stacker_state.pool_adapter_definition.parameters.loadName,
stacker_state.pool_adapter_definition.version,
)
if stacker_state.pool_adapter_definition is not None
else None,
lidLabwareURI=uri_from_details(
stacker_state.pool_lid_definition.namespace,
stacker_state.pool_lid_definition.parameters.loadName,
stacker_state.pool_lid_definition.version,
)
if stacker_state.pool_lid_definition is not None
else None,
),
state_update=state_update,
)
Expand Down
Loading
Loading