RFC 7386 JSON Merge Patch serialization with compact format

pylabrobot/centrifuge/centrifuge.py

@@ -135,17 +135,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,
       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=Rotation.deserialize(rotation_data) if rotation_data else None,
+      category=data.get("category", "centrifuge"),
+      model=data.get("model"),
       buckets=buckets,
     )
 
@@ -228,15 +233,19 @@ def serialize(self) -> dict:
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False):
+    resource_data = data.get("resource", {})
+    machine_data = data.get("machine", {})
     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(resource_data.get("rotation"))
+      if resource_data.get("rotation")
+      else None,
+      category=resource_data.get("category", "loader"),
+      model=resource_data.get("model"),
     )

pylabrobot/io/serial.py

@@ -345,15 +345,15 @@ def serialize(self):
   @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"],
+      port=data.get("port"),
+      baudrate=data.get("baudrate", 9600),
+      bytesize=data.get("bytesize", 8),
+      parity=data.get("parity", "N"),
+      stopbits=data.get("stopbits", 1),
+      write_timeout=data.get("write_timeout", 1),
+      timeout=data.get("timeout", 1),
+      rtscts=data.get("rtscts", False),
+      dsrdtr=data.get("dsrdtr", False),
     )
 
 

pylabrobot/io/socket.py

@@ -98,15 +98,11 @@ def serialize(self):
 
   @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,
+      read_timeout=data.get("read_timeout", 30),
+      write_timeout=data.get("write_timeout", 30),
     )
 
   async def write(self, data: bytes, timeout: Optional[float] = None) -> None:

pylabrobot/pumps/calibration.py

@@ -103,7 +103,7 @@ def serialize(self) -> dict:
   def deserialize(cls, data: dict) -> PumpCalibration:
     return cls(
       calibration=data["calibration"],
-      calibration_mode=data["calibration_mode"],
+      calibration_mode=data.get("calibration_mode", "duration"),
     )
 
   @classmethod

pylabrobot/resources/hamilton/tip_creators.py

@@ -100,12 +100,12 @@ def serialize(self):
   @classmethod
   def deserialize(cls, data):
     return HamiltonTip(
-      name=data["name"],
+      name=data.get("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"]],
+      pickup_method=TipPickupMethod[data.get("pickup_method", "OUT_OF_RACK")],
     )
 
 

pylabrobot/resources/plate_adapter.py

@@ -129,9 +129,9 @@ def deserialize(cls, data: dict, allow_marshal: bool = False) -> PlateAdapter:
       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"],
+      adapter_hole_dx=data.get("adapter_hole_dx", 9.0),
+      adapter_hole_dy=data.get("adapter_hole_dy", 9.0),
+      plate_z_offset=data.get("plate_z_offset", 0.0),
       category=data.get("category"),
       model=data.get("model"),
     )

pylabrobot/resources/resource.py

@@ -6,7 +6,7 @@
 import sys
 from typing import Any, Callable, Dict, List, Optional, Union, cast
 
-from pylabrobot.serializer import deserialize, serialize
+from pylabrobot.serializer import compact, deserialize, serialize
 from pylabrobot.utils.linalg import matrix_vector_multiply_3x3
 from pylabrobot.utils.object_parsing import find_subclass
 
@@ -717,7 +717,7 @@ def save(self, fn: str, indent: Optional[int] = None):
       >>> deck.save("my_layout.json")
     """
 
-    serialized = self.serialize()
+    serialized = compact(self.serialize())
     with open(fn, "w", encoding="utf-8") as f:
       json.dump(serialized, f, indent=indent)
 
@@ -750,13 +750,14 @@ 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 = Rotation.deserialize(rotation)  # not pretty, should be done in init.
     if barcode is not None:
       resource.barcode = Barcode.deserialize(barcode)
     if preferred_pickup_location is not None:

pylabrobot/resources/rotation.py

@@ -64,7 +64,7 @@ def __add__(self, other) -> "Rotation":
 
   @staticmethod
   def deserialize(data) -> "Rotation":
-    return Rotation(data["x"], data["y"], data["z"])
+    return Rotation(data.get("x", 0), data.get("y", 0), data.get("z", 0))
 
   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/serializer.py

@@ -6,7 +6,7 @@
 import math
 import sys
 import types
-from typing import Any, Dict, List, Union, cast
+from typing import Any, Dict, List, Optional, Union, cast
 
 if sys.version_info >= (3, 10):
   from typing import TypeAlias
@@ -145,8 +145,161 @@ def deserialize(data: JSON, allow_marshal: bool = False) -> Any:
       params = {k: deserialize(v, allow_marshal=allow_marshal) for k, v in data.items()}
       if "deserialize" in klass.__dict__:
         return klass.deserialize(params)
+      params = _fill_defaults(klass, params)
       return klass(**params)
     return {k: deserialize(v, allow_marshal=allow_marshal) for k, v in data.items()}
   if isinstance(data, object):
     return data
   raise TypeError(f"Cannot deserialize {data} of type {type(data)}")
+
+
+def apply_merge_patch(target: JSON, patch: JSON) -> JSON:
+  """Apply a JSON Merge Patch (RFC 7386) to a target document.
+
+  Rules:
+  - If patch is not a dict, it replaces the target entirely.
+  - If a patch value is None (JSON null), the key is removed from target.
+  - If a patch value is a dict, recurse.
+  - Otherwise, replace the key.
+  """
+  if not isinstance(patch, dict):
+    return patch
+
+  if not isinstance(target, dict):
+    target = {}
+
+  result = dict(target)
+  for key, value in patch.items():
+    if value is None:
+      result.pop(key, None)
+    else:
+      result[key] = apply_merge_patch(result.get(key), value)
+  return result
+
+
+def create_merge_patch(source: JSON, target: JSON) -> Optional[JSON]:
+  """Create a JSON Merge Patch (RFC 7386) that transforms source into target.
+
+  Returns None if source and target are equal (no patch needed).
+  """
+  if isinstance(source, dict) and isinstance(target, dict):
+    patch: Dict[str, JSON] = {}
+    all_keys = set(source.keys()) | set(target.keys())
+    for key in all_keys:
+      if key not in target:
+        patch[key] = None  # remove
+      elif key not in source:
+        patch[key] = target[key]  # add
+      else:
+        sub_patch = create_merge_patch(source[key], target[key])
+        if sub_patch is not None:
+          patch[key] = sub_patch
+    return patch if patch else None
+
+  if source == target:
+    # Handle NaN: NaN != NaN, but we treat string "nan" as equal
+    if isinstance(source, float) and isinstance(target, float):
+      if math.isnan(source) and math.isnan(target):
+        return None
+    else:
+      return None
+  return target
+
+
+def compact(data: JSON) -> JSON:
+  """Strip fields from a serialized dict tree that match constructor defaults.
+
+  For every dict with a ``"type"`` field, looks up the class via
+  :func:`get_plr_class_from_string`, inspects ``__init__`` defaults, serializes
+  each default, and omits fields where the serialized value matches.
+
+  Meta keys (``type``, ``location``, ``parent_name``) are always kept.
+  ``children`` is kept only when non-empty.
+  """
+  if isinstance(data, list):
+    return [compact(item) for item in data]
+  if not isinstance(data, dict):
+    return data
+
+  if "type" not in data:
+    return {k: compact(v) for k, v in data.items()}
+
+  type_name = data["type"]
+
+  # Try to look up the class; if not found, just recurse children
+  try:
+    klass = get_plr_class_from_string(type_name)
+  except ValueError:
+    return {k: compact(v) for k, v in data.items()}
+
+  # Get constructor defaults
+  defaults = _get_init_defaults(klass)
+
+  meta_keys = {"type", "location", "parent_name", "name"}
+  result: Dict[str, JSON] = {}
+
+  for key, value in data.items():
+    # Always keep meta keys
+    if key in meta_keys:
+      result[key] = compact(value) if isinstance(value, (dict, list)) else value
+      continue
+
+    # children: keep only when non-empty
+    if key == "children":
+      if isinstance(value, list) and len(value) > 0:
+        result[key] = compact(value)
+      continue
+
+    # If we have a default for this key and it matches, skip it
+    if key in defaults:
+      default_serialized = serialize(defaults[key])
+      if _json_equal(value, default_serialized):
+        continue
+
+    # Recurse into nested dicts/lists
+    result[key] = compact(value)
+
+  return result
+
+
+def _get_init_defaults(klass: type) -> Dict[str, Any]:
+  """Get the default values for a class's __init__ parameters."""
+  try:
+    sig = inspect.signature(klass.__init__)
+  except (ValueError, TypeError):
+    return {}
+
+  defaults: Dict[str, Any] = {}
+  for name, param in sig.parameters.items():
+    if name == "self":
+      continue
+    if param.default is not inspect.Parameter.empty:
+      defaults[name] = param.default
+  return defaults
+
+
+def _fill_defaults(klass: type, params: Dict[str, Any]) -> Dict[str, Any]:
+  """Fill in missing params from constructor defaults."""
+  defaults = _get_init_defaults(klass)
+  for key, default_value in defaults.items():
+    if key not in params:
+      params[key] = default_value
+  return params
+
+
+def _json_equal(a: JSON, b: JSON) -> bool:
+  """Compare two JSON values for equality, handling NaN."""
+  if isinstance(a, float) and isinstance(b, float):
+    if math.isnan(a) and math.isnan(b):
+      return True
+  if type(a) != type(b):
+    return False
+  if isinstance(a, dict) and isinstance(b, dict):
+    if set(a.keys()) != set(b.keys()):
+      return False
+    return all(_json_equal(a[k], b[k]) for k in a)
+  if isinstance(a, list) and isinstance(b, list):
+    if len(a) != len(b):
+      return False
+    return all(_json_equal(x, y) for x, y in zip(a, b))
+  return a == b

pylabrobot/storage/incubator.py

@@ -194,7 +194,9 @@ def serialize(self):
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False):
+    data = data.copy()
     backend = IncubatorBackend.deserialize(data.pop("backend"))
+    rotation_data = data.get("rotation")
     return cls(
       backend=backend,
       name=data["name"],
@@ -203,7 +205,7 @@ def deserialize(cls, data: dict, allow_marshal: bool = False):
       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=Rotation.deserialize(rotation_data) if rotation_data else None,
+      category=data.get("category"),
+      model=data.get("model"),
     )