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()