File size: 5,011 Bytes
13fc29d | 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """proxy_a_score — composite photosynthesis-health metric, 0..1.
LiCor (direct A measurement) is not available this season, so the control
loop and chatbot need a single scalar that says "how healthy is the canopy
right now" from the sensors we already collect. This module produces that
scalar by combining 5 already-streamed metrics on Crop_2Soil devices:
- PRI (Photochemical Reflectance Index — light-use efficiency)
- PSRI (Plant Senescence Reflectance Index — inverted)
- leaf-air ΔT (transpiration cooling — inverted)
- shallow soil moisture (water availability)
- NDVI (greenness baseline)
Each metric is mapped linearly from its `stress`→`healthy` bound into
[0, 1] (clipped), then combined as a weighted average. Weights and bounds
live in `config/settings.py` (`PROXY_A_WEIGHTS`, `PROXY_A_BOUNDS`) so they
can be tuned after the null-intervention baseline period without code change.
Higher score = healthier / more productive canopy. The control loop should
prefer interventions that *don't* drop the score by more than ε.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, Optional
from config import settings
@dataclass
class ProxyAScore:
"""Composite photosynthesis-health metric, plus per-term breakdown.
Attributes
----------
score : float
Weighted average of the per-metric sub-scores, in [0, 1].
NaN if no metrics were available.
contributions : dict[str, float]
Per-metric sub-score in [0, 1] (only metrics that had data).
weights_used : dict[str, float]
Per-metric weight actually applied (renormalised over available
metrics so missing inputs don't bias the score downward).
"""
score: float
contributions: Dict[str, float]
weights_used: Dict[str, float]
def _linear_score(value: float, stress: float, healthy: float) -> float:
"""Map a raw measurement into [0, 1] given its stress/healthy bounds.
Works whether `healthy > stress` (PRI, NDVI, soil — higher is better)
or `healthy < stress` (PSRI, ΔT — lower is better).
"""
if stress == healthy:
return 0.5 # degenerate config, neutral
raw = (value - stress) / (healthy - stress)
return max(0.0, min(1.0, raw))
def compute_proxy_a(
pri: Optional[float] = None,
psri: Optional[float] = None,
delta_t: Optional[float] = None,
soil_moisture_pct: Optional[float] = None,
ndvi: Optional[float] = None,
) -> ProxyAScore:
"""Compute the proxy-A score from any subset of the 5 input metrics.
Missing metrics are dropped and the weights for the remaining metrics
are renormalised so they sum to 1. If all metrics are missing, the
score is NaN.
"""
raw_inputs: Dict[str, Optional[float]] = {
"pri": pri,
"psri": psri,
"delta_t": delta_t,
"soil": soil_moisture_pct,
"ndvi": ndvi,
}
contributions: Dict[str, float] = {}
weights_used: Dict[str, float] = {}
for key, val in raw_inputs.items():
if val is None:
continue
b = settings.PROXY_A_BOUNDS.get(key)
w = settings.PROXY_A_WEIGHTS.get(key, 0.0)
if b is None or w <= 0:
continue
contributions[key] = _linear_score(val, b["stress"], b["healthy"])
weights_used[key] = w
if not weights_used:
return ProxyAScore(score=float("nan"), contributions={}, weights_used={})
total_w = sum(weights_used.values())
renormalised = {k: w / total_w for k, w in weights_used.items()}
score = sum(contributions[k] * renormalised[k] for k in contributions)
return ProxyAScore(
score=round(score, 3),
contributions={k: round(v, 3) for k, v in contributions.items()},
weights_used=renormalised,
)
def compute_from_snapshot(snapshot: dict) -> ProxyAScore:
"""Convenience: pull the 5 inputs from a VineSnapshot.to_dict()."""
return compute_proxy_a(
pri=snapshot.get("treatment_crop_pri"),
psri=snapshot.get("treatment_crop_psri"),
delta_t=snapshot.get("treatment_air_leaf_delta_t"),
soil_moisture_pct=snapshot.get("treatment_soil_moisture_pct"),
ndvi=snapshot.get("treatment_crop_ndvi"),
)
def slot_cost(
*,
energy_value_ils: float,
proxy_a: ProxyAScore,
proxy_a_floor: float = 0.5,
proxy_a_weight: float = 5.0,
) -> float:
"""Slot-level cost for the tracker control loop.
Combines instantaneous energy revenue with a penalty for any drop of
`proxy_a.score` below `proxy_a_floor`. Higher = better.
cost = energy_value_ils − proxy_a_weight × max(0, floor − score)
Tune `proxy_a_weight` from the calibration period so that the penalty
at the 95th-percentile stress reading equals roughly half a mid-day
hour of energy revenue.
"""
shortfall = max(0.0, proxy_a_floor - proxy_a.score)
return round(energy_value_ils - proxy_a_weight * shortfall, 3)
|