Skills › Scientific Agent Skills › omero-integration › plan_transfer.py
plan_transfer.py
400 lines · skills/omero-integration/scripts/plan_transfer.py at 5f2bc03
1#!/usr/bin/env python3
2"""Build a local-only, bounded OMERO import or export plan."""
4from __future__ import annotations
6import argparse
7import os
8import re
9import sys
10from datetime import datetime, timezone
11from pathlib import Path
12from typing import Any
14from omero_common import (
15 OutputPathError,
16 bounded_int,
17 emit_json,
18 scrubbed_error,
19)
22TARGET_PATTERN = re.compile(r"^(Dataset|Screen):id:([1-9][0-9]*)$")
23IMAGE_PATTERN = re.compile(r"^Image:([1-9][0-9]*)$")
26def positive_argument(value: str) -> int:
27 try:
28 parsed = int(value, 10)
29 except ValueError as exc:
30 raise argparse.ArgumentTypeError("expected an integer") from exc
31 if parsed <= 0:
32 raise argparse.ArgumentTypeError("expected a positive integer")
33 return parsed
36def import_target(value: str) -> str:
37 if not TARGET_PATTERN.fullmatch(value):
38 raise argparse.ArgumentTypeError(
39 "target must be Dataset:id:<positive-id> or "
40 "Screen:id:<positive-id>"
41 )
42 return value
45def image_selector(value: str) -> str:
46 if not IMAGE_PATTERN.fullmatch(value):
47 raise argparse.ArgumentTypeError(
48 "selector must be Image:<positive-id>"
49 )
50 return value
53def build_parser() -> argparse.ArgumentParser:
54 parser = argparse.ArgumentParser(
55 description=(
56 "Create a bounded local OMERO import/export plan. This command "
57 "never imports omero, connects to a server, or executes a plan."
58 )
59 )
60 parser.add_argument(
61 "--output",
62 help="Optional .json plan in an existing directory.",
63 )
64 parser.add_argument(
65 "--overwrite",
66 action="store_true",
67 help="Replace an existing regular plan file, never a symlink.",
68 )
70 subparsers = parser.add_subparsers(dest="operation", required=True)
72 import_parser = subparsers.add_parser(
73 "import",
74 help="Scan explicit local paths and propose importer commands.",
75 )
76 import_parser.add_argument(
77 "paths",
78 nargs="+",
79 help="Explicit existing files/directories; symlinks are refused.",
80 )
81 import_parser.add_argument(
82 "--target",
83 type=import_target,
84 help="Optional numeric Dataset:id:<id> or Screen:id:<id> target.",
85 )
86 import_parser.add_argument(
87 "--max-paths",
88 type=positive_argument,
89 default=25,
90 help="Maximum explicit top-level paths, 1..100.",
91 )
92 import_parser.add_argument(
93 "--max-files",
94 type=positive_argument,
95 default=100,
96 help="Maximum discovered regular files, 1..10000.",
97 )
98 import_parser.add_argument(
99 "--scan-depth",
100 type=positive_argument,
101 default=4,
102 help="Maximum directory depth, 1..20 (default: 4).",
103 )
105 export_parser = subparsers.add_parser(
106 "export",
107 help="Propose one documented OMERO export command per image.",
108 )
109 export_parser.add_argument(
110 "selectors",
111 nargs="+",
112 type=image_selector,
113 help="Explicit Image:<positive-id> selectors.",
114 )
115 export_parser.add_argument(
116 "--format",
117 choices=("ome-tiff", "xml"),
118 default="ome-tiff",
119 help="Documented export format (default: ome-tiff).",
120 )
121 export_parser.add_argument(
122 "--output-dir",
123 required=True,
124 help="Existing non-symlink directory for future exports.",
125 )
126 export_parser.add_argument(
127 "--max-images",
128 type=positive_argument,
129 default=25,
130 help="Maximum image selectors, 1..100.",
131 )
132 return parser
135def path_depth(root: Path, current: Path) -> int:
136 relative = current.relative_to(root)
137 return 0 if relative == Path(".") else len(relative.parts)
140def scan_import_path(
141 raw_path: str,
142 *,
143 max_files_remaining: int,
144 scan_depth: int,
145) -> dict[str, Any]:
146 """Scan metadata only; never read file contents or follow symlinks."""
148 requested = Path(raw_path).expanduser()
149 if requested.is_symlink():
150 raise ValueError("top-level import paths must not be symlinks")
151 try:
152 resolved = requested.resolve(strict=True)
153 except FileNotFoundError as exc:
154 raise ValueError("an explicit import path does not exist") from exc
156 if resolved.is_file():
157 if max_files_remaining < 1:
158 raise ValueError("discovered file count exceeds --max-files")
159 return {
160 "path": str(resolved),
161 "kind": "file",
162 "regular_files": 1,
163 "depth_limit_reached": False,
164 }
165 if not resolved.is_dir():
166 raise ValueError("import paths must be regular files or directories")
168 regular_files = 0
169 skipped_symlinks = 0
170 depth_limit_reached = False
171 for root_text, directories, files in os.walk(
172 resolved,
173 topdown=True,
174 followlinks=False,
175 ):
176 root = Path(root_text)
177 depth = path_depth(resolved, root)
179 kept_directories = []
180 for directory in directories:
181 candidate = root / directory
182 if candidate.is_symlink():
183 skipped_symlinks += 1
184 else:
185 kept_directories.append(directory)
186 directories[:] = kept_directories
188 if depth >= scan_depth:
189 if directories:
190 depth_limit_reached = True
191 directories[:] = []
193 for filename in files:
194 candidate = root / filename
195 if candidate.is_symlink():
196 skipped_symlinks += 1
197 continue
198 if candidate.is_file():
199 regular_files += 1
200 if regular_files > max_files_remaining:
201 raise ValueError(
202 "discovered file count exceeds --max-files"
203 )
205 return {
206 "path": str(resolved),
207 "kind": "directory",
208 "regular_files": regular_files,
209 "skipped_symlinks": skipped_symlinks,
210 "depth_limit_reached": depth_limit_reached,
211 }
214def plan_import(args: argparse.Namespace) -> dict[str, Any]:
215 bounded_int(
216 args.max_paths,
217 name="max-paths",
218 minimum=1,
219 maximum=100,
220 )
221 bounded_int(
222 args.max_files,
223 name="max-files",
224 minimum=1,
225 maximum=10_000,
226 )
227 bounded_int(
228 args.scan_depth,
229 name="scan-depth",
230 minimum=1,
231 maximum=20,
232 )
233 if len(args.paths) > args.max_paths:
234 raise ValueError("explicit path count exceeds --max-paths")
236 entries: list[dict[str, Any]] = []
237 total_files = 0
238 for raw_path in args.paths:
239 entry = scan_import_path(
240 raw_path,
241 max_files_remaining=args.max_files - total_files,
242 scan_depth=args.scan_depth,
243 )
244 total_files += entry["regular_files"]
245 scan_command = [
246 "omero", "import", "--depth", str(args.scan_depth),
247 "-f", entry["path"],
248 ]
249 import_command = ["omero", "import", "--depth", str(args.scan_depth)]
250 if args.target:
251 import_command.extend(["-T", args.target])
252 import_command.append(entry["path"])
253 entry["local_omero_scan_command"] = scan_command
254 entry["future_remote_import_command"] = import_command
255 entries.append(entry)
257 depth_limited = any(
258 entry["depth_limit_reached"]
259 for entry in entries
260 if entry["kind"] == "directory"
261 )
262 skipped_symlinks = sum(entry.get("skipped_symlinks", 0) for entry in entries)
263 return {
264 "operation": "import",
265 "mode": "local-dry-run",
266 "server_contacted": False,
267 "commands_executed": False,
268 "credential_flags_included": False,
269 "target": args.target,
270 "limits": {
271 "max_paths": args.max_paths,
272 "max_files": args.max_files,
273 "scan_depth": args.scan_depth,
274 },
275 "total_regular_files": total_files,
276 "depth_limit_reached": depth_limited,
277 "skipped_symlinks": skipped_symlinks,
278 "ready_for_remote_import_review": not depth_limited and not skipped_symlinks,
279 "entries": entries,
280 "notes": [
281 "Run each local 'omero import -f' scan before any remote import.",
282 "Use a separately prompted 'omero login'; do not add -w or -k.",
283 "Importer grouping/file-format support is determined by Bio-Formats.",
284 "Local counts do not bound files added by Bio-Formats fileset discovery.",
285 "Review the actual -f fileset list, including paths outside the scan.",
286 ],
287 }
290def validated_output_directory(raw_path: str) -> Path:
291 requested = Path(raw_path).expanduser()
292 if requested.is_symlink():
293 raise ValueError("output directory must not be a symlink")
294 try:
295 resolved = requested.resolve(strict=True)
296 except FileNotFoundError as exc:
297 raise ValueError("output directory does not exist") from exc
298 if not resolved.is_dir():
299 raise ValueError("output directory is not a directory")
300 return resolved
303def plan_export(args: argparse.Namespace) -> dict[str, Any]:
304 bounded_int(
305 args.max_images,
306 name="max-images",
307 minimum=1,
308 maximum=100,
309 )
310 if len(args.selectors) > args.max_images:
311 raise ValueError("selector count exceeds --max-images")
312 if len(set(args.selectors)) != len(args.selectors):
313 raise ValueError("duplicate image selectors are not allowed")
315 output_directory = validated_output_directory(args.output_dir)
316 entries = []
317 for selector in args.selectors:
318 match = IMAGE_PATTERN.fullmatch(selector)
319 if match is None:
320 raise ValueError("invalid image selector")
321 image_id = int(match.group(1))
322 if args.format == "xml":
323 destination = output_directory / f"image-{image_id}.ome.xml"
324 command = [
325 "omero",
326 "export",
327 "--file",
328 str(destination),
329 "--type",
330 "XML",
331 selector,
332 ]
333 else:
334 destination = output_directory / f"image-{image_id}.ome.tiff"
335 command = [
336 "omero",
337 "export",
338 "--file",
339 str(destination),
340 selector,
341 ]
342 entries.append(
343 {
344 "selector": selector,
345 "destination": str(destination),
346 "collision": destination.exists()
347 or destination.is_symlink(),
348 "future_remote_export_command": command,
349 }
350 )
352 collisions = sum(bool(entry["collision"]) for entry in entries)
353 return {
354 "operation": "export",
355 "mode": "local-dry-run",
356 "server_contacted": False,
357 "commands_executed": False,
358 "credential_flags_included": False,
359 "format": args.format,
360 "output_directory": str(output_directory),
361 "limits": {"max_images": args.max_images},
362 "image_count": len(entries),
363 "collisions": collisions,
364 "ready_for_remote_export_review": collisions == 0,
365 "entries": entries,
366 "notes": [
367 "The documented export formats are OME-TIFF and XML.",
368 "This plan intentionally excludes experimental Dataset iteration.",
369 "Use a separately prompted 'omero login'; do not add -w or -k.",
370 ],
371 }
374def main(argv: list[str] | None = None) -> int:
375 parser = build_parser()
376 args = parser.parse_args(argv)
377 try:
378 if args.operation == "import":
379 payload = plan_import(args)
380 else:
381 payload = plan_export(args)
382 payload["generated_at"] = datetime.now(timezone.utc).isoformat()
383 emit_json(
384 payload,
385 output=args.output,
386 overwrite=args.overwrite,
387 )
388 return 0
389 except (
390 OutputPathError,
391 FileExistsError,
392 ValueError,
393 OSError,
394 ) as error:
395 print(scrubbed_error(error), file=sys.stderr)
396 return 2
399if __name__ == "__main__":
400 raise SystemExit(main())