"""Manual ground-truth observations log for the 2026 season. Replaces what LiCor measurements would have provided. One row per manual measurement event (SPAD chlorophyll, refractometer brix, lab anthocyanin, yield, pruning weight, etc.), keyed by `(date, row, position, metric)` so multiple metrics on the same day/row append cleanly. Append-only, version-controlled, schema-validated. CSV (not parquet) so agronomists can edit in Excel or Google Sheets if needed. Usage:: from src.season_ground_truth import ( Observation, append_observation, load_observations, ) append_observation(Observation( date="2026-06-10", row=502, position="north", phenology="fruit-set", observer="ES", spad=42.3, notes="post-irrigation", )) df = load_observations(row=502) # filter to one row df = load_observations(date_from="2026-08-01") # filter by date range """ from __future__ import annotations import csv from dataclasses import asdict, dataclass, field, fields from datetime import date, datetime, timezone from pathlib import Path from typing import Iterable, List, Optional from config import settings # All optional float / int columns default to None; only date+row are mandatory. @dataclass class Observation: """Single ground-truth measurement event. Mandatory: date ISO date string (YYYY-MM-DD) row int — vineyard row id (501/502/503/504/509/202) Recommended: position cardinal/composite (north/south/center-east/...) phenology free text BBCH stage or label observer initials of the agronomist Per-metric (any subset; leave others None): spad SPAD-502 chlorophyll units brix °Brix from handheld refractometer ta_g_l titratable acidity, g/L (lab) ph must pH (lab) yield_kg kg / vine, harvest day berry_count berries per cluster (sample) berry_weight_g avg single berry weight, g cluster_count clusters per vine anthocyanin_mg_g mg / g skin, lab assay phenolics_mg_g total phenolics, mg / g sunburn_pct % of clusters with visible damage pruning_weight_kg dormant pruning, kg / vine lai leaf area index (LAI-2200 / hemiphoto) notes free text photo_url optional reference to a photo (local path or URL) """ date: str row: int position: Optional[str] = None phenology: Optional[str] = None observer: Optional[str] = None spad: Optional[float] = None brix: Optional[float] = None ta_g_l: Optional[float] = None ph: Optional[float] = None yield_kg: Optional[float] = None berry_count: Optional[int] = None berry_weight_g: Optional[float] = None cluster_count: Optional[int] = None anthocyanin_mg_g: Optional[float] = None phenolics_mg_g: Optional[float] = None sunburn_pct: Optional[float] = None pruning_weight_kg: Optional[float] = None lai: Optional[float] = None notes: Optional[str] = None photo_url: Optional[str] = None recorded_at: str = field( default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"), ) # --------------------------------------------------------------------------- # CSV column order (stable, append-friendly) # --------------------------------------------------------------------------- _COLUMNS: List[str] = [f.name for f in fields(Observation)] def _validate(obs: Observation) -> None: """Cheap structural checks. Range / sanity checks live in load.""" try: date.fromisoformat(obs.date) except (TypeError, ValueError) as e: raise ValueError(f"Observation.date must be ISO YYYY-MM-DD: {obs.date!r}") from e valid_rows = {202, 501, 502, 503, 504, 509} if obs.row not in valid_rows: raise ValueError(f"Observation.row {obs.row} not in {sorted(valid_rows)}") def append_observation(obs: Observation, path: Optional[Path] = None) -> Path: """Append a single Observation to the CSV. Creates the file + header on first call. Returns the path written to. """ _validate(obs) path = path or settings.MANUAL_OBSERVATIONS_PATH path.parent.mkdir(parents=True, exist_ok=True) write_header = not path.exists() or path.stat().st_size == 0 with path.open("a", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=_COLUMNS, extrasaction="ignore") if write_header: writer.writeheader() writer.writerow({k: v if v is not None else "" for k, v in asdict(obs).items()}) return path def append_many(observations: Iterable[Observation], path: Optional[Path] = None) -> Path: """Bulk-append helper. Same validation per-row.""" path = path or settings.MANUAL_OBSERVATIONS_PATH for obs in observations: append_observation(obs, path) return path def load_observations( path: Optional[Path] = None, *, row: Optional[int] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, ): """Load observations into a pandas DataFrame. Optional filters. Returns an empty DataFrame (typed columns) if the file doesn't exist or is empty. """ import pandas as pd path = path or settings.MANUAL_OBSERVATIONS_PATH if not path.exists() or path.stat().st_size == 0: return pd.DataFrame(columns=_COLUMNS) df = pd.read_csv(path) df["date"] = pd.to_datetime(df["date"]).dt.date if row is not None: df = df[df["row"] == row] if date_from is not None: df = df[df["date"] >= date.fromisoformat(date_from)] if date_to is not None: df = df[df["date"] <= date.fromisoformat(date_to)] return df.reset_index(drop=True) def summary(path: Optional[Path] = None) -> dict: """Quick stats: total rows, rows per metric, date range.""" df = load_observations(path) if df.empty: return {"rows": 0, "per_metric": {}, "date_range": None} counts = { col: int(df[col].notna().sum()) for col in _COLUMNS if col not in {"date", "row", "position", "phenology", "observer", "notes", "photo_url", "recorded_at"} } return { "rows": int(len(df)), "per_metric": {k: v for k, v in counts.items() if v > 0}, "date_range": (str(df["date"].min()), str(df["date"].max())), "rows_observed": sorted(df["row"].unique().tolist()), }