| """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 |
|
|
|
|
| |
| @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"), |
| ) |
|
|
|
|
| |
| |
| |
| _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()), |
| } |
|
|
