Skip to content

Commit

Permalink
fix types in robot-server and tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@ahiuchingau
ahiuchingau committed Jan 9, 2025
1 parent 5d94f16 commit 281fb67
Show file tree
Hide file tree
Showing 3 changed files with 63 additions and 53 deletions.
1 change: 1 addition & 0 deletions api/src/opentrons/hardware_control/modules/types.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
cast,
TYPE_CHECKING,
TypeGuard,
Literal,
)
from typing_extensions import TypedDict
from pathlib import Path
Expand Down
7 changes: 6 additions & 1 deletion robot-server/robot_server/modules/module_data_mapper.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from opentrons.hardware_control.modules import (
LiveData,
ModuleDataValidator,
ModuleType,
MagneticStatus,
TemperatureStatus,
Expand Down Expand Up @@ -71,7 +72,7 @@ def map_data(
# rely on Pydantic to check/coerce data fields from dicts at run time
if module_type == ModuleType.MAGNETIC:
module_cls = MagneticModule

assert ModuleDataValidator.is_magnetic_module_data(live_data["data"])
live_data_height = live_data["data"].get("height")
assert isinstance(
live_data_height, (int, float)
Expand All @@ -92,6 +93,7 @@ def map_data(

elif module_type == ModuleType.TEMPERATURE:
module_cls = TemperatureModule
assert ModuleDataValidator.is_temperature_module_data(live_data["data"])
module_data = TemperatureModuleData(
status=TemperatureStatus(live_data["status"]),
targetTemperature=cast(float, live_data["data"].get("targetTemp")),
Expand All @@ -100,6 +102,7 @@ def map_data(

elif module_type == ModuleType.THERMOCYCLER:
module_cls = ThermocyclerModule
assert ModuleDataValidator.is_thermocycler_data(live_data["data"])
module_data = ThermocyclerModuleData(
status=TemperatureStatus(live_data["status"]),
targetTemperature=cast(float, live_data["data"].get("targetTemp")),
Expand All @@ -120,6 +123,7 @@ def map_data(

elif module_type == ModuleType.HEATER_SHAKER:
module_cls = HeaterShakerModule
assert ModuleDataValidator.is_heater_shaker_data(live_data["data"])
module_data = HeaterShakerModuleData(
status=HeaterShakerStatus(live_data["status"]),
labwareLatchStatus=cast(
Expand All @@ -137,6 +141,7 @@ def map_data(
errorDetails=cast(str, live_data["data"].get("errorDetails")),
)
elif module_type == ModuleType.ABSORBANCE_READER:
assert ModuleDataValidator.is_absorbance_reader_data(live_data["data"])
module_cls = AbsorbanceReaderModule
module_data = AbsorbanceReaderModuleData(
status=AbsorbanceReaderStatus(live_data["status"]),
Expand Down
108 changes: 56 additions & 52 deletions robot-server/tests/modules/test_module_data_mapper.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Tests for robot_server.modules.module_data_mapper."""
import pytest
from typing import Dict

from opentrons.protocol_engine import ModuleModel, DeckType
from opentrons.protocol_engine.types import Vec3f
Expand All @@ -10,6 +11,7 @@
MagneticStatus,
TemperatureStatus,
HeaterShakerStatus,
types as hc_types,
)


Expand Down Expand Up @@ -174,22 +176,21 @@ def test_maps_magnetic_module_data(
],
)
@pytest.mark.parametrize(
"input_data",
"status,data",
[
{"status": "idle", "data": {"currentTemp": 42.0, "targetTemp": None}},
{
"status": "holding at target",
"data": {"currentTemp": 84.0, "targetTemp": 84.0},
},
("idle", {"currentTemp": 42.0, "targetTemp": None}),
("holding at target", {"currentTemp": 84.0, "targetTemp": 84.0}),
],
)
def test_maps_temperature_module_data(
input_model: str,
deck_type: DeckType,
expected_compatible: bool,
input_data: LiveData,
status: str,
data: hc_types.TemperatureModuleData,
) -> None:
"""It should map hardware data to a magnetic module."""
input_data: LiveData = {"status": status, "data": data}
module_identity = ModuleIdentity(
module_id="module-id",
serial_number="serial-number",
Expand Down Expand Up @@ -236,9 +237,9 @@ def test_maps_temperature_module_data(
),
moduleOffset=ModuleCalibrationData(offset=Vec3f(x=0.0, y=0.0, z=0.0)),
data=TemperatureModuleData(
status=TemperatureStatus(input_data["status"]),
currentTemperature=input_data["data"]["currentTemp"], # type: ignore[arg-type]
targetTemperature=input_data["data"]["targetTemp"], # type: ignore[arg-type]
status=TemperatureStatus(status),
currentTemperature=data["currentTemp"],
targetTemperature=data["targetTemp"],
),
)

Expand All @@ -253,11 +254,11 @@ def test_maps_temperature_module_data(
],
)
@pytest.mark.parametrize(
"input_data",
"status,data",
[
{
"status": "idle",
"data": {
(
"idle",
{
"lid": "open",
"lidTarget": None,
"lidTemp": None,
Expand All @@ -271,10 +272,10 @@ def test_maps_temperature_module_data(
"currentStepIndex": None,
"totalStepCount": None,
},
},
{
"status": "heating",
"data": {
),
(
"heating",
{
"lid": "open",
"lidTarget": 1,
"lidTemp": 2,
Expand All @@ -288,16 +289,18 @@ def test_maps_temperature_module_data(
"currentStepIndex": 9,
"totalStepCount": 10,
},
},
),
],
)
def test_maps_thermocycler_module_data(
input_model: str,
deck_type: DeckType,
expected_compatible: bool,
input_data: LiveData,
status: str,
data: hc_types.ThermocyclerData,
) -> None:
"""It should map hardware data to a magnetic module."""
input_data: LiveData = {"status": status, "data": data}
module_identity = ModuleIdentity(
module_id="module-id",
serial_number="serial-number",
Expand Down Expand Up @@ -344,19 +347,19 @@ def test_maps_thermocycler_module_data(
),
moduleOffset=ModuleCalibrationData(offset=Vec3f(x=0.0, y=0.0, z=0.0)),
data=ThermocyclerModuleData(
status=TemperatureStatus(input_data["status"]),
currentTemperature=input_data["data"]["currentTemp"], # type: ignore[arg-type]
targetTemperature=input_data["data"]["targetTemp"], # type: ignore[arg-type]
lidStatus=input_data["data"]["lid"], # type: ignore[arg-type]
lidTemperatureStatus=input_data["data"]["lidTempStatus"], # type: ignore[arg-type]
lidTemperature=input_data["data"]["lidTemp"], # type: ignore[arg-type]
lidTargetTemperature=input_data["data"]["lidTarget"], # type: ignore[arg-type]
holdTime=input_data["data"]["holdTime"], # type: ignore[arg-type]
rampRate=input_data["data"]["rampRate"], # type: ignore[arg-type]
currentCycleIndex=input_data["data"]["currentCycleIndex"], # type: ignore[arg-type]
totalCycleCount=input_data["data"]["totalCycleCount"], # type: ignore[arg-type]
currentStepIndex=input_data["data"]["currentStepIndex"], # type: ignore[arg-type]
totalStepCount=input_data["data"]["totalStepCount"], # type: ignore[arg-type]
status=TemperatureStatus(status),
currentTemperature=data["currentTemp"],
targetTemperature=data["targetTemp"],
lidStatus=data["lid"], # type: ignore[arg-type]
lidTemperatureStatus=data["lidTempStatus"], # type: ignore[arg-type]
lidTemperature=data["lidTemp"],
lidTargetTemperature=data["lidTarget"],
holdTime=data["holdTime"],
rampRate=data["rampRate"],
currentCycleIndex=data["currentCycleIndex"],
totalCycleCount=data["totalCycleCount"],
currentStepIndex=data["currentStepIndex"],
totalStepCount=data["totalStepCount"],
),
)

Expand All @@ -369,11 +372,11 @@ def test_maps_thermocycler_module_data(
],
)
@pytest.mark.parametrize(
"input_data",
"status,data",
[
{
"status": "idle",
"data": {
(
"idle",
{
"temperatureStatus": "idle",
"speedStatus": "idle",
"labwareLatchStatus": "idle_open",
Expand All @@ -383,10 +386,10 @@ def test_maps_thermocycler_module_data(
"targetSpeed": None,
"errorDetails": None,
},
},
{
"status": "running",
"data": {
),
(
"running",
{
"temperatureStatus": "heating",
"speedStatus": "speeding up",
"labwareLatchStatus": "idle_closed",
Expand All @@ -396,13 +399,14 @@ def test_maps_thermocycler_module_data(
"targetSpeed": 9001,
"errorDetails": "oh no",
},
},
),
],
)
def test_maps_heater_shaker_module_data(
input_model: str, deck_type: DeckType, input_data: LiveData
input_model: str, deck_type: DeckType, status: str, data: hc_types.HeaterShakerData
) -> None:
"""It should map hardware data to a magnetic module."""
input_data: LiveData = {"status": status, "data": data}
module_identity = ModuleIdentity(
module_id="module-id",
serial_number="serial-number",
Expand Down Expand Up @@ -449,14 +453,14 @@ def test_maps_heater_shaker_module_data(
),
moduleOffset=ModuleCalibrationData(offset=Vec3f(x=0.0, y=0.0, z=0.0)),
data=HeaterShakerModuleData(
status=HeaterShakerStatus(input_data["status"]),
labwareLatchStatus=input_data["data"]["labwareLatchStatus"], # type: ignore[arg-type]
speedStatus=input_data["data"]["speedStatus"], # type: ignore[arg-type]
currentSpeed=input_data["data"]["currentSpeed"], # type: ignore[arg-type]
targetSpeed=input_data["data"]["targetSpeed"], # type: ignore[arg-type]
temperatureStatus=input_data["data"]["temperatureStatus"], # type: ignore[arg-type]
currentTemperature=input_data["data"]["currentTemp"], # type: ignore[arg-type]
targetTemperature=input_data["data"]["targetTemp"], # type: ignore[arg-type]
errorDetails=input_data["data"]["errorDetails"], # type: ignore[arg-type]
status=HeaterShakerStatus(status),
labwareLatchStatus=data["labwareLatchStatus"], # type: ignore[arg-type]
speedStatus=data["speedStatus"], # type: ignore[arg-type]
currentSpeed=data["currentSpeed"],
targetSpeed=data["targetSpeed"],
temperatureStatus=data["temperatureStatus"], # type: ignore[arg-type]
currentTemperature=data["currentTemp"],
targetTemperature=data["targetTemp"],
errorDetails=data["errorDetails"],
),
)

0 comments on commit 281fb67

Please sign in to comment.