Skills › Scientific Agent Skills › bids › update_schema.py
update_schema.py
101 lines · skills/bids/scripts/update_schema.py at 5f2bc03
1#!/usr/bin/env python3
2"""Update the bundled BIDS schema and BEP catalogue from official public sources.
4From the skill directory:
5 python scripts/update_schema.py
7For a reproducible release, or if ReadTheDocs rejects automated downloads:
8 python scripts/update_schema.py --schema-url https://raw.githubusercontent.com/bids-standard/bids-schema/main/versions/1.11.2/schema.json
10For draft BEP schemas, discover the actual path under the upstream BEPs directory;
11do not assume that BEPs/BEP032/schema.json exists. Drafts are not stable requirements.
12No external dependencies beyond the Python standard library.
13"""
15import argparse
16import json
17import os
18import re
19import tempfile
20import urllib.request
21from pathlib import Path
23REFERENCES_DIR = Path(__file__).resolve().parent.parent / "references"
24SCHEMA_URL = "https://bids-specification.readthedocs.io/en/stable/schema.json"
25BEPS_URL = "https://raw.githubusercontent.com/bids-standard/bids-website/main/data/beps/beps.yml"
26MAX_BYTES = 20 * 1024 * 1024
27TIMEOUT_SECONDS = 30
30def fetch(url):
31 """Fetch a bounded public response; reject non-HTTPS sources."""
32 if not url.startswith("https://"):
33 raise ValueError("Use an HTTPS URL for upstream data")
34 print(f"Fetching {url} ...")
35 req = urllib.request.Request(url, headers={"User-Agent": "bids-skill-updater/1.1"})
36 with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp:
37 data = resp.read(MAX_BYTES + 1)
38 if len(data) > MAX_BYTES:
39 raise ValueError(f"Upstream response exceeds {MAX_BYTES} bytes")
40 return data
43def atomic_write(output, data):
44 """Replace one validated artifact without truncating its previous copy."""
45 temporary = None
46 try:
47 with tempfile.NamedTemporaryFile(dir=output.parent, delete=False) as stream:
48 temporary = Path(stream.name)
49 stream.write(data)
50 os.replace(temporary, output)
51 finally:
52 if temporary is not None:
53 temporary.unlink(missing_ok=True)
56def update_schema(url):
57 """Download and check the schema's envelope before replacing the snapshot."""
58 d = json.loads(fetch(url))
59 if not isinstance(d, dict):
60 raise ValueError("Expected a BIDS schema object")
61 for field in ("schema_version", "bids_version"):
62 if not isinstance(d.get(field), str) or not d[field]:
63 raise ValueError(f"BIDS schema missing {field}")
64 if not isinstance(d.get("objects"), dict) or not isinstance(d.get("rules"), dict):
65 raise ValueError("BIDS schema must contain objects and rules mappings")
66 output = REFERENCES_DIR / "bids_schema.json"
67 atomic_write(output, (json.dumps(d, indent=2) + "\n").encode("utf-8"))
68 print(f" -> {output.name}: schema {d['schema_version']} / BIDS {d['bids_version']}")
71def bep_numbers(data):
72 """Check the known upstream catalogue envelope (not a general YAML parser)."""
73 text = data.decode("utf-8")
74 numbers = re.findall(r"^\s*-\s+number:\s*['\"]?(\d{3})['\"]?\s*$", text, re.MULTILINE)
75 if not numbers or len(numbers) != len(set(numbers)):
76 raise ValueError("Expected a BEP catalogue with unique three-digit number entries")
77 return numbers
80def update_beps():
81 """Preserve upstream YAML after an envelope check; reject HTML/error payloads."""
82 data = fetch(BEPS_URL)
83 count = len(bep_numbers(data))
84 output = REFERENCES_DIR / "beps.yml"
85 atomic_write(output, data)
86 print(f" -> {output.name}: {count} BEPs")
89def main():
90 parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
91 parser.add_argument("--schema-url", default=SCHEMA_URL, help=f"Schema JSON URL (default: {SCHEMA_URL})")
92 parser.add_argument("--skip-beps", action="store_true", help="Skip fetching beps.yml")
93 args = parser.parse_args()
94 update_schema(args.schema_url)
95 if not args.skip_beps:
96 update_beps()
97 print("Done. Review version changes and dependent documentation before committing.")
100if __name__ == "__main__":
101 main()