Add extinfo command

Author: dominikl

src/omero_zarr/cli.py

@@ -41,6 +41,11 @@
     image_to_zarr,
     plate_to_zarr,
 )
+from .extinfo import (
+    get_images,
+    set_external_info,
+    get_path,
+)
 
 HELP = """Export data in zarr format.
 
@@ -108,6 +113,7 @@
 
 POLYGONS_HELP = """Export ROI Polygons on the Image or Plate in zarr format"""
 
+EXTINFO_HELP = """Set the external info path for an ome.zarr image."""
 
 def gateway_required(func: Callable) -> Callable:
     """
@@ -284,6 +290,11 @@ def _configure(self, parser: Parser) -> None:
             help="The Image to export.",
         )
 
+        exinfo = parser.add(sub, self.extinfo, EXTINFO_HELP)
+        exinfo.add_argument("object",
+            type=ProxyStringType(),
+            help="Object in Class:ID format")
+
         for subcommand in (polygons, masks, export):
             subcommand.add_argument(
                 "--output", type=str, default="", help="The output directory"
@@ -335,6 +346,21 @@ def export(self, args: argparse.Namespace) -> None:
             plate = self._lookup(self.gateway, "Plate", args.object.id)
             plate_to_zarr(plate, args)
 
+    @gateway_required
+    def extinfo(self, args: argparse.Namespace) -> None:
+        for img in get_images(self.gateway, args.object):
+            path = get_path(self.gateway, img.getId())
+            img = img._obj
+            if path.endswith("OME/METADATA.ome.xml"):
+                path = path.replace("OME/METADATA.ome.xml", "0")
+                path = f"/{path}"
+                img = set_external_info(img, path)
+                img = self.gateway.getUpdateService().saveAndReturnObject(img)
+                self.ctx.out(f"Set path to '{path}' for image {img.id._val}")
+            else:
+                self.ctx.out(f"'{path}' for image {img.id._val} doesn't seem to be an ome.zarr, skipping.")
+
+
     def _lookup(
         self, gateway: BlitzGateway, otype: str, oid: int
     ) -> BlitzObjectWrapper:

src/omero_zarr/extinfo.py

@@ -0,0 +1,71 @@
+from omero.gateway import BlitzGateway
+from omero.gateway import BlitzObjectWrapper
+from omero.model import ExternalInfoI, ImageI
+from omero.rtypes import rstring, rlong
+from omero_sys_ParametersI import ParametersI
+from omero.model import Image
+from omero.model import Plate
+from omero.model import Screen
+from omero.model import Dataset
+from omero.model import Project
+
+
+def get_path(conn: BlitzGateway, image_id: int) -> str:
+    params = ParametersI()
+    params.addId(image_id)
+    query = """
+        select fs from Fileset as fs
+        join fetch fs.images as image
+        left outer join fetch fs.usedFiles as usedFile
+        join fetch usedFile.originalFile as f
+        join fetch f.hasher where image.id = :id
+    """
+    fs = conn.getQueryService().findByQuery(query, params)
+    res = fs._getUsedFiles()[0]._clientPath._val
+    return res
+
+
+def set_external_info(img: ImageI, path: str) -> ImageI:
+    info = ExternalInfoI()
+    info.entityType = rstring("com.glencoesoftware.ngff:multiscales")
+    info.entityId = rlong(3)
+    info.lsid = rstring(path)
+    img.details.externalInfo = info
+    return img
+
+
+def _lookup(conn: BlitzGateway, type: str, oid: int) -> BlitzObjectWrapper:
+    conn.SERVICE_OPTS.setOmeroGroup("-1")
+    obj = conn.getObject(type, oid)
+    if not obj:
+        raise ValueError(f"No such {type}: {oid}")
+    return obj
+
+
+def get_images(conn: BlitzGateway, object):
+    if isinstance(object, list):
+        for x in object:
+            yield from get_images(conn, x)
+    elif isinstance(object, Screen):
+        scr = _lookup(conn, "Screen", object.id)
+        for plate in scr.listChildren():
+            yield from get_images(conn, plate._obj)
+    elif isinstance(object, Plate):
+        plt = _lookup(conn, "Plate", object.id)
+        for well in plt.listChildren():
+            for idx in range(0, well.countWellSample()):
+                img = well.getImage(idx)
+                yield img
+    elif isinstance(object, Project):
+        prj = _lookup(conn, "Project", object.id)
+        for ds in prj.listChildren():
+            yield from get_images(conn, ds._obj)
+    elif isinstance(object, Dataset):
+        ds = _lookup(conn, "Dataset", object.id)
+        for img in ds.listChildren():
+            yield img
+    elif isinstance(object, Image):
+        img = _lookup(conn, "Image", object.id)
+        yield img
+    else:
+        raise ValueError(f"Unsupported type: {object.__class__.__name__}")