Skip to content

Commit

Permalink
Consolidate parallel dicts.
Browse files Browse the repository at this point in the history
This makes it harder to forget to update one of them.
  • Loading branch information
SyntaxColoring committed Oct 11, 2024
1 parent 646487c commit 9121c0b
Showing 1 changed file with 29 additions and 21 deletions.
50 changes: 29 additions & 21 deletions api/src/opentrons/protocol_engine/state/tips.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,16 +35,21 @@ class TipRackWellState(Enum):
TipRackStateByWellName = Dict[str, TipRackWellState]


@dataclass
class _PipetteInfo:
channels: int
active_channels: int
nozzle_map: NozzleMap


@dataclass
class TipState:
"""State of all tips."""

tips_by_labware_id: Dict[str, TipRackStateByWellName]
column_by_labware_id: Dict[str, List[List[str]]]

channels_by_pipette_id: Dict[str, int]
active_channels_by_pipette_id: Dict[str, int]
nozzle_map_by_pipette_id: Dict[str, NozzleMap]
pipette_info_by_pipette_id: Dict[str, _PipetteInfo]


class TipStore(HasState[TipState], HandlesActions):
Expand All @@ -57,9 +62,7 @@ def __init__(self) -> None:
self._state = TipState(
tips_by_labware_id={},
column_by_labware_id={},
channels_by_pipette_id={},
active_channels_by_pipette_id={},
nozzle_map_by_pipette_id={},
pipette_info_by_pipette_id={},
)

def handle_action(self, action: Action) -> None:
Expand All @@ -68,23 +71,25 @@ def handle_action(self, action: Action) -> None:
if isinstance(action.private_result, PipetteConfigUpdateResultMixin):
pipette_id = action.private_result.pipette_id
config = action.private_result.config
self._state.channels_by_pipette_id[pipette_id] = config.channels
self._state.active_channels_by_pipette_id[pipette_id] = config.channels
self._state.nozzle_map_by_pipette_id[pipette_id] = config.nozzle_map
self._state.pipette_info_by_pipette_id[pipette_id] = _PipetteInfo(
channels=config.channels,
active_channels=config.channels,
nozzle_map=config.nozzle_map,
)

self._handle_succeeded_command(action.command)

if isinstance(action.private_result, PipetteNozzleLayoutResultMixin):
pipette_id = action.private_result.pipette_id
nozzle_map = action.private_result.nozzle_map
pipette_info = self._state.pipette_info_by_pipette_id[pipette_id]
if nozzle_map:
self._state.active_channels_by_pipette_id[
pipette_id
] = nozzle_map.tip_count
self._state.nozzle_map_by_pipette_id[pipette_id] = nozzle_map
pipette_info.active_channels = nozzle_map.tip_count
pipette_info.nozzle_map = nozzle_map
else:
self._state.active_channels_by_pipette_id[
pipette_id
] = self._state.channels_by_pipette_id[pipette_id]
# todo(mm, 2024-10-10): nozzle_map looks always truthy--can this
# else-block actually run?
pipette_info.active_channels = pipette_info.channels

elif isinstance(action, FailCommandAction):
self._handle_failed_command(action)
Expand Down Expand Up @@ -148,7 +153,7 @@ def _set_used_tips( # noqa: C901
) -> None:
columns = self._state.column_by_labware_id.get(labware_id, [])
wells = self._state.tips_by_labware_id.get(labware_id, {})
nozzle_map = self._state.nozzle_map_by_pipette_id[pipette_id]
nozzle_map = self._state.pipette_info_by_pipette_id[pipette_id].nozzle_map

# TODO (cb, 02-28-2024): Transition from using partial nozzle map to full instrument map for the set used logic
num_nozzle_cols = len(nozzle_map.columns)
Expand Down Expand Up @@ -463,19 +468,22 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]:

def get_pipette_channels(self, pipette_id: str) -> int:
"""Return the given pipette's number of channels."""
return self._state.channels_by_pipette_id[pipette_id]
return self._state.pipette_info_by_pipette_id[pipette_id].channels

def get_pipette_active_channels(self, pipette_id: str) -> int:
"""Get the number of channels being used in the given pipette's configuration."""
return self._state.active_channels_by_pipette_id[pipette_id]
return self._state.pipette_info_by_pipette_id[pipette_id].active_channels

def get_pipette_nozzle_map(self, pipette_id: str) -> NozzleMap:
"""Get the current nozzle map the given pipette's configuration."""
return self._state.nozzle_map_by_pipette_id[pipette_id]
return self._state.pipette_info_by_pipette_id[pipette_id].nozzle_map

def get_pipette_nozzle_maps(self) -> Dict[str, NozzleMap]:
"""Get current nozzle maps keyed by pipette id."""
return self._state.nozzle_map_by_pipette_id
return {
pipette_id: pipette_info.nozzle_map
for pipette_id, pipette_info in self._state.pipette_info_by_pipette_id.items()
}

def has_clean_tip(self, labware_id: str, well_name: str) -> bool:
"""Get whether a well in a labware has a clean tip.
Expand Down

0 comments on commit 9121c0b

Please sign in to comment.