Spaces:
Sleeping
Sleeping
| """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() | |