Blame for scripts/risk_visualize.py

#!/usr/bin/env python3
"""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")  # headless environment (CLI / WSL): no display available
    import matplotlib.pyplot as plt
    import pandas as pd
except ImportError as exc:  # enhanced-script dependencies missing -> report clearly
    matplotlib = plt = pd = None  # type: ignore[assignment]
    _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}"

    # risk ranking: descending by value, keep top-N, then ascending so the highest sits on top
    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:  # guard against a zero range (single value / all equal)
        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,
    )

    # annotate each bar end with its value
    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())