diff --git a/api/src/opentrons/protocol_engine/commands/aspirate.py b/api/src/opentrons/protocol_engine/commands/aspirate.py index 14b59248216..8e3dcde80eb 100644 --- a/api/src/opentrons/protocol_engine/commands/aspirate.py +++ b/api/src/opentrons/protocol_engine/commands/aspirate.py @@ -140,6 +140,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn: command_note_adder=self._command_note_adder, ) except PipetteOverpressureError as e: + # can we get aspirated_amount_prior_to_error? If not, can we assume no liquid was removed from well? Ask Ryan return DefinedErrorData( public=OverpressureError( id=self._model_utils.generate_id(), @@ -156,6 +157,11 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn: state_update=state_update, ) else: + state_update.set_operated_liquid( + labware_id=labware_id, + well_name=well_name, + volume=-volume_aspirated, + ) return SuccessData( public=AspirateResult( volume=volume_aspirated, diff --git a/api/src/opentrons/protocol_engine/commands/dispense.py b/api/src/opentrons/protocol_engine/commands/dispense.py index 7e18cc6560b..1285d159360 100644 --- a/api/src/opentrons/protocol_engine/commands/dispense.py +++ b/api/src/opentrons/protocol_engine/commands/dispense.py @@ -107,6 +107,7 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn: push_out=params.pushOut, ) except PipetteOverpressureError as e: + # can we get aspirated_amount_prior_to_error? If not, can we assume no liquid was removed from well? Ask Ryan return DefinedErrorData( public=OverpressureError( id=self._model_utils.generate_id(), @@ -123,6 +124,11 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn: state_update=state_update, ) else: + state_update.set_operated_liquid( + labware_id=labware_id, + well_name=well_name, + volume=volume, + ) return SuccessData( public=DispenseResult(volume=volume, position=deck_point), private=None, diff --git a/api/src/opentrons/protocol_engine/commands/liquid_probe.py b/api/src/opentrons/protocol_engine/commands/liquid_probe.py index 1a8597f9c03..b58fea92219 100644 --- a/api/src/opentrons/protocol_engine/commands/liquid_probe.py +++ b/api/src/opentrons/protocol_engine/commands/liquid_probe.py @@ -205,6 +205,13 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn: self._state_view, self._movement, self._pipetting, params ) if isinstance(z_pos_or_error, PipetteLiquidNotFoundError): + state_update.set_probed_liquid( + labware_id=params.labwareId, + well_name=params.wellName, + height=None, + volume=None, + last_probed=self._model_utils.get_timestamp(), + ) return DefinedErrorData( public=LiquidNotFoundError( id=self._model_utils.generate_id(), @@ -220,6 +227,18 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn: state_update=state_update, ) else: + well_volume = self._state_view.geometry.get_well_volume_at_height( + labware_id=params.labwareId, + well_name=params.wellName, + height=z_pos_or_error, + ) + state_update.set_probed_liquid( + labware_id=params.labwareId, + well_name=params.wellName, + height=z_pos_or_error, + volume=well_volume, + last_probed=self._model_utils.get_timestamp(), + ) return SuccessData( public=LiquidProbeResult( z_position=z_pos_or_error, position=deck_point @@ -239,11 +258,13 @@ def __init__( state_view: StateView, movement: MovementHandler, pipetting: PipettingHandler, + model_utils: ModelUtils, **kwargs: object, ) -> None: self._state_view = state_view self._movement = movement self._pipetting = pipetting + self._model_utils = model_utils async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn: """Execute a `tryLiquidProbe` command. @@ -256,11 +277,23 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn: self._state_view, self._movement, self._pipetting, params ) - z_pos = ( - None - if isinstance(z_pos_or_error, PipetteLiquidNotFoundError) - else z_pos_or_error + if isinstance(z_pos_or_error, PipetteLiquidNotFoundError): + z_pos = None + well_volume = None + else: + z_pos = z_pos_or_error + well_volume = self._state_view.geometry.get_well_volume_at_height( + labware_id=params.labwareId, well_name=params.wellName, height=z_pos + ) + + state_update.set_probed_liquid( + labware_id=params.labwareId, + well_name=params.wellName, + height=z_pos, + volume=well_volume, + last_probed=self._model_utils.get_timestamp(), ) + return SuccessData( public=TryLiquidProbeResult( z_position=z_pos, diff --git a/api/src/opentrons/protocol_engine/commands/load_liquid.py b/api/src/opentrons/protocol_engine/commands/load_liquid.py index 856cf3ee127..d5fdfd13ed2 100644 --- a/api/src/opentrons/protocol_engine/commands/load_liquid.py +++ b/api/src/opentrons/protocol_engine/commands/load_liquid.py @@ -4,11 +4,14 @@ from typing import Optional, Type, Dict, TYPE_CHECKING from typing_extensions import Literal +from opentrons.protocol_engine.state.update_types import StateUpdate + from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData from ..errors.error_occurrence import ErrorOccurrence if TYPE_CHECKING: from ..state.state import StateView + from ..resources import ModelUtils LoadLiquidCommandType = Literal["loadLiquid"] @@ -41,8 +44,11 @@ class LoadLiquidImplementation( ): """Load liquid command implementation.""" - def __init__(self, state_view: StateView, **kwargs: object) -> None: + def __init__( + self, state_view: StateView, model_utils: ModelUtils, **kwargs: object + ) -> None: self._state_view = state_view + self._model_utils = model_utils async def execute( self, params: LoadLiquidParams @@ -54,7 +60,16 @@ async def execute( labware_id=params.labwareId, wells=params.volumeByWell ) - return SuccessData(public=LoadLiquidResult(), private=None) + state_update = StateUpdate() + state_update.set_loaded_liquid( + labware_id=params.labwareId, + volumes=params.volumeByWell, + last_loaded=self._model_utils.get_timestamp(), + ) + + return SuccessData( + public=LoadLiquidResult(), private=None, state_update=state_update + ) class LoadLiquid(BaseCommand[LoadLiquidParams, LoadLiquidResult, ErrorOccurrence]): diff --git a/api/src/opentrons/protocol_engine/state/geometry.py b/api/src/opentrons/protocol_engine/state/geometry.py index 9783508800c..0ebc5eca720 100644 --- a/api/src/opentrons/protocol_engine/state/geometry.py +++ b/api/src/opentrons/protocol_engine/state/geometry.py @@ -1457,6 +1457,15 @@ def get_well_height_at_volume( target_volume=volume, well_geometry=well_geometry ) + def get_well_volume_at_height( + self, labware_id: str, well_name: str, height: float + ) -> float: + """Convert well height to volume.""" + well_geometry = self._labware.get_well_geometry(labware_id, well_name) + return find_volume_at_well_height( + target_height=height, well_geometry=well_geometry + ) + def validate_dispense_volume_into_well( self, labware_id: str, diff --git a/api/src/opentrons/protocol_engine/state/update_types.py b/api/src/opentrons/protocol_engine/state/update_types.py index 5d941d33933..91bc8c7d0d2 100644 --- a/api/src/opentrons/protocol_engine/state/update_types.py +++ b/api/src/opentrons/protocol_engine/state/update_types.py @@ -3,7 +3,8 @@ import dataclasses import enum -import typing +from typing import Final, TypeAlias, Literal, Dict, Optional, overload +from datetime import datetime from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_engine.resources import pipette_data_provider @@ -17,14 +18,14 @@ class _NoChangeEnum(enum.Enum): NO_CHANGE = enum.auto() -NO_CHANGE: typing.Final = _NoChangeEnum.NO_CHANGE +NO_CHANGE: Final = _NoChangeEnum.NO_CHANGE """A sentinel value to indicate that a value shouldn't be changed. Useful when `None` is semantically unclear or already has some other meaning. """ -NoChangeType: typing.TypeAlias = typing.Literal[_NoChangeEnum.NO_CHANGE] +NoChangeType: TypeAlias = Literal[_NoChangeEnum.NO_CHANGE] """The type of `NO_CHANGE`, as `NoneType` is to `None`. Unfortunately, mypy doesn't let us write `Literal[NO_CHANGE]`. Use this instead. @@ -35,14 +36,14 @@ class _ClearEnum(enum.Enum): CLEAR = enum.auto() -CLEAR: typing.Final = _ClearEnum.CLEAR +CLEAR: Final = _ClearEnum.CLEAR """A sentinel value to indicate that a value should be cleared. Useful when `None` is semantically unclear or has some other meaning. """ -ClearType: typing.TypeAlias = typing.Literal[_ClearEnum.CLEAR] +ClearType: TypeAlias = Literal[_ClearEnum.CLEAR] """The type of `CLEAR`, as `NoneType` is to `None`. Unfortunately, mypy doesn't let us write `Literal[CLEAR]`. Use this instead. @@ -91,7 +92,7 @@ class LabwareLocationUpdate: new_location: LabwareLocation """The labware's new location.""" - offset_id: typing.Optional[str] + offset_id: Optional[str] """The ID of the labware's new offset, for its new location.""" @@ -105,10 +106,10 @@ class LoadedLabwareUpdate: new_location: LabwareLocation """The labware's initial location.""" - offset_id: typing.Optional[str] + offset_id: Optional[str] """The ID of the labware's offset.""" - display_name: typing.Optional[str] + display_name: Optional[str] definition: LabwareDefinition @@ -126,7 +127,7 @@ class LoadPipetteUpdate: pipette_name: PipetteNameType mount: MountType - liquid_presence_detection: typing.Optional[bool] + liquid_presence_detection: Optional[bool] @dataclasses.dataclass @@ -155,7 +156,7 @@ class PipetteTipStateUpdate: """Update pipette tip state.""" pipette_id: str - tip_geometry: typing.Optional[TipGeometry] + tip_geometry: Optional[TipGeometry] @dataclasses.dataclass @@ -175,6 +176,35 @@ class TipsUsedUpdate: """ +@dataclasses.dataclass +class LoadLiquidUpdate: + """An update from loading a liquid.""" + + labware_id: str + volumes: Dict[str, float] + last_loaded: datetime + + +@dataclasses.dataclass +class ProbeLiquidUpdate: + """An update from probing a liquid.""" + + labware_id: str + well_name: str + last_probed: datetime + height: Optional[float] = None + volume: Optional[float] = None + + +@dataclasses.dataclass +class OperateLiquidUpdate: + """An update from operating a liquid.""" + + labware_id: str + well_name: str + volume: float + + @dataclasses.dataclass class StateUpdate: """Represents an update to perform on engine state.""" @@ -195,10 +225,16 @@ class StateUpdate: tips_used: TipsUsedUpdate | NoChangeType = NO_CHANGE + loaded_liquid: LoadLiquidUpdate | NoChangeType = NO_CHANGE + + probed_liquid: ProbeLiquidUpdate | NoChangeType = NO_CHANGE + + operated_liquid: OperateLiquidUpdate | NoChangeType = NO_CHANGE + # These convenience functions let the caller avoid the boilerplate of constructing a # complicated dataclass tree. - @typing.overload + @overload def set_pipette_location( self, *, @@ -209,7 +245,7 @@ def set_pipette_location( ) -> None: """Schedule a pipette's location to be set to a well.""" - @typing.overload + @overload def set_pipette_location( self, *, @@ -270,8 +306,8 @@ def set_loaded_labware( self, definition: LabwareDefinition, labware_id: str, - offset_id: typing.Optional[str], - display_name: typing.Optional[str], + offset_id: Optional[str], + display_name: Optional[str], location: LabwareLocation, ) -> None: """Add a new labware to state. See `LoadedLabwareUpdate`.""" @@ -288,7 +324,7 @@ def set_load_pipette( pipette_id: str, pipette_name: PipetteNameType, mount: MountType, - liquid_presence_detection: typing.Optional[bool], + liquid_presence_detection: Optional[bool], ) -> None: """Add a new pipette to state. See `LoadPipetteUpdate`.""" self.loaded_pipette = LoadPipetteUpdate( @@ -316,7 +352,7 @@ def update_pipette_nozzle(self, pipette_id: str, nozzle_map: NozzleMap) -> None: ) def update_pipette_tip_state( - self, pipette_id: str, tip_geometry: typing.Optional[TipGeometry] + self, pipette_id: str, tip_geometry: Optional[TipGeometry] ) -> None: """Update a pipette's tip state. See `PipetteTipStateUpdate`.""" self.pipette_tip_state = PipetteTipStateUpdate( @@ -330,3 +366,46 @@ def mark_tips_as_used( self.tips_used = TipsUsedUpdate( pipette_id=pipette_id, labware_id=labware_id, well_name=well_name ) + + def set_loaded_liquid( + self, + labware_id: str, + volumes: Dict[str, float], + last_loaded: datetime, + ) -> None: + """Add liquid volumes to well state. See `LoadLiquidUpdate`.""" + self.loaded_liquid = LoadLiquidUpdate( + labware_id=labware_id, + volumes=volumes, + last_loaded=last_loaded, + ) + + def set_probed_liquid( + self, + labware_id: str, + well_name: str, + last_probed: datetime, + height: Optional[float] = None, + volume: Optional[float] = None, + ) -> None: + """Add a liquid height and volume to well state. See `ProbeLiquidUpdate`.""" + self.probed_liquid = ProbeLiquidUpdate( + labware_id=labware_id, + well_name=well_name, + height=height, + volume=volume, + last_probed=last_probed, + ) + + def set_operated_liquid( + self, + labware_id: str, + well_name: str, + volume: float, + ) -> None: + """Update liquid volumes in well state. See `OperateLiquidUpdate`.""" + self.operated_liquid = OperateLiquidUpdate( + labware_id=labware_id, + well_name=well_name, + volume=volume, + ) diff --git a/api/src/opentrons/protocol_engine/state/wells.py b/api/src/opentrons/protocol_engine/state/wells.py index 75783d3308f..7a33a79d5ac 100644 --- a/api/src/opentrons/protocol_engine/state/wells.py +++ b/api/src/opentrons/protocol_engine/state/wells.py @@ -1,29 +1,29 @@ """Basic well data state and store.""" from dataclasses import dataclass -from datetime import datetime from typing import Dict, List, Optional from opentrons.protocol_engine.actions.actions import ( FailCommandAction, SucceedCommandAction, ) -from opentrons.protocol_engine.commands.liquid_probe import LiquidProbeResult -from opentrons.protocol_engine.commands.load_liquid import LoadLiquidResult -from opentrons.protocol_engine.commands.aspirate import AspirateResult -from opentrons.protocol_engine.commands.dispense import DispenseResult -from opentrons.protocol_engine.commands.pipetting_common import LiquidNotFoundError -from opentrons.protocol_engine.types import LiquidHeightInfo, LiquidHeightSummary +from opentrons.protocol_engine.types import ( + ProbedHeightInfo, + ProbedVolumeInfo, + LoadedVolumeInfo, + LiquidHeightSummary, +) +from . import update_types from ._abstract_store import HasState, HandlesActions -from ..actions import Action -from ..commands import Command -from .geometry import get_well_height_at_volume, get_well_height_after_volume +from ..actions import Action, get_state_update @dataclass class WellState: """State of all wells.""" - measured_liquid_heights: Dict[str, Dict[str, LiquidHeightInfo]] + loaded_volumes: Dict[str, Dict[str, LoadedVolumeInfo]] + probed_heights: Dict[str, Dict[str, ProbedHeightInfo]] + probed_volumes: Dict[str, Dict[str, ProbedVolumeInfo]] class WellStore(HasState[WellState], HandlesActions): @@ -33,109 +33,63 @@ class WellStore(HasState[WellState], HandlesActions): def __init__(self) -> None: """Initialize a well store and its state.""" - self._state = WellState(measured_liquid_heights={}) + self._state = WellState(loaded_volumes={}, probed_heights={}, probed_volumes={}) def handle_action(self, action: Action) -> None: """Modify state in reaction to an action.""" - if isinstance(action, SucceedCommandAction): - self._handle_succeeded_command(action.command) - if isinstance(action, FailCommandAction): - self._handle_failed_command(action) - - def _handle_succeeded_command(self, command: Command) -> None: - if isinstance(command.result, LiquidProbeResult): - self._set_liquid_height_after_probe( - labware_id=command.params.labwareId, - well_name=command.params.wellName, - height=command.result.z_position, - time=command.completedAt - if command.completedAt is not None - else command.createdAt, - ) - if isinstance(command.result, LoadLiquidResult): - self._set_liquid_height_after_load( - labware_id=command.params.labwareId, - well_name=next(iter(command.params.volumeByWell)), - volume=next(iter(command.params.volumeByWell.values())), - time=command.completedAt - if command.completedAt is not None - else command.createdAt, - ) - if isinstance(command.result, AspirateResult): - self._update_liquid_height_after_operation( - labware_id=command.params.labwareId, - well_name=command.params.wellName, - volume=-command.result.volume, - ) - if isinstance(command.result, DispenseResult): - self._update_liquid_height_after_operation( - labware_id=command.params.labwareId, - well_name=command.params.wellName, - volume=command.result.volume, - ) + state_update = get_state_update(action) + if state_update is not None: + self._set_loaded_liquid(state_update) + self._set_probed_liquid(state_update) + self._set_operated_liquid(state_update) + + def _set_loaded_liquid(self, state_update: update_types.StateUpdate) -> None: + if state_update.loaded_liquid != update_types.NO_CHANGE: + labware_id = state_update.loaded_liquid.labware_id + for (well, volume) in state_update.loaded_liquid.volumes.items(): + self._state.loaded_volumes[labware_id][well] = LoadedVolumeInfo( + volume=volume, + last_loaded=state_update.loaded_liquid.last_loaded, + operations_since_load=0, + ) - def _handle_failed_command(self, action: FailCommandAction) -> None: - if isinstance(action.error, LiquidNotFoundError): - self._set_liquid_height_after_probe( - labware_id=action.error.private.labware_id, - well_name=action.error.private.well_name, - height=None, - time=action.failed_at, + def _set_probed_liquid(self, state_update: update_types.StateUpdate) -> None: + if state_update.probed_liquid != update_types.NO_CHANGE: + labware_id = state_update.probed_liquid.labware_id + well_name = state_update.probed_liquid.well_name + self._state.probed_heights[labware_id][well_name] = ProbedHeightInfo( + height=state_update.probed_liquid.height, + last_probed=state_update.probed_liquid.last_probed, + ) + self._state.probed_volumes[labware_id][well_name] = ProbedVolumeInfo( + volume=state_update.probed_liquid.volume, + last_probed=state_update.probed_liquid.last_probed, + operations_since_probe=0, ) - def _set_liquid_height_after_probe( - self, labware_id: str, well_name: str, height: float, time: datetime - ) -> None: - """Set the liquid height of the well from a LiquidProbe command.""" - lhi = LiquidHeightInfo( - height=height, last_measured=time, operations_since_measurement=0 - ) - if labware_id not in self._state.measured_liquid_heights: - self._state.measured_liquid_heights[labware_id] = {} - self._state.measured_liquid_heights[labware_id][well_name] = lhi - - def _set_liquid_height_after_load( - self, labware_id: str, well_name: str, volume: float, time: datetime - ) -> None: - """Set the liquid height of the well from a LoadLiquid command.""" - height = get_well_height_at_volume( - labware_id=labware_id, well_name=well_name, volume=volume - ) - lhi = LiquidHeightInfo( - height=height, last_measured=time, operations_since_measurement=0 - ) - if labware_id not in self._state.measured_liquid_heights: - self._state.measured_liquid_heights[labware_id] = {} - self._state.measured_liquid_heights[labware_id][well_name] = lhi - - def _update_liquid_height_after_operation( - self, labware_id: str, well_name: str, volume: float - ) -> None: - """Update the liquid height of the well after an Aspirate or Dispense command.""" - time = self._state.measured_liquid_heights[labware_id][well_name].last_measured - operations_since_measurement = ( - self._state.measured_liquid_heights[labware_id][ - well_name - ].operations_since_measurement - + 1 - ) - initial_height = self._state.measured_liquid_heights[labware_id][ - well_name - ].height - height = get_well_height_after_volume( - labware_id=labware_id, - well_name=well_name, - initial_height=initial_height, - volume=volume, - ) - lhi = LiquidHeightInfo( - height=height, - last_measured=time, - operations_since_measurement=operations_since_measurement, - ) - if labware_id not in self._state.measured_liquid_heights: - self._state.measured_liquid_heights[labware_id] = {} - self._state.measured_liquid_heights[labware_id][well_name] = lhi + def _set_operated_liquid(self, state_update: update_types.StateUpdate) -> None: + if state_update.operated_liquid != update_types.NO_CHANGE: + labware_id = state_update.operated_liquid.labware_id + well_name = state_update.operated_liquid.well_name + # clear well probed_height info? Update types.py docstring + prev_loaded_vol_info = self._state.loaded_volumes[labware_id][well_name] + if prev_loaded_vol_info.volume: + self._state.loaded_volumes[labware_id][well_name] = LoadedVolumeInfo( + volume=prev_loaded_vol_info.volume + + state_update.operated_liquid.volume, + last_loaded=prev_loaded_vol_info.last_loaded, + operations_since_load=prev_loaded_vol_info.operations_since_load + + 1, + ) + prev_probed_vol_info = self._state.probed_volumes[labware_id][well_name] + if prev_probed_vol_info.volume: + self._state.probed_volumes[labware_id][well_name] = ProbedVolumeInfo( + volume=prev_probed_vol_info.volume + + state_update.operated_liquid.volume, + last_probed=prev_probed_vol_info.last_probed, + operations_since_probe=prev_probed_vol_info.operations_since_probe + + 1, + ) class WellView(HasState[WellState]): @@ -151,29 +105,35 @@ def __init__(self, state: WellState) -> None: """ self._state = state + # if volume requested, loaded_volumes or probed_volumes + # if height requested, probed_heights or loaded_vols_to_height or probed_vols_to_height + # to get height, call GeometryView.get_well_height, which does conversion if needed + + # update this def get_all(self) -> List[LiquidHeightSummary]: """Get all well liquid heights.""" all_heights: List[LiquidHeightSummary] = [] - for labware, wells in self._state.measured_liquid_heights.items(): + for labware, wells in self._state.probed_heights.items(): for well, lhi in wells.items(): lhs = LiquidHeightSummary( labware_id=labware, well_name=well, height=lhi.height, - last_measured=lhi.last_measured, + last_measured=lhi.last_probed, ) all_heights.append(lhs) return all_heights + # update this def get_all_in_labware(self, labware_id: str) -> List[LiquidHeightSummary]: """Get all well liquid heights for a particular labware.""" all_heights: List[LiquidHeightSummary] = [] - for well, lhi in self._state.measured_liquid_heights[labware_id].items(): + for well, lhi in self._state.probed_heights[labware_id].items(): lhs = LiquidHeightSummary( labware_id=labware_id, well_name=well, height=lhi.height, - last_measured=lhi.last_measured, + last_measured=lhi.last_probed, ) all_heights.append(lhs) return all_heights @@ -186,7 +146,7 @@ def get_last_measured_liquid_height( Returns None if no liquid probe has been done. """ try: - height = self._state.measured_liquid_heights[labware_id][well_name].height + height = self._state.probed_heights[labware_id][well_name].height return height except KeyError: return None @@ -194,8 +154,6 @@ def get_last_measured_liquid_height( def has_measured_liquid_height(self, labware_id: str, well_name: str) -> bool: """Returns True if the well has been liquid level probed previously.""" try: - return bool( - self._state.measured_liquid_heights[labware_id][well_name].height - ) + return bool(self._state.probed_heights[labware_id][well_name].height) except KeyError: return False diff --git a/api/src/opentrons/protocol_engine/types.py b/api/src/opentrons/protocol_engine/types.py index d8b9345ab71..1bd4e7afd88 100644 --- a/api/src/opentrons/protocol_engine/types.py +++ b/api/src/opentrons/protocol_engine/types.py @@ -355,12 +355,27 @@ class CurrentWell: well_name: str -class LiquidHeightInfo(BaseModel): - """Payload required to store recent measured liquid heights.""" +class ProbedHeightInfo(BaseModel): + """A well's liquid height, initialized by a LiquidProbe, cleared by Aspirate and Dispense.""" - height: float - last_measured: datetime - operations_since_measurement: int + height: Optional[float] = None + last_probed: datetime + + +class ProbedVolumeInfo(BaseModel): + """A well's liquid volume, initialized by a LiquidProbe, updated by Aspirate and Dispense.""" + + volume: Optional[float] = None + last_probed: datetime + operations_since_probe: int + + +class LoadedVolumeInfo(BaseModel): + """A well's liquid volume, initialized by a LoadLiquid, updated by Aspirate and Dispense.""" + + volume: Optional[float] = None + last_loaded: datetime + operations_since_load: int class LiquidHeightSummary(BaseModel): @@ -368,8 +383,8 @@ class LiquidHeightSummary(BaseModel): labware_id: str well_name: str - height: float last_measured: datetime + height: Optional[float] = None @dataclass(frozen=True)