Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add absorbance reader command to protocol engine 1
Browse files Browse the repository at this point in the history
ahiuchingau committed May 10, 2024
1 parent a96add8 commit ced2175
Showing 4 changed files with 95 additions and 1 deletion.
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Command models for Absorbance Reader commands."""
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
"""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
Original file line number Diff line number Diff line change
@@ -12,14 +12,15 @@
ThermocyclerModuleId,
)
from .magnetic_block_substate import MagneticBlockSubState, MagneticBlockId

from .absorbance_reader_substate import AbsorbanceReaderSubState, AbsorbanceReaderId

ModuleSubStateType = Union[
HeaterShakerModuleSubState,
MagneticModuleSubState,
TemperatureModuleSubState,
ThermocyclerModuleSubState,
MagneticBlockSubState,
AbsorbanceReaderSubState,
]

__all__ = [
@@ -33,6 +34,8 @@
"ThermocyclerModuleId",
"MagneticBlockSubState",
"MagneticBlockId",
"AbsorbanceReaderSubState",
"AbsorbanceReaderId",
# Union of all module substates
"ModuleSubStateType",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
"""Heater-Shaker Module sub-state."""
from dataclasses import dataclass
from typing import NewType, List

from opentrons.protocol_engine.types import (
TemperatureRange,
SpeedRange,
HeaterShakerLatchStatus,
)

AbsorbanceReaderId = NewType("AbsorbanceReaderId", str)


@dataclass(frozen=True)
class AbsorbanceReaderSubState:
"""Absorbance-Plate-Reader-specific state."""

module_id: AbsorbanceReaderId
is_lid_open: bool
is_loaded: bool
sample_wavelength: int
supported_wavelengths: List[int]

0 comments on commit ced2175

Please sign in to comment.