211 lines
7.5 KiB
Python
211 lines
7.5 KiB
Python
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()
|