Skills › Scientific Agent Skills › gtars › execution_plan.py
execution_plan.py
384 lines · skills/gtars/scripts/execution_plan.py at 5f2bc03
1#!/usr/bin/env python3
2"""Build a bounded local Gtars overlap/coverage execution plan."""
4from __future__ import annotations
6import argparse
7import sys
8from pathlib import Path
10from _common import (
11 HARD_MAX_BYTES,
12 HARD_MAX_RECORDS,
13 HARD_MAX_WORKERS,
14 SafetyError,
15 add_path_mode_argument,
16 display_path,
17 fail_json,
18 inspect_bed,
19 int_type,
20 load_chrom_sizes,
21 local_path,
22 print_json,
23 prefix_collision_count,
24 sha256_file,
25)
28TOOL = "gtars-execution-plan"
29GTARS_CLI_VERSION = "0.10.0"
30GTARS_PYTHON_VERSION = "0.10.0"
33def build_parser() -> argparse.ArgumentParser:
34 parser = argparse.ArgumentParser(
35 description=(
36 "Validate local inputs and emit a fixed Gtars 0.10 execution plan. "
37 "The helper never imports gtars or runs a subprocess."
38 )
39 )
40 parser.add_argument(
41 "--operation",
42 required=True,
43 choices=("overlap", "count", "coverage", "consensus", "fragment-score"),
44 )
45 parser.add_argument("--assembly", required=True)
46 parser.add_argument("--query", help="Query BED or fragment file.")
47 parser.add_argument("--universe", help="Universe/consensus BED file.")
48 parser.add_argument(
49 "--input",
50 action="append",
51 default=[],
52 help="Consensus input BED; repeat at least twice.",
53 )
54 parser.add_argument("--chrom-sizes", help="Local chromosome-sizes file.")
55 parser.add_argument("--output", help="Planned local output or output prefix.")
56 parser.add_argument("--backend", choices=("bits", "ailist"), default="bits")
57 parser.add_argument(
58 "--coverage-format",
59 choices=("bw", "wig", "bedgraph", "npy"),
60 default="bw",
61 )
62 parser.add_argument(
63 "--coverage-streaming",
64 action="store_true",
65 help="Plan BED streaming; only wig/bedgraph output is supported.",
66 )
67 parser.add_argument(
68 "--smooth-size",
69 type=int_type(minimum=1, maximum=10_000_000, label="smooth-size"),
70 default=5,
71 )
72 parser.add_argument(
73 "--step-size",
74 type=int_type(minimum=1, maximum=10_000_000, label="step-size"),
75 default=1,
76 )
77 parser.add_argument(
78 "--count-type",
79 choices=("start", "end", "core", "all"),
80 default="core",
81 )
82 parser.add_argument(
83 "--threads",
84 type=int_type(minimum=1, maximum=HARD_MAX_WORKERS, label="threads"),
85 default=1,
86 )
87 parser.add_argument(
88 "--min-count",
89 type=int_type(minimum=1, maximum=1_000_000, label="min-count"),
90 default=1,
91 )
92 parser.add_argument("--scoring-mode", choices=("atac", "chip"), default="atac")
93 parser.add_argument(
94 "--barcode",
95 action="store_true",
96 help="Plan fscoring sparse barcode mode.",
97 )
98 parser.add_argument(
99 "--max-bytes",
100 type=int_type(minimum=1, maximum=HARD_MAX_BYTES, label="max-bytes"),
101 default=1024**3,
102 )
103 parser.add_argument(
104 "--max-records",
105 type=int_type(
106 minimum=1,
107 maximum=HARD_MAX_RECORDS,
108 label="max-records",
109 ),
110 default=2_000_000,
111 )
112 add_path_mode_argument(parser)
113 return parser
116def _require(value: str | None, option: str) -> str:
117 if value is None:
118 raise SafetyError(f"{option} is required for this operation")
119 return value
122def _output(raw: str | None) -> Path:
123 return local_path(_require(raw, "--output"), must_exist=False)
126def build_plan(args: argparse.Namespace) -> tuple[dict, int]:
127 assembly = args.assembly.strip()
128 if not assembly or len(assembly) > 200:
129 raise SafetyError("assembly must contain 1-200 characters")
131 chrom_sizes = None
132 chrom_order = None
133 chrom_report = None
134 if args.chrom_sizes:
135 chrom_path = local_path(args.chrom_sizes, kind="file")
136 chrom_sizes, chrom_order = load_chrom_sizes(
137 chrom_path,
138 max_bytes=min(args.max_bytes, 128 * 1024**2),
139 max_records=min(args.max_records, 1_000_000),
140 )
141 digest, size = sha256_file(
142 chrom_path,
143 max_bytes=min(args.max_bytes, 128 * 1024**2),
144 )
145 chrom_report = {
146 "path": display_path(chrom_path, 1, args.path_mode),
147 "sha256": digest,
148 "size_bytes": size,
149 "contig_count": len(chrom_sizes),
150 }
152 inputs: list[dict] = []
153 errors: dict[str, int] = {}
154 next_index = 2
156 def add_bed(raw: str, role: str, *, sorted_required: bool = False) -> Path:
157 nonlocal next_index
158 path = local_path(raw, kind="file")
159 summary = inspect_bed(
160 path,
161 max_bytes=args.max_bytes,
162 max_records=args.max_records,
163 chrom_sizes=chrom_sizes,
164 chrom_order=chrom_order,
165 )
166 for code, count in summary["errors"].items():
167 errors[f"{role}:{code}"] = count
168 if summary["records"] == 0:
169 errors[f"{role}:no_data_records"] = 1
170 if args.operation == "overlap" and summary["skipped_records"]:
171 errors[f"{role}:cli_overlap_requires_header_free_bed"] = summary[
172 "skipped_records"
173 ]
174 if args.operation == "fragment-score" and role == "consensus" and summary["duplicate_intervals"]:
175 errors["consensus:duplicate_intervals"] = summary["duplicate_intervals"]
176 if sorted_required and summary["out_of_order_records"]:
177 errors[f"{role}:sorting_required"] = summary["out_of_order_records"]
178 inputs.append(
179 {
180 "role": role,
181 "path": display_path(path, next_index, args.path_mode),
182 "sha256": summary["sha256"],
183 "size_bytes": summary["size_bytes"],
184 "records": summary["records"],
185 "minimum_columns": summary["minimum_columns"],
186 "out_of_order_records": summary["out_of_order_records"],
187 }
188 )
189 next_index += 1
190 return path
192 operation = args.operation
193 plan: dict[str, object]
194 output_report = None
196 if operation in {"overlap", "count"}:
197 add_bed(_require(args.query, "--query"), "query")
198 add_bed(_require(args.universe, "--universe"), "universe")
199 if operation == "overlap":
200 plan = {
201 "interface": "cli",
202 "argv_template": [
203 "gtars",
204 "overlaprs",
205 "--query",
206 "<query-bed>",
207 "--universe",
208 "<universe-bed>",
209 "--backend",
210 args.backend,
211 ],
212 "stdout": (
213 "one BED3 universe-hit row per overlap; query IDs and counts "
214 "are not emitted"
215 ),
216 }
217 else:
218 plan = {
219 "interface": "python",
220 "imports": ["from gtars.models import RegionSet"],
221 "call": "query.count_overlaps(universe)",
222 "return": "one integer per query region",
223 }
224 elif operation == "coverage":
225 if chrom_sizes is None:
226 raise SafetyError("--chrom-sizes is required for coverage")
227 add_bed(_require(args.query, "--query"), "coverage_input", sorted_required=True)
228 out = _output(args.output)
229 output_report = {
230 "path": display_path(out, next_index, args.path_mode),
231 "already_exists": out.exists(),
232 }
233 if out.exists():
234 errors["output_already_exists"] = 1
235 collisions = prefix_collision_count(out)
236 if collisions:
237 errors["output_prefix_collisions"] = collisions
238 if args.coverage_streaming and args.coverage_format not in {"wig", "bedgraph"}:
239 raise SafetyError(
240 "streaming coverage supports only wig or bedgraph, not bw/npy"
241 )
242 argv = [
243 "gtars",
244 "uniwig",
245 "--file",
246 "<coverage-input>",
247 "--filetype",
248 "bed",
249 "--chromref",
250 "<chrom-sizes>",
251 "--smoothsize",
252 str(args.smooth_size),
253 "--stepsize",
254 str(args.step_size),
255 "--fileheader",
256 "<output-prefix>",
257 "--outputtype",
258 args.coverage_format,
259 "--counttype",
260 args.count_type,
261 ]
262 if args.coverage_streaming:
263 argv.append("--streaming")
264 else:
265 argv.extend(["--threads", str(args.threads)])
266 plan = {
267 "interface": "cli",
268 "argv_template": argv,
269 "note": (
270 "batch mode is required for bigWig; run coverage_preflight.py "
271 "before approval"
272 ),
273 }
274 elif operation == "consensus":
275 if len(args.input) < 2:
276 raise SafetyError("consensus requires at least two --input BED files")
277 if args.min_count > len(args.input):
278 raise SafetyError("--min-count cannot exceed the number of input sets")
279 for index, raw in enumerate(args.input, start=1):
280 add_bed(raw, f"consensus_{index}")
281 out = _output(args.output)
282 output_report = {
283 "path": display_path(out, next_index, args.path_mode),
284 "already_exists": out.exists(),
285 }
286 if out.exists():
287 errors["output_already_exists"] = 1
288 plan = {
289 "interface": "cli",
290 "argv_template": [
291 "gtars",
292 "consensus",
293 "--beds",
294 *[f"<bed-{index}>" for index in range(1, len(args.input) + 1)],
295 "--min-count",
296 str(args.min_count),
297 "--output",
298 "<output-bed>",
299 ],
300 "semantics": (
301 "reduce the union (including adjacent intervals), then count "
302 "input sets having any overlap with each union interval"
303 ),
304 }
305 else:
306 if not args.barcode and args.scoring_mode == "atac":
307 # gtars-scoring 0.5.2 in the CLI 0.10.0 lockfile constructs its
308 # right-cut query with end < start. Do not green-light those counts.
309 errors["upstream_atac_right_cut_interval_invalid"] = 1
310 fragment = add_bed(_require(args.query, "--query"), "fragments")
311 add_bed(_require(args.universe, "--universe"), "consensus")
312 fragment_summary = next(item for item in inputs if item["role"] == "fragments")
313 if (fragment_summary["minimum_columns"] or 0) < 5:
314 errors["fragments:fewer_than_five_columns"] = 1
315 out = _output(args.output)
316 output_report = {
317 "path": display_path(out, next_index, args.path_mode),
318 "already_exists": out.exists(),
319 }
320 if out.exists():
321 errors["output_already_exists"] = 1
322 argv = [
323 "gtars",
324 "fscoring",
325 "<fragment-file>",
326 "<consensus-bed>",
327 "--output",
328 "<output-or-prefix>",
329 ]
330 if args.barcode:
331 collisions = prefix_collision_count(out)
332 if collisions:
333 errors["output_prefix_collisions"] = collisions
334 argv.append("--barcode")
335 else:
336 argv.extend(["--mode", args.scoring_mode])
337 plan = {
338 "interface": "cli",
339 "argv_template": argv,
340 "fragment_input_is_single_local_file": fragment.is_file(),
341 }
343 report = {
344 "ok": not errors,
345 "ready_to_execute": not errors,
346 "tool": TOOL,
347 "contract": {
348 "assembly": assembly,
349 "coordinate_system": "0-based-half-open",
350 "gtars_cli_version": GTARS_CLI_VERSION,
351 "gtars_python_version": GTARS_PYTHON_VERSION,
352 "commands_executed": False,
353 "packages_imported": False,
354 "network_used": False,
355 "files_written": False,
356 "symlinks_allowed": False,
357 },
358 "operation": operation,
359 "chromosome_sizes": chrom_report,
360 "inputs": inputs,
361 "output": output_report,
362 "errors": errors,
363 "execution_plan": plan,
364 "approval_gate": [
365 "review checksums, assembly, contigs, sorting, and output collision",
366 "set CPU, RAM, disk, file-count, and wall-time limits",
367 "run only the exact pinned interface after explicit approval",
368 ],
369 }
370 return report, 0 if not errors else 2
373def main() -> int:
374 args = build_parser().parse_args()
375 try:
376 report, status = build_plan(args)
377 print_json(report)
378 return status
379 except (OSError, SafetyError, UnicodeError) as exc:
380 return fail_json(TOOL, exc)
383if __name__ == "__main__":
384 sys.exit(main())