#!/usr/bin/env python3 """MASHLipidDroplet — CLI entry point. End-to-end pipeline: 1. Load LD-object table + plate map (or generate synthetic). 2. Per-cell aggregation, subtype classification. 3. Per-well summary, co-localization, PLIN coat shift. 4. Optional MOA fingerprint matching. 5. Optional 4PL dose-response + IC50/EC50 fit. 6. Korean/English PDF report + figure-ready PNG + CSV exports. """ from __future__ import annotations import argparse import os import sys import json from datetime import datetime # Project lib import HERE = os.path.dirname(os.path.abspath(__file__)) if HERE not in sys.path: sys.path.insert(0, HERE) import pandas as pd # noqa: E402 from lib import schemas # noqa: E402 from lib import quant # noqa: E402 from lib import coloc # noqa: E402 from lib import dose_response # noqa: E402 from lib import moa # noqa: E402 from lib import report # noqa: E402 from lib import synth # noqa: E402 def _info(msg: str) -> None: print(f"[mash-ld] {msg}") def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser( prog="mash-lipid-droplet", description=( "MASHLipidDroplet — hepatocyte LD imaging quantification + drug " "screening MVP (research / reference use only)." ), ) p.add_argument("--demo", action="store_true", help="Generate synthetic data and run end-to-end.") p.add_argument("--ld-table", type=str, default=None, help="Per-LD object CSV path.") p.add_argument("--plate-map", type=str, default=None, help="Plate map CSV path (well, drug, dose_uM, replicate).") p.add_argument("--cell-meta", type=str, default=None, help="Cell-meta CSV path (optional).") p.add_argument("--reference-moa", type=str, default=None, help="Reference MOA fingerprint CSV (optional).") p.add_argument("--report-dir", type=str, default=os.path.join(HERE, "output"), help="Output dir.") p.add_argument("--data-dir", type=str, default=os.path.join(HERE, "data"), help="Data dir for --demo seed files.") p.add_argument("--moa-match", action="store_true", help="Run MOA fingerprint matching.") p.add_argument("--ic50", action="store_true", help="Run 4PL dose-response fit + IC50/EC50.") p.add_argument("--language", choices=["ko", "en"], default="ko") p.add_argument("--seed", type=int, default=7) return p.parse_args() def _ensure_dirs(*paths: str) -> None: for path in paths: os.makedirs(path, exist_ok=True) def _load_input(args: argparse.Namespace): """Return (ld_df, plate_df, cell_df, ref_df). Generates demo if --demo.""" if args.demo: _info("--demo: generating synthetic LD-object table…") paths = synth.write_synthetic_to(args.data_dir, seed=args.seed) ld = pd.read_csv(paths["ld_objects"]) plate = pd.read_csv(paths["plate_map"]) cell = pd.read_csv(paths["cell_meta"]) ref = pd.read_csv(paths["reference_moa"]) return ld, plate, cell, ref if not args.ld_table: raise SystemExit("--ld-table is required unless --demo is set") ld = pd.read_csv(args.ld_table) plate = pd.read_csv(args.plate_map) if args.plate_map else pd.DataFrame() cell = pd.read_csv(args.cell_meta) if args.cell_meta else pd.DataFrame() ref = pd.read_csv(args.reference_moa) if args.reference_moa else pd.DataFrame() return ld, plate, cell, ref def main() -> int: args = parse_args() _ensure_dirs(args.report_dir, args.data_dir) ld_df, plate_df, cell_df, ref_df = _load_input(args) # Validation problems = schemas.validate_ld_table(ld_df) for p in problems: _info(f"WARN: {p}") if not plate_df.empty: for p in schemas.validate_plate_map(plate_df): _info(f"WARN: {p}") n_lds = len(ld_df) n_cells = ld_df[["well", "cell_id"]].drop_duplicates().shape[0] if n_lds else 0 n_wells = ld_df["well"].nunique() if n_lds else 0 _info(f"loaded: {n_lds} LD objects across {n_cells} cells / {n_wells} wells") # Per-cell + per-well per_cell = quant.per_cell_summary(ld_df) per_well = quant.per_well_summary(per_cell, plate_df if not plate_df.empty else None) # Co-loc + PLIN coloc_well = coloc.per_well_coloc(ld_df) plin_well = coloc.plin_coat_composition(ld_df) vehicle_wells: list[str] = [] if not plate_df.empty and "drug" in plate_df.columns: vehicle_wells = plate_df.loc[plate_df["drug"] == "vehicle", "well"].tolist() plin_well_fc = coloc.detect_coat_shift(plin_well, vehicle_wells) # MOA matching moa_long = pd.DataFrame() moa_top = pd.DataFrame() if args.moa_match and not ref_df.empty: _info("MOA: fingerprint matching (cosine similarity)…") moa_long = moa.match_moa(per_well, ref_df) moa_top = moa_long[moa_long.get("top1", False)] if not moa_long.empty else moa_long # IC50 / EC50 fits fit_table = pd.DataFrame() if args.ic50 and "drug" in per_well.columns and "dose_uM" in per_well.columns: _info("IC50: 4PL fits on key responses…") responses = ["ld_count_per_cell", "total_LD_area_per_cell", "macro_pct", "manders_m1"] rows = [] for r in responses: if r in per_well.columns: tab = dose_response.fit_per_drug(per_well, r) if not tab.empty: rows.append(tab) if rows: fit_table = pd.concat(rows, ignore_index=True) # Persist CSVs per_cell.to_csv(os.path.join(args.report_dir, "per_cell_summary.csv"), index=False) per_well.to_csv(os.path.join(args.report_dir, "per_well_summary.csv"), index=False) coloc_well.to_csv(os.path.join(args.report_dir, "per_well_coloc.csv"), index=False) plin_well_fc.to_csv(os.path.join(args.report_dir, "per_well_plin_coat.csv"), index=False) if not moa_long.empty: moa_long.to_csv(os.path.join(args.report_dir, "moa_match_long.csv"), index=False) moa_top.to_csv(os.path.join(args.report_dir, "moa_match_top1.csv"), index=False) if not fit_table.empty: fit_table.to_csv(os.path.join(args.report_dir, "dose_response_fits.csv"), index=False) # Figures figs = [] f1 = report.write_subtype_heatmap_png(per_well, os.path.join(args.report_dir, "fig_subtype_heatmap.png")) if f1: figs.append(f1) f2 = report.write_dose_response_png( per_well, fit_table, "ld_count_per_cell", os.path.join(args.report_dir, "fig_dose_response_ld_count.png") ) if f2: figs.append(f2) f3 = report.write_dose_response_png( per_well, fit_table, "total_LD_area_per_cell", os.path.join(args.report_dir, "fig_dose_response_total_area.png"), ) if f3: figs.append(f3) f4 = report.write_moa_png(moa_long, os.path.join(args.report_dir, "fig_moa_top1.png")) if f4: figs.append(f4) f5 = report.write_plin_shift_png( plin_well_fc, per_well, os.path.join(args.report_dir, "fig_plin_coat_shift.png") ) if f5: figs.append(f5) # Summary stats summary_stats = { "분석 일시 / Generated": datetime.now().strftime("%Y-%m-%d %H:%M"), "분석 well 수 / wells": n_wells, "총 세포 수 / cells": n_cells, "총 LD object 수 / LDs": n_lds, "약물 수 / drugs": int(plate_df["drug"].nunique()) if "drug" in plate_df.columns and not plate_df.empty else 0, "MOA matching": "ON" if args.moa_match else "off", "IC50 4PL fit": "ON" if args.ic50 else "off", } pdf_path = os.path.join(args.report_dir, "MASHLipidDroplet_report.pdf") report.write_pdf_report( out_pdf=pdf_path, summary_stats=summary_stats, fit_table=fit_table, moa_top=moa_top, figure_paths=figs, language=args.language, ) _info(f"PDF written: {pdf_path}") # Also dump a JSON manifest manifest = { "summary": summary_stats, "outputs": sorted(os.listdir(args.report_dir)), "language": args.language, } with open(os.path.join(args.report_dir, "manifest.json"), "w") as fh: json.dump(manifest, fh, ensure_ascii=False, indent=2) _info("done.") return 0 if __name__ == "__main__": raise SystemExit(main())