"""Build the per-plot modeling dataset from per-tuber morphometrics and ground truth. QC-filter the per-tuber dataframe, aggregate to (trial, Plot_std), merge with the cleaned ground truth, and compute the regression target obs_sum_B_A1_A2_OV. Usage: uv run python scripts/build_plot_model_dataset.py \\ --tubers outputs/df_class_real_scale_ALL.csv \\ --out outputs/df_model_yield_by_plot.csv """ from __future__ import annotations import argparse from pathlib import Path import pandas as pd from potatopheno.aggregation import aggregate_by_plot from potatopheno.filtering import ( ADC_QC_DROP_PLOTS, derive_axis_ratio_pca, drop_qc_plots, filter_axis_ratio, ) from potatopheno.harvest import add_plot_std, merge_with_ground_truth def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) p.add_argument( "--tubers", type=Path, required=True, help="Per-tuber morphometrics CSV produced by extract_morphometrics.py.", ) p.add_argument( "--gt", type=Path, required=True, help="Cleaned ground-truth CSV.", ) p.add_argument( "--out", type=Path, required=True, help="Path of the plot-level model CSV to write.", ) p.add_argument( "--qc-drop-plots", type=lambda s: tuple(t.strip() for t in s.split(",") if t.strip()), default=ADC_QC_DROP_PLOTS, help="Comma-separated list of Plot IDs (image_id[-8:]) to drop as QC. " f"Default: {','.join(ADC_QC_DROP_PLOTS)}.", ) p.add_argument( "--no-ar-filter", action="store_true", help="Skip the axis-ratio-PCA filter entirely.", ) return p.parse_args() def main() -> None: args = _parse_args() df_tubers = pd.read_csv(args.tubers) n0 = len(df_tubers) df = drop_qc_plots(df_tubers, drop_plots=args.qc_drop_plots) n1 = len(df) df = derive_axis_ratio_pca(df) if not args.no_ar_filter: df = filter_axis_ratio(df) n2 = len(df) df = add_plot_std(df) df_agg = aggregate_by_plot(df) df_gt = pd.read_csv(args.gt) df_model = merge_with_ground_truth(df_agg, df_gt) args.out.parent.mkdir(parents=True, exist_ok=True) df_model.to_csv(args.out, index=False) print(f"qc_drop_plots={list(args.qc_drop_plots)} ar_filter={'OFF' if args.no_ar_filter else 'ON'}") print( f"tubers: {n0} → after qc-drop: {n1} → " f"after AR filter: {n2}" ) print( f"plots: {df_model['Plot_std'].nunique()} across " f"{df_model['trial'].nunique()} trials" ) print(f"[DONE] wrote {len(df_model)} rows → {args.out}") if __name__ == "__main__": main()