RFC 7386 JSON Merge Patch serialization with compact format

pylabrobot/centrifuge/centrifuge.py

@@ -134,18 +134,22 @@ def serialize(self) -> dict:
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False):
-    backend = CentrifugeBackend.deserialize(data["backend"])
-    buckets = tuple(ResourceHolder.deserialize(bucket) for bucket in data["buckets"])
-    assert len(buckets) == 2
+    buckets_data = data.get("buckets")
+    buckets = (
+      tuple(ResourceHolder.deserialize(bucket) for bucket in buckets_data) if buckets_data else None
+    )
+    if buckets is not None:
+      assert len(buckets) == 2
+    rotation_data = data.get("rotation")
     return cls(
-      backend=backend,
+      backend=CentrifugeBackend.deserialize(data["backend"]),
       name=data["name"],
       size_x=data["size_x"],
       size_y=data["size_y"],
       size_z=data["size_z"],
-      rotation=Rotation.deserialize(data["rotation"]),
-      category=data["category"],
-      model=data["model"],
+      rotation=deserialize(rotation_data) if rotation_data else None,
+      category=data.get("category"),
+      model=data.get("model"),
       buckets=buckets,
     )
 
@@ -228,15 +232,18 @@ def serialize(self) -> dict:
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False):
+    resource_data = data.get("resource", {})
+    machine_data = data.get("machine", {})
+    rotation_data = resource_data.get("rotation")
     return cls(
-      backend=LoaderBackend.deserialize(data["machine"]["backend"]),
+      backend=LoaderBackend.deserialize(machine_data["backend"]),
       centrifuge=Centrifuge.deserialize(data["centrifuge"]),
-      name=data["resource"]["name"],
-      size_x=data["resource"]["size_x"],
-      size_y=data["resource"]["size_y"],
-      size_z=data["resource"]["size_z"],
-      child_location=deserialize(data["resource"]["child_location"]),
-      rotation=deserialize(data["resource"]["rotation"]),
-      category=data["resource"]["category"],
-      model=data["resource"]["model"],
+      name=resource_data["name"],
+      size_x=resource_data["size_x"],
+      size_y=resource_data["size_y"],
+      size_z=resource_data["size_z"],
+      child_location=deserialize(resource_data["child_location"]),
+      rotation=deserialize(rotation_data) if rotation_data else None,
+      category=resource_data.get("category"),
+      model=resource_data.get("model"),
     )

pylabrobot/io/serial.py

@@ -342,20 +342,6 @@ def serialize(self):
       "dsrdtr": self.dsrdtr,
     }
 
-  @classmethod
-  def deserialize(cls, data: dict) -> "Serial":
-    return cls(
-      port=data["port"],
-      baudrate=data["baudrate"],
-      bytesize=data["bytesize"],
-      parity=data["parity"],
-      stopbits=data["stopbits"],
-      write_timeout=data["write_timeout"],
-      timeout=data["timeout"],
-      rtscts=data["rtscts"],
-      dsrdtr=data["dsrdtr"],
-    )
-
 
 class SerialValidator(Serial):
   def __init__(

pylabrobot/io/socket.py

@@ -96,19 +96,6 @@ def serialize(self):
       "write_timeout": self._write_timeout,
     }
 
