Skills › Scientific Agent Skills › omero-integration › export_image_metadata.py
export_image_metadata.py
576 lines · skills/omero-integration/scripts/export_image_metadata.py at 5f2bc03
1#!/usr/bin/env python3
2"""Export bounded annotations and ROI geometry for explicit image IDs."""
4from __future__ import annotations
6import argparse
7import json
8import sys
9from datetime import datetime, timezone
10from typing import Any
12from omero_common import (
13 ConfigError,
14 DependencyError,
15 OutputPathError,
16 bounded_int,
17 config_summary,
18 emit_json,
19 gateway_session,
20 json_safe,
21 load_connection_config,
22 scrubbed_error,
23 take_bounded,
24 unwrap_omero,
25)
28SHAPE_GEOMETRY = {
29 "EllipseI": ("X", "Y", "RadiusX", "RadiusY"),
30 "LabelI": ("X", "Y"),
31 "LineI": ("X1", "Y1", "X2", "Y2"),
32 "MaskI": ("X", "Y", "Width", "Height"),
33 "PointI": ("X", "Y"),
34 "PolygonI": ("Points",),
35 "PolylineI": ("Points",),
36 "RectangleI": ("X", "Y", "Width", "Height"),
37}
40def positive_argument(value: str) -> int:
41 try:
42 parsed = int(value, 10)
43 except ValueError as exc:
44 raise argparse.ArgumentTypeError("expected an integer") from exc
45 if parsed <= 0:
46 raise argparse.ArgumentTypeError("expected a positive integer")
47 return parsed
50def build_parser() -> argparse.ArgumentParser:
51 parser = argparse.ArgumentParser(
52 description=(
53 "Plan or execute a read-only JSON export of annotations and ROI "
54 "geometry for explicit image IDs. No pixels or attachment bytes "
55 "are retrieved."
56 )
57 )
58 parser.add_argument(
59 "--image-id",
60 action="append",
61 required=True,
62 type=positive_argument,
63 help="Explicit image ID; repeat for additional images.",
64 )
65 parser.add_argument(
66 "--group-id",
67 type=positive_argument,
68 help="Optional explicit group ID; cross-group -1 is not supported.",
69 )
70 parser.add_argument(
71 "--max-images",
72 type=positive_argument,
73 default=25,
74 help="Maximum explicit images, 1..100 (default: 25).",
75 )
76 parser.add_argument(
77 "--max-annotations-per-image",
78 type=positive_argument,
79 default=100,
80 help="Per-image annotation cap, 1..1000 (default: 100).",
81 )
82 parser.add_argument(
83 "--max-rois-per-image",
84 type=positive_argument,
85 default=100,
86 help="Per-image ROI serialization cap, 1..1000 (default: 100).",
87 )
88 parser.add_argument(
89 "--max-shapes-per-roi",
90 type=positive_argument,
91 default=500,
92 help="Per-ROI shape cap, 1..5000 (default: 500).",
93 )
94 parser.add_argument(
95 "--max-string-length",
96 type=positive_argument,
97 default=512,
98 help="Maximum exported string length, 32..4096.",
99 )
100 parser.add_argument(
101 "--max-value-items",
102 type=positive_argument,
103 default=100,
104 help="Maximum items in an included annotation value, 1..1000.",
105 )
106 parser.add_argument(
107 "--include-annotation-values",
108 action="store_true",
109 help="Include bounded annotation values; redacted by default.",
110 )
111 parser.add_argument(
112 "--include-owner-names",
113 action="store_true",
114 help="Include annotation owner usernames; IDs only by default.",
115 )
116 parser.add_argument(
117 "--include-file-names",
118 action="store_true",
119 help="Include FileAnnotation filenames; redacted by default.",
120 )
121 parser.add_argument(
122 "--include-roi-labels",
123 action="store_true",
124 help="Include shape text labels; redacted by default.",
125 )
126 parser.add_argument(
127 "--output",
128 required=True,
129 help="Required .json destination in an existing directory.",
130 )
131 parser.add_argument(
132 "--overwrite",
133 action="store_true",
134 help="Replace an existing regular output file, never a symlink.",
135 )
136 parser.add_argument(
137 "--execute",
138 action="store_true",
139 help="Connect and perform the read-only export.",
140 )
141 parser.add_argument(
142 "--allow-insecure-transport",
143 action="store_true",
144 help=(
145 "Permit OMERO_SECURE=false after explicit policy review. "
146 "Encrypted transport remains the default."
147 ),
148 )
149 return parser
152def call_or_none(obj: Any, method: str) -> Any:
153 function = getattr(obj, method, None)
154 if not callable(function):
155 return None
156 return function()
159def normalized(
160 value: Any,
161 *,
162 max_string_length: int,
163 max_value_items: int,
164) -> Any:
165 return json_safe(
166 unwrap_omero(value),
167 max_string_length=max_string_length,
168 max_collection_length=max_value_items,
169 )
172def owner_record(
173 obj: Any,
174 *,
175 include_owner_names: bool,
176 max_string_length: int,
177) -> dict[str, Any]:
178 details = call_or_none(obj, "getDetails")
179 owner = call_or_none(details, "getOwner") if details is not None else None
180 record = {
181 "owner_id": normalized(
182 call_or_none(owner, "getId"),
183 max_string_length=max_string_length,
184 max_value_items=1,
185 )
186 }
187 if include_owner_names:
188 record["owner_username"] = normalized(
189 call_or_none(owner, "getOmeName"),
190 max_string_length=max_string_length,
191 max_value_items=1,
192 )
193 else:
194 record["owner_name_redacted"] = True
195 return record
198def annotation_record(
199 annotation: Any,
200 *,
201 include_values: bool,
202 include_owner_names: bool,
203 include_file_names: bool,
204 max_string_length: int,
205 max_value_items: int,
206) -> dict[str, Any]:
207 # AnnotationWrapper subclasses inherit OMERO_CLASS=None in OMERO.py.
208 kind = getattr(annotation, "OMERO_CLASS", None) or (
209 type(annotation).__name__.removesuffix("Wrapper")
210 )
211 record: dict[str, Any] = {
212 "id": call_or_none(annotation, "getId"),
213 "type": kind,
214 "namespace": normalized(
215 call_or_none(annotation, "getNs"),
216 max_string_length=max_string_length,
217 max_value_items=1,
218 ),
219 **owner_record(
220 annotation,
221 include_owner_names=include_owner_names,
222 max_string_length=max_string_length,
223 ),
224 }
226 if include_values:
227 record["value"] = normalized(
228 call_or_none(annotation, "getValue"),
229 max_string_length=max_string_length,
230 max_value_items=max_value_items,
231 )
232 else:
233 record["value_redacted"] = True
235 if kind == "FileAnnotation" or type(annotation).__name__ == (
236 "FileAnnotationWrapper"
237 ):
238 original = call_or_none(annotation, "getFile")
239 file_record = {
240 "original_file_id": call_or_none(original, "getId"),
241 "size_bytes": call_or_none(original, "getSize"),
242 "mimetype": normalized(
243 call_or_none(original, "getMimetype"),
244 max_string_length=128,
245 max_value_items=1,
246 ),
247 "bytes_downloaded": False,
248 }
249 if include_file_names:
250 file_record["name"] = normalized(
251 call_or_none(original, "getName"),
252 max_string_length=max_string_length,
253 max_value_items=1,
254 )
255 else:
256 file_record["name_redacted"] = True
257 record["file"] = file_record
258 return record
261def shape_record(
262 shape: Any,
263 *,
264 include_labels: bool,
265 max_string_length: int,
266) -> dict[str, Any]:
267 model_name = type(shape).__name__
268 shape_type = model_name.removesuffix("I")
269 record: dict[str, Any] = {
270 "id": normalized(
271 call_or_none(shape, "getId"),
272 max_string_length=max_string_length,
273 max_value_items=1,
274 ),
275 "type": shape_type,
276 "the_z": normalized(
277 call_or_none(shape, "getTheZ"),
278 max_string_length=max_string_length,
279 max_value_items=1,
280 ),
281 "the_t": normalized(
282 call_or_none(shape, "getTheT"),
283 max_string_length=max_string_length,
284 max_value_items=1,
285 ),
286 "the_c": normalized(
287 call_or_none(shape, "getTheC"),
288 max_string_length=max_string_length,
289 max_value_items=1,
290 ),
291 }
292 if include_labels:
293 record["label"] = normalized(
294 call_or_none(shape, "getTextValue"),
295 max_string_length=max_string_length,
296 max_value_items=1,
297 )
298 else:
299 record["label_redacted"] = True
301 geometry: dict[str, Any] = {}
302 for field in SHAPE_GEOMETRY.get(model_name, ()):
303 geometry[field[0].lower() + field[1:]] = normalized(
304 call_or_none(shape, f"get{field}"),
305 max_string_length=max_string_length,
306 max_value_items=1,
307 )
308 record["geometry"] = geometry
309 # Shape coordinates are in the shape's local coordinate system. Omitting
310 # its affine transform changes the ROI when reconstructing it on an image.
311 transform = call_or_none(shape, "getTransform")
312 record["transform"] = None if transform is None else {
313 field.lower(): normalized(
314 call_or_none(transform, f"get{field}"),
315 max_string_length=max_string_length,
316 max_value_items=1,
317 )
318 for field in ("A00", "A01", "A02", "A10", "A11", "A12")
319 }
320 points = unwrap_omero(call_or_none(shape, "getPoints"))
321 record["geometry_truncated"] = (
322 isinstance(points, str) and len(points) > max_string_length
323 )
324 if model_name == "MaskI":
325 record["mask_bytes_omitted"] = True
326 return record
329def roi_record(
330 roi: Any,
331 *,
332 max_shapes: int,
333 include_labels: bool,
334 max_string_length: int,
335) -> dict[str, Any]:
336 shapes = take_bounded(roi.copyShapes(), max_shapes)
337 return {
338 "id": normalized(
339 call_or_none(roi, "getId"),
340 max_string_length=max_string_length,
341 max_value_items=1,
342 ),
343 "shapes": [
344 shape_record(
345 shape,
346 include_labels=include_labels,
347 max_string_length=max_string_length,
348 )
349 for shape in shapes.items
350 ],
351 "returned_shapes": len(shapes.items),
352 "shapes_truncated": shapes.truncated,
353 }
356def export_one_image(
357 connection: Any,
358 roi_service: Any,
359 image_id: int,
360 args: argparse.Namespace,
361) -> dict[str, Any]:
362 image = connection.getObject("Image", image_id)
363 if image is None:
364 return {"id": image_id, "accessible": False}
366 annotation_result = take_bounded(
367 image.listAnnotations(),
368 args.max_annotations_per_image,
369 )
370 roi_result = roi_service.findByImage(
371 image_id, None, connection.SERVICE_OPTS,
372 )
373 roi_items = take_bounded(
374 roi_result.rois,
375 args.max_rois_per_image,
376 )
378 details = call_or_none(image, "getDetails")
379 group = call_or_none(details, "getGroup") if details is not None else None
380 owner = call_or_none(details, "getOwner") if details is not None else None
381 return {
382 "id": image_id,
383 "accessible": True,
384 "name_redacted": True,
385 "group_id": call_or_none(group, "getId"),
386 "owner_id": call_or_none(owner, "getId"),
387 "dimensions": {
388 "x": call_or_none(image, "getSizeX"),
389 "y": call_or_none(image, "getSizeY"),
390 "z": call_or_none(image, "getSizeZ"),
391 "c": call_or_none(image, "getSizeC"),
392 "t": call_or_none(image, "getSizeT"),
393 },
394 "annotations": [
395 annotation_record(
396 annotation,
397 include_values=args.include_annotation_values,
398 include_owner_names=args.include_owner_names,
399 include_file_names=args.include_file_names,
400 max_string_length=args.max_string_length,
401 max_value_items=args.max_value_items,
402 )
403 for annotation in annotation_result.items
404 ],
405 "returned_annotations": len(annotation_result.items),
406 "annotations_truncated": annotation_result.truncated,
407 "rois": [
408 roi_record(
409 roi,
410 max_shapes=args.max_shapes_per_roi,
411 include_labels=args.include_roi_labels,
412 max_string_length=args.max_string_length,
413 )
414 for roi in roi_items.items
415 ],
416 "returned_rois": len(roi_items.items),
417 "rois_truncated": roi_items.truncated,
418 }
421def scope_payload(
422 args: argparse.Namespace,
423 image_ids: list[int],
424) -> dict[str, Any]:
425 return {
426 "image_ids": image_ids,
427 "group_id": args.group_id,
428 "cross_group": False,
429 "max_images": args.max_images,
430 "max_annotations_per_image": args.max_annotations_per_image,
431 "max_rois_per_image": args.max_rois_per_image,
432 "max_shapes_per_roi": args.max_shapes_per_roi,
433 "max_string_length": args.max_string_length,
434 "max_value_items": args.max_value_items,
435 }
438def redaction_payload(args: argparse.Namespace) -> dict[str, bool]:
439 return {
440 "annotation_values_included": args.include_annotation_values,
441 "owner_names_included": args.include_owner_names,
442 "file_names_included": args.include_file_names,
443 "roi_labels_included": args.include_roi_labels,
444 "file_bytes_included": False,
445 "pixel_data_included": False,
446 "mask_bytes_included": False,
447 }
450def main(argv: list[str] | None = None) -> int:
451 parser = build_parser()
452 args = parser.parse_args(argv)
453 try:
454 bounded_int(
455 args.max_images,
456 name="max-images",
457 minimum=1,
458 maximum=100,
459 )
460 bounded_int(
461 args.max_annotations_per_image,
462 name="max-annotations-per-image",
463 minimum=1,
464 maximum=1000,
465 )
466 bounded_int(
467 args.max_rois_per_image,
468 name="max-rois-per-image",
469 minimum=1,
470 maximum=1000,
471 )
472 bounded_int(
473 args.max_shapes_per_roi,
474 name="max-shapes-per-roi",
475 minimum=1,
476 maximum=5000,
477 )
478 bounded_int(
479 args.max_string_length,
480 name="max-string-length",
481 minimum=32,
482 maximum=4096,
483 )
484 bounded_int(
485 args.max_value_items,
486 name="max-value-items",
487 minimum=1,
488 maximum=1000,
489 )
491 image_ids = list(dict.fromkeys(args.image_id))
492 if len(image_ids) != len(args.image_id):
493 raise ValueError("duplicate --image-id values are not allowed")
494 if len(image_ids) > args.max_images:
495 raise ValueError("explicit image count exceeds --max-images")
497 config = load_connection_config(require_auth=args.execute)
498 if not args.execute:
499 print(
500 json.dumps(
501 {
502 "mode": "dry-run",
503 "server_contacted": False,
504 "output_not_written": True,
505 "requested_output": args.output,
506 "configuration": config_summary(config),
507 "scope": scope_payload(args, image_ids),
508 "redaction": redaction_payload(args),
509 "next_step": (
510 "Review scope/redaction, then add --execute."
511 ),
512 },
513 indent=2,
514 sort_keys=True,
515 )
516 )
517 return 0
519 with gateway_session(
520 config,
521 allow_insecure_transport=args.allow_insecure_transport,
522 ) as connection:
523 if args.group_id is not None:
524 connection.SERVICE_OPTS.setOmeroGroup(str(args.group_id))
525 roi_service = connection.getRoiService()
526 images = [
527 export_one_image(
528 connection,
529 roi_service,
530 image_id,
531 args,
532 )
533 for image_id in image_ids
534 ]
536 payload = {
537 "mode": "executed-read-only",
538 "generated_at": datetime.now(timezone.utc).isoformat(),
539 "configuration": config_summary(config),
540 "scope": scope_payload(args, image_ids),
541 "redaction": redaction_payload(args),
542 "api_notes": [
543 (
544 "IRoi.findByImage is used by current official Python "
545 "examples but IRoi is deprecated."
546 ),
547 (
548 "ROI limits bound serialization; findByImage itself "
549 "does not expose pagination."
550 ),
551 ],
552 "images": images,
553 }
554 emit_json(
555 payload,
556 output=args.output,
557 overwrite=args.overwrite,
558 )
559 return 0
560 except (
561 ConfigError,
562 DependencyError,
563 OutputPathError,
564 FileExistsError,
565 ValueError,
566 RuntimeError,
567 ) as error:
568 print(scrubbed_error(error), file=sys.stderr)
569 return 2
570 except Exception as error:
571 print(scrubbed_error(error), file=sys.stderr)
572 return 1
575if __name__ == "__main__":
576 raise SystemExit(main())