diff --git a/api/src/opentrons/protocol_engine/errors/__init__.py b/api/src/opentrons/protocol_engine/errors/__init__.py index e9f1acddeed..c6adf4bfc16 100644 --- a/api/src/opentrons/protocol_engine/errors/__init__.py +++ b/api/src/opentrons/protocol_engine/errors/__init__.py @@ -78,6 +78,7 @@ InvalidDispenseVolumeError, StorageLimitReachedError, InvalidLiquidError, + LiquidClassDoesNotExistError, ) from .error_occurrence import ErrorOccurrence, ProtocolCommandFailedError @@ -164,4 +165,5 @@ "OperationLocationNotInWellError", "InvalidDispenseVolumeError", "StorageLimitReachedError", + "LiquidClassDoesNotExistError", ] diff --git a/api/src/opentrons/protocol_engine/errors/exceptions.py b/api/src/opentrons/protocol_engine/errors/exceptions.py index 36b0d2ccbef..e5a17ea4da2 100644 --- a/api/src/opentrons/protocol_engine/errors/exceptions.py +++ b/api/src/opentrons/protocol_engine/errors/exceptions.py @@ -1155,3 +1155,15 @@ def __init__( ) -> None: """Build an StorageLimitReached.""" super().__init__(ErrorCodes.GENERAL_ERROR, message, detail, wrapping) + + +class LiquidClassDoesNotExistError(ProtocolEngineError): + """Raised when referencing a liquid class that has not been loaded.""" + + def __init__( + self, + message: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + wrapping: Optional[Sequence[EnumeratedError]] = None, + ) -> None: + super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping) diff --git a/api/src/opentrons/protocol_engine/state/liquid_classes.py b/api/src/opentrons/protocol_engine/state/liquid_classes.py new file mode 100644 index 00000000000..7992735fecd --- /dev/null +++ b/api/src/opentrons/protocol_engine/state/liquid_classes.py @@ -0,0 +1,82 @@ +"""A data store of liquid classes.""" + +from __future__ import annotations + +import dataclasses +from typing import Dict +from typing_extensions import Optional + +from .. import errors +from ..actions import Action, get_state_updates +from ..types import LiquidClassRecord +from . import update_types +from ._abstract_store import HasState, HandlesActions + + +@dataclasses.dataclass +class LiquidClassState: + """Our state is a bidirectional mapping between IDs <-> LiquidClassRecords.""" + + # We use the bidirectional map to see if we've already assigned an ID to a liquid class when the + # engine is asked to store a new liquid class. + liquid_class_record_by_id: Dict[str, LiquidClassRecord] + liquid_class_record_to_id: Dict[LiquidClassRecord, str] + + +class LiquidClassStore(HasState[LiquidClassState], HandlesActions): + """Container for LiquidClassState.""" + + _state: LiquidClassState + + def __init__(self) -> None: + self._state = LiquidClassState( + liquid_class_record_by_id={}, + liquid_class_record_to_id={}, + ) + + def handle_action(self, action: Action) -> None: + """Update the state in response to the action.""" + for state_update in get_state_updates(action): + if state_update.liquid_class_loaded != update_types.NO_CHANGE: + self._handle_liquid_class_loaded_update( + state_update.liquid_class_loaded + ) + + def _handle_liquid_class_loaded_update( + self, state_update: update_types.LiquidClassLoadedUpdate + ) -> None: + # We're just a data store. All the validation and ID generation happens in the command implementation. + self._state.liquid_class_record_by_id[ + state_update.liquid_class_id + ] = state_update.liquid_class_record + self._state.liquid_class_record_to_id[ + state_update.liquid_class_record + ] = state_update.liquid_class_id + + +class LiquidClassView(HasState[LiquidClassState]): + """Read-only view of the LiquidClassState.""" + + _state: LiquidClassState + + def __init__(self, state: LiquidClassState) -> None: + self._state = state + + def get(self, liquid_class_id: str) -> LiquidClassRecord: + """Get the LiquidClassRecord with the given identifier.""" + try: + return self._state.liquid_class_record_by_id[liquid_class_id] + except KeyError as e: + raise errors.LiquidClassDoesNotExistError( + f"Liquid class ID {liquid_class_id} not found." + ) from e + + def get_id_for_liquid_class_record( + self, liquid_class_record: LiquidClassRecord + ) -> Optional[str]: + """See if the given LiquidClassRecord if already in the store, and if so, return its identifier.""" + return self._state.liquid_class_record_to_id.get(liquid_class_record) + + def get_all(self) -> Dict[str, LiquidClassRecord]: + """Get all the LiquidClassRecords in the store.""" + return self._state.liquid_class_record_by_id.copy() diff --git a/api/src/opentrons/protocol_engine/state/update_types.py b/api/src/opentrons/protocol_engine/state/update_types.py index a3b20bde28a..aed6e637f3f 100644 --- a/api/src/opentrons/protocol_engine/state/update_types.py +++ b/api/src/opentrons/protocol_engine/state/update_types.py @@ -13,6 +13,7 @@ LabwareLocation, TipGeometry, AspiratedFluid, + LiquidClassRecord, ) from opentrons.types import MountType from opentrons_shared_data.labware.labware_definition import LabwareDefinition @@ -252,6 +253,14 @@ class AbsorbanceReaderLidUpdate: is_lid_on: bool +@dataclasses.dataclass +class LiquidClassLoadedUpdate: + """The state update from loading a liquid class.""" + + liquid_class_id: str + liquid_class_record: LiquidClassRecord + + @dataclasses.dataclass class StateUpdate: """Represents an update to perform on engine state.""" @@ -284,6 +293,8 @@ class StateUpdate: absorbance_reader_lid: AbsorbanceReaderLidUpdate | NoChangeType = NO_CHANGE + liquid_class_loaded: LiquidClassLoadedUpdate | NoChangeType = NO_CHANGE + # These convenience functions let the caller avoid the boilerplate of constructing a # complicated dataclass tree. diff --git a/api/src/opentrons/protocol_engine/types.py b/api/src/opentrons/protocol_engine/types.py index 73a2df980f9..1a11a99df86 100644 --- a/api/src/opentrons/protocol_engine/types.py +++ b/api/src/opentrons/protocol_engine/types.py @@ -36,7 +36,9 @@ from opentrons.hardware_control.modules import ( ModuleType as ModuleType, ) - +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + ByTipTypeSetting, +) from opentrons_shared_data.pipette.types import ( # noqa: F401 # convenience re-export of LabwareUri type LabwareUri as LabwareUri, @@ -842,6 +844,49 @@ class Liquid(BaseModel): displayColor: Optional[HexColor] +class LiquidClassRecord(ByTipTypeSetting, frozen=True): + """LiquidClassRecord is our internal representation of an (immutable) liquid class. + + Conceptually, a liquid class record is the tuple (name, pipette, tip, transfer properties). + We consider two liquid classes to be the same if every entry in that tuple is the same; and liquid + classes are different if any entry in the tuple is different. + + This class defines the tuple via inheritance so that we can reuse the definitions from shared_data. + """ + + liquidClassName: str = Field( + ..., + description="Identifier for the liquid of this liquid class, e.g. glycerol50.", + ) + pipetteModel: str = Field( + ..., + description="Identifier for the pipette of this liquid class.", + ) + # The other fields like tiprack ID, aspirate properties, etc. are pulled in from ByTipTypeSetting. + + def __hash__(self) -> int: + """Hash function for LiquidClassRecord.""" + # Within the Protocol Engine, LiquidClassRecords are immutable, and we'd like to be able to + # look up LiquidClassRecords by value, which involves hashing. However, Pydantic does not + # generate a usable hash function if any of the subfields (like Coordinate) are not frozen. + # So we have to implement the hash function ourselves. + # Our strategy is to recursively convert this object into a list of (key, value) tuples. + def dict_to_tuple(d: dict[str, Any]) -> tuple[tuple[str, Any], ...]: + return tuple( + ( + field_name, + dict_to_tuple(value) + if isinstance(value, dict) + else tuple(value) + if isinstance(value, list) + else value, + ) + for field_name, value in d.items() + ) + + return hash(dict_to_tuple(self.dict())) + + class SpeedRange(NamedTuple): """Minimum and maximum allowed speeds for a shaking module.""" diff --git a/api/tests/opentrons/protocol_engine/state/test_liquid_class_store.py b/api/tests/opentrons/protocol_engine/state/test_liquid_class_store.py new file mode 100644 index 00000000000..f9032acdb94 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/state/test_liquid_class_store.py @@ -0,0 +1,60 @@ +"""Liquid state store tests.""" +import pytest + +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + LiquidClassSchemaV1, +) +from opentrons.protocol_engine import actions +from opentrons.protocol_engine.commands import Comment +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.liquid_classes import LiquidClassStore +from opentrons.protocol_engine.types import LiquidClassRecord + + +@pytest.fixture +def subject() -> LiquidClassStore: + """The LiquidClassStore test subject.""" + return LiquidClassStore() + + +def test_handles_add_liquid_class( + subject: LiquidClassStore, minimal_liquid_class_def2: LiquidClassSchemaV1 +) -> None: + """Should add the LiquidClassRecord to the store.""" + pipette_0 = minimal_liquid_class_def2.byPipette[0] + by_tip_type_0 = pipette_0.byTipType[0] + liquid_class_record = LiquidClassRecord( + liquidClassName=minimal_liquid_class_def2.liquidClassName, + pipetteModel=pipette_0.pipetteModel, + tiprack=by_tip_type_0.tiprack, + aspirate=by_tip_type_0.aspirate, + singleDispense=by_tip_type_0.singleDispense, + multiDispense=by_tip_type_0.multiDispense, + ) + + subject.handle_action( + actions.SucceedCommandAction( + # TODO(dc): this is a placeholder command, LoadLiquidClassCommand coming soon + command=Comment.construct(), # type: ignore[call-arg] + state_update=update_types.StateUpdate( + liquid_class_loaded=update_types.LiquidClassLoadedUpdate( + liquid_class_id="liquid-class-id", + liquid_class_record=liquid_class_record, + ), + ), + ) + ) + + assert len(subject.state.liquid_class_record_by_id) == 1 + assert ( + subject.state.liquid_class_record_by_id["liquid-class-id"] + == liquid_class_record + ) + + assert len(subject.state.liquid_class_record_to_id) == 1 + # Make sure that LiquidClassRecords are hashable, and that we can query for LiquidClassRecords by value: + assert ( + subject.state.liquid_class_record_to_id[liquid_class_record] + == "liquid-class-id" + ) + # If this fails with an error like "TypeError: unhashable type: AspirateProperties", then you broke something. diff --git a/api/tests/opentrons/protocol_engine/state/test_liquid_class_view.py b/api/tests/opentrons/protocol_engine/state/test_liquid_class_view.py new file mode 100644 index 00000000000..d80f40a5d0c --- /dev/null +++ b/api/tests/opentrons/protocol_engine/state/test_liquid_class_view.py @@ -0,0 +1,62 @@ +"""Liquid view tests.""" +import pytest + +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + LiquidClassSchemaV1, +) + +from opentrons.protocol_engine.state.liquid_classes import ( + LiquidClassState, + LiquidClassView, +) +from opentrons.protocol_engine.types import LiquidClassRecord + + +@pytest.fixture +def liquid_class_record( + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> LiquidClassRecord: + """An example LiquidClassRecord for tests.""" + pipette_0 = minimal_liquid_class_def2.byPipette[0] + by_tip_type_0 = pipette_0.byTipType[0] + return LiquidClassRecord( + liquidClassName=minimal_liquid_class_def2.liquidClassName, + pipetteModel=pipette_0.pipetteModel, + tiprack=by_tip_type_0.tiprack, + aspirate=by_tip_type_0.aspirate, + singleDispense=by_tip_type_0.singleDispense, + multiDispense=by_tip_type_0.multiDispense, + ) + + +@pytest.fixture +def subject(liquid_class_record: LiquidClassRecord) -> LiquidClassView: + """The LiquidClassView test subject.""" + state = LiquidClassState( + liquid_class_record_by_id={"liquid-class-id": liquid_class_record}, + liquid_class_record_to_id={liquid_class_record: "liquid-class-id"}, + ) + return LiquidClassView(state) + + +def test_get_by_id( + subject: LiquidClassView, liquid_class_record: LiquidClassRecord +) -> None: + """Should look up LiquidClassRecord by ID.""" + assert subject.get("liquid-class-id") == liquid_class_record + + +def test_get_by_liquid_class_record( + subject: LiquidClassView, liquid_class_record: LiquidClassRecord +) -> None: + """Should look up existing ID given a LiquidClassRecord.""" + assert ( + subject.get_id_for_liquid_class_record(liquid_class_record) == "liquid-class-id" + ) + + +def test_get_all( + subject: LiquidClassView, liquid_class_record: LiquidClassRecord +) -> None: + """Should get all LiquidClassRecords in the store.""" + assert subject.get_all() == {"liquid-class-id": liquid_class_record}