Spaces:

SolarWine
/

api

Sleeping

api / src /baseline_predictor.py

Eli Safra

Deploy SolarWine API (FastAPI + Docker, port 7860)

938949f 8 months ago

8.25 kB

"""
BaselinePredictor: hybrid FvCB + ML photosynthesis baseline for day-ahead planning.
Provides a single ``predict_day()`` method that:
  1. Runs FvCB (Farquhar–Greer–Weedon) for each slot using forecast weather
  2. Optionally runs a trained ML model for the same slots
  3. Uses the RoutingAgent's rule-based logic to pick the better prediction per slot
  4. Returns a 96-slot profile of predicted photosynthesis rate A (µmol CO₂ m⁻² s⁻¹)
This feeds into the DayAheadPlanner to estimate crop value for each slot,
replacing the current temperature-only heuristic with an actual photosynthesis
prediction that captures the Rubisco transition more accurately.
"""

from __future__ import annotations

import logging
import math
from datetime import date
from typing import List, Optional

import numpy as np

from config.settings import SEMILLON_TRANSITION_TEMP_C

logger = logging.getLogger(__name__)


class BaselinePredictor:
    """Hybrid FvCB + ML photosynthesis prediction for day-ahead planning.
    Parameters
    ----------
    fvcb_model : FarquharModel, optional
        Lazy-initialised if not provided.
    ml_predictor : PhotosynthesisPredictor, optional
        Trained ML model. If None, FvCB-only mode is used.
    routing_agent : RoutingAgent, optional
        Model router for per-slot FvCB/ML selection.
        If None, uses rule-based routing only (no API calls).
    """

    def __init__(
        self,
        fvcb_model=None,
        ml_predictor=None,
        routing_agent=None,
    ):
        self._fvcb = fvcb_model
        self._ml = ml_predictor
        self._router = routing_agent

    @property
    def fvcb(self):
        if self._fvcb is None:
            from src.models.farquhar_model import FarquharModel
            self._fvcb = FarquharModel()
        return self._fvcb

    # ------------------------------------------------------------------
    # Main API
    # ------------------------------------------------------------------

    def predict_day(
        self,
        forecast_temps: List[float],
        forecast_ghi: List[float],
        co2_ppm: float = 400.0,
        rh_pct: float = 40.0,
    ) -> List[float]:
        """Predict photosynthesis rate A for each 15-min slot.
        Parameters
        ----------
        forecast_temps : list of 96 floats
            Forecast air temperature (°C) per slot.
        forecast_ghi : list of 96 floats
            Forecast GHI (W/m²) per slot.
        co2_ppm : float
            Atmospheric CO₂ concentration (default 400 ppm).
        rh_pct : float
            Relative humidity (%) for VPD estimation (default 40%).
        Returns
        -------
        list of 96 floats
            Predicted net photosynthesis A (µmol CO₂ m⁻² s⁻¹) per slot.
            0.0 for nighttime slots.
        """
        assert len(forecast_temps) == 96 and len(forecast_ghi) == 96

        # FvCB predictions for all 96 slots
        fvcb_predictions = self._predict_fvcb(
            forecast_temps, forecast_ghi, co2_ppm, rh_pct,
        )

        # If no ML model, return FvCB-only
        if self._ml is None:
            return fvcb_predictions

        # ML predictions for all 96 slots
        ml_predictions = self._predict_ml(forecast_temps, forecast_ghi)

        # Route each slot
        predictions = self._route_predictions(
            forecast_temps, forecast_ghi,
            fvcb_predictions, ml_predictions,
        )

        return predictions

    # ------------------------------------------------------------------
    # FvCB predictions
    # ------------------------------------------------------------------

    def _predict_fvcb(
        self,
        temps: List[float],
        ghis: List[float],
        co2_ppm: float,
        rh_pct: float,
    ) -> List[float]:
        """Run FvCB for each slot. Returns 96 A values."""
        predictions = []
        for i in range(96):
            temp = temps[i]
            ghi = ghis[i]

            # Nighttime or negligible light
            if ghi < 50:
                predictions.append(0.0)
                continue

            # Estimate PAR from GHI (roughly 2× conversion for photosynthetically active)
            par = ghi * 2.0

            # Estimate Tleaf from Tair (proxy: +2°C under sun)
            tleaf = temp + 2.0

            # Estimate VPD from temperature and RH
            vpd = self._estimate_vpd(temp, rh_pct)

            try:
                result = self.fvcb.calc_photosynthesis_semillon(
                    PAR=par,
                    Tleaf=tleaf,
                    CO2=co2_ppm,
                    VPD=vpd,
                    Tair=temp,
                )
                # Returns (A, limiting_state, shading_helps)
                A = result[0] if isinstance(result, tuple) else result
                predictions.append(max(0.0, float(A)))
            except Exception as exc:
                logger.debug("FvCB failed at slot %d: %s", i, exc)
                predictions.append(0.0)

        return predictions

    @staticmethod
    def _estimate_vpd(tair_c: float, rh_pct: float) -> float:
        """Estimate VPD (kPa) from air temperature and relative humidity."""
        # Tetens formula for saturated vapor pressure
        es = 0.6108 * math.exp(17.27 * tair_c / (tair_c + 237.3))
        ea = es * rh_pct / 100.0
        return max(0.0, es - ea)

    # ------------------------------------------------------------------
    # ML predictions
    # ------------------------------------------------------------------

    def _predict_ml(
        self,
        temps: List[float],
        ghis: List[float],
    ) -> List[float]:
        """Run ML model for each slot. Returns 96 A values."""
        if self._ml is None:
            return [0.0] * 96

        try:
            import pandas as pd

            # Build feature DataFrame matching ML model's expected features
            hours = [i * 0.25 for i in range(96)]
            df = pd.DataFrame({
                "air_temperature_c": temps,
                "ghi_w_m2": ghis,
                "hour": [int(h) for h in hours],
                "minute": [int((h % 1) * 60) for h in hours],
            })

            # Try prediction with the best model
            best_model = None
            best_mae = float("inf")
            for name, result in self._ml.results.items():
                if result.get("mae", float("inf")) < best_mae:
                    best_mae = result["mae"]
                    best_model = name

            if best_model and best_model in self._ml.models:
                model = self._ml.models[best_model]
                # Use whatever features the model was trained on
                feature_cols = [c for c in df.columns if c in getattr(model, "feature_names_in_", df.columns)]
                if feature_cols:
                    preds = model.predict(df[feature_cols])
                    return [max(0.0, float(p)) for p in preds]

        except Exception as exc:
            logger.warning("ML prediction failed: %s", exc)

        return [0.0] * 96

    # ------------------------------------------------------------------
    # Routing
    # ------------------------------------------------------------------

    def _route_predictions(
        self,
        temps: List[float],
        ghis: List[float],
        fvcb_preds: List[float],
        ml_preds: List[float],
    ) -> List[float]:
        """Pick FvCB or ML per slot using routing logic."""
        from src.chatbot.routing_agent import RoutingAgent

        predictions = []
        for i in range(96):
            telemetry = {
                "temp_c": temps[i],
                "ghi_w_m2": ghis[i],
                "hour": i // 4,
            }

            # Use rule-based routing only (no API calls for batch prediction)
            choice = RoutingAgent._rule_based_route(telemetry)
            if choice is None:
                # Transition zone: weight FvCB 60% / ML 40% as compromise
                a = 0.6 * fvcb_preds[i] + 0.4 * ml_preds[i]
            elif choice == "ml":
                a = ml_preds[i]
            else:
                a = fvcb_preds[i]

            predictions.append(a)

        return predictions

	"""
	BaselinePredictor: hybrid FvCB + ML photosynthesis baseline for day-ahead planning.

	Provides a single ``predict_day()`` method that:
	1. Runs FvCB (Farquhar–Greer–Weedon) for each slot using forecast weather
	2. Optionally runs a trained ML model for the same slots
	3. Uses the RoutingAgent's rule-based logic to pick the better prediction per slot
	4. Returns a 96-slot profile of predicted photosynthesis rate A (µmol CO₂ m⁻² s⁻¹)

	This feeds into the DayAheadPlanner to estimate crop value for each slot,
	replacing the current temperature-only heuristic with an actual photosynthesis
	prediction that captures the Rubisco transition more accurately.
	"""

	from __future__ import annotations

	import logging
	import math
	from datetime import date
	from typing import List, Optional

	import numpy as np

	from config.settings import SEMILLON_TRANSITION_TEMP_C

	logger = logging.getLogger(__name__)


	class BaselinePredictor:
	"""Hybrid FvCB + ML photosynthesis prediction for day-ahead planning.

	Parameters
	----------
	fvcb_model : FarquharModel, optional
	Lazy-initialised if not provided.
	ml_predictor : PhotosynthesisPredictor, optional
	Trained ML model. If None, FvCB-only mode is used.
	routing_agent : RoutingAgent, optional
	Model router for per-slot FvCB/ML selection.
	If None, uses rule-based routing only (no API calls).
	"""

	def __init__(
	self,
	fvcb_model=None,
	ml_predictor=None,
	routing_agent=None,
	):
	self._fvcb = fvcb_model
	self._ml = ml_predictor
	self._router = routing_agent

	@property
	def fvcb(self):
	if self._fvcb is None:
	from src.models.farquhar_model import FarquharModel
	self._fvcb = FarquharModel()
	return self._fvcb

	# ------------------------------------------------------------------
	# Main API
	# ------------------------------------------------------------------

	def predict_day(
	self,
	forecast_temps: List[float],
	forecast_ghi: List[float],
	co2_ppm: float = 400.0,
	rh_pct: float = 40.0,
	) -> List[float]:
	"""Predict photosynthesis rate A for each 15-min slot.

	Parameters
	----------
	forecast_temps : list of 96 floats
	Forecast air temperature (°C) per slot.
	forecast_ghi : list of 96 floats
	Forecast GHI (W/m²) per slot.
	co2_ppm : float
	Atmospheric CO₂ concentration (default 400 ppm).
	rh_pct : float
	Relative humidity (%) for VPD estimation (default 40%).

	Returns
	-------
	list of 96 floats
	Predicted net photosynthesis A (µmol CO₂ m⁻² s⁻¹) per slot.
	0.0 for nighttime slots.
	"""
	assert len(forecast_temps) == 96 and len(forecast_ghi) == 96

	# FvCB predictions for all 96 slots
	fvcb_predictions = self._predict_fvcb(
	forecast_temps, forecast_ghi, co2_ppm, rh_pct,
	)

	# If no ML model, return FvCB-only
	if self._ml is None:
	return fvcb_predictions

	# ML predictions for all 96 slots
	ml_predictions = self._predict_ml(forecast_temps, forecast_ghi)

	# Route each slot
	predictions = self._route_predictions(
	forecast_temps, forecast_ghi,
	fvcb_predictions, ml_predictions,
	)

	return predictions

	# ------------------------------------------------------------------
	# FvCB predictions
	# ------------------------------------------------------------------

	def _predict_fvcb(
	self,
	temps: List[float],
	ghis: List[float],
	co2_ppm: float,
	rh_pct: float,
	) -> List[float]:
	"""Run FvCB for each slot. Returns 96 A values."""
	predictions = []
	for i in range(96):
	temp = temps[i]
	ghi = ghis[i]

	# Nighttime or negligible light
	if ghi < 50:
	predictions.append(0.0)
	continue

	# Estimate PAR from GHI (roughly 2× conversion for photosynthetically active)
	par = ghi * 2.0

	# Estimate Tleaf from Tair (proxy: +2°C under sun)
	tleaf = temp + 2.0

	# Estimate VPD from temperature and RH
	vpd = self._estimate_vpd(temp, rh_pct)

	try:
	result = self.fvcb.calc_photosynthesis_semillon(
	PAR=par,
	Tleaf=tleaf,
	CO2=co2_ppm,
	VPD=vpd,
	Tair=temp,
	)
	# Returns (A, limiting_state, shading_helps)
	A = result[0] if isinstance(result, tuple) else result
	predictions.append(max(0.0, float(A)))
	except Exception as exc:
	logger.debug("FvCB failed at slot %d: %s", i, exc)
	predictions.append(0.0)

	return predictions

	@staticmethod
	def _estimate_vpd(tair_c: float, rh_pct: float) -> float:
	"""Estimate VPD (kPa) from air temperature and relative humidity."""
	# Tetens formula for saturated vapor pressure
	es = 0.6108 * math.exp(17.27 * tair_c / (tair_c + 237.3))
	ea = es * rh_pct / 100.0
	return max(0.0, es - ea)

	# ------------------------------------------------------------------
	# ML predictions
	# ------------------------------------------------------------------

	def _predict_ml(
	self,
	temps: List[float],
	ghis: List[float],
	) -> List[float]:
	"""Run ML model for each slot. Returns 96 A values."""
	if self._ml is None:
	return [0.0] * 96

	try:
	import pandas as pd

	# Build feature DataFrame matching ML model's expected features
	hours = [i * 0.25 for i in range(96)]
	df = pd.DataFrame({
	"air_temperature_c": temps,
	"ghi_w_m2": ghis,
	"hour": [int(h) for h in hours],
	"minute": [int((h % 1) * 60) for h in hours],
	})

	# Try prediction with the best model
	best_model = None
	best_mae = float("inf")
	for name, result in self._ml.results.items():
	if result.get("mae", float("inf")) < best_mae:
	best_mae = result["mae"]
	best_model = name

	if best_model and best_model in self._ml.models:
	model = self._ml.models[best_model]
	# Use whatever features the model was trained on
	feature_cols = [c for c in df.columns if c in getattr(model, "feature_names_in_", df.columns)]
	if feature_cols:
	preds = model.predict(df[feature_cols])
	return [max(0.0, float(p)) for p in preds]

	except Exception as exc:
	logger.warning("ML prediction failed: %s", exc)

	return [0.0] * 96

	# ------------------------------------------------------------------
	# Routing
	# ------------------------------------------------------------------

	def _route_predictions(
	self,
	temps: List[float],
	ghis: List[float],
	fvcb_preds: List[float],
	ml_preds: List[float],
	) -> List[float]:
	"""Pick FvCB or ML per slot using routing logic."""
	from src.chatbot.routing_agent import RoutingAgent

	predictions = []
	for i in range(96):
	telemetry = {
	"temp_c": temps[i],
	"ghi_w_m2": ghis[i],
	"hour": i // 4,
	}

	# Use rule-based routing only (no API calls for batch prediction)
	choice = RoutingAgent._rule_based_route(telemetry)
	if choice is None:
	# Transition zone: weight FvCB 60% / ML 40% as compromise
	a = 0.6 * fvcb_preds[i] + 0.4 * ml_preds[i]
	elif choice == "ml":
	a = ml_preds[i]
	else:
	a = fvcb_preds[i]

	predictions.append(a)

	return predictions