-
Notifications
You must be signed in to change notification settings - Fork 178
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(api): track volumes from multichannel configs
This PR adds the capability to properly track volume changes made by multichannel pipettes (and partial tip loadings of multichannel pipettes) to the engine. There are two ways in which we need to handle multichannel nozzle configurations specially compared to single-channel configurations. First, and what EXEC-795 is about, is that pipettes with multiple active nozzles will aspirate out of or dispense into multiple wells in an aspirate/dispense/in_place command. Which wells the pipette touches is a matter of projecting the pipette nozzle map out over the layout of the labware and predicting which wells are interacted with. This is itself non-trivial because labware can have many formats. What we can do is make the math work correctly when possible - when the labware is laid out normally enough that we can do projections of this type - and fall back to pretending to be a single channel if we fail. Since we're computing the logical equivalent of actual physical state, and if the labware is irregular it's unlikely that a multiple nozzle layout will physically work with the labware, I think this is safe. Specifically the thing we need to do is generalize the logic used in the tip store to project which tips are picked up by a multichannel to labware of different formats. Our multichannel pipette nozzles are laid out to match SBS 96-well plates, and so that's our "default" labware. On labware that follows SBS patterns but is more dense - a 384 plate, for instance - then we have to subsample, picking a single well in each group of (well_count / 96) that occupies the same space as a 96-well well to interact with. On labware that follows SBS patterns but is less dense - a 12-column reservoir, for instance - then we have to supersample, letting a labware well be touched by multiple nozzles. The second thing we have to deal with is that if the labware is a reservoir or reservoir-like - it has fewer wells than we have nozzles - then the common case is that multiple nozzles are in a well, and in that case if we're keeping track of the volume taken out of or added into a well we have to multiply the operation volume by the number of nozzles per well, which we can get by just dividing sizes without taking into account pattern overlap. Closes EXEC-795
- Loading branch information
Showing
10 changed files
with
317 additions
and
60 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,186 @@ | ||
| """Utilities for doing coverage math on wells.""" | ||
|
|
||
| from typing import Iterator | ||
| from typing_extensions import assert_never | ||
| from opentrons_shared_data.errors.exceptions import ( | ||
| InvalidStoredData, | ||
| InvalidProtocolData, | ||
| ) | ||
|
|
||
| from opentrons.hardware_control.nozzle_manager import NozzleMap | ||
|
|
||
|
|
||
| def wells_covered_by_pipette_configuration( | ||
| nozzle_map: NozzleMap, | ||
| target_well: str, | ||
| labware_wells_by_column: list[list[str]], | ||
| ) -> Iterator[str]: | ||
| """Compute the wells covered by a pipette nozzle configuration.""" | ||
| if len(labware_wells_by_column) >= 12 and len(labware_wells_by_column[0]) >= 8: | ||
| yield from wells_covered_dense( | ||
| nozzle_map, | ||
| target_well, | ||
| labware_wells_by_column, | ||
| ) | ||
| elif len(labware_wells_by_column) < 12 and len(labware_wells_by_column[0]) < 8: | ||
| yield from wells_covered_sparse( | ||
| nozzle_map, target_well, labware_wells_by_column | ||
| ) | ||
| else: | ||
| raise InvalidStoredData( | ||
| "Labware of non-SBS and non-reservoir format cannot be handled" | ||
| ) | ||
|
|
||
|
|
||
| def row_col_ordinals_from_column_major_map( | ||
| target_well: str, column_major_wells: list[list[str]] | ||
| ) -> tuple[int, int]: | ||
| """Turn a well name into the index of its row and column (in that order) within the labware.""" | ||
| for column_index, column in enumerate(column_major_wells): | ||
| if target_well in column: | ||
| return column.index(target_well), column_index | ||
| raise InvalidStoredData(f"Well name {target_well} is not present in labware") | ||
|
|
||
|
|
||
| def wells_covered_dense( | ||
| nozzle_map: NozzleMap, target_well: str, target_wells_by_column: list[list[str]] | ||
| ) -> Iterator[str]: | ||
| """Get the list of wells covered by a nozzle map on an SBS format labware with a specified multiplier of 96 into the number of wells. | ||
| This will handle the offsetting of the nozzle map into higher-density well plates. For instance, a full column config target at A1 of a | ||
| 96 plate would cover wells A1, B1, C1, D1, E1, F1, G1, H1, and use downsample_factor 1.0 (96*1 = 96). A full column config target on a | ||
| 384 plate would cover wells A1, C1, E1, G1, I1, K1, M1, O1 and use downsample_factor 4.0 (96*4 = 384), while a full column config | ||
| targeting B1 would cover wells B1, D1, F1, H1, J1, L1, N1, P1 - still using downsample_factor 4.0, with the offset gathered from the | ||
| target well. | ||
| The function may also handle sub-96 regular labware with fractional downsample factors, but that's physically improbable and it's not | ||
| tested. If you have a regular labware with fewer than 96 wells that is still regularly-spaced and has little enough space between well | ||
| walls that it's reasonable to use with multiple channels, you probably want wells_covered_trough. | ||
| """ | ||
| target_row_index, target_column_index = row_col_ordinals_from_column_major_map( | ||
| target_well, target_wells_by_column | ||
| ) | ||
| column_downsample = len(target_wells_by_column) // 12 | ||
| row_downsample = len(target_wells_by_column[0]) // 8 | ||
| if column_downsample < 1 or row_downsample < 1: | ||
| raise InvalidStoredData( | ||
| "This labware cannot be used wells_covered_dense because it is less dense than an SBS 96 standard" | ||
| ) | ||
|
|
||
| for nozzle_column in range(len(nozzle_map.columns)): | ||
| target_column_offset = nozzle_column * column_downsample | ||
| for nozzle_row in range(len(nozzle_map.rows)): | ||
| target_row_offset = nozzle_row * row_downsample | ||
| if nozzle_map.starting_nozzle == "A1": | ||
| if ( | ||
| target_column_index + target_column_offset | ||
| < len(target_wells_by_column) | ||
| ) and ( | ||
| target_row_index + target_row_offset | ||
| < len(target_wells_by_column[target_column_index]) | ||
| ): | ||
| yield target_wells_by_column[ | ||
| target_column_index + target_column_offset | ||
| ][target_row_index + target_row_offset] | ||
| elif nozzle_map.starting_nozzle == "A12": | ||
| if (target_column_index - target_column_offset >= 0) and ( | ||
| target_row_index + target_row_offset | ||
| < len(target_wells_by_column[target_column_index]) | ||
| ): | ||
| yield target_wells_by_column[ | ||
| target_column_index - target_column_offset | ||
| ][target_row_index + target_row_offset] | ||
| elif nozzle_map.starting_nozzle == "H1": | ||
| if ( | ||
| target_column_index + target_column_offset | ||
| < len(target_wells_by_column) | ||
| ) and (target_row_index - target_row_offset >= 0): | ||
| yield target_wells_by_column[ | ||
| target_column_index + target_column_offset | ||
| ][target_row_index - target_row_offset] | ||
| elif nozzle_map.starting_nozzle == "H12": | ||
| if (target_column_index - target_column_offset >= 0) and ( | ||
| target_row_index - target_row_offset >= 0 | ||
| ): | ||
| yield target_wells_by_column[ | ||
| target_column_index - target_column_offset | ||
| ][target_row_index - target_row_offset] | ||
| else: | ||
| raise InvalidProtocolData( | ||
| f"A pipette nozzle configuration may not having a starting nozzle of {nozzle_map.starting_nozzle}" | ||
| ) | ||
|
|
||
|
|
||
| def wells_covered_sparse( | ||
| nozzle_map: NozzleMap, target_well: str, target_wells_by_column: list[list[str]] | ||
| ) -> Iterator[str]: | ||
| """Get the list of wells covered by a nozzle map on a column-oriented reservoir. | ||
| This function handles reservoirs whose wells span multiple rows and columns - the most common case is something like a | ||
| 12-well reservoir, whose wells are the height of an SBS column and the width of an SBS row, or a 1-well reservoir whose well | ||
| is the size of an SBS active area. | ||
| """ | ||
| target_row_index, target_column_index = row_col_ordinals_from_column_major_map( | ||
| target_well, target_wells_by_column | ||
| ) | ||
| column_upsample = 12 // len(target_wells_by_column) | ||
| row_upsample = 8 // len(target_wells_by_column[0]) | ||
| if column_upsample < 1 or row_upsample < 1: | ||
| raise InvalidStoredData( | ||
| "This labware cannot be uased with wells_covered_sparse because it is more dense than an SBS 96 standard." | ||
| ) | ||
| for nozzle_column in range(max(1, len(nozzle_map.columns) // column_upsample)): | ||
| for nozzle_row in range(max(1, len(nozzle_map.rows) // row_upsample)): | ||
| if nozzle_map.starting_nozzle == "A1": | ||
| if ( | ||
| target_column_index + nozzle_column < len(target_wells_by_column) | ||
| ) and ( | ||
| target_row_index + nozzle_row | ||
| < len(target_wells_by_column[target_column_index]) | ||
| ): | ||
| yield target_wells_by_column[target_column_index + nozzle_column][ | ||
| target_row_index + nozzle_row | ||
| ] | ||
| elif nozzle_map.starting_nozzle == "A12": | ||
| if (target_column_index - nozzle_column >= 0) and ( | ||
| target_row_index + nozzle_row | ||
| < len(target_wells_by_column[target_column_index]) | ||
| ): | ||
| yield target_wells_by_column[ | ||
| target_column_index - nozzle_column | ||
| ][target_row_index + nozzle_row] | ||
| elif nozzle_map.starting_nozzle == "H1": | ||
| if ( | ||
| target_column_index + nozzle_column | ||
| < len(target_wells_by_column[target_column_index]) | ||
| ) and (target_row_index - nozzle_row >= 0): | ||
| yield target_wells_by_column[ | ||
| target_column_index + nozzle_column | ||
| ][target_row_index - nozzle_row] | ||
| elif nozzle_map.starting_nozzle == "H12": | ||
| if (target_column_index - nozzle_column >= 0) and ( | ||
| target_row_index - nozzle_row >= 0 | ||
| ): | ||
| yield target_wells_by_column[ | ||
| target_column_index - nozzle_column | ||
| ][target_row_index - nozzle_row] | ||
| else: | ||
| raise InvalidProtocolData( | ||
| f"A pipette nozzle configuration may not having a starting nozzle of {nozzle_map.starting_nozzle}" | ||
| ) | ||
|
|
||
|
|
||
| def nozzles_per_well( | ||
| nozzle_map: NozzleMap, target_well: str, target_wells_by_column: list[list[str]] | ||
| ) -> int: | ||
| _, target_column_index = row_col_ordinals_from_column_major_map( | ||
| target_well, target_wells_by_column | ||
| ) | ||
| # labware as or more dense than a 96 plate will only ever have 1 nozzle per well (and some wells won't be touched) | ||
| if len(target_wells_by_column) >= len(nozzle_map.columns) and len( | ||
| target_wells_by_column[target_column_index] | ||
| ) >= len(nozzle_map.rows): | ||
| return 1 | ||
| return max(1, len(nozzle_map.columns) // len(target_wells_by_column)) * max( | ||
| 1, len(nozzle_map.rows) // len(target_wells_by_column[target_column_index]) | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.