Spaces:
Sleeping
Sleeping
File size: 2,787 Bytes
3ecb175 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | """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()
|