pv forecaster added

simulators/pv_plant_simulator.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Optional, Dict, List, Literal, Tuple, Union
+
+import numpy as np
+import pandas as pd
+import pvlib
+import matplotlib.pyplot as plt
+from pvlib.location import Location
+from pvlib.pvsystem import PVSystem
+from pvlib.modelchain import ModelChain
+
+SeriesOrArray = Union[pd.Series, np.ndarray]
+
+# ----------------------------- Konfiguration -----------------------------
+
+@dataclass
+class PvWattsSubarrayConfig:
+    name: str
+    pdc0_w: float                   # STC-DC-Leistung [W]
+    tilt_deg: float                 # Neigung (0=horizontal)
+    azimuth_deg: float              # Azimut (180=Süd)
+    gamma_pdc: float = -0.004       # Tempkoeff. [1/K]
+    eta_inv_nom: float = 0.96       # WR-Wirkungsgrad (nominal)
+    albedo: float = 0.2             # Bodenreflexion
+
+    # Pauschale Verluste (PVWatts-Losses)
+    dc_loss: float = 0.0
+    ac_loss: float = 0.0
+    soiling: float = 0.0
+
+    # Modell
+    transposition_model: Literal["perez","haydavies","isotropic","klucher","reindl"] = "perez"
+
+
+# ------------------------------ Subarray ---------------------------------
+
+class PvWattsSubarray:
+    """
+    Ein Subarray mit pvlib.ModelChain (PVWatts).
+    Berechnet automatisch DNI/DHI aus GHI (ERBS-Methode)
+    und nutzt ein SAPM-Temperaturmodell.
+    """
+    def __init__(self, cfg: PvWattsSubarrayConfig, location: Location):
+        self.cfg = cfg
+        self.location = location
+        self._mc: Optional[ModelChain] = None
+
+    # ---------------------------------------------------------------------
+    def _create_modelchain(self) -> ModelChain:
+        """Erzeuge eine pvlib.ModelChain-Instanz mit PVWatts-Parametern."""
+        temp_params = pvlib.temperature.TEMPERATURE_MODEL_PARAMETERS["sapm"]["open_rack_glass_polymer"]
+
+        system = PVSystem(
+            surface_tilt=self.cfg.tilt_deg,
+            surface_azimuth=self.cfg.azimuth_deg,
+            module_parameters={"pdc0": self.cfg.pdc0_w, "gamma_pdc": self.cfg.gamma_pdc},
+            inverter_parameters={"pdc0": self.cfg.pdc0_w, "eta_inv_nom": self.cfg.eta_inv_nom},
+            albedo=self.cfg.albedo,
+            temperature_model_parameters=temp_params,
+            module_type="glass_polymer",
+            racking_model="open_rack",
+        )
+
+        mc = ModelChain(
+            system, self.location,
+            transposition_model=self.cfg.transposition_model,
+            solar_position_method="nrel_numpy",
+            airmass_model="kastenyoung1989",
+            dc_model="pvwatts",
+            ac_model="pvwatts",
+            aoi_model="physical",
+            spectral_model=None,
+            losses_model="pvwatts",
+            temperature_model="sapm",
+        )
+
+        mc.losses_parameters = {
+            "dc_loss": float(self.cfg.dc_loss),
+            "ac_loss": float(self.cfg.ac_loss),
+            "soiling": float(self.cfg.soiling),
+        }
+
+        self._mc = mc
+        return mc
+
+    # ---------------------------------------------------------------------
+    def calc_dni_and_dhi(self, weather: pd.DataFrame) -> pd.DataFrame:
+        """
+        Berechnet DNI & DHI aus GHI über die ERBS-Methode.
+        Gibt ein neues DataFrame mit 'ghi', 'dni', 'dhi' zurück.
+        """
+        if "ghi" not in weather:
+            raise ValueError("Wetterdaten benötigen mindestens 'ghi'.")
+        # Sonnenstand bestimmen
+        sp = self.location.get_solarposition(weather.index)
+        erbs = pvlib.irradiance.erbs(weather["ghi"], sp["zenith"], weather.index)
+        out = weather.copy()
+        out["dni"] = erbs["dni"].clip(lower=0)
+        out["dhi"] = erbs["dhi"].clip(lower=0)
+        return out
+
+    # ---------------------------------------------------------------------
+    def _prepare_weather(self, weather: pd.DataFrame) -> pd.DataFrame:
+        """Sichert vollständige Spalten (ghi, dni, dhi, temp_air, wind_speed)."""
+        if "ghi" not in weather or "temp_air" not in weather:
+            raise ValueError("weather benötigt Spalten: 'ghi' und 'temp_air'.")
+
+        w = weather.copy()
+
+        # Zeitzone prüfen
+        if w.index.tz is None:
+            w.index = w.index.tz_localize(self.location.tz)
+        else:
+            if str(w.index.tz) != str(self.location.tz):
+                w = w.tz_convert(self.location.tz)
+
+        # Wind default
+        if "wind_speed" not in w:
+            w["wind_speed"] = 1.0
+
+        # DNI/DHI ergänzen (immer mit ERBS)
+        if "dni" not in w or "dhi" not in w:
+            w = self.calc_dni_and_dhi(w)
+
+        return w
+
+    # ---------------------------------------------------------------------
+    def get_power(self, weather: pd.DataFrame) -> pd.Series:
+        """
+        Berechnet AC-Leistung aus Wetterdaten.
+        """
+        w = self._prepare_weather(weather)
+        mc = self._create_modelchain()
+        mc.run_model(weather=w)
+        return mc.results.ac.rename(self.cfg.name)
+
+
+# ------------------------------- Anlage ----------------------------------
+
+class PvWattsPlant:
+    """
+    Eine PV-Anlage mit mehreren Subarrays, die ein gemeinsames Wetter-DataFrame nutzt.
+    """
+    def __init__(self, site: Location, subarray_cfgs: List[PvWattsSubarrayConfig]):
+        self.site = site
+        self.subs: Dict[str, PvWattsSubarray] = {c.name: PvWattsSubarray(c, site) for c in subarray_cfgs}
+
+    def get_power(
+        self,
+        weather: pd.DataFrame,
+        *,
+        return_breakdown: bool = False
+    ) -> pd.Series | Tuple[pd.Series, Dict[str, pd.Series]]:
+        """Berechne Gesamtleistung und optional Einzel-Subarrays."""
+        parts: Dict[str, pd.Series] = {name: sub.get_power(weather) for name, sub in self.subs.items()}
+
+        # gemeinsamen Index bilden
+        idx = list(parts.values())[0].index
+        for s in parts.values():
+            idx = idx.intersection(s.index)
+        parts = {k: v.reindex(idx).fillna(0.0) for k, v in parts.items()}
+
+        total = sum(parts.values())
+        total.name = "total_ac"
+
+        if return_breakdown:
+            return total, parts
+        return total
+
+
+# --------------------------- Beispielnutzung -----------------------------
+if __name__ == "__main__":
+    # Standort
+    site = Location(latitude=52.52, longitude=13.405, altitude=35, tz="Europe/Berlin", name="Berlin")
+
+    # Zeitachse: 1 Tag, 15-minütig
+    times = pd.date_range("2025-06-21 00:00", "2025-06-21 23:45", freq="15min", tz=site.tz)
+
+    # Dummy-Wetter
+    ghi = 1000 * np.clip(np.sin(np.linspace(0, np.pi, len(times)))**1.2, 0, None)
+    temp_air = 16 + 8 * np.clip(np.sin(np.linspace(-np.pi/2, np.pi/2, len(times))), 0, None)
+    wind = np.full(len(times), 1.0)
+    weather = pd.DataFrame(index=times)
+    weather["ghi"] = ghi
+    weather["temp_air"] = temp_air
+    weather["wind_speed"] = wind
+
+    # Zwei Subarrays
+    cfgs = [
+        PvWattsSubarrayConfig(name="Sued_30", pdc0_w=6000, tilt_deg=30, azimuth_deg=180, dc_loss=0.02, ac_loss=0.01),
+        PvWattsSubarrayConfig(name="West_20", pdc0_w=4000, tilt_deg=20, azimuth_deg=270, soiling=0.02),
+    ]
+    plant = PvWattsPlant(site, cfgs)
+
+    # Simulation
+    total, parts = plant.get_power(weather, return_breakdown=True)
+
+    # Plot
+    plt.figure(figsize=(10, 6))
+    plt.plot(total.index, total / 1000, label="Gesamtleistung (AC)", linewidth=2, color="black")
+    for name, s in parts.items():
+        plt.plot(s.index, s / 1000, label=name)
+    plt.title("PV-Leistung (PVWatts, ERBS-Methode für DNI/DHI)")
+    plt.ylabel("Leistung [kW]")
+    plt.xlabel("Zeit")
+    plt.legend()
+    plt.grid(True, linestyle="--", alpha=0.5)
+    plt.tight_layout()
+    plt.show()