| |
| """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() |
|
|
| |
| 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 |
| 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()) |
|
|