-  @classmethod
-  def deserialize(cls, data: dict) -> "Socket":
-    kwargs = {}
-    if "read_timeout" in data:
-      kwargs["read_timeout"] = data["read_timeout"]
-    if "write_timeout" in data:
-      kwargs["write_timeout"] = data["write_timeout"]
-    return cls(
-      host=data["host"],
-      port=data["port"],
-      **kwargs,
-    )
-
   async def write(self, data: bytes, timeout: Optional[float] = None) -> None:
     """Wrapper around StreamWriter.write with lock and io logging.
     Does not retry on timeouts.

pylabrobot/machines/backend.py

@@ -26,12 +26,13 @@ def serialize(self) -> dict:
 
   @classmethod
   def deserialize(cls, data: dict):
+    data = data.copy()
     class_name = data.pop("type")
     subclass = find_subclass(class_name, cls=cls)
     if subclass is None:
-      raise ValueError(f'Could not find subclass with name "{data["type"]}"')
+      raise ValueError(f'Could not find subclass with name "{class_name}"')
     if inspect.isabstract(subclass):
-      raise ValueError(f'Subclass with name "{data["type"]}" is abstract')
+      raise ValueError(f'Subclass with name "{class_name}" is abstract')
     assert issubclass(subclass, cls)
     return subclass(**data)
 

pylabrobot/pumps/calibration.py

@@ -99,13 +99,6 @@ def serialize(self) -> dict:
       "calibration_mode": self.calibration_mode,
     }
 
-  @classmethod
-  def deserialize(cls, data: dict) -> PumpCalibration:
-    return cls(
-      calibration=data["calibration"],
-      calibration_mode=data["calibration_mode"],
-    )
-
   @classmethod
   def load_from_json(
     cls,

pylabrobot/pumps/pump.py

@@ -34,8 +34,7 @@ def deserialize(cls, data: dict):
     data_copy = data.copy()
     calibration_data = data_copy.pop("calibration", None)
     if calibration_data is not None:
-      calibration = PumpCalibration.deserialize(calibration_data)
-      data_copy["calibration"] = calibration
+      data_copy["calibration"] = PumpCalibration(**calibration_data)
     return super().deserialize(data_copy)
 
   async def run_revolutions(self, num_revolutions: float):

pylabrobot/pumps/pumparray.py

@@ -50,8 +50,7 @@ def deserialize(cls, data: dict):
     data_copy = data.copy()
     calibration_data = data_copy.pop("calibration", None)
     if calibration_data is not None:
-      calibration = PumpCalibration.deserialize(calibration_data)
-      data_copy["calibration"] = calibration
+      data_copy["calibration"] = PumpCalibration(**calibration_data)
     return super().deserialize(data_copy)
 
   async def run_revolutions(

pylabrobot/resources/barcode.py

@@ -34,13 +34,5 @@ def serialize(self) -> dict:
       "position_on_resource": self.position_on_resource,
     }
 
-  @staticmethod
-  def deserialize(data: dict) -> "Barcode":
-    return Barcode(
-      data=data["data"],
-      symbology=data["symbology"],
-      position_on_resource=data["position_on_resource"],
-    )
-
   def __str__(self) -> str:
     return f'Barcode(data="{self.data}", symbology="{self.symbology}", position_on_resource="{self.position_on_resource}")'

pylabrobot/resources/hamilton/tip_creators.py

@@ -97,17 +97,6 @@ def serialize(self):
       "tip_size": self.tip_size.name,
     }
 
-  @classmethod
-  def deserialize(cls, data):
-    return HamiltonTip(
-      name=data["name"],
-      has_filter=data["has_filter"],
-      total_tip_length=data["total_tip_length"],
-      maximal_volume=data["maximal_volume"],
-      tip_size=TipSize[data["tip_size"]],
-      pickup_method=TipPickupMethod[data["pickup_method"]],
-    )
-
 
 def standard_volume_tip_no_filter(name: Optional[str] = None) -> HamiltonTip:
   """Deprecated. Use :func:`hamilton_tip_300uL` instead."""

pylabrobot/resources/plate_adapter.py

@@ -117,25 +117,6 @@ def serialize(self) -> dict:
       "plate_z_offset": self.plate_z_offset,
     }
 
-  @classmethod
-  def deserialize(cls, data: dict, allow_marshal: bool = False) -> PlateAdapter:
-    return cls(
-      name=data["name"],
-      size_x=data["size_x"],
-      size_y=data["size_y"],
-      size_z=data["size_z"],
-      dx=data["dx"],
-      dy=data["dy"],
-      dz=data["dz"],
-      adapter_hole_size_x=data["adapter_hole_size_x"],
-      adapter_hole_size_y=data["adapter_hole_size_y"],
-      adapter_hole_dx=data["adapter_hole_dx"],
-      adapter_hole_dy=data["adapter_hole_dy"],
-      plate_z_offset=data["plate_z_offset"],
-      category=data.get("category"),
-      model=data.get("model"),
-    )
-
   def compute_plate_location(self, resource: Plate) -> Coordinate:
     """Compute the location of the `Plate` child resource in relationship to the `PlateAdapter` to
     align the `Plate` well-grid with the adapter's hole grid."""

pylabrobot/resources/resource.py

@@ -750,15 +750,16 @@ def deserialize(cls, data: dict, allow_marshal: bool = False) -> Self:
       "parent_name",
       "location",
     ]:  # delete meta keys
-      del data_copy[key]
-    children_data = data_copy.pop("children")
-    rotation = data_copy.pop("rotation")
+      data_copy.pop(key, None)
+    children_data = data_copy.pop("children", [])
+    rotation = data_copy.pop("rotation", None)
     barcode = data_copy.pop("barcode", None)
     preferred_pickup_location = data_copy.pop("preferred_pickup_location", None)
     resource = subclass(**deserialize(data_copy, allow_marshal=allow_marshal))
-    resource.rotation = Rotation.deserialize(rotation)  # not pretty, should be done in init.
+    if rotation is not None:
+      resource.rotation = deserialize(rotation)  # not pretty, should be done in init.
     if barcode is not None:
-      resource.barcode = Barcode.deserialize(barcode)
+      resource.barcode = Barcode(**barcode)
     if preferred_pickup_location is not None:
       resource.preferred_pickup_location = cast(Coordinate, deserialize(preferred_pickup_location))
 

pylabrobot/resources/rotation.py

@@ -62,10 +62,6 @@ def __str__(self) -> str:
   def __add__(self, other) -> "Rotation":
     return Rotation(x=self.x + other.x, y=self.y + other.y, z=self.z + other.z)
 
