| |
| """iHARP ER/AhR RQmodified deterministic calculation entry point (self-contained, runnable standalone). |
| |
| Does not depend on the project-root core/ engine; directly calls the self-contained |
| calculation modules `er_rqmodified.py` / `ahr_rqmodified.py` in the same directory. |
| |
| Usage: |
| python run_rqmodified.py --pathway er --input <endpoint_details.csv> \ |
| [--reference <reference.csv>] [--output-dir <output_dir>] |
| python run_rqmodified.py --pathway ahr --input <endpoint_details.csv> \ |
| [--reference <reference.csv>] [--output-dir <output_dir>] |
| |
| If --reference is not passed, the reference files bundled under assets/ are used by default: |
| assets/er_reference_endpoint_details.csv |
| assets/ahr_reference_endpoint_details.csv |
| |
| Output files (written to output-dir, default current directory): |
| <pathway>_detail.csv -- compound-level detail results |
| <pathway>_summary.csv -- SUM_RQmodified aggregate |
| <pathway>_warnings.txt -- warning log |
| |
| stdout prints a JSON summary for upstream auditing. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import json |
| import sys |
| from pathlib import Path |
| from typing import Any |
|
|
| |
| import ahr_rqmodified |
| import er_rqmodified |
|
|
| RULE_VERSION = "iharp_skill_v1" |
|
|
| |
| _ASSETS_DIR = Path(__file__).resolve().parent.parent / "assets" |
| DEFAULT_ER_REFERENCE = _ASSETS_DIR / "er_reference_endpoint_details.csv" |
| DEFAULT_AHR_REFERENCE = _ASSETS_DIR / "ahr_reference_endpoint_details.csv" |
|
|
|
|
| def format_number(value: float) -> str: |
| if abs(value) < 1e-15: |
| value = 0.0 |
| return f"{value:.12g}" |
|
|
|
|
| def format_value(value: Any) -> str: |
| if value is None: |
| return "" |
| if isinstance(value, str): |
| return value |
| if isinstance(value, (int, float)): |
| return format_number(float(value)) |
| return str(value) |
|
|
|
|
| def write_csv(path: Path, rows: list[dict[str, Any]], columns: list[str]) -> None: |
| with path.open("w", newline="", encoding="utf-8") as f: |
| writer = csv.DictWriter(f, fieldnames=columns) |
| writer.writeheader() |
| for row in rows: |
| writer.writerow({col: format_value(row.get(col)) for col in columns}) |
|
|
|
|
| def _emit_error(message: str) -> int: |
| print(json.dumps({"ok": False, "error": message}, ensure_ascii=False)) |
| return 1 |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser( |
| description="iHARP ER/AhR RQmodified calculation entry point (self-contained)" |
| ) |
| parser.add_argument("--pathway", required=True, choices=["er", "ahr"], help="pathway (er or ahr)") |
| parser.add_argument("--input", required=True, help="endpoint details CSV path") |
| parser.add_argument("--reference", help="reference endpoint CSV path; defaults to the bundled assets/ file") |
| parser.add_argument("--output-dir", default=".", help="output directory (default current directory)") |
| parser.add_argument("--pnec-e2", type=float, default=er_rqmodified.DEFAULT_PNEC_E2) |
| parser.add_argument("--pnec-4oht", type=float, default=er_rqmodified.DEFAULT_PNEC_4OHT) |
| parser.add_argument("--pnec-k", type=float, default=ahr_rqmodified.DEFAULT_PNEC_K) |
| args = parser.parse_args() |
|
|
| input_path = Path(args.input) |
| if not input_path.is_file(): |
| return _emit_error(f"input file not found: {input_path}") |
|
|
| if args.reference: |
| reference_path = Path(args.reference) |
| elif args.pathway == "er": |
| reference_path = DEFAULT_ER_REFERENCE |
| else: |
| reference_path = DEFAULT_AHR_REFERENCE |
| if not reference_path.is_file(): |
| return _emit_error(f"reference file not found: {reference_path}") |
|
|
| output_dir = Path(args.output_dir) |
| output_dir.mkdir(parents=True, exist_ok=True) |
|
|
| if args.pathway == "er": |
| rows, warnings, totals = er_rqmodified.calculate_rqmodified( |
| er_rqmodified.read_csv(input_path), |
| er_rqmodified.read_csv(reference_path), |
| pnec_e2=args.pnec_e2, |
| pnec_4oht=args.pnec_4oht, |
| ) |
| detail_columns = er_rqmodified.DETAIL_COLUMNS |
| summary_columns = er_rqmodified.SUMMARY_COLUMNS |
| else: |
| rows, warnings, totals = ahr_rqmodified.calculate_ahrrqmodified( |
| ahr_rqmodified.read_csv(input_path), |
| ahr_rqmodified.read_csv(reference_path), |
| pnec_k=args.pnec_k, |
| ) |
| detail_columns = ahr_rqmodified.DETAIL_COLUMNS |
| summary_columns = ahr_rqmodified.SUMMARY_COLUMNS |
|
|
| detail_path = output_dir / f"{args.pathway}_detail.csv" |
| summary_path = output_dir / f"{args.pathway}_summary.csv" |
| warning_path = output_dir / f"{args.pathway}_warnings.txt" |
|
|
| write_csv(detail_path, rows, detail_columns) |
| write_csv(summary_path, [{col: totals.get(col) for col in summary_columns}], summary_columns) |
| warning_path.write_text("\n".join(warnings) + ("\n" if warnings else ""), encoding="utf-8") |
|
|
| summary_out = { |
| "ok": True, |
| "pathway": args.pathway.upper(), |
| "rule_version": RULE_VERSION, |
| "files": { |
| "detail": str(detail_path), |
| "summary": str(summary_path), |
| "warnings": str(warning_path), |
| }, |
| "totals": totals, |
| "warning_count": len(warnings), |
| } |
| print(json.dumps(summary_out, ensure_ascii=False, indent=2)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
