fix(api): fix SimulatedProbeResult ser/deser

api/src/opentrons/protocol_engine/types/__init__.py

@@ -135,6 +135,7 @@
     WellInfoSummary,
     WellLiquidInfo,
     LiquidTrackingType,
+    SimulatedProbeResult,
 )
 from .liquid_handling import FlowRates
 from .labware_movement import LabwareMovementStrategy, LabwareMovementOffsetData
@@ -280,6 +281,7 @@
     "WellInfoSummary",
     "WellLiquidInfo",
     "LiquidTrackingType",
+    "SimulatedProbeResult",
     # Liquid handling
     "FlowRates",
     # Labware movement

api/src/opentrons/protocol_engine/types/liquid_level_detection.py

@@ -1,9 +1,10 @@
 """Protocol Engine types to do with liquid level detection."""
+
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Optional, List
-from pydantic import BaseModel, model_serializer, field_validator
+from typing import Optional, List, Any
+from pydantic import BaseModel, model_serializer, model_validator
 
 
 class SimulatedProbeResult(BaseModel):
@@ -17,6 +18,14 @@ def serialize_model(self) -> str:
         """Serialize instances of this class as a string."""
         return "SimulatedProbeResult"
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_model(cls, data: object) -> Any:
+        """Handle deserializing from a simulated probe result."""
+        if isinstance(data, str) and data == "SimulatedProbeResult":
+            return {}
+        return data
+
     def __add__(
         self, other: float | SimulatedProbeResult
     ) -> float | SimulatedProbeResult:
@@ -75,7 +84,9 @@ def simulate_probed_aspirate_dispense(self, volume: float) -> None:
         self.operations_after_probe.append(volume)
 
 
-LiquidTrackingType = SimulatedProbeResult | float
+# Work around https://github.com/pydantic/pydantic/issues/6830 - do not change the order of
+# this union
+LiquidTrackingType = float | SimulatedProbeResult
 
 
 class LoadedVolumeInfo(BaseModel):
@@ -104,23 +115,6 @@ class ProbedVolumeInfo(BaseModel):
 class WellInfoSummary(BaseModel):
     """Payload for a well's liquid info in StateSummary."""
 
-    # TODO(cm): 3/21/25: refactor SimulatedLiquidProbe in a way that
-    # doesn't require models like this one that are just using it to
-    # need a custom validator
-    @field_validator("probed_height", "probed_volume", mode="before")
-    @classmethod
-    def validate_simulated_probe_result(
-        cls, input_val: object
-    ) -> LiquidTrackingType | None:
-        """Return the appropriate input to WellInfoSummary from json data."""
-        if input_val is None:
-            return None
-        if isinstance(input_val, LiquidTrackingType):
-            return input_val
-        if isinstance(input_val, str) and input_val == "SimulatedProbeResult":
-            return SimulatedProbeResult()
-        raise ValueError(f"Invalid input value {input_val} to WellInfoSummary")
-
     labware_id: str
     well_name: str
     loaded_volume: Optional[float] = None

api/tests/opentrons/protocol_engine/test_types.py

@@ -1,8 +1,14 @@
 """Test protocol engine types."""
+
 import pytest
-from pydantic import ValidationError
+from pydantic import ValidationError, BaseModel
 
-from opentrons.protocol_engine.types import HexColor
+from opentrons.protocol_engine.types import (
+    HexColor,
+    SimulatedProbeResult,
+    LiquidTrackingType,
+    WellInfoSummary,
+)
 
 
 @pytest.mark.parametrize("hex_color", ["#F00", "#FFCC00CC", "#FC0C", "#98e2d1"])
@@ -20,3 +26,51 @@ def test_handles_invalid_hex(invalid_hex_color: str) -> None:
         HexColor(invalid_hex_color)
     with pytest.raises(ValidationError):
         HexColor.model_validate_json(f'"{invalid_hex_color}"')
+
+
+class _TestModel(BaseModel):
+    """Test model for deserializing SimulatedProbeResults."""
+
+    value: LiquidTrackingType
+
+
+def test_roundtrips_simulated_liquid_probe() -> None:
+    """Should be able to roundtrip our simulated results."""
+    base = _TestModel(value=SimulatedProbeResult())
+    serialized = base.model_dump_json()
+    deserialized = _TestModel.model_validate_json(serialized)
+    assert isinstance(deserialized.value, SimulatedProbeResult)
+
+
+def test_roundtrips_nonsimulated_liquid_probe() -> None:
+    """Should be able to roundtrip our simulated results."""
+    base = _TestModel(value=10.0)
+    serialized = base.model_dump_json()
+    deserialized = _TestModel.model_validate_json(serialized)
+    assert deserialized.value == 10.0
+
+
+def test_fails_deser_wrong_string() -> None:
+    """Should fail to deserialize the wrong string."""
+    with pytest.raises(ValidationError):
+        _TestModel.model_validate_json('{"value": "not the right string"}')
+
+
+@pytest.mark.parametrize("height", [None, 10.0, SimulatedProbeResult()])
+def test_roundtrips_well_info_summary(height: LiquidTrackingType | None) -> None:
+    """It should round trip a WellInfoSummary."""
+    inp = WellInfoSummary(
+        labware_id="hi",
+        well_name="lo",
+        loaded_volume=None,
+        probed_height=height,
+        probed_volume=height,
+    )
+    outp = WellInfoSummary.model_validate_json(inp.model_dump_json())
+    if isinstance(height, SimulatedProbeResult):
+        assert outp.labware_id == inp.labware_id
+        assert outp.well_name == inp.well_name
+        assert isinstance(outp.probed_height, SimulatedProbeResult)
+        assert isinstance(outp.probed_volume, SimulatedProbeResult)
+    else:
+        assert outp == inp