Add support for HCS 0.4 specification

ome_zarr/format.py

@@ -2,7 +2,7 @@
 
 import logging
 from abc import ABC, abstractmethod
-from typing import Iterator, Optional
+from typing import Dict, Iterator, List, Optional
 
 from zarr.storage import FSStore
 
@@ -72,12 +72,24 @@ def __repr__(self) -> str:
     def __eq__(self, other: object) -> bool:
         return self.__class__ == other.__class__
 
+    @abstractmethod
+    def generate_well_dict(
+        self, well: str, rows: List[str], columns: List[str]
+    ) -> dict:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def validate_well_dict(self, well: dict) -> None:
+        raise NotImplementedError()
+
 
 class FormatV01(Format):
     """
     Initial format. (2020)
     """
 
+    REQUIRED_PLATE_WELL_KEYS: Dict[str, type] = {"path": str}
+
     @property
     def version(self) -> str:
         return "0.1"
@@ -92,8 +104,24 @@ def init_store(self, path: str, mode: str = "r") -> FSStore:
         LOGGER.debug(f"Created legacy flat FSStore({path}, {mode})")
         return store
 
+    def generate_well_dict(
+        self, well: str, rows: List[str], columns: List[str]
+    ) -> dict:
+        return {"path": str(well)}
+
+    def validate_well_dict(self, well: dict) -> None:
+        if any(e not in self.REQUIRED_PLATE_WELL_KEYS for e in well.keys()):
+            LOGGER.debug("f{well} contains unspecified keys")
+        for key, key_type in self.REQUIRED_PLATE_WELL_KEYS.items():
+            if key not in well:
+                raise ValueError(f"{well} must contain a {key} key of type {key_type}")
+            if not isinstance(well[key], key_type):
+                raise ValueError(f"{well} path must be of {key_type} type")
+        if len(well["path"].split("/")) != 2:
+            raise ValueError(f"{well} path must exactly be composed of 2 groups")
+
 
-class FormatV02(Format):
+class FormatV02(FormatV01):
     """
     Changelog: move to nested storage (April 2021)
     """
@@ -151,9 +179,23 @@ class FormatV04(FormatV03):
     introduce transformations in multiscales (Nov 2021)
     """
 
+    REQUIRED_PLATE_WELL_KEYS = {"path": str, "rowIndex": int, "columnIndex": int}
+
     @property
     def version(self) -> str:
         return "0.4"
 
+    def generate_well_dict(
+        self, well: str, rows: List[str], columns: List[str]
+    ) -> dict:
+        row, column = well.split("/")
+        if row not in rows:
+            raise ValueError(f"{row} is not defined in the list of rows")
+        rowIndex = rows.index(row)
+        if column not in columns:
+            raise ValueError(f"{column} is not defined in the list of columns")
+        columnIndex = columns.index(column)
+        return {"path": str(well), "rowIndex": rowIndex, "columnIndex": columnIndex}
+
 
 CurrentFormat = FormatV04

ome_zarr/writer.py

@@ -107,26 +107,38 @@ def _validate_plate_acquisitions(
     return acquisitions
 
 
+def _validate_plate_rows_columns(
+    rows_or_columns: List[str],
+    fmt: Format = CurrentFormat(),
+) -> List[dict]:
+
+    if len(set(rows_or_columns)) != len(rows_or_columns):
+        raise ValueError(f"{rows_or_columns} must contain unique elements")
+    validated_list = []
+    for element in rows_or_columns:
+        if not element.isalnum():
+            raise ValueError(f"{element} must contain alphanumeric characters")
+        validated_list.append({"name": str(element)})
+    return validated_list
+
+
 def _validate_plate_wells(
-    wells: List[Union[str, dict]], fmt: Format = CurrentFormat()
+    wells: List[Union[str, dict]],
+    rows: List[str],
+    columns: List[str],
+    fmt: Format = CurrentFormat(),
 ) -> List[dict]:
 
-    VALID_KEYS = [
-        "path",
-    ]
     validated_wells = []
     if wells is None or len(wells) == 0:
         raise ValueError("Empty wells list")
     for well in wells:
         if isinstance(well, str):
-            validated_wells.append({"path": str(well)})
+            well_dict = fmt.generate_well_dict(well, rows, columns)
+            fmt.validate_well_dict(well_dict)
+            validated_wells.append(well_dict)
         elif isinstance(well, dict):
-            if any(e not in VALID_KEYS for e in well.keys()):
-                LOGGER.debug("f{well} contains unspecified keys")
-            if "path" not in well:
-                raise ValueError(f"{well} must contain a path key")
-            if not isinstance(well["path"], str):
-                raise ValueError(f"{well} path must be of str type")
+            fmt.validate_well_dict(well)
             validated_wells.append(well)
         else:
             raise ValueError(f"Unrecognized type for {well}")
@@ -259,9 +271,9 @@ def write_plate_metadata(
     """
 
     plate: Dict[str, Union[str, int, List[Dict]]] = {
-        "columns": [{"name": str(c)} for c in columns],
-        "rows": [{"name": str(r)} for r in rows],
-        "wells": _validate_plate_wells(wells),
+        "columns": _validate_plate_rows_columns(columns),
+        "rows": _validate_plate_rows_columns(rows),
+        "wells": _validate_plate_wells(wells, rows, columns, fmt=fmt),
         "version": fmt.version,
     }
     if name is not None:

tests/test_node.py

@@ -1,3 +1,5 @@
+import uuid
+
 import pytest
 import zarr
 from numpy import zeros
@@ -141,3 +143,31 @@ def test_plate_2D5D(self, axes, dims):
         assert node.metadata
         assert len(node.specs) == 1
         assert isinstance(node.specs[0], Well)
+
+    def test_uuid_plate(self):
+        group1 = uuid.uuid4()
+        group2 = uuid.uuid4()
+        well = {"path": f"{group1}/{group2}", "rowIndex": 0, "columnIndex": 0}
+        write_plate_metadata(self.root, ["A"], ["1"], [well])
+        row_group = self.root.require_group(group1)
+        well = row_group.require_group(group2)
+        write_well_metadata(well, ["0"])
+        image = well.require_group("0")
+        write_image(zeros((1, 1, 1, 256, 256)), image)
+
+        node = Node(parse_url(str(self.path)), list())
+        assert node.data
+        assert node.metadata
+        assert len(node.specs) == 1
+        assert isinstance(node.specs[0], Plate)
+        assert node.specs[0].row_names == ["A"]
+        assert node.specs[0].col_names == ["1"]
+        assert node.specs[0].well_paths == [f"{group1}/{group2}"]
+        assert node.specs[0].row_count == 1
+        assert node.specs[0].column_count == 1
+
+        node = Node(parse_url(str(self.path / f"{group1}/{group2}")), list())
+        assert node.data
+        assert node.metadata
+        assert len(node.specs) == 1
+        assert isinstance(node.specs[0], Well)