From 576501a83a95fc6a4db95b2a4efabaf8aaabbdfd Mon Sep 17 00:00:00 2001 From: TamarZanzouri Date: Thu, 2 Jan 2025 13:48:46 -0500 Subject: [PATCH] chore(api): port absorbance reader commands to state update (#17113) --- .../commands/absorbance_reader/close_lid.py | 1 - .../commands/absorbance_reader/initialize.py | 11 +- .../commands/absorbance_reader/read.py | 14 +- .../protocol_engine/state/modules.py | 109 ++++---- .../protocol_engine/state/update_types.py | 70 +++++- .../commands/absorbance_reader/__init__.py | 1 + .../absorbance_reader/test_close_lid.py | 221 ++++++++++++++++ .../absorbance_reader/test_initialize.py | 238 ++++++++++++++++++ .../absorbance_reader/test_open_lid.py | 222 ++++++++++++++++ .../commands/absorbance_reader/test_read.py | 176 +++++++++++++ .../protocol_engine/commands/conftest.py | 7 + .../state/test_update_types.py | 58 +++-- 12 files changed, 1033 insertions(+), 95 deletions(-) create mode 100644 api/tests/opentrons/protocol_engine/commands/absorbance_reader/__init__.py create mode 100644 api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_close_lid.py create mode 100644 api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_initialize.py create mode 100644 api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_open_lid.py create mode 100644 api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_read.py diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/close_lid.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/close_lid.py index 14c1f0f9ea3..41c1e65bce9 100644 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/close_lid.py +++ b/api/src/opentrons/protocol_engine/commands/absorbance_reader/close_lid.py @@ -60,7 +60,6 @@ async def execute(self, params: CloseLidParams) -> SuccessData[CloseLidResult]: hardware_lid_status = AbsorbanceReaderLidStatus.OFF if not self._state_view.config.use_virtual_modules: abs_reader = self._equipment.get_module_hardware_api(mod_substate.module_id) - if abs_reader is not None: hardware_lid_status = await abs_reader.get_current_lid_status() else: diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/initialize.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/initialize.py index 43937c07ab2..bcb2639d0a5 100644 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/initialize.py +++ b/api/src/opentrons/protocol_engine/commands/absorbance_reader/initialize.py @@ -12,6 +12,7 @@ from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData from ...errors.error_occurrence import ErrorOccurrence from ...errors import InvalidWavelengthError +from ...state import update_types if TYPE_CHECKING: from opentrons.protocol_engine.state.state import StateView @@ -60,6 +61,7 @@ def __init__( async def execute(self, params: InitializeParams) -> SuccessData[InitializeResult]: """Initiate a single absorbance measurement.""" + state_update = update_types.StateUpdate() abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate( module_id=params.moduleId ) @@ -120,10 +122,15 @@ async def execute(self, params: InitializeParams) -> SuccessData[InitializeResul reference_wavelength=params.referenceWavelength, ) - return SuccessData( - public=InitializeResult(), + state_update.initialize_absorbance_reader( + abs_reader_substate.module_id, + params.measureMode, + params.sampleWavelengths, + params.referenceWavelength, ) + return SuccessData(public=InitializeResult(), state_update=state_update) + class Initialize(BaseCommand[InitializeParams, InitializeResult, ErrorOccurrence]): """A command to initialize an Absorbance Reader.""" diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py index b06a2527cc8..c8f7dca8706 100644 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py +++ b/api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py @@ -74,6 +74,7 @@ async def execute( # noqa: C901 self, params: ReadAbsorbanceParams ) -> SuccessData[ReadAbsorbanceResult]: """Initiate an absorbance measurement.""" + state_update = update_types.StateUpdate() abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate( module_id=params.moduleId ) @@ -149,6 +150,9 @@ async def execute( # noqa: C901 "Plate Reader data cannot be requested with a module that has not been initialized." ) + state_update.set_absorbance_reader_data( + module_id=abs_reader_substate.module_id, read_result=asbsorbance_result + ) # TODO (cb, 10-17-2024): FILE PROVIDER - Some day we may want to break the file provider behavior into a seperate API function. # When this happens, we probably will to have the change the command results handler we utilize to track file IDs in engine. # Today, the action handler for the FileStore looks for a ReadAbsorbanceResult command action, this will need to be delinked. @@ -181,18 +185,20 @@ async def execute( # noqa: C901 # Return success data to api return SuccessData( public=ReadAbsorbanceResult( - data=asbsorbance_result, fileIds=file_ids + data=asbsorbance_result, + fileIds=file_ids, ), + state_update=state_update, ) + state_update.files_added = update_types.FilesAddedUpdate(file_ids=file_ids) + return SuccessData( public=ReadAbsorbanceResult( data=asbsorbance_result, fileIds=file_ids, ), - state_update=update_types.StateUpdate( - files_added=update_types.FilesAddedUpdate(file_ids=file_ids) - ), + state_update=state_update, ) diff --git a/api/src/opentrons/protocol_engine/state/modules.py b/api/src/opentrons/protocol_engine/state/modules.py index a0b22f14fcb..ce75473265e 100644 --- a/api/src/opentrons/protocol_engine/state/modules.py +++ b/api/src/opentrons/protocol_engine/state/modules.py @@ -35,6 +35,7 @@ AbsorbanceReaderMeasureMode, ) from opentrons.types import DeckSlotName, MountType, StagingSlotName +from .update_types import AbsorbanceReaderStateUpdate from ..errors import ModuleNotConnectedError from ..types import ( @@ -63,7 +64,6 @@ heater_shaker, temperature_module, thermocycler, - absorbance_reader, ) from ..actions import ( Action, @@ -296,40 +296,10 @@ def _handle_command(self, command: Command) -> None: ): self._handle_thermocycler_module_commands(command) - if isinstance( - command.result, - ( - absorbance_reader.InitializeResult, - absorbance_reader.ReadAbsorbanceResult, - ), - ): - self._handle_absorbance_reader_commands(command) - def _handle_state_update(self, state_update: update_types.StateUpdate) -> None: - if state_update.absorbance_reader_lid != update_types.NO_CHANGE: - module_id = state_update.absorbance_reader_lid.module_id - is_lid_on = state_update.absorbance_reader_lid.is_lid_on - - # Get current values: - absorbance_reader_substate = self._state.substate_by_module_id[module_id] - assert isinstance( - absorbance_reader_substate, AbsorbanceReaderSubState - ), f"{module_id} is not an absorbance plate reader." - configured = absorbance_reader_substate.configured - measure_mode = absorbance_reader_substate.measure_mode - configured_wavelengths = absorbance_reader_substate.configured_wavelengths - reference_wavelength = absorbance_reader_substate.reference_wavelength - data = absorbance_reader_substate.data - - self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState( - module_id=AbsorbanceReaderId(module_id), - configured=configured, - measured=True, - is_lid_on=is_lid_on, - measure_mode=measure_mode, - configured_wavelengths=configured_wavelengths, - reference_wavelength=reference_wavelength, - data=data, + if state_update.absorbance_reader_state_update != update_types.NO_CHANGE: + self._handle_absorbance_reader_commands( + state_update.absorbance_reader_state_update ) def _add_module_substate( @@ -589,47 +559,58 @@ def _handle_thermocycler_module_commands( ) def _handle_absorbance_reader_commands( - self, - command: Union[ - absorbance_reader.Initialize, - absorbance_reader.ReadAbsorbance, - ], + self, absorbance_reader_state_update: AbsorbanceReaderStateUpdate ) -> None: - module_id = command.params.moduleId + # Get current values: + module_id = absorbance_reader_state_update.module_id absorbance_reader_substate = self._state.substate_by_module_id[module_id] assert isinstance( absorbance_reader_substate, AbsorbanceReaderSubState ), f"{module_id} is not an absorbance plate reader." - - # Get current values + is_lid_on = absorbance_reader_substate.is_lid_on + measured = True configured = absorbance_reader_substate.configured measure_mode = absorbance_reader_substate.measure_mode configured_wavelengths = absorbance_reader_substate.configured_wavelengths reference_wavelength = absorbance_reader_substate.reference_wavelength - is_lid_on = absorbance_reader_substate.is_lid_on - - if isinstance(command.result, absorbance_reader.InitializeResult): - self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState( - module_id=AbsorbanceReaderId(module_id), - configured=True, - measured=False, - is_lid_on=is_lid_on, - measure_mode=AbsorbanceReaderMeasureMode(command.params.measureMode), - configured_wavelengths=command.params.sampleWavelengths, - reference_wavelength=command.params.referenceWavelength, - data=None, + data = absorbance_reader_substate.data + if ( + absorbance_reader_state_update.absorbance_reader_lid + != update_types.NO_CHANGE + ): + is_lid_on = absorbance_reader_state_update.absorbance_reader_lid.is_lid_on + elif ( + absorbance_reader_state_update.initialize_absorbance_reader_update + != update_types.NO_CHANGE + ): + configured = True + measured = False + is_lid_on = is_lid_on + measure_mode = AbsorbanceReaderMeasureMode( + absorbance_reader_state_update.initialize_absorbance_reader_update.measure_mode ) - elif isinstance(command.result, absorbance_reader.ReadAbsorbanceResult): - self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState( - module_id=AbsorbanceReaderId(module_id), - configured=configured, - measured=True, - is_lid_on=is_lid_on, - measure_mode=measure_mode, - configured_wavelengths=configured_wavelengths, - reference_wavelength=reference_wavelength, - data=command.result.data, + configured_wavelengths = ( + absorbance_reader_state_update.initialize_absorbance_reader_update.sample_wave_lengths ) + reference_wavelength = ( + absorbance_reader_state_update.initialize_absorbance_reader_update.reference_wave_length + ) + data = None + elif ( + absorbance_reader_state_update.absorbance_reader_data + != update_types.NO_CHANGE + ): + data = absorbance_reader_state_update.absorbance_reader_data.read_result + self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState( + module_id=AbsorbanceReaderId(module_id), + configured=configured, + measured=measured, + is_lid_on=is_lid_on, + measure_mode=measure_mode, + configured_wavelengths=configured_wavelengths, + reference_wavelength=reference_wavelength, + data=data, + ) class ModuleView: diff --git a/api/src/opentrons/protocol_engine/state/update_types.py b/api/src/opentrons/protocol_engine/state/update_types.py index 568519b2784..9f44fcfee11 100644 --- a/api/src/opentrons/protocol_engine/state/update_types.py +++ b/api/src/opentrons/protocol_engine/state/update_types.py @@ -15,6 +15,7 @@ TipGeometry, AspiratedFluid, LiquidClassRecord, + ABSMeasureMode, ) from opentrons.types import MountType from opentrons_shared_data.labware.labware_definition import LabwareDefinition @@ -284,10 +285,37 @@ class PipetteEmptyFluidUpdate: class AbsorbanceReaderLidUpdate: """An update to an absorbance reader's lid location.""" - module_id: str is_lid_on: bool +@dataclasses.dataclass +class AbsorbanceReaderDataUpdate: + """An update to an absorbance reader's lid location.""" + + read_result: typing.Dict[int, typing.Dict[str, float]] + + +@dataclasses.dataclass(frozen=True) +class AbsorbanceReaderInitializeUpdate: + """An update to an absorbance reader's initialization.""" + + measure_mode: ABSMeasureMode + sample_wave_lengths: typing.List[int] + reference_wave_length: typing.Optional[int] + + +@dataclasses.dataclass +class AbsorbanceReaderStateUpdate: + """An update to the absorbance reader module state.""" + + module_id: str + absorbance_reader_lid: AbsorbanceReaderLidUpdate | NoChangeType = NO_CHANGE + absorbance_reader_data: AbsorbanceReaderDataUpdate | NoChangeType = NO_CHANGE + initialize_absorbance_reader_update: AbsorbanceReaderInitializeUpdate | NoChangeType = ( + NO_CHANGE + ) + + @dataclasses.dataclass class LiquidClassLoadedUpdate: """The state update from loading a liquid class.""" @@ -348,7 +376,9 @@ class StateUpdate: liquid_operated: LiquidOperatedUpdate | NoChangeType = NO_CHANGE - absorbance_reader_lid: AbsorbanceReaderLidUpdate | NoChangeType = NO_CHANGE + absorbance_reader_state_update: AbsorbanceReaderStateUpdate | NoChangeType = ( + NO_CHANGE + ) liquid_class_loaded: LiquidClassLoadedUpdate | NoChangeType = NO_CHANGE @@ -644,8 +674,40 @@ def set_fluid_empty(self: Self, pipette_id: str) -> Self: def set_absorbance_reader_lid(self: Self, module_id: str, is_lid_on: bool) -> Self: """Update an absorbance reader's lid location. See `AbsorbanceReaderLidUpdate`.""" - self.absorbance_reader_lid = AbsorbanceReaderLidUpdate( - module_id=module_id, is_lid_on=is_lid_on + assert self.absorbance_reader_state_update == NO_CHANGE + self.absorbance_reader_state_update = AbsorbanceReaderStateUpdate( + module_id=module_id, + absorbance_reader_lid=AbsorbanceReaderLidUpdate(is_lid_on=is_lid_on), + ) + return self + + def set_absorbance_reader_data( + self, module_id: str, read_result: typing.Dict[int, typing.Dict[str, float]] + ) -> Self: + """Update an absorbance reader's read data. See `AbsorbanceReaderReadDataUpdate`.""" + assert self.absorbance_reader_state_update == NO_CHANGE + self.absorbance_reader_state_update = AbsorbanceReaderStateUpdate( + module_id=module_id, + absorbance_reader_data=AbsorbanceReaderDataUpdate(read_result=read_result), + ) + return self + + def initialize_absorbance_reader( + self, + module_id: str, + measure_mode: ABSMeasureMode, + sample_wave_lengths: typing.List[int], + reference_wave_length: typing.Optional[int], + ) -> Self: + """Initialize absorbance reader.""" + assert self.absorbance_reader_state_update == NO_CHANGE + self.absorbance_reader_state_update = AbsorbanceReaderStateUpdate( + module_id=module_id, + initialize_absorbance_reader_update=AbsorbanceReaderInitializeUpdate( + measure_mode=measure_mode, + sample_wave_lengths=sample_wave_lengths, + reference_wave_length=reference_wave_length, + ), ) return self diff --git a/api/tests/opentrons/protocol_engine/commands/absorbance_reader/__init__.py b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/__init__.py new file mode 100644 index 00000000000..ebb91ede166 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/__init__.py @@ -0,0 +1 @@ +"""Tests for absorbance reader commands.""" diff --git a/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_close_lid.py b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_close_lid.py new file mode 100644 index 00000000000..c40a825d9c6 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_close_lid.py @@ -0,0 +1,221 @@ +"""Test absorbance reader initilize command.""" +import pytest +from decoy import Decoy + +from opentrons.drivers.types import ( + AbsorbanceReaderLidStatus, +) +from opentrons.hardware_control.modules import AbsorbanceReader +from opentrons.protocol_engine import ModuleModel, DeckSlotLocation +from opentrons.protocol_engine.errors import CannotPerformModuleAction + +from opentrons.protocol_engine.execution import EquipmentHandler, LabwareMovementHandler +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state.module_substates import ( + AbsorbanceReaderSubState, + AbsorbanceReaderId, +) +from opentrons.protocol_engine.commands.command import SuccessData +from opentrons.protocol_engine.commands.absorbance_reader import ( + CloseLidResult, + CloseLidParams, +) +from opentrons.protocol_engine.commands.absorbance_reader.close_lid import ( + CloseLidImpl, +) +from opentrons.protocol_engine.types import ( + LabwareMovementOffsetData, + LabwareOffsetVector, +) +from opentrons.types import DeckSlotName +from opentrons_shared_data.labware.labware_definition import ( + LabwareDefinition, + Parameters, +) + + +@pytest.fixture +def absorbance_def() -> LabwareDefinition: + """Get a tip rack Pydantic model definition value object.""" + return LabwareDefinition.model_construct( # type: ignore[call-arg] + namespace="test", + version=1, + parameters=Parameters.model_construct( # type: ignore[call-arg] + loadName="cool-labware", + tipOverlap=None, # add a None value to validate serialization to dictionary + ), + ) + + +@pytest.fixture +def subject( + state_view: StateView, + equipment: EquipmentHandler, + labware_movement: LabwareMovementHandler, +) -> CloseLidImpl: + """Subject fixture.""" + return CloseLidImpl( + state_view=state_view, equipment=equipment, labware_movement=labware_movement + ) + + +@pytest.mark.parametrize( + "hardware_lid_status", + (AbsorbanceReaderLidStatus.ON, AbsorbanceReaderLidStatus.OFF), +) +async def test_absorbance_reader_close_lid_implementation( + decoy: Decoy, + subject: CloseLidImpl, + state_view: StateView, + equipment: EquipmentHandler, + hardware_lid_status: AbsorbanceReaderLidStatus, + absorbance_def: LabwareDefinition, +) -> None: + """It should validate, find hardware module if not virtualized, and close lid.""" + params = CloseLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(await absorbance_module_hw.get_current_lid_status()).then_return( + hardware_lid_status + ) + decoy.when(state_view.modules.get_requested_model(params.moduleId)).then_return( + ModuleModel.ABSORBANCE_READER_V1 + ) + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return() + + decoy.when(state_view.modules.get_location(params.moduleId)).then_return( + DeckSlotLocation(slotName=DeckSlotName.SLOT_D3) + ) + decoy.when( + state_view.modules.ensure_and_convert_module_fixture_location( + DeckSlotName.SLOT_D3, ModuleModel.ABSORBANCE_READER_V1 + ) + ).then_return("absorbanceReaderV1D3") + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return( + absorbance_def + ) + decoy.when( + state_view.labware.get_child_gripper_offsets( + labware_definition=absorbance_def, + slot_name=None, + ) + ).then_return( + LabwareMovementOffsetData( + pickUpOffset=LabwareOffsetVector(x=0, y=0, z=0), + dropOffset=LabwareOffsetVector(x=0, y=0, z=0), + ) + ) + + result = await subject.execute(params=params) + + assert result == SuccessData( + public=CloseLidResult(), + state_update=update_types.StateUpdate( + absorbance_reader_state_update=update_types.AbsorbanceReaderStateUpdate( + module_id="module-id", + absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( + is_lid_on=True + ), + ), + ), + ) + + +async def test_close_lid_raises_no_module( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: CloseLidImpl, +) -> None: + """Should raise an error that the hardware module not found.""" + params = CloseLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + decoy.when(state_view.config.use_virtual_modules).then_return(False) + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(None) + with pytest.raises(CannotPerformModuleAction): + await subject.execute(params=params) + + +async def test_close_lid_raises_no_gripper_offset( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: CloseLidImpl, + absorbance_def: LabwareDefinition, +) -> None: + """Should raise an error that gripper offset not found.""" + params = CloseLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(await absorbance_module_hw.get_current_lid_status()).then_return( + AbsorbanceReaderLidStatus.OFF + ) + decoy.when(state_view.modules.get_requested_model(params.moduleId)).then_return( + ModuleModel.ABSORBANCE_READER_V1 + ) + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return() + + decoy.when(state_view.modules.get_location(params.moduleId)).then_return( + DeckSlotLocation(slotName=DeckSlotName.SLOT_D3) + ) + decoy.when( + state_view.modules.ensure_and_convert_module_fixture_location( + DeckSlotName.SLOT_D3, ModuleModel.ABSORBANCE_READER_V1 + ) + ).then_return("absorbanceReaderV1D3") + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return( + absorbance_def + ) + decoy.when( + state_view.labware.get_child_gripper_offsets( + labware_definition=absorbance_def, + slot_name=None, + ) + ).then_return(None) + with pytest.raises(ValueError): + await subject.execute(params=params) diff --git a/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_initialize.py b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_initialize.py new file mode 100644 index 00000000000..efdd8908583 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_initialize.py @@ -0,0 +1,238 @@ +"""Test absorbance reader initilize command.""" +import pytest +from decoy import Decoy +from typing import List + +from opentrons.drivers.types import ABSMeasurementMode +from opentrons.hardware_control.modules import AbsorbanceReader +from opentrons.protocol_engine.errors import InvalidWavelengthError + +from opentrons.protocol_engine.execution import EquipmentHandler +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state.module_substates import ( + AbsorbanceReaderSubState, + AbsorbanceReaderId, +) +from opentrons.protocol_engine.commands.command import SuccessData +from opentrons.protocol_engine.commands.absorbance_reader import ( + InitializeParams, + InitializeResult, +) +from opentrons.protocol_engine.commands.absorbance_reader.initialize import ( + InitializeImpl, +) + + +@pytest.fixture +def subject( + state_view: StateView, + equipment: EquipmentHandler, +) -> InitializeImpl: + """Subject command implementation to test.""" + return InitializeImpl(state_view=state_view, equipment=equipment) + + +@pytest.mark.parametrize( + "input_sample_wave_length, input_measure_mode", [([1, 2], "multi"), ([1], "single")] +) +async def test_absorbance_reader_implementation( + decoy: Decoy, + input_sample_wave_length: List[int], + input_measure_mode: str, + subject: InitializeImpl, + state_view: StateView, + equipment: EquipmentHandler, +) -> None: + """It should validate, find hardware module if not virtualized, and disengage.""" + params = InitializeParams( + moduleId="unverified-module-id", + measureMode=input_measure_mode, # type: ignore[arg-type] + sampleWavelengths=input_sample_wave_length, + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2]) + + result = await subject.execute(params=params) + + decoy.verify( + await absorbance_module_hw.set_sample_wavelength( + ABSMeasurementMode(params.measureMode), + params.sampleWavelengths, + reference_wavelength=params.referenceWavelength, + ), + times=1, + ) + assert result == SuccessData( + public=InitializeResult(), + state_update=update_types.StateUpdate( + absorbance_reader_state_update=update_types.AbsorbanceReaderStateUpdate( + module_id="module-id", + initialize_absorbance_reader_update=update_types.AbsorbanceReaderInitializeUpdate( + measure_mode=input_measure_mode, # type: ignore[arg-type] + sample_wave_lengths=input_sample_wave_length, + reference_wave_length=None, + ), + ) + ), + ) + + +@pytest.mark.parametrize( + "input_sample_wave_length, input_measure_mode", + [([1, 2, 3], "multi"), ([3], "single")], +) +async def test_initialize_raises_invalid_wave_length( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: InitializeImpl, + input_sample_wave_length: List[int], + input_measure_mode: str, +) -> None: + """Should raise an InvalidWavelengthError error.""" + params = InitializeParams( + moduleId="unverified-module-id", + measureMode=input_measure_mode, # type: ignore[arg-type] + sampleWavelengths=input_sample_wave_length, + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2]) + + with pytest.raises(InvalidWavelengthError): + await subject.execute(params=params) + + +@pytest.mark.parametrize( + "input_sample_wave_length, input_measure_mode", + [([], "multi"), ([], "single")], +) +async def test_initialize_raises_measure_mode_not_matching( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: InitializeImpl, + input_sample_wave_length: List[int], + input_measure_mode: str, +) -> None: + """Should raise an error that the measure mode does not match sample wave.""" + params = InitializeParams( + moduleId="unverified-module-id", + measureMode=input_measure_mode, # type: ignore[arg-type] + sampleWavelengths=input_sample_wave_length, + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2]) + + with pytest.raises(ValueError): + await subject.execute(params=params) + + +async def test_initialize_single_raises_reference_wave_length_not_matching( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: InitializeImpl, +) -> None: + """Should raise an error that the measure mode does not match sample wave.""" + params = InitializeParams( + moduleId="unverified-module-id", + measureMode="single", + sampleWavelengths=[1], + referenceWavelength=3, + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2]) + + with pytest.raises(InvalidWavelengthError): + await subject.execute(params=params) + + +async def test_initialize_multi_raises_no_reference_wave_length( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: InitializeImpl, +) -> None: + """Should raise an error that the measure mode does not match sample wave.""" + params = InitializeParams( + moduleId="unverified-module-id", + measureMode="multi", + sampleWavelengths=[1, 2], + referenceWavelength=3, + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2]) + + with pytest.raises(ValueError): + await subject.execute(params=params) diff --git a/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_open_lid.py b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_open_lid.py new file mode 100644 index 00000000000..bc555a9bb18 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_open_lid.py @@ -0,0 +1,222 @@ +"""Test absorbance reader initilize command.""" +import pytest +from decoy import Decoy + +from opentrons.drivers.types import ( + AbsorbanceReaderLidStatus, +) +from opentrons.hardware_control.modules import AbsorbanceReader +from opentrons.protocol_engine import ModuleModel, DeckSlotLocation +from opentrons.protocol_engine.errors import CannotPerformModuleAction + +from opentrons.protocol_engine.execution import EquipmentHandler, LabwareMovementHandler +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state.module_substates import ( + AbsorbanceReaderSubState, + AbsorbanceReaderId, +) +from opentrons.protocol_engine.commands.command import SuccessData +from opentrons.protocol_engine.commands.absorbance_reader import ( + OpenLidResult, + OpenLidParams, +) +from opentrons.protocol_engine.commands.absorbance_reader.open_lid import ( + OpenLidImpl, +) +from opentrons.protocol_engine.types import ( + LabwareMovementOffsetData, + LabwareOffsetVector, +) +from opentrons.types import DeckSlotName +from opentrons_shared_data.labware.labware_definition import ( + LabwareDefinition, + Parameters, +) + + +@pytest.fixture +def absorbance_def() -> LabwareDefinition: + """Get a tip rack Pydantic model definition value object.""" + return LabwareDefinition.model_construct( # type: ignore[call-arg] + namespace="test", + version=1, + parameters=Parameters.model_construct( # type: ignore[call-arg] + loadName="cool-labware", + tipOverlap=None, # add a None value to validate serialization to dictionary + ), + ) + + +@pytest.fixture +def subject( + state_view: StateView, + equipment: EquipmentHandler, + labware_movement: LabwareMovementHandler, +) -> OpenLidImpl: + """Command implementation subject for testing.""" + return OpenLidImpl( + state_view=state_view, equipment=equipment, labware_movement=labware_movement + ) + + +@pytest.mark.parametrize( + "hardware_lid_status", + (AbsorbanceReaderLidStatus.ON, AbsorbanceReaderLidStatus.OFF), +) +async def test_absorbance_reader_implementation( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + labware_movement: LabwareMovementHandler, + hardware_lid_status: AbsorbanceReaderLidStatus, + absorbance_def: LabwareDefinition, + subject: OpenLidImpl, +) -> None: + """It should validate, find hardware module if not virtualized, and disengage.""" + params = OpenLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(await absorbance_module_hw.get_current_lid_status()).then_return( + hardware_lid_status + ) + decoy.when(state_view.modules.get_requested_model(params.moduleId)).then_return( + ModuleModel.ABSORBANCE_READER_V1 + ) + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return() + + decoy.when(state_view.modules.get_location(params.moduleId)).then_return( + DeckSlotLocation(slotName=DeckSlotName.SLOT_D3) + ) + decoy.when( + state_view.modules.ensure_and_convert_module_fixture_location( + DeckSlotName.SLOT_D3, ModuleModel.ABSORBANCE_READER_V1 + ) + ).then_return("absorbanceReaderV1D3") + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return( + absorbance_def + ) + decoy.when( + state_view.labware.get_child_gripper_offsets( + labware_definition=absorbance_def, + slot_name=None, + ) + ).then_return( + LabwareMovementOffsetData( + pickUpOffset=LabwareOffsetVector(x=0, y=0, z=0), + dropOffset=LabwareOffsetVector(x=0, y=0, z=0), + ) + ) + + result = await subject.execute(params=params) + + assert result == SuccessData( + public=OpenLidResult(), + state_update=update_types.StateUpdate( + absorbance_reader_state_update=update_types.AbsorbanceReaderStateUpdate( + module_id="module-id", + absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( + is_lid_on=False + ), + ), + ), + ) + + +async def test_open_lid_raises_no_module( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: OpenLidImpl, +) -> None: + """Should raise an error that the hardware module not found.""" + params = OpenLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + decoy.when(state_view.config.use_virtual_modules).then_return(False) + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(None) + with pytest.raises(CannotPerformModuleAction): + await subject.execute(params=params) + + +async def test_open_lid_raises_no_gripper_offset( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + subject: OpenLidImpl, + absorbance_def: LabwareDefinition, +) -> None: + """Should raise an error that gripper offset not found.""" + params = OpenLidParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(await absorbance_module_hw.get_current_lid_status()).then_return( + AbsorbanceReaderLidStatus.OFF + ) + decoy.when(state_view.modules.get_requested_model(params.moduleId)).then_return( + ModuleModel.ABSORBANCE_READER_V1 + ) + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return() + + decoy.when(state_view.modules.get_location(params.moduleId)).then_return( + DeckSlotLocation(slotName=DeckSlotName.SLOT_D3) + ) + decoy.when( + state_view.modules.ensure_and_convert_module_fixture_location( + DeckSlotName.SLOT_D3, ModuleModel.ABSORBANCE_READER_V1 + ) + ).then_return("absorbanceReaderV1D3") + + decoy.when(state_view.labware.get_absorbance_reader_lid_definition()).then_return( + absorbance_def + ) + decoy.when( + state_view.labware.get_child_gripper_offsets( + labware_definition=absorbance_def, + slot_name=None, + ) + ).then_return(None) + with pytest.raises(ValueError): + await subject.execute(params=params) diff --git a/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_read.py b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_read.py new file mode 100644 index 00000000000..6ba7619f219 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/absorbance_reader/test_read.py @@ -0,0 +1,176 @@ +"""Test absorbance reader initilize command.""" +import pytest +from decoy import Decoy + +from opentrons.drivers.types import ABSMeasurementMode, ABSMeasurementConfig +from opentrons.hardware_control.modules import AbsorbanceReader +from opentrons.protocol_engine.errors import ( + CannotPerformModuleAction, + StorageLimitReachedError, +) + +from opentrons.protocol_engine.execution import EquipmentHandler +from opentrons.protocol_engine.resources import FileProvider +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state.module_substates import ( + AbsorbanceReaderSubState, + AbsorbanceReaderId, +) +from opentrons.protocol_engine.commands.command import SuccessData +from opentrons.protocol_engine.commands.absorbance_reader import ( + ReadAbsorbanceResult, + ReadAbsorbanceParams, +) +from opentrons.protocol_engine.commands.absorbance_reader.read import ( + ReadAbsorbanceImpl, +) + + +async def test_absorbance_reader_implementation( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + file_provider: FileProvider, +) -> None: + """It should validate, find hardware module if not virtualized, and disengage.""" + subject = ReadAbsorbanceImpl( + state_view=state_view, equipment=equipment, file_provider=file_provider + ) + + params = ReadAbsorbanceParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + verified_module_id = AbsorbanceReaderId("module-id") + asbsorbance_result = {1: {"A1": 1.2}} + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(await absorbance_module_hw.start_measure()).then_return([[1.2, 1.3]]) + decoy.when(absorbance_module_hw._measurement_config).then_return( + ABSMeasurementConfig( + measure_mode=ABSMeasurementMode.SINGLE, + sample_wavelengths=[1, 2], + reference_wavelength=None, + ) + ) + decoy.when( + state_view.modules.convert_absorbance_reader_data_points([1.2, 1.3]) + ).then_return({"A1": 1.2}) + + result = await subject.execute(params=params) + + assert result == SuccessData( + public=ReadAbsorbanceResult( + data=asbsorbance_result, + fileIds=[], + ), + state_update=update_types.StateUpdate( + files_added=update_types.FilesAddedUpdate(file_ids=[]), + absorbance_reader_state_update=update_types.AbsorbanceReaderStateUpdate( + module_id="module-id", + absorbance_reader_data=update_types.AbsorbanceReaderDataUpdate( + read_result=asbsorbance_result + ), + ), + ), + ) + + +async def test_read_raises_cannot_preform_action( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + file_provider: FileProvider, +) -> None: + """It should raise CannotPerformModuleAction when not configured/lid is not on.""" + subject = ReadAbsorbanceImpl( + state_view=state_view, equipment=equipment, file_provider=file_provider + ) + + params = ReadAbsorbanceParams( + moduleId="unverified-module-id", + ) + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + + decoy.when(mabsorbance_module_substate.configured).then_return(False) + + with pytest.raises(CannotPerformModuleAction): + await subject.execute(params=params) + + decoy.when(mabsorbance_module_substate.configured).then_return(True) + + decoy.when(mabsorbance_module_substate.is_lid_on).then_return(False) + + with pytest.raises(CannotPerformModuleAction): + await subject.execute(params=params) + + +async def test_read_raises_storage_limit( + decoy: Decoy, + state_view: StateView, + equipment: EquipmentHandler, + file_provider: FileProvider, +) -> None: + """It should raise StorageLimitReachedError when not configured/lid is not on.""" + subject = ReadAbsorbanceImpl( + state_view=state_view, equipment=equipment, file_provider=file_provider + ) + + params = ReadAbsorbanceParams(moduleId="unverified-module-id", fileName="test") + + mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState) + absorbance_module_hw = decoy.mock(cls=AbsorbanceReader) + + verified_module_id = AbsorbanceReaderId("module-id") + + decoy.when( + state_view.modules.get_absorbance_reader_substate("unverified-module-id") + ).then_return(mabsorbance_module_substate) + + decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id) + + decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return( + absorbance_module_hw + ) + decoy.when(await absorbance_module_hw.start_measure()).then_return([[1.2, 1.3]]) + + decoy.when(absorbance_module_hw._measurement_config).then_return( + ABSMeasurementConfig( + measure_mode=ABSMeasurementMode.SINGLE, + sample_wavelengths=[1, 2], + reference_wavelength=None, + ) + ) + decoy.when(mabsorbance_module_substate.configured_wavelengths).then_return( + [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11] + ) + + decoy.when(state_view.files.get_filecount()).then_return(390) + with pytest.raises(StorageLimitReachedError): + await subject.execute(params=params) diff --git a/api/tests/opentrons/protocol_engine/commands/conftest.py b/api/tests/opentrons/protocol_engine/commands/conftest.py index 1d27dea0536..cf2d36b092e 100644 --- a/api/tests/opentrons/protocol_engine/commands/conftest.py +++ b/api/tests/opentrons/protocol_engine/commands/conftest.py @@ -15,6 +15,7 @@ TipHandler, GantryMover, ) +from opentrons.protocol_engine.resources import FileProvider from opentrons.protocol_engine.resources.model_utils import ModelUtils from opentrons.protocol_engine.state.state import StateView @@ -83,3 +84,9 @@ def status_bar(decoy: Decoy) -> StatusBarHandler: def gantry_mover(decoy: Decoy) -> GantryMover: """Get a mocked out GantryMover.""" return decoy.mock(cls=GantryMover) + + +@pytest.fixture +def file_provider(decoy: Decoy) -> FileProvider: + """Get a mocked out StateView.""" + return decoy.mock(cls=FileProvider) diff --git a/api/tests/opentrons/protocol_engine/state/test_update_types.py b/api/tests/opentrons/protocol_engine/state/test_update_types.py index 741df813e19..325f2611f37 100644 --- a/api/tests/opentrons/protocol_engine/state/test_update_types.py +++ b/api/tests/opentrons/protocol_engine/state/test_update_types.py @@ -1,14 +1,16 @@ """Unit tests for the utilities in `update_types`.""" - - +from opentrons.protocol_engine import DeckSlotLocation, ModuleLocation from opentrons.protocol_engine.state import update_types +from opentrons.types import DeckSlotName def test_append() -> None: """Test `StateUpdate.append()`.""" state_update = update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=True + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=DeckSlotLocation(slotName=DeckSlotName.SLOT_A1), + offset_id=None, ) ) @@ -17,22 +19,28 @@ def test_append() -> None: update_types.StateUpdate(pipette_location=update_types.CLEAR) ) assert result is state_update - assert state_update.absorbance_reader_lid == update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=True + assert state_update.labware_location == update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=DeckSlotLocation(slotName=DeckSlotName.SLOT_A1), + offset_id=None, ) assert state_update.pipette_location == update_types.CLEAR # Populating a field that's already been populated should overwrite it. result = state_update.append( update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=False + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=ModuleLocation(moduleId="module-123"), + offset_id=None, ) ) ) assert result is state_update - assert state_update.absorbance_reader_lid == update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=False + assert state_update.labware_location == update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=ModuleLocation(moduleId="module-123"), + offset_id=None, ) assert state_update.pipette_location == update_types.CLEAR @@ -44,14 +52,18 @@ def test_reduce() -> None: # It should union all the set fields together. assert update_types.StateUpdate.reduce( update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=True + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=DeckSlotLocation(slotName=DeckSlotName.SLOT_A1), + offset_id=None, ) ), update_types.StateUpdate(pipette_location=update_types.CLEAR), ) == update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=True + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=DeckSlotLocation(slotName=DeckSlotName.SLOT_A1), + offset_id=None, ), pipette_location=update_types.CLEAR, ) @@ -59,17 +71,23 @@ def test_reduce() -> None: # When one field appears multiple times, the last write wins. assert update_types.StateUpdate.reduce( update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=True + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=DeckSlotLocation(slotName=DeckSlotName.SLOT_A1), + offset_id=None, ) ), update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=False + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=ModuleLocation(moduleId="module-123"), + offset_id=None, ) ), ) == update_types.StateUpdate( - absorbance_reader_lid=update_types.AbsorbanceReaderLidUpdate( - module_id="module_id", is_lid_on=False + labware_location=update_types.LabwareLocationUpdate( + labware_id="test-123", + new_location=ModuleLocation(moduleId="module-123"), + offset_id=None, ) )