From 64856a9c18c8321153b4969dc58722247f1dabc8 Mon Sep 17 00:00:00 2001 From: ahiuchingau <20424172+ahiuchingau@users.noreply.github.com> Date: Tue, 14 May 2024 13:10:40 -0400 Subject: [PATCH] draft --- .../drivers/absorbance_reader/hid_protocol.py | 1 + .../modules/absorbance_reader.py | 5 +- .../commands/absorbance_reader/__init__.py | 20 +++--- .../commands/absorbance_reader/measure.py | 45 ++++++------ .../commands/absorbance_reader/run_profile.py | 68 ------------------- .../commands/command_unions.py | 6 ++ .../protocol_engine/execution/equipment.py | 9 +++ .../absorbance_reader_substate.py | 9 +-- .../protocol_engine/state/modules.py | 22 ++++++ api/src/opentrons/protocol_engine/types.py | 11 +++ 10 files changed, 88 insertions(+), 108 deletions(-) delete mode 100644 api/src/opentrons/protocol_engine/commands/absorbance_reader/run_profile.py diff --git a/api/src/opentrons/drivers/absorbance_reader/hid_protocol.py b/api/src/opentrons/drivers/absorbance_reader/hid_protocol.py index 6ddd82aa030d..87d453c7631d 100644 --- a/api/src/opentrons/drivers/absorbance_reader/hid_protocol.py +++ b/api/src/opentrons/drivers/absorbance_reader/hid_protocol.py @@ -7,6 +7,7 @@ ClassVar, runtime_checkable, TypeVar, + Literal, ) Response = TypeVar("Response") diff --git a/api/src/opentrons/hardware_control/modules/absorbance_reader.py b/api/src/opentrons/hardware_control/modules/absorbance_reader.py index 8f3c65e3acea..aea7f0d88578 100644 --- a/api/src/opentrons/hardware_control/modules/absorbance_reader.py +++ b/api/src/opentrons/hardware_control/modules/absorbance_reader.py @@ -159,9 +159,10 @@ async def set_sample_wavelength(self, wavelength: int) -> None: """Set the Absorbance Reader's active wavelength.""" await self._driver.initialize_measurement(wavelength) - async def start_measure(self, wavelength: int) -> None: + async def start_measure(self, wavelength: int) -> List[float]: """Initiate a single measurement.""" - await self._driver.get_single_measurement(wavelength) + measurement = await self._driver.get_single_measurement(wavelength) + return measurement async def get_supported_wavelengths(self) -> List[int]: """Get the Absorbance Reader's supported wavelengths.""" diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/__init__.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/__init__.py index b1bd90510d89..c575fa96efb4 100644 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/__init__.py +++ b/api/src/opentrons/protocol_engine/commands/absorbance_reader/__init__.py @@ -1,17 +1,17 @@ """Command models for Absorbance Reader commands.""" from .measure import ( - AbsorbanceMeasureCommandType, - AbsorbanceMeasureParams, - AbsorbanceMeasureResult, - AbsorbanceMeasure, - AbsorbanceMeasureCreate, + MeasureAbsorbanceCommandType, + MeasureAbsorbanceParams, + MeasureAbsorbanceResult, + MeasureAbsorbance, + MeasureAbsorbanceCreate, ) __all__ = [ - "AbsorbanceMeasureCommandType", - "AbsorbanceMeasureParams", - "AbsorbanceMeasureResult", - "AbsorbanceMeasure", - "AbsorbanceMeasureCreate", + "MeasureAbsorbanceCommandType", + "MeasureAbsorbanceParams", + "MeasureAbsorbanceResult", + "MeasureAbsorbance", + "MeasureAbsorbanceCreate", ] diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/measure.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/measure.py index a8d12d163d19..f39e2e080dc9 100644 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/measure.py +++ b/api/src/opentrons/protocol_engine/commands/absorbance_reader/measure.py @@ -5,8 +5,6 @@ from pydantic import BaseModel, Field -from opentrons.hardware_control.modules.types import ThermocyclerStep - from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate if TYPE_CHECKING: @@ -14,24 +12,26 @@ from opentrons.protocol_engine.execution import EquipmentHandler -AbsorbanceMeasureCommandType = Literal["absorbanceReader/measure"] +MeasureAbsorbanceCommandType = Literal["absorbanceReader/measure"] -class AbsorbanceMeasureParams(BaseModel): +class MeasureAbsorbanceParams(BaseModel): """Input parameters for a single absorbance reading.""" moduleId: str = Field(..., description="Unique ID of the Thermocycler.") - sampleWavelength: float = Field(..., description="Sample wavelength in nm.") + sampleWavelength: int = Field(..., description="Sample wavelength in nm.") -class AbsorbanceMeasureResult(BaseModel): +class MeasureAbsorbanceResult(BaseModel): """Result data from running an aborbance reading.""" - data: List[float] = Field(..., description="Absorbance data points.") + data: Optional[List[float]] = Field( + ..., min_items=96, max_items=96, description="Absorbance data points." + ) -class AbsorbanceMeasureImpl( - AbstractCommandImpl[AbsorbanceMeasureParams, AbsorbanceMeasureResult] +class MeasureAbsorbanceImpl( + AbstractCommandImpl[MeasureAbsorbanceParams, MeasureAbsorbanceResult] ): """Execution implementation of a Thermocycler's run profile command.""" @@ -44,7 +44,7 @@ def __init__( self._state_view = state_view self._equipment = equipment - async def execute(self, params: AbsorbanceMeasureParams) -> AbsorbanceMeasureResult: + async def execute(self, params: MeasureAbsorbanceParams) -> MeasureAbsorbanceResult: """Initiate a single absorbance measurement.""" abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate( module_id=params.moduleId @@ -53,23 +53,28 @@ async def execute(self, params: AbsorbanceMeasureParams) -> AbsorbanceMeasureRes abs_reader = self._equipment.get_module_hardware_api( abs_reader_substate.module_id ) - return AbsorbanceMeasureResult(data=[]) + + if abs_reader is not None: + result = await abs_reader.start_measure(wavelength=params.sampleWavelength) + return MeasureAbsorbanceResult(data=result) + + return MeasureAbsorbanceResult(data=None) -class AbsorbanceMeasure(BaseCommand[AbsorbanceMeasureParams, AbsorbanceMeasureResult]): +class MeasureAbsorbance(BaseCommand[MeasureAbsorbanceParams, MeasureAbsorbanceResult]): """A command to execute an Absorbance Reader measurement.""" - commandType: AbsorbanceMeasureCommandType = "absorbanceReader/measure" - params: AbsorbanceMeasureParams - result: Optional[AbsorbanceMeasureResult] + commandType: MeasureAbsorbanceCommandType = "absorbanceReader/measure" + params: MeasureAbsorbanceParams + result: Optional[MeasureAbsorbanceResult] - _ImplementationCls: Type[AbsorbanceMeasureImpl] = AbsorbanceMeasureImpl + _ImplementationCls: Type[MeasureAbsorbanceImpl] = MeasureAbsorbanceImpl -class AbsorbanceMeasureCreate(BaseCommandCreate[AbsorbanceMeasureParams]): +class MeasureAbsorbanceCreate(BaseCommandCreate[MeasureAbsorbanceParams]): """A request to execute an Absorbance Reader measurement.""" - commandType: AbsorbanceMeasureCommandType = "absorbanceReader/measure" - params: AbsorbanceMeasureParams + commandType: MeasureAbsorbanceCommandType = "absorbanceReader/measure" + params: MeasureAbsorbanceParams - _CommandCls: Type[AbsorbanceMeasure] = AbsorbanceMeasure + _CommandCls: Type[MeasureAbsorbance] = MeasureAbsorbance diff --git a/api/src/opentrons/protocol_engine/commands/absorbance_reader/run_profile.py b/api/src/opentrons/protocol_engine/commands/absorbance_reader/run_profile.py deleted file mode 100644 index 467f1d7cac04..000000000000 --- a/api/src/opentrons/protocol_engine/commands/absorbance_reader/run_profile.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Command models to execute a Thermocycler profile.""" -from __future__ import annotations -from typing import List, Optional, TYPE_CHECKING -from typing_extensions import Literal, Type - -from pydantic import BaseModel, Field - -from opentrons.hardware_control.modules.types import ThermocyclerStep - -from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate - -if TYPE_CHECKING: - from opentrons.protocol_engine.state import StateView - from opentrons.protocol_engine.execution import EquipmentHandler - - -RunProfileCommandType = Literal["thermocycler/runProfile"] - - -class AbsorbanceReadParams(BaseModel): - """Input parameters for a single absorbance reading.""" - - moduleId: str = Field(..., description="Unique ID of the Thermocycler.") - sampleWavelength: float = Field(..., description="Sample wavelength in nm.") - - -class AbsorbanceReadResult(BaseModel): - """Result data from running an aborbance reading.""" - - data: List[float] = Field(..., description="Absorbance data points.") - - -class AbsorbanceReadImpl(AbstractCommandImpl[AbsorbanceReadParams, AbsorbanceReadResult]): - """Execution implementation of a Thermocycler's run profile command.""" - - def __init__( - self, - state_view: StateView, - equipment: EquipmentHandler, - **unused_dependencies: object, - ) -> None: - self._state_view = state_view - self._equipment = equipment - - async def execute(self, params: AbsorbanceReadParams) -> AbsorbanceReadResult: - """Initiate a single absorbance measurement.""" - # TODO: Implement this - return AbsorbanceReadResult(data=[]) - - -class AbsorbanceRead(BaseCommand[AbsorbanceReadParams, AbsorbanceReadResult]): - """A command to execute a Thermocycler profile run.""" - - # TODO: fix this - commandType: AbsorbanceReadCommandType = "absorbanceReader/measure" - params: AbsorbanceReadParams - result: Optional[AbsorbanceReadResult] - - _ImplementationCls: Type[AbsorbanceReadImpl] = AbsorbanceReadImpl - - -class RunProfileCreate(BaseCommandCreate[AbsorbanceReadParams]): - """A request to execute a Thermocycler profile run.""" - - commandType: AbsorbanceReadCommandType = "absorbanceReader/measure" - params: AbsorbanceReadParams - - _CommandCls: Type[AbsorbanceRead] = AbsorbanceRead diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py index 7674508cc96f..f84861f01946 100644 --- a/api/src/opentrons/protocol_engine/commands/command_unions.py +++ b/api/src/opentrons/protocol_engine/commands/command_unions.py @@ -5,6 +5,7 @@ from pydantic import Field +from . import absorbance_reader from . import heater_shaker from . import magnetic_module from . import temperature_module @@ -353,6 +354,7 @@ thermocycler.OpenLid, thermocycler.CloseLid, thermocycler.RunProfile, + absorbance_reader.MeasureAbsorbance, calibration.CalibrateGripper, calibration.CalibratePipette, calibration.CalibrateModule, @@ -419,6 +421,7 @@ thermocycler.CloseLidParams, thermocycler.RunProfileParams, thermocycler.RunProfileStepParams, + absorbance_reader.MeasureAbsorbanceParams, calibration.CalibrateGripperParams, calibration.CalibratePipetteParams, calibration.CalibrateModuleParams, @@ -482,6 +485,7 @@ thermocycler.OpenLidCommandType, thermocycler.CloseLidCommandType, thermocycler.RunProfileCommandType, + absorbance_reader.MeasureAbsorbanceCommandType, calibration.CalibrateGripperCommandType, calibration.CalibratePipetteCommandType, calibration.CalibrateModuleCommandType, @@ -546,6 +550,7 @@ thermocycler.OpenLidCreate, thermocycler.CloseLidCreate, thermocycler.RunProfileCreate, + absorbance_reader.MeasureAbsorbanceCreate, calibration.CalibrateGripperCreate, calibration.CalibratePipetteCreate, calibration.CalibrateModuleCreate, @@ -611,6 +616,7 @@ thermocycler.OpenLidResult, thermocycler.CloseLidResult, thermocycler.RunProfileResult, + absorbance_reader.MeasureAbsorbanceResult, calibration.CalibrateGripperResult, calibration.CalibratePipetteResult, calibration.CalibrateModuleResult, diff --git a/api/src/opentrons/protocol_engine/execution/equipment.py b/api/src/opentrons/protocol_engine/execution/equipment.py index 7dc2f3bcfaaf..d41894a4cb21 100644 --- a/api/src/opentrons/protocol_engine/execution/equipment.py +++ b/api/src/opentrons/protocol_engine/execution/equipment.py @@ -14,6 +14,7 @@ HeaterShaker, TempDeck, Thermocycler, + AbsorbanceReader, ) from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_engine.state.module_substates import ( @@ -21,6 +22,7 @@ HeaterShakerModuleId, TemperatureModuleId, ThermocyclerModuleId, + AbsorbanceReaderId, ) from ..errors import ( FailedToLoadPipetteError, @@ -488,6 +490,13 @@ def get_module_hardware_api( ) -> Optional[Thermocycler]: ... + @overload + def get_module_hardware_api( + self, + module_id: AbsorbanceReaderId, + ) -> Optional[AbsorbanceReader]: + ... + def get_module_hardware_api(self, module_id: str) -> Optional[AbstractModule]: """Get the hardware API for a given module.""" use_virtual_modules = self._state_store.config.use_virtual_modules diff --git a/api/src/opentrons/protocol_engine/state/module_substates/absorbance_reader_substate.py b/api/src/opentrons/protocol_engine/state/module_substates/absorbance_reader_substate.py index f694f798a71e..140e391cfaf7 100644 --- a/api/src/opentrons/protocol_engine/state/module_substates/absorbance_reader_substate.py +++ b/api/src/opentrons/protocol_engine/state/module_substates/absorbance_reader_substate.py @@ -1,6 +1,6 @@ """Heater-Shaker Module sub-state.""" from dataclasses import dataclass -from typing import List, NewType, Optional +from typing import NewType AbsorbanceReaderId = NewType("AbsorbanceReaderId", str) @@ -11,10 +11,3 @@ class AbsorbanceReaderSubState: """Absorbance-Plate-Reader-specific state.""" module_id: AbsorbanceReaderId - initialized: bool - is_lid_open: bool - is_loaded: bool - is_measuring: bool - temperature: float - sample_wavelength: Optional[int] - supported_wavelengths: Optional[List[int]] diff --git a/api/src/opentrons/protocol_engine/state/modules.py b/api/src/opentrons/protocol_engine/state/modules.py index 0e79dd53cf25..5814d77623d9 100644 --- a/api/src/opentrons/protocol_engine/state/modules.py +++ b/api/src/opentrons/protocol_engine/state/modules.py @@ -62,10 +62,12 @@ HeaterShakerModuleSubState, TemperatureModuleSubState, ThermocyclerModuleSubState, + AbsorbanceReaderSubState, MagneticModuleId, HeaterShakerModuleId, TemperatureModuleId, ThermocyclerModuleId, + AbsorbanceReaderId, MagneticBlockSubState, MagneticBlockId, ModuleSubStateType, @@ -321,6 +323,10 @@ def _add_module_substate( self._state.substate_by_module_id[module_id] = MagneticBlockSubState( module_id=MagneticBlockId(module_id) ) + elif ModuleModel.is_absorbance_reader(actual_model): + self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState( + module_id=AbsorbanceReaderId(module_id) + ) def _update_additional_slots_occupied_by_thermocycler( self, @@ -644,6 +650,22 @@ def get_thermocycler_module_substate( expected_name="Thermocycler Module", ) + def get_absorbance_reader_substate( + self, module_id: str + ) -> AbsorbanceReaderSubState: + """Return a `AbsorbanceReaderSubState` for the given Absorbance Reader. + + Raises: + ModuleNotLoadedError: If module_id has not been loaded. + WrongModuleTypeError: If module_id has been loaded, + but it's not an Absorbance Reader. + """ + return self._get_module_substate( + module_id=module_id, + expected_type=AbsorbanceReaderSubState, + expected_name="Thermocycler Module", + ) + def get_location(self, module_id: str) -> DeckSlotLocation: """Get the slot location of the given module.""" location = self.get(module_id).location diff --git a/api/src/opentrons/protocol_engine/types.py b/api/src/opentrons/protocol_engine/types.py index 13e9515e4479..617e119f777e 100644 --- a/api/src/opentrons/protocol_engine/types.py +++ b/api/src/opentrons/protocol_engine/types.py @@ -306,6 +306,7 @@ class ModuleModel(str, Enum): THERMOCYCLER_MODULE_V2 = "thermocyclerModuleV2" HEATER_SHAKER_MODULE_V1 = "heaterShakerModuleV1" MAGNETIC_BLOCK_V1 = "magneticBlockV1" + ABSORBANCE_READER_V1 = "absorbanceReaderV1" def as_type(self) -> ModuleType: """Get the ModuleType of this model.""" @@ -319,6 +320,8 @@ def as_type(self) -> ModuleType: return ModuleType.HEATER_SHAKER elif ModuleModel.is_magnetic_block(self): return ModuleType.MAGNETIC_BLOCK + elif ModuleModel.is_absorbance_reader(self): + return ModuleType.ABSORBANCE_READER assert False, f"Invalid ModuleModel {self}" @@ -355,6 +358,13 @@ def is_magnetic_block(cls, model: ModuleModel) -> TypeGuard[MagneticBlockModel]: """Whether a given model is a Magnetic block.""" return model == cls.MAGNETIC_BLOCK_V1 + @classmethod + def is_absorbance_reader( + cls, model: ModuleModel + ) -> TypeGuard[AbsorbanceReaderModel]: + """Whether a given model is a Magnetic block.""" + return model == cls.ABSORBANCE_READER_V1 + TemperatureModuleModel = Literal[ ModuleModel.TEMPERATURE_MODULE_V1, ModuleModel.TEMPERATURE_MODULE_V2 @@ -367,6 +377,7 @@ def is_magnetic_block(cls, model: ModuleModel) -> TypeGuard[MagneticBlockModel]: ] HeaterShakerModuleModel = Literal[ModuleModel.HEATER_SHAKER_MODULE_V1] MagneticBlockModel = Literal[ModuleModel.MAGNETIC_BLOCK_V1] +AbsorbanceReaderModel = Literal[ModuleModel.ABSORBANCE_READER_V1] class ModuleDimensions(BaseModel):