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
 
@@ -44,19 +44,19 @@ def detect_format(metadata: dict) -> "Format":
 class Format(ABC):
     @property
     @abstractmethod
-    def version(self) -> str:
+    def version(self) -> str:  # pragma: no cover
         raise NotImplementedError()
 
     @abstractmethod
-    def matches(self, metadata: dict) -> bool:
+    def matches(self, metadata: dict) -> bool:  # pragma: no cover
         raise NotImplementedError()
 
     @abstractmethod
     def init_store(self, path: str, mode: str = "r") -> FSStore:
         raise NotImplementedError()
 
     # @abstractmethod
-    def init_channels(self) -> None:
+    def init_channels(self) -> None:  # pragma: no cover
         raise NotImplementedError()
 
     def _get_multiscale_version(self, metadata: dict) -> Optional[str]:
@@ -72,12 +72,26 @@ 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:  # pragma: no cover
+        raise NotImplementedError()
+
+    @abstractmethod
+    def validate_well_dict(
+        self, well: dict, rows: List[str], columns: List[str]
+    ) -> None:  # pragma: no cover
+        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 +106,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, rows: List[str], columns: List[str]
+    ) -> 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")
 
-class FormatV02(Format):
+
+class FormatV02(FormatV01):
     """
     Changelog: move to nested storage (April 2021)
     """
@@ -151,9 +181,39 @@ 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}
+
+    def validate_well_dict(
+        self, well: dict, rows: List[str], columns: List[str]
+    ) -> None:
+        super().validate_well_dict(well, rows, columns)
+        if len(well["path"].split("/")) != 2:
+            raise ValueError(f"{well} path must exactly be composed of 2 groups")
+        row, column = well["path"].split("/")
+        if row not in rows:
+            raise ValueError(f"{row} is not defined in the plate rows")
+        if well["rowIndex"] != rows.index(row):
+            raise ValueError(f"Mismatching row index for {well}")
+        if column not in columns:
+            raise ValueError(f"{column} is not defined in the plate columns")
+        if well["columnIndex"] != columns.index(column):
+            raise ValueError(f"Mismatching column index for {well}")
+
 
 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, rows, columns)
+            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, rows, columns)
             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: