rpotter6298
c8e3016fd6
- Implement `sigen_daemon.py` to poll Sigenergy plant metrics and store snapshots. - Create `web_daemon.py` for serving a web interface with various endpoints. - Add debug scripts: - `debug_duplicates.py` to find duplicate target times in forecast data. - `debug_energy_forecast.py` to print baseline energy forecast curves. - `debug_oracle_evaluations.py` to run the oracle evaluator. - `debug_sigen.py` to inspect stored Sigenergy plant snapshots. - `debug_weather.py` to trace resolved truth data. - `modbus_test.py` for exploring Sigenergy plants or inverters over Modbus TCP. - Introduce `oracle_evaluator.py` for evaluating stored oracle predictions against actuals. - Add TCN training scripts in `tcn` directory for training usage sequence models.
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from gibil.classes.models import ForecastKind, PowerForecastPoint, PowerForecastRun, WeatherForecastPoint
|
|
from gibil.classes.oracle.config import EnergyForecastConfig
|
|
from gibil.classes.sigen.store import SigenStore
|
|
from gibil.classes.weather.store import WeatherStore
|
|
|
|
|
|
class BaselineSolarProductionOracle:
|
|
"""Forecasts solar production from shortwave radiation and recent plant peak."""
|
|
|
|
model_version = "baseline_solar_radiation_v1"
|
|
|
|
def __init__(
|
|
self,
|
|
weather_store: WeatherStore,
|
|
sigen_store: SigenStore,
|
|
config: EnergyForecastConfig,
|
|
) -> None:
|
|
self.weather_store = weather_store
|
|
self.sigen_store = sigen_store
|
|
self.config = config
|
|
|
|
def forecast(self, issued_at: datetime | None = None) -> PowerForecastRun:
|
|
if issued_at is None:
|
|
issued_at = datetime.now(timezone.utc)
|
|
|
|
weather_points = self.weather_store.load_latest_forecast_points(
|
|
start_at=issued_at,
|
|
end_at=issued_at + timedelta(hours=self.config.horizon_hours),
|
|
)
|
|
peak_w = self._solar_peak_w()
|
|
points = [
|
|
self._forecast_point(
|
|
weather_point=point,
|
|
issued_at=issued_at,
|
|
peak_w=peak_w,
|
|
)
|
|
for point in weather_points
|
|
]
|
|
|
|
return PowerForecastRun(
|
|
issued_at=issued_at,
|
|
kind=ForecastKind.SOLAR,
|
|
source="baseline_solar_oracle",
|
|
model_version=self.model_version,
|
|
points=points,
|
|
)
|
|
|
|
def _forecast_point(
|
|
self,
|
|
weather_point: WeatherForecastPoint,
|
|
issued_at: datetime,
|
|
peak_w: float,
|
|
) -> PowerForecastPoint:
|
|
radiation = max(weather_point.shortwave_radiation_w_m2 or 0.0, 0.0)
|
|
expected = min(peak_w, peak_w * (radiation / 1000.0) * self.config.solar_scale)
|
|
cloud_cover = weather_point.cloud_cover_pct
|
|
cloud_uncertainty = 1.0
|
|
if cloud_cover is not None:
|
|
cloud_uncertainty += min(max(cloud_cover, 0.0), 100.0) / 200.0
|
|
|
|
p10 = max(0.0, expected * (0.75 / cloud_uncertainty))
|
|
p90 = min(peak_w, expected * (1.15 * cloud_uncertainty))
|
|
|
|
return PowerForecastPoint(
|
|
target_at=weather_point.target_at,
|
|
horizon_minutes=self._horizon_minutes(issued_at, weather_point.target_at),
|
|
expected_power_w=expected,
|
|
p10_power_w=p10,
|
|
p50_power_w=expected,
|
|
p90_power_w=p90,
|
|
confidence=0.25,
|
|
source="open_meteo_shortwave",
|
|
model_version=self.model_version,
|
|
metadata={
|
|
"shortwave_radiation_w_m2": weather_point.shortwave_radiation_w_m2,
|
|
"cloud_cover_pct": weather_point.cloud_cover_pct,
|
|
"temperature_c": weather_point.temperature_c,
|
|
"solar_peak_w": peak_w,
|
|
"fallback_reason": "not_enough_solar_training_samples",
|
|
},
|
|
)
|
|
|
|
def _solar_peak_w(self) -> float:
|
|
recent_peak = self.sigen_store.load_recent_solar_peak_w()
|
|
if recent_peak is None or recent_peak <= 0:
|
|
return self.config.fallback_solar_peak_w
|
|
return recent_peak * max(self.config.solar_peak_headroom, 1.0)
|
|
|
|
def _horizon_minutes(self, issued_at: datetime, target_at: datetime) -> int:
|
|
return max(0, round((target_at - issued_at).total_seconds() / 60))
|