-  @staticmethod
-  def deserialize(data) -> "Rotation":
-    return Rotation(data["x"], data["y"], data["z"])
-
   def __repr__(self) -> str:
     return self.__str__()
 

pylabrobot/resources/tip_rack.py

@@ -122,7 +122,7 @@ def make_tip(name: str) -> Tip:
       name=data["name"],
       size_x=data["size_x"],
       size_y=data["size_y"],
-      size_z=data["size_z"],
+      size_z=data.get("size_z", 0),
       make_tip=make_tip,
       category=data.get("category", "tip_spot"),
     )

pylabrobot/resources/tip_tests.py

@@ -61,4 +61,4 @@ def test_deserialize_subclass(self):
       TipPickupMethod.OUT_OF_RACK,
       name="test_tip",
     )
-    self.assertEqual(HamiltonTip.deserialize(tip.serialize()), tip)
+    self.assertEqual(deserialize(tip.serialize()), tip)

pylabrobot/storage/cytomat/constants.py

@@ -137,10 +137,6 @@ class CytomatRack:
   num_slots: int  # number of plate locations in rack
   pitch: int  # distance between 2 plate locations
 
-  @classmethod
-  def deserialize(cls, data: dict):
-    return cls(num_slots=data["num_slots"], pitch=data["pitch"])
-
 
 class CytomatType(Enum):
   C6000 = "C6000"

pylabrobot/storage/cytomat/heraeus_cytomat_backend.py

@@ -192,7 +192,3 @@ def serialize(self) -> dict:
       **super().serialize(),
       "port": self.io.port,
     }
-
-  @classmethod
-  def deserialize(cls, data: dict):
-    return cls(port=data["port"])

pylabrobot/storage/incubator.py

@@ -194,16 +194,16 @@ def serialize(self):
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False):
-    backend = IncubatorBackend.deserialize(data.pop("backend"))
+    rotation_data = data.get("rotation")
     return cls(
-      backend=backend,
+      backend=IncubatorBackend.deserialize(data["backend"]),
       name=data["name"],
       size_x=data["size_x"],
       size_y=data["size_y"],
       size_z=data["size_z"],
       racks=[PlateCarrier.deserialize(rack) for rack in data["racks"]],
       loading_tray_location=cast(Coordinate, deserialize(data["loading_tray_location"])),
-      rotation=Rotation.deserialize(data["rotation"]),
-      category=data["category"],
-      model=data["model"],
+      rotation=deserialize(rotation_data) if rotation_data else None,
+      category=data.get("category"),
+      model=data.get("model"),
     )

pylabrobot/storage/inheco/scila/scila_backend.py

@@ -131,7 +131,3 @@ def serialize(self) -> dict[str, Any]:
       "scila_ip": self._sila_interface.machine_ip,
       "client_ip": self._sila_interface.client_ip,
     }
-
-  @classmethod
-  def deserialize(cls, data: dict[str, Any]) -> "SCILABackend":
-    return cls(scila_ip=data["scila_ip"], client_ip=data.get("client_ip"))

pylabrobot/storage/inheco/scila/scila_backend_tests.py

@@ -2,6 +2,7 @@
 import xml.etree.ElementTree as ET
 from unittest.mock import AsyncMock, patch
 
+from pylabrobot.machines.backend import MachineBackend
 from pylabrobot.storage.inheco.scila.inheco_sila_interface import InhecoSiLAInterface
 from pylabrobot.storage.inheco.scila.scila_backend import SCILABackend
 
@@ -220,15 +221,15 @@ def test_serialize_no_client_ip(self):
     self.assertIsNone(data["client_ip"])
 
   def test_deserialize(self):
-    data = {"scila_ip": "169.254.1.117", "client_ip": "192.168.1.10"}
-    SCILABackend.deserialize(data)
+    data = {"type": "SCILABackend", "scila_ip": "169.254.1.117", "client_ip": "192.168.1.10"}
+    MachineBackend.deserialize(data)
     self.MockInhecoSiLAInterface.assert_called_with(
       client_ip="192.168.1.10", machine_ip="169.254.1.117"
     )
 
   def test_deserialize_no_client_ip(self):
-    data = {"scila_ip": "169.254.1.117"}
-    SCILABackend.deserialize(data)
+    data = {"type": "SCILABackend", "scila_ip": "169.254.1.117"}
+    MachineBackend.deserialize(data)
     self.MockInhecoSiLAInterface.assert_called_with(client_ip=None, machine_ip="169.254.1.117")
 
 

pylabrobot/tests/serializer_tests.py

@@ -1,6 +1,9 @@
 import math
 
-from pylabrobot.serializer import deserialize, serialize
+from pylabrobot.serializer import (
+  deserialize,
+  serialize,
+)
 
 
 def test_serialize_deserialize_closure():