#!/usr/bin/env python3 """Match and query AC50/HitCall details from the ToxCast_AC50 folder. Data source: the ToxCast_AC50 folder (each CSV = all-compound AC50 for one endpoint). Column structure: DTXSID, PREFERRED NAME, CASRN, MOLECULAR FORMULA, MONOISOTOPIC MASS, ToxCast Active, ToxCast Total, % ToxCast Active, HIT CALL, TOP, SCALED TOP, AC50, LOGAC50 Purpose: the invitrodb MCP can only return an endpoint name list, not AC50/HitCall; this script goes back to the ToxCast folder and matches the AC50/HitCall details by compound name or CASRN. Usage: python query_toxcast.py --query "Bisphenol A" python query_toxcast.py --query "80-05-7" python query_toxcast.py --query "Bisphenol A" --toxcast-dir /path/to/ToxCast_AC50 python query_toxcast.py --query "80-05-7" --output /tmp/bpa_toxcast.csv External data dependency: - The ToxCast_AC50 folder is bundled with this skill under assets/ToxCast_AC50. - When --toxcast-dir is not passed, auto-detection order: assets/ToxCast_AC50 -> environment variable TOXCAST_DIR -> project-root ToxCast_AC50; error if not found. Notes: - standard library only, no third-party dependencies. - prefer reusing the skill-root `.cache/toxcast_index.sqlite` cache (if present and fresh); otherwise scan the folder and rebuild. """ from __future__ import annotations import argparse import csv import glob import json import os import re import sqlite3 from pathlib import Path def normalize_name(value: object) -> str: """Matching key kept consistent with chemical_annotator5.normalize_name.""" text = "" if value is None else str(value) text = text.strip().strip('"').strip("'").lower() text = re.sub(r"\s+", " ", text) text = text.replace("–", "-").replace("—", "-") text = re.sub(r"\s*([,;:/()\[\]{}+-])\s*", r"\1", text) return text def is_probable_cas(value: str) -> bool: return bool(re.fullmatch(r"\d{2,7}-\d{2}-\d", value.strip())) def skill_root() -> Path: """This script lives in skill/iharp/scripts/; go up two levels to the skill root.""" return Path(__file__).resolve().parents[1] def project_root() -> Path: """Go up three levels to the project root (skill/iharp/scripts/ -> project root).""" return Path(__file__).resolve().parents[3] def probe_toxcast_dir() -> Path | None: """Auto-detect the ToxCast folder, by priority: 1. skill-bundled data: assets/ToxCast_AC50 2. environment variable TOXCAST_DIR 3. project-root ToxCast_AC50 (legacy layout compatibility) """ bundled = skill_root() / "assets" / "ToxCast_AC50" if bundled.is_dir(): return bundled env = os.environ.get("TOXCAST_DIR") if env and Path(env).is_dir(): return Path(env) candidate = project_root() / "ToxCast_AC50" if candidate.is_dir(): return candidate return None def cache_path_for(toxcast_dir: Path) -> Path: """Cache is placed under the skill root .cache directory (separate from bundled data).""" return skill_root() / ".cache" / "toxcast_index.sqlite" def csv_signature(toxcast_dir: Path) -> str: """Use total file size + newest mtime as the folder signature to decide whether the cache is stale.""" total_size = 0 newest = 0 for path in glob.glob(str(toxcast_dir / "*.csv")): stat = os.stat(path) total_size += stat.st_size newest = max(newest, int(stat.st_mtime)) return f"{total_size}:{newest}" def scan_folder(toxcast_dir: Path) -> list[dict[str, str]]: """Scan the folder directly and return a list of detail rows.""" rows: list[dict[str, str]] = [] for path in sorted(glob.glob(str(toxcast_dir / "*.csv"))): endpoint = os.path.splitext(os.path.basename(path))[0] with open(path, newline="", encoding="utf-8-sig", errors="replace") as f: reader = csv.DictReader(f) for row in reader: name = (row.get("PREFERRED NAME") or row.get("Preferred Name") or "").strip() casrn = (row.get("CASRN") or row.get("CAS") or "").strip() if not name and not casrn: continue rows.append( { "name_key": normalize_name(name), "casrn": casrn, "compound_name": name, "endpoint": endpoint, "hit_call": (row.get("HIT CALL") or "").strip(), "ac50": (row.get("AC50") or "").strip(), } ) return rows def load_index(toxcast_dir: Path) -> list[dict[str, str]]: """Prefer reusing the sqlite cache; scan and rebuild when missing or stale.""" cache_path = cache_path_for(toxcast_dir) sig = csv_signature(toxcast_dir) use_cache = False if cache_path.exists(): try: conn = sqlite3.connect(cache_path) cur = conn.cursor() cur.execute("SELECT value FROM meta WHERE key='sig'") row = cur.fetchone() if row is not None and row[0] == sig: use_cache = True conn.close() except sqlite3.Error: use_cache = False if use_cache: conn = sqlite3.connect(cache_path) try: cur = conn.cursor() cur.execute( "SELECT name_key, casrn, compound_name, endpoint, hit_call, ac50 FROM toxcast" ) rows = [ { "name_key": r[0], "casrn": r[1], "compound_name": r[2], "endpoint": r[3], "hit_call": r[4] or "", "ac50": r[5] or "", } for r in cur.fetchall() ] return rows finally: conn.close() # scan the folder and write back the cache rows = scan_folder(toxcast_dir) try: cache_path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(cache_path) cur = conn.cursor() cur.execute( "CREATE TABLE IF NOT EXISTS toxcast " "(name_key TEXT, casrn TEXT, compound_name TEXT, endpoint TEXT, hit_call TEXT, ac50 TEXT)" ) cur.execute("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)") cur.execute("DELETE FROM toxcast") cur.executemany( "INSERT INTO toxcast VALUES (?, ?, ?, ?, ?, ?)", [ (r["name_key"], r["casrn"], r["compound_name"], r["endpoint"], r["hit_call"], r["ac50"]) for r in rows ], ) cur.execute("DELETE FROM meta") cur.execute("INSERT INTO meta(key, value) VALUES('sig', ?)", (sig,)) conn.commit() conn.close() except sqlite3.Error: pass # cache write failure does not affect the main flow return rows def query_details(rows: list[dict[str, str]], query: str) -> list[dict[str, str]]: """Match by CAS or name, return de-duplicated details sorted by endpoint.""" q = query.strip() if is_probable_cas(q): matched = [r for r in rows if r["casrn"] == q] else: key = normalize_name(q) matched = [r for r in rows if r["name_key"] == key] if not matched: return [] seen: set[tuple[str, str]] = set() details: list[dict[str, str]] = [] for r in sorted(matched, key=lambda x: (x["endpoint"], x["ac50"])): key = (r["endpoint"], r["ac50"]) if key in seen: continue seen.add(key) details.append( { "CompoundName": r["compound_name"], "CASRN": r["casrn"], "Endpoint": r["endpoint"], "AC50": r["ac50"], "HitCall": r["hit_call"], } ) return details def main() -> int: parser = argparse.ArgumentParser(description="Match and query AC50/HitCall from the ToxCast folder") parser.add_argument("--query", required=True, help="compound name or CASRN") parser.add_argument( "--toxcast-dir", default=None, help="ToxCast_AC50 folder path; auto-detect when omitted (project root or env TOXCAST_DIR)", ) parser.add_argument("--output", help="optional detail CSV output path") args = parser.parse_args() toxcast_dir = Path(args.toxcast_dir) if args.toxcast_dir else probe_toxcast_dir() if toxcast_dir is None or not toxcast_dir.is_dir(): print( json.dumps( { "ok": False, "error": ( "ToxCast_AC50 folder not found." "Specify it with --toxcast-dir, or set the TOXCAST_DIR environment variable." ), }, ensure_ascii=False, ) ) return 2 rows = load_index(toxcast_dir) details = query_details(rows, args.query) if not details: print(json.dumps({"ok": True, "query": args.query, "match_count": 0, "details": []}, ensure_ascii=False)) return 0 active_count = sum(1 for d in details if d["HitCall"].strip().lower() == "active") summary = { "ok": True, "query": args.query, "compound_name": details[0]["CompoundName"], "casrn": details[0]["CASRN"], "endpoint_count": len(details), "active_count": active_count, "details": details, } if args.output: out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) cols = ["CompoundName", "CASRN", "Endpoint", "AC50", "HitCall"] with out.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=cols) writer.writeheader() for d in details: writer.writerow({c: d[c] for c in cols}) summary["output"] = str(out) print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())