Skills › Scientific Agent Skills › gtars › bed_validator.py
bed_validator.py
183 lines · skills/gtars/scripts/bed_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a local BED file without rewriting or uploading it."""
4from __future__ import annotations
6import argparse
7import sys
9from _common import (
10 HARD_MAX_BYTES,
11 HARD_MAX_RECORDS,
12 SafetyError,
13 add_path_mode_argument,
14 display_path,
15 fail_json,
16 inspect_bed,
17 int_type,
18 load_chrom_sizes,
19 local_path,
20 print_json,
21 sha256_file,
22)
25TOOL = "gtars-bed-validator"
28def build_parser() -> argparse.ArgumentParser:
29 parser = argparse.ArgumentParser(
30 description=(
31 "Validate one local BED/BED.GZ file as 0-based half-open intervals. "
32 "No input is rewritten and no network operation is available."
33 )
34 )
35 parser.add_argument("--input", required=True, help="Local BED or BED.GZ file.")
36 parser.add_argument(
37 "--assembly",
38 required=True,
39 help="Declared reference assembly/accession for the report.",
40 )
41 parser.add_argument(
42 "--chrom-sizes",
43 help="Local two-column chrom.sizes file for contig and bounds checks.",
44 )
45 parser.add_argument(
46 "--require-sorted",
47 action="store_true",
48 help="Fail if rows are not in chrom.sizes order then numeric start/end.",
49 )
50 parser.add_argument(
51 "--allow-empty",
52 action="store_true",
53 help="Allow a BED with no data records.",
54 )
55 parser.add_argument(
56 "--max-bytes",
57 type=int_type(minimum=1, maximum=HARD_MAX_BYTES, label="max-bytes"),
58 default=512 * 1024**2,
59 help="Compressed and expanded byte cap (default: 512 MiB).",
60 )
61 parser.add_argument(
62 "--max-records",
63 type=int_type(
64 minimum=1,
65 maximum=HARD_MAX_RECORDS,
66 label="max-records",
67 ),
68 default=1_000_000,
69 )
70 parser.add_argument(
71 "--max-examples",
72 type=int_type(minimum=0, maximum=100, label="max-examples"),
73 default=10,
74 )
75 add_path_mode_argument(parser)
76 return parser
79def validate(args: argparse.Namespace) -> tuple[dict, int]:
80 assembly = args.assembly.strip()
81 if not assembly or len(assembly) > 200:
82 raise SafetyError("assembly must contain 1-200 characters")
84 bed_path = local_path(args.input, kind="file")
85 chrom_sizes = None
86 chrom_order = None
87 chrom_sizes_report = None
88 if args.chrom_sizes:
89 chrom_path = local_path(args.chrom_sizes, kind="file")
90 chrom_sizes, chrom_order = load_chrom_sizes(
91 chrom_path,
92 max_bytes=min(args.max_bytes, 128 * 1024**2),
93 max_records=min(args.max_records, 1_000_000),
94 )
95 chrom_digest, chrom_bytes = sha256_file(
96 chrom_path,
97 max_bytes=min(args.max_bytes, 128 * 1024**2),
98 )
99 chrom_sizes_report = {
100 "path": display_path(chrom_path, 2, args.path_mode),
101 "sha256": chrom_digest,
102 "size_bytes": chrom_bytes,
103 "contig_count": len(chrom_sizes),
104 }
106 summary = inspect_bed(
107 bed_path,
108 max_bytes=args.max_bytes,
109 max_records=args.max_records,
110 chrom_sizes=chrom_sizes,
111 chrom_order=chrom_order,
112 max_examples=args.max_examples,
113 )
114 errors = dict(summary.pop("errors"))
115 warnings = dict(summary.pop("warnings"))
116 examples = summary.pop("issue_examples")
117 if summary["records"] == 0 and not args.allow_empty:
118 errors["no_data_records"] = 1
119 if args.require_sorted and summary["out_of_order_records"]:
120 errors["sorting_required"] = summary["out_of_order_records"]
122 plan: list[dict[str, str]] = [
123 {
124 "action": "preserve_coordinate_contract",
125 "detail": "BED is 0-based, start-inclusive, end-exclusive",
126 },
127 {
128 "action": "preserve_assembly_and_contig_names",
129 "detail": "do not rename contigs or liftover implicitly",
130 },
131 ]
132 if chrom_sizes is None:
133 plan.append(
134 {
135 "action": "supply_chromosome_sizes",
136 "detail": "required before contig and interval-bound validation",
137 }
138 )
139 if summary["out_of_order_records"]:
140 plan.append(
141 {
142 "action": "stable_sort_a_copy",
143 "detail": "chrom.sizes order, numeric start, numeric end; retain original",
144 }
145 )
147 report = {
148 "ok": not errors,
149 "tool": TOOL,
150 "contract": {
151 "assembly": assembly,
152 "coordinate_system": "0-based-half-open",
153 "input_mutated": False,
154 "network_used": False,
155 "subprocess_used": False,
156 "symlinks_allowed": False,
157 "gtars_coordinate_limit": "u32",
158 },
159 "input": {
160 "path": display_path(bed_path, 1, args.path_mode),
161 **summary,
162 },
163 "chromosome_sizes": chrom_sizes_report,
164 "errors": errors,
165 "warnings": warnings,
166 "issue_examples": examples,
167 "normalization_plan": plan,
168 }
169 return report, 0 if not errors else 2
172def main() -> int:
173 args = build_parser().parse_args()
174 try:
175 report, status = validate(args)
176 print_json(report)
177 return status
178 except (OSError, SafetyError, UnicodeError) as exc:
179 return fail_json(TOOL, exc)
182if __name__ == "__main__":
183 sys.exit(main())