| |
| """iHARP risk visualization entry point (enhanced script). |
| |
| Reads a batch_annotate.py summary CSV and renders a publication-style horizontal |
| bar chart of the top-N compounds by an RQmodified column (risk ranking + visualization). |
| |
| This is an ENHANCED script: it depends on matplotlib + pandas (not part of the |
| standard library). The core calculation scripts (er_rqmodified / ahr_rqmodified / |
| batch_annotate / query_toxcast / xlsx_reader) remain pure standard-library and do |
| not require these dependencies. |
| |
| Usage: |
| python risk_visualize.py --input ER_CS_summary.csv --column RQmodified_ag --column RQmodified_ant |
| python risk_visualize.py --input ER_CS_summary.csv --column RQmodified_ag --top 20 |
| |
| Default --column: RQmodified_ag, RQmodified_ant (ER). AhR summaries only contain |
| RQmodified_ag; missing columns are skipped with a warning. |
| |
| Output: one PNG per requested column, named <input_stem>_<column>.png under |
| --output-dir (default current directory). PNG only. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
| try: |
| import matplotlib |
|
|
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import pandas as pd |
| except ImportError as exc: |
| matplotlib = plt = pd = None |
| _IMPORT_ERROR = exc |
| else: |
| _IMPORT_ERROR = None |
|
|
| RULE_VERSION = "iharp_skill_v1" |
|
|
| DEFAULT_COLUMNS = ["RQmodified_ag", "RQmodified_ant"] |
|
|
|
|
| def render_hbar( |
| df: pd.DataFrame, |
| column: str, |
| compound_col: str, |
| top: int, |
| out_path: Path, |
| ) -> tuple[int, str | None]: |
| """Render a horizontal bar chart of top-N compounds by `column`. |
| |
| Returns (plotted_count, error_or_None). |
| """ |
| sub = df[[compound_col, column]].copy() |
| sub[column] = pd.to_numeric(sub[column], errors="coerce") |
| sub = sub.dropna(subset=[compound_col, column]) |
| sub = sub[sub[compound_col].astype(str).str.strip() != ""] |
| if sub.empty: |
| return 0, f"no valid numeric values in column {column!r}" |
|
|
| |
| top_df = ( |
| sub.sort_values(by=column, ascending=False) |
| .head(top) |
| .sort_values(by=column, ascending=True) |
| ) |
|
|
| values = top_df[column].astype(float) |
| vmin = float(values.min()) |
| vmax = float(values.max()) |
| if vmax - vmin < 1e-12: |
| vmin = 0.0 if vmax > 0 else vmax - 1.0 |
| vmax = vmax if vmax > 0 else 0.0 |
| norm = plt.Normalize(vmin, vmax) |
| colors = plt.cm.YlOrRd(norm(values)) |
|
|
| plt.rcParams["font.sans-serif"] = ["Arial", "DejaVu Sans"] |
| plt.rcParams["axes.unicode_minus"] = False |
| fig, ax = plt.subplots(figsize=(7, 6.5), dpi=300) |
|
|
| bars = ax.barh( |
| top_df[compound_col].astype(str), |
| values, |
| color=colors, |
| height=0.68, |
| edgecolor="#333333", |
| linewidth=0.5, |
| ) |
|
|
| |
| for bar in bars: |
| width = bar.get_width() |
| ax.text( |
| width + (vmax * 0.01), |
| bar.get_y() + bar.get_height() / 2, |
| f"{width:.3f}", |
| va="center", |
| ha="left", |
| fontsize=14, |
| color="#333333", |
| ) |
|
|
| ax.set_xlabel(column, fontsize=10, fontweight="bold") |
| ax.set_ylabel("Top Candidate Compounds", fontsize=10, fontweight="bold") |
| ax.set_xlim(0, vmax * 1.15) |
| ax.spines["top"].set_visible(False) |
| ax.spines["right"].set_visible(False) |
| ax.tick_params(axis="both", which="major", labelsize=8.5) |
|
|
| plt.tight_layout() |
| plt.savefig(out_path, dpi=300) |
| plt.close(fig) |
| return int(len(top_df)), None |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser( |
| description="iHARP risk visualization (top-N RQmodified horizontal bar chart)" |
| ) |
| parser.add_argument("--input", required=True, help="summary CSV from batch_annotate.py") |
| parser.add_argument( |
| "--column", |
| action="append", |
| help="RQmodified column to plot; may be repeated (default: RQmodified_ag, RQmodified_ant)", |
| ) |
| parser.add_argument("--top", type=int, default=20, help="number of top compounds (default 20)") |
| parser.add_argument( |
| "--compound-col", default="CompoundName", help="compound name column (default CompoundName)" |
| ) |
| parser.add_argument("--output-dir", default=".", help="output directory (default current directory)") |
| args = parser.parse_args() |
|
|
| if _IMPORT_ERROR is not None: |
| print(json.dumps({ |
| "ok": False, |
| "error": "missing required dependencies (matplotlib, pandas); install with: pip install matplotlib pandas", |
| "detail": str(_IMPORT_ERROR), |
| }, ensure_ascii=False)) |
| return 3 |
|
|
| input_path = Path(args.input) |
| if not input_path.is_file(): |
| print(json.dumps({"ok": False, "error": f"input file not found: {input_path}"}, ensure_ascii=False)) |
| return 2 |
|
|
| columns = args.column or DEFAULT_COLUMNS |
| output_dir = Path(args.output_dir) |
| output_dir.mkdir(parents=True, exist_ok=True) |
|
|
| try: |
| df = pd.read_csv(input_path) |
| except Exception as exc: |
| print(json.dumps({"ok": False, "error": f"failed to read CSV: {exc}"}, ensure_ascii=False)) |
| return 2 |
|
|
| if args.compound_col not in df.columns: |
| print( |
| json.dumps( |
| { |
| "ok": False, |
| "error": f"compound column {args.compound_col!r} not found; actual columns: {list(df.columns)}", |
| }, |
| ensure_ascii=False, |
| ) |
| ) |
| return 2 |
|
|
| input_stem = input_path.stem |
| files: dict[str, str] = {} |
| skipped: list[str] = [] |
| for col in columns: |
| if col not in df.columns: |
| skipped.append(f"{col} (column not found)") |
| continue |
| out_path = output_dir / f"{input_stem}_{col}.png" |
| count, err = render_hbar(df, col, args.compound_col, args.top, out_path) |
| if err: |
| skipped.append(f"{col}: {err}") |
| continue |
| files[col] = str(out_path) |
|
|
| result = { |
| "ok": True, |
| "rule_version": RULE_VERSION, |
| "input": str(input_path), |
| "top": args.top, |
| "compound_col": args.compound_col, |
| "files": files, |
| "skipped": skipped, |
| } |
| print(json.dumps(result, ensure_ascii=False, indent=2)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
