| """ |
| ROIService: budget utilisation, intervention statistics, and LER computation. |
| |
| Provides the data layer for the ROI Dashboard tab. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from dataclasses import dataclass, field |
| from datetime import date, datetime, timedelta |
| from typing import Dict, List, Optional |
|
|
| from config.settings import ( |
| MAX_ENERGY_REDUCTION_PCT, |
| TARGET_LER, |
| ) |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| |
| |
| |
|
|
| @dataclass |
| class InterventionStats: |
| """Statistics for a single period (day/week/month).""" |
|
|
| period_label: str |
| total_slots: int = 0 |
| intervention_slots: int = 0 |
| avg_offset_deg: float = 0.0 |
| max_offset_deg: float = 0.0 |
| energy_sacrificed_kwh: float = 0.0 |
| budget_allocated_kwh: float = 0.0 |
| budget_utilisation_pct: float = 0.0 |
|
|
| @property |
| def intervention_rate_pct(self) -> float: |
| if self.total_slots == 0: |
| return 0.0 |
| return self.intervention_slots / self.total_slots * 100 |
|
|
|
|
| @dataclass |
| class LERResult: |
| """Land Equivalent Ratio computation.""" |
|
|
| energy_fraction: float |
| crop_fraction: float |
| ler: float |
| meets_target: bool = False |
|
|
| def summary(self) -> str: |
| status = "MEETS" if self.meets_target else "BELOW" |
| return ( |
| f"LER = {self.ler:.2f} ({status} target {TARGET_LER:.1f}): " |
| f"energy {self.energy_fraction:.2f} + crop {self.crop_fraction:.2f}" |
| ) |
|
|
|
|
| @dataclass |
| class BudgetStatus: |
| """Current budget utilisation snapshot.""" |
|
|
| annual_budget_kwh: float |
| annual_spent_kwh: float |
| annual_remaining_kwh: float |
| monthly_budget_kwh: float = 0.0 |
| monthly_spent_kwh: float = 0.0 |
| weekly_budget_kwh: float = 0.0 |
| weekly_spent_kwh: float = 0.0 |
| daily_budget_kwh: float = 0.0 |
| daily_spent_kwh: float = 0.0 |
|
|
| @property |
| def annual_utilisation_pct(self) -> float: |
| if self.annual_budget_kwh == 0: |
| return 0.0 |
| return self.annual_spent_kwh / self.annual_budget_kwh * 100 |
|
|
| def is_over_budget(self) -> bool: |
| return self.annual_spent_kwh > self.annual_budget_kwh |
|
|
|
|
| |
| |
| |
|
|
| class ROIService: |
| """Compute budget utilisation, intervention stats, and LER from tick logs. |
| |
| Parameters |
| ---------- |
| annual_generation_kwh : float |
| Expected total annual PV generation (kWh). |
| Default: 48 kW × 1800 peak-sun-hours ≈ 86,400 kWh. |
| max_reduction_pct : float |
| Maximum energy sacrifice ceiling (%). |
| """ |
|
|
| def __init__( |
| self, |
| annual_generation_kwh: float = 86_400.0, |
| max_reduction_pct: float = MAX_ENERGY_REDUCTION_PCT, |
| ): |
| self.annual_gen = annual_generation_kwh |
| self.annual_budget = annual_generation_kwh * max_reduction_pct / 100.0 |
| self._tick_log: List[dict] = [] |
|
|
| def load_tick_log(self, tick_log: List[dict]) -> None: |
| """Load a tick log (list of TickResult.to_dict() entries).""" |
| self._tick_log = list(tick_log) |
| logger.info("Loaded %d tick entries", len(self._tick_log)) |
|
|
| |
| |
| |
|
|
| def get_budget_status( |
| self, |
| target_date: Optional[date] = None, |
| ) -> BudgetStatus: |
| """Compute current budget utilisation from the tick log.""" |
| today = target_date or date.today() |
| year = today.year |
|
|
| annual_spent = 0.0 |
| monthly_spent = 0.0 |
| weekly_spent = 0.0 |
| daily_spent = 0.0 |
|
|
| week_start = today - timedelta(days=today.weekday()) |
|
|
| for tick in self._tick_log: |
| cost = tick.get("energy_cost_kwh", 0.0) or 0.0 |
| ts = tick.get("timestamp", "") |
| try: |
| if isinstance(ts, str): |
| tick_date = datetime.fromisoformat(ts).date() |
| elif isinstance(ts, datetime): |
| tick_date = ts.date() |
| else: |
| continue |
| except (ValueError, AttributeError): |
| continue |
|
|
| if tick_date.year == year: |
| annual_spent += cost |
| if tick_date.year == year and tick_date.month == today.month: |
| monthly_spent += cost |
| if tick_date >= week_start: |
| weekly_spent += cost |
| if tick_date == today: |
| daily_spent += cost |
|
|
| return BudgetStatus( |
| annual_budget_kwh=self.annual_budget, |
| annual_spent_kwh=annual_spent, |
| annual_remaining_kwh=max(0, self.annual_budget - annual_spent), |
| monthly_spent_kwh=monthly_spent, |
| weekly_spent_kwh=weekly_spent, |
| daily_spent_kwh=daily_spent, |
| ) |
|
|
| |
| |
| |
|
|
| def compute_intervention_stats( |
| self, |
| start_date: Optional[date] = None, |
| end_date: Optional[date] = None, |
| label: str = "period", |
| ) -> InterventionStats: |
| """Compute intervention statistics for a date range.""" |
| stats = InterventionStats(period_label=label) |
|
|
| for tick in self._tick_log: |
| ts = tick.get("timestamp", "") |
| try: |
| if isinstance(ts, str): |
| tick_date = datetime.fromisoformat(ts).date() |
| elif isinstance(ts, datetime): |
| tick_date = ts.date() |
| else: |
| continue |
| except (ValueError, AttributeError): |
| continue |
|
|
| if start_date and tick_date < start_date: |
| continue |
| if end_date and tick_date > end_date: |
| continue |
|
|
| stats.total_slots += 1 |
| offset = tick.get("plan_offset_deg", 0.0) or 0.0 |
| if offset > 0: |
| stats.intervention_slots += 1 |
| stats.max_offset_deg = max(stats.max_offset_deg, offset) |
|
|
| stats.energy_sacrificed_kwh += tick.get("energy_cost_kwh", 0.0) or 0.0 |
|
|
| if stats.intervention_slots > 0: |
| |
| total_offset = sum( |
| (t.get("plan_offset_deg", 0.0) or 0.0) |
| for t in self._tick_log |
| if (t.get("plan_offset_deg", 0.0) or 0.0) > 0 |
| ) |
| stats.avg_offset_deg = total_offset / stats.intervention_slots |
|
|
| return stats |
|
|
| |
| |
| |
|
|
| def compute_ler( |
| self, |
| actual_energy_kwh: float, |
| mono_energy_kwh: Optional[float] = None, |
| actual_crop_yield: float = 1.0, |
| mono_crop_yield: float = 1.0, |
| ) -> LERResult: |
| """Compute Land Equivalent Ratio. |
| |
| LER = (E_agri / E_mono) + (Y_agri / Y_mono) |
| |
| LER > 1.0 means the combined system is more productive |
| than growing either crop alone. |
| |
| Parameters |
| ---------- |
| actual_energy_kwh : float |
| Actual PV generation under agrivoltaic operation. |
| mono_energy_kwh : float, optional |
| Theoretical generation without any shading interventions. |
| Defaults to annual_generation_kwh. |
| actual_crop_yield : float |
| Agrivoltaic crop yield (any unit, must match mono). |
| mono_crop_yield : float |
| Monoculture crop yield (same unit). |
| """ |
| e_mono = mono_energy_kwh or self.annual_gen |
| e_frac = actual_energy_kwh / e_mono if e_mono > 0 else 0.0 |
| c_frac = actual_crop_yield / mono_crop_yield if mono_crop_yield > 0 else 0.0 |
| ler = e_frac + c_frac |
|
|
| return LERResult( |
| energy_fraction=e_frac, |
| crop_fraction=c_frac, |
| ler=ler, |
| meets_target=ler >= TARGET_LER, |
| ) |
|
|
| |
| |
| |
|
|
| def summary(self, target_date: Optional[date] = None) -> dict: |
| """Return a combined summary dict for the dashboard.""" |
| budget = self.get_budget_status(target_date) |
| stats = self.compute_intervention_stats(label="all_time") |
|
|
| return { |
| "budget": { |
| "annual_budget_kwh": budget.annual_budget_kwh, |
| "annual_spent_kwh": budget.annual_spent_kwh, |
| "annual_remaining_kwh": budget.annual_remaining_kwh, |
| "utilisation_pct": budget.annual_utilisation_pct, |
| "over_budget": budget.is_over_budget(), |
| }, |
| "interventions": { |
| "total_slots": stats.total_slots, |
| "intervention_slots": stats.intervention_slots, |
| "intervention_rate_pct": stats.intervention_rate_pct, |
| "avg_offset_deg": stats.avg_offset_deg, |
| "max_offset_deg": stats.max_offset_deg, |
| "energy_sacrificed_kwh": stats.energy_sacrificed_kwh, |
| }, |
| } |
|
|
