rename-wells CLI utility

docs/examples/well-mapping.csv

@@ -0,0 +1,4 @@
+0/0,A/1
+0/1,A/2
+0/2,B/1
+0/3,B/2

iohub/cli/cli.py

@@ -7,6 +7,7 @@
 from iohub.cli.parsing import input_position_dirpaths
 from iohub.convert import TIFFConverter
 from iohub.reader import print_info
+from iohub.rename_wells import rename_wells
 
 VERSION = __version__
 
@@ -152,3 +153,35 @@ def set_scale(
                     f"{old_value} to {value}."
                 )
                 dataset.set_scale("0", name, value)
+
+
+@cli.command(name="rename-wells")
+@click.help_option("-h", "--help")
+@click.option(
+    "-i",
+    "--input",
+    "zarrfile",
+    type=click.Path(exists=True, file_okay=True, dir_okay=True),
+    required=True,
+    help="Path to the input Zarr file.",
+)
+@click.option(
+    "-c",
+    "--csv",
+    "csvfile",
+    type=click.Path(exists=True, file_okay=True, dir_okay=False),
+    required=True,
+    help="Path to the CSV file containing old and new well names.",
+)
+def rename_wells_command(zarrfile, csvfile):
+    """Rename wells in an plate.
+
+    >> iohub rename-wells -i plate.zarr -c names.csv
+
+    The CSV file must have two columns with old and new names in the form:
+    ```
+    A/1,B/2
+    A/2,B/2
+    ```
+    """
+    rename_wells(zarrfile, csvfile)

iohub/ngff/nodes.py

@@ -1653,6 +1653,61 @@ def positions(self) -> Generator[tuple[str, Position], None, None]:
             for _, position in well.positions():
                 yield position.zgroup.path, position
 
+    def rename_well(
+        self,
+        old: str,
+        new: str,
+    ):
+        """Rename a well.
+
+        Parameters
+        ----------
+        old : str
+            Old name of well, e.g. "A/1"
+        new : str
+            New name of well, e.g. "B/2"
+        """
+
+        # normalize inputs
+        old = normalize_storage_path(old)
+        new = normalize_storage_path(new)
+        old_row, old_column = old.split("/")
+        new_row, new_column = new.split("/")
+        new_row_meta = PlateAxisMeta(name=new_row)
+        new_col_meta = PlateAxisMeta(name=new_column)
+
+        # raises ValueError if old well does not exist
+        # or if new well already exists
+        self.zgroup.move(old, new)
+
+        # update well metadata
+        old_well_index = [
+            well_name.path for well_name in self.metadata.wells
+        ].index(old)
+        self.metadata.wells[old_well_index].path = new
+        new_well_names = [well.path for well in self.metadata.wells]
+
+        # update row/col metadata
+        # check for new row/col
+        if new_row not in [row.name for row in self.metadata.rows]:
+            self.metadata.rows.append(new_row_meta)
+        if new_column not in [col.name for col in self.metadata.columns]:
+            self.metadata.columns.append(new_col_meta)
+
+        # check for empty row/col
+        if old_row not in [well.split("/")[0] for well in new_well_names]:
+            # delete empty row from zarr
+            del self.zgroup[old_row]
+            self.metadata.rows = [
+                row for row in self.metadata.rows if row.name != old_row
+            ]
+        if old_column not in [well.split("/")[1] for well in new_well_names]:
+            self.metadata.columns = [
+                col for col in self.metadata.columns if col.name != old_column
+            ]
+
+        self.dump_meta()
+
 
 def open_ome_zarr(
     store_path: StrOrBytesPath,

iohub/rename_wells.py

@@ -0,0 +1,57 @@
+import csv
+from pathlib import Path
+
+from iohub.ngff import open_ome_zarr
+
+
+def rename_wells(zarr_store_path: str | Path, csv_file_path: str | Path):
+    """
+    Rename wells in a Zarr store based on a CSV file containing old and new
+    well names.
+
+    Parameters
+    ----------
+    zarr_store_path : str or Path
+        Path to the Zarr store.
+    csv_file_path : str or Path
+        Path to the CSV file containing the old and new well names.
+
+    Raises
+    ------
+    ValueError
+        If a row in the CSV file does not have exactly two columns.
+        If there is an error renaming a well in the Zarr store.
+
+    Notes
+    -----
+    The CSV file should have two columns:
+        - The first column contains the old well names.
+        - The second column contains the new well names.
+
+    Examples
+    --------
+    CSV file content:
+    A/1,B/2
+    A/2,B/2
+
+    """
+
+    # read and check csv
+    name_pair_list = []
+    with open(csv_file_path, "r") as csv_file:
+        for row in csv.reader(csv_file):
+            if len(row) != 2:
+                raise ValueError(
+                    f"Invalid row format: {row}."
+                    f"Each row must have two columns."
+                )
+            name_pair_list.append([row[0].strip(), row[1].strip()])
+
+    # rename each well while catching errors
+    with open_ome_zarr(zarr_store_path, mode="a") as plate:
+        for old, new in name_pair_list:
+            print(f"Renaming {old} to {new}")
+            try:
+                plate.rename_well(old, new)
+            except ValueError as e:
+                print(f"Error renaming {old} to {new}: {e}")

tests/cli/test_cli.py

@@ -171,3 +171,29 @@ def test_cli_set_scale():
         with open_ome_zarr(position_path, layout="fov") as output_dataset:
             assert output_dataset.scale[-1] == 0.1
             assert output_dataset.zattrs["iohub"]["prior_x_scale"] == 0.5
+
+
+def test_cli_rename_wells_help():
+    runner = CliRunner()
+    cmd = ["rename-wells"]
+    for option in ("-h", "--help"):
+        cmd.append(option)
+        result = runner.invoke(cli, cmd)
+        assert result.exit_code == 0
+        assert ">> iohub rename-wells" in result.output
+
+
+def test_cli_rename_wells(csv_data_file_1):
+    with _temp_copy(hcs_ref) as store_path:
+        runner = CliRunner()
+        cmd = [
+            "rename-wells",
+            "-i",
+            str(store_path),
+            "-c",
+            str(csv_data_file_1),
+        ]
+        result = runner.invoke(cli, cmd)
+
+        assert result.exit_code == 0
+        assert "Renaming" in result.output

tests/conftest.py

@@ -1,7 +1,9 @@
+import csv
 import shutil
 from pathlib import Path
 
 import fsspec
+import pytest
 from wget import download
 
 
@@ -100,3 +102,27 @@ def subdirs(parent: Path, name: str) -> list[Path]:
 
 
 ndtiff_v3_labeled_positions = test_datasets / "ndtiff_v3_labeled_positions"
+
+
+@pytest.fixture
+def csv_data_file_1(tmpdir):
+    test_csv_1 = tmpdir / "well_names_1.csv"
+    csv_data_1 = [
+        ["B/03", "D/4"],
+    ]
+    with open(test_csv_1, mode="w", newline="") as csvfile:
+        writer = csv.writer(csvfile)
+        writer.writerows(csv_data_1)
+    return test_csv_1
+
+
+@pytest.fixture
+def csv_data_file_2(tmpdir):
+    test_csv_2 = tmpdir / "well_names_2.csv"
+    csv_data_2 = [
+        ["D/4", "B/03"],
+    ]
+    with open(test_csv_2, mode="w", newline="") as csvfile:
+        writer = csv.writer(csvfile)
+        writer.writerows(csv_data_2)
+    return test_csv_2

tests/ngff/test_ngff.py

@@ -183,6 +183,46 @@ def _temp_ome_zarr(
         temp_dir.cleanup()
 
 
+@contextmanager
+def _temp_ome_zarr_plate(
+    image_5d: NDArray,
+    channel_names: list[str],
+    arr_name: str,
+    position_list: list[tuple[str, str, str]],
+    **kwargs,
+):
+    """Helper function to generate a temporary OME-Zarr store.
+
+    Parameters
+    ----------
+    image_5d : NDArray
+    channel_names : list[str]
+    arr_name : str
+    position_list : list[tuple[str, str, str]]
+
+    Yields
+    ------
+    Position
+    """
+    try:
+        temp_dir = TemporaryDirectory()
+        dataset = open_ome_zarr(
+            os.path.join(temp_dir.name, "ome.zarr"),
+            layout="hcs",
+            mode="a",
+            channel_names=channel_names,
+        )
+        for position in position_list:
+            pos = dataset.create_position(
+                position[0], position[1], position[2]
+            )
+            pos.create_image(arr_name, image_5d, **kwargs)
+        yield dataset
+    finally:
+        dataset.close()
+        temp_dir.cleanup()
+
+
 @given(
     channels_and_random_5d=_channels_and_random_5d(),
     arr_name=short_alpha_numeric,
@@ -291,6 +331,60 @@ def test_rename_channel(channels_and_random_5d, arr_name, new_channel):
         assert dataset.metadata.omero.channels[0].label == new_channel
 
 
+@given(
+    channels_and_random_5d=_channels_and_random_5d(),
+    arr_name=short_alpha_numeric,
+)
+@settings(deadline=None)
+def test_rename_well(channels_and_random_5d, arr_name):
+    """Test `iohub.ngff.Position.rename_well()`"""
+    channel_names, random_5d = channels_and_random_5d
+
+    position_list = [["A", "1", "0"], ["C", "4", "0"]]
+    with _temp_ome_zarr_plate(
+        random_5d, channel_names, arr_name, position_list
+    ) as dataset:
+        assert dataset.zgroup["A/1"]
+        with pytest.raises(KeyError):
+            dataset.zgroup["B/2"]
+        assert "A" in [r[0] for r in dataset.rows()]
+        assert "B" not in [r[0] for r in dataset.rows()]
+        assert "A" in [row.name for row in dataset.metadata.rows]
+        assert "B" not in [row.name for row in dataset.metadata.rows]
+        assert "1" in [col.name for col in dataset.metadata.columns]
+        assert "2" not in [col.name for col in dataset.metadata.columns]
+        assert "C" in [row.name for row in dataset.metadata.rows]
+        assert "4" in [col.name for col in dataset.metadata.columns]
+
+        dataset.rename_well("A/1", "B/2")
+
+        assert dataset.zgroup["B/2"]
+        with pytest.raises(KeyError):
+            dataset.zgroup["A/1"]
+        assert "A" not in [r[0] for r in dataset.rows()]
+        assert "B" in [r[0] for r in dataset.rows()]
+        assert "A" not in [row.name for row in dataset.metadata.rows]
+        assert "B" in [row.name for row in dataset.metadata.rows]
+        assert "1" not in [col.name for col in dataset.metadata.columns]
+        assert "2" in [col.name for col in dataset.metadata.columns]
+        assert "C" in [row.name for row in dataset.metadata.rows]
+        assert "4" in [col.name for col in dataset.metadata.columns]
+
+        # destination exists
+        with pytest.raises(ValueError):
+            dataset.rename_well("B/2", "C/4")
+
+        # source doesn't exist
+        with pytest.raises(ValueError):
+            dataset.rename_well("Q/1", "Q/2")
+
+        # invalid well names
+        with pytest.raises(ValueError):
+            dataset.rename_well("B/2", " A/1")
+        with pytest.raises(ValueError):
+            dataset.rename_well("B/2", "A/?")
+
+
 @given(
     channels_and_random_5d=_channels_and_random_5d(),
     arr_name=short_alpha_numeric,
@@ -406,6 +500,7 @@ def test_set_transform_fov(ch_shape_dtype, arr_name):
 @given(
     ch_shape_dtype=_channels_and_random_5d_shape_and_dtype(),
 )
+@settings(deadline=None)
 def test_set_scale(ch_shape_dtype):
     channel_names, shape, dtype = ch_shape_dtype
     transform = [

tests/test_rename_wells.py

@@ -0,0 +1,32 @@
+from iohub import open_ome_zarr
+from iohub.rename_wells import rename_wells
+from tests.conftest import hcs_ref
+from tests.ngff.test_ngff import _temp_copy
+
+
+def test_cli_rename_wells(csv_data_file_1, csv_data_file_2):
+    with _temp_copy(hcs_ref) as store_path:
+        rename_wells(store_path, csv_data_file_1)
+        with open_ome_zarr(store_path, mode="r") as plate:
+            well_names = [well[0] for well in plate.wells()]
+            assert "D/4" in well_names
+            assert "B/03" not in well_names
+            assert len(plate.metadata.wells) == 1
+            assert len(plate.metadata.rows) == 1
+            assert len(plate.metadata.columns) == 1
+            assert plate.metadata.wells[0].path == "D/4"
+            assert plate.metadata.rows[0].name == "D"
+            assert plate.metadata.columns[0].name == "4"
+
+        # Test round trip
+        rename_wells(store_path, csv_data_file_2)
+        with open_ome_zarr(store_path, mode="r") as plate:
+            well_names = [well[0] for well in plate.wells()]
+            assert "D/4" not in well_names
+            assert "B/03" in well_names
+            assert len(plate.metadata.wells) == 1
+            assert len(plate.metadata.rows) == 1
+            assert len(plate.metadata.columns) == 1
+            assert plate.metadata.wells[0].path == "B/03"
+            assert plate.metadata.rows[0].name == "B"
+            assert plate.metadata.columns[0].name == "03"