Files
PV_Simulator/pvsim/simulator.py

1142 lines
24 KiB
Python

```python
"""
simulator.py
Motore centrale della simulazione fotovoltaica.
Collega:
SunModel
|
v
WeatherModel
|
v
FaultManager
|
v
PVPlant
|
+-- Inverters
| |
| +-- CombinerBoxes
| |
| +-- PVPanels
|
v
SimulationResult
Il simulatore esegue la simulazione timestep per timestep.
Per ogni timestamp vengono calcolati:
- posizione del sole;
- condizioni meteorologiche;
- irraggiamento;
- temperatura;
- fault attivi;
- produzione dei pannelli;
- aggregazione Combiner Box;
- aggregazione Inverter;
- aggregazione Plant.
Il modulo è progettato per simulazioni:
- giornaliere;
- mensili;
- stagionali;
- annuali;
- multi-annuali.
Il timestep può essere configurato liberamente.
Esempio:
5 minuti
15 minuti
1 ora
Nota:
Il simulatore mantiene due livelli di output:
1. summary:
dati aggregati a livello impianto.
2. hierarchy:
dati dettagliati di pannelli,
combiner e inverter.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
import pandas as pd
from .plant import PVPlant
from .sun import SunModel
from .weather import WeatherModel
from .faults import FaultManager
# ======================================================================
# CONFIGURAZIONE SIMULAZIONE
# ======================================================================
@dataclass
class SimulationConfig:
"""
Configurazione temporale della simulazione.
Parameters
----------
start:
Timestamp iniziale.
end:
Timestamp finale.
timestep_minutes:
Durata del timestep in minuti.
years_from_start:
Anno relativo utilizzato per la degradazione.
store_panel_data:
Salva i dati dei singoli pannelli.
store_combiner_data:
Salva i dati delle Combiner Box.
store_inverter_data:
Salva i dati degli inverter.
store_plant_data:
Salva i dati aggregati dell'impianto.
generate_random_faults:
Abilita la generazione automatica dei fault.
"""
start: datetime
end: datetime
timestep_minutes: int = 5
years_from_start: float = 0.0
store_panel_data: bool = True
store_combiner_data: bool = True
store_inverter_data: bool = True
store_plant_data: bool = True
generate_random_faults: bool = False
def __post_init__(
self
) -> None:
if self.end <= self.start:
raise ValueError(
"end deve essere maggiore di start"
)
if self.timestep_minutes <= 0:
raise ValueError(
"timestep_minutes deve essere > 0"
)
# ======================================================================
# RISULTATO DELLA SIMULAZIONE
# ======================================================================
@dataclass
class SimulationResult:
"""
Contenitore dei risultati della simulazione.
I dati sono organizzati in quattro livelli:
plant
inverter
combiner
panel
Ogni elemento contiene una lista di record
successivamente convertibili in DataFrame.
"""
plant: List[Dict[str, Any]] = field(
default_factory=list
)
inverter: List[Dict[str, Any]] = field(
default_factory=list
)
combiner: List[Dict[str, Any]] = field(
default_factory=list
)
panel: List[Dict[str, Any]] = field(
default_factory=list
)
faults: List[Dict[str, Any]] = field(
default_factory=list
)
# ------------------------------------------------------------------
# Conversione DataFrame
# ------------------------------------------------------------------
def plant_dataframe(
self
) -> pd.DataFrame:
return pd.DataFrame(
self.plant
)
# ------------------------------------------------------------------
def inverter_dataframe(
self
) -> pd.DataFrame:
return pd.DataFrame(
self.inverter
)
# ------------------------------------------------------------------
def combiner_dataframe(
self
) -> pd.DataFrame:
return pd.DataFrame(
self.combiner
)
# ------------------------------------------------------------------
def panel_dataframe(
self
) -> pd.DataFrame:
return pd.DataFrame(
self.panel
)
# ------------------------------------------------------------------
def faults_dataframe(
self
) -> pd.DataFrame:
return pd.DataFrame(
self.faults
)
# ------------------------------------------------------------------
# Export CSV
# ------------------------------------------------------------------
def export_csv(
self,
output_dir: str
) -> None:
"""
Esporta tutti i livelli in file CSV.
"""
import os
os.makedirs(
output_dir,
exist_ok=True
)
if self.plant:
self.plant_dataframe().to_csv(
os.path.join(
output_dir,
"plant.csv"
),
index=False
)
if self.inverter:
self.inverter_dataframe().to_csv(
os.path.join(
output_dir,
"inverter.csv"
),
index=False
)
if self.combiner:
self.combiner_dataframe().to_csv(
os.path.join(
output_dir,
"combiner.csv"
),
index=False
)
if self.panel:
self.panel_dataframe().to_csv(
os.path.join(
output_dir,
"panel.csv"
),
index=False
)
if self.faults:
self.faults_dataframe().to_csv(
os.path.join(
output_dir,
"faults.csv"
),
index=False
)
# ======================================================================
# SIMULATORE
# ======================================================================
@dataclass
class PVSimulator:
"""
Motore principale della simulazione.
Parameters
----------
plant:
Modello gerarchico dell'impianto.
sun:
Modello della posizione solare.
weather:
Modello meteorologico.
fault_manager:
Gestore dei fault.
"""
plant: PVPlant
sun: SunModel
weather: WeatherModel
fault_manager: FaultManager
# ------------------------------------------------------------------
# Preparazione ID
# ------------------------------------------------------------------
def _get_component_ids(
self
) -> Dict[str, List[str]]:
"""
Estrae tutti gli ID della gerarchia.
"""
panel_ids = []
combiner_ids = []
inverter_ids = []
for inverter in self.plant.inverters:
inverter_ids.append(
inverter.inverter_id
)
for combiner in (
inverter.combiners
):
combiner_ids.append(
combiner.combiner_id
)
for panel in (
combiner.panels
):
panel_ids.append(
panel.panel_id
)
return {
"panel":
panel_ids,
"combiner":
combiner_ids,
"inverter":
inverter_ids
}
# ------------------------------------------------------------------
# Applicazione fault
# ------------------------------------------------------------------
def _get_fault_factor(
self,
component_id: str,
timestamp: datetime
) -> float:
"""
Restituisce il fattore di produzione dovuto ai fault.
"""
return (
self.fault_manager
.get_component_factor(
component_id,
timestamp
)
)
# ------------------------------------------------------------------
# Dati fault
# ------------------------------------------------------------------
def _collect_fault_records(
self,
timestamp: datetime
) -> List[Dict[str, Any]]:
"""
Converte i fault attivi in record serializzabili.
"""
records = []
active_faults = (
self.fault_manager
.get_active_faults(
timestamp
)
)
for fault in active_faults:
records.append({
"timestamp":
timestamp,
"fault_id":
fault.fault_id,
"fault_type":
fault.fault_type.value,
"component_level":
fault.component_level.value,
"component_id":
fault.component_id,
"severity":
fault.severity,
"reduction_factor":
fault.reduction_factor(),
"description":
fault.description
})
return records
# ------------------------------------------------------------------
# Simulazione singolo timestep
# ------------------------------------------------------------------
def simulate_timestep(
self,
timestamp: datetime,
config: SimulationConfig
) -> Dict[str, Any]:
"""
Esegue un singolo timestep.
Il metodo restituisce i dati gerarchici del timestep.
"""
# --------------------------------------------------------------
# Condizioni meteorologiche
# --------------------------------------------------------------
weather = (
self.weather.get_conditions(
timestamp=
timestamp
)
)
irradiance = (
weather[
"poa_global_Wm2"
]
)
ambient_temperature = (
weather[
"ambient_temperature_C"
]
)
# --------------------------------------------------------------
# Fault automatici
# --------------------------------------------------------------
if config.generate_random_faults:
ids = (
self._get_component_ids()
)
self.fault_manager.simulate_random_faults(
timestamp=
timestamp,
panel_ids=
ids[
"panel"
],
combiner_ids=
ids[
"combiner"
],
inverter_ids=
ids[
"inverter"
]
)
# --------------------------------------------------------------
# Aggiornamento Plant
# --------------------------------------------------------------
plant_data = (
self.plant.update(
timestamp=
timestamp,
irradiance=
irradiance,
ambient_temperature=
ambient_temperature,
years_from_start=
config.years_from_start,
timestep_minutes=
config.timestep_minutes
)
)
# --------------------------------------------------------------
# Applicazione fault Plant
# --------------------------------------------------------------
plant_factor = (
self._get_fault_factor(
self.plant.plant_id,
timestamp
)
)
plant_data[
"fault_factor"
] = plant_factor
plant_data[
"effective_ac_power_W"
] = (
plant_data[
"ac_power_W"
]
* plant_factor
)
# --------------------------------------------------------------
# Output
# --------------------------------------------------------------
return {
"timestamp":
timestamp,
"weather":
weather,
"plant":
plant_data,
"faults":
self._collect_fault_records(
timestamp
)
}
# ------------------------------------------------------------------
# Simulazione completa
# ------------------------------------------------------------------
def run(
self,
config: SimulationConfig
) -> SimulationResult:
"""
Esegue la simulazione completa.
Returns
-------
SimulationResult
Risultati a tutti i livelli.
"""
result = (
SimulationResult()
)
timestamp = (
config.start
)
# --------------------------------------------------------------
# Loop temporale
# --------------------------------------------------------------
while timestamp < config.end:
timestep = (
self.simulate_timestep(
timestamp,
config
)
)
weather = (
timestep[
"weather"
]
)
plant_data = (
timestep[
"plant"
]
)
# ----------------------------------------------------------
# Plant level
# ----------------------------------------------------------
if config.store_plant_data:
plant_record = {
**plant_data,
"ambient_temperature_C":
weather[
"ambient_temperature_C"
],
"ghi_Wm2":
weather[
"ghi_Wm2"
],
"dni_Wm2":
weather[
"dni_Wm2"
],
"dhi_Wm2":
weather[
"dhi_Wm2"
],
"poa_global_Wm2":
weather[
"poa_global_Wm2"
],
"cloud_factor":
weather[
"cloud_factor"
],
"rain_factor":
weather[
"rain_factor"
]
}
result.plant.append(
plant_record
)
# ----------------------------------------------------------
# Inverter level
# ----------------------------------------------------------
if config.store_inverter_data:
for inverter in (
self.plant.inverters
):
inverter_factor = (
self._get_fault_factor(
inverter.inverter_id,
timestamp
)
)
inverter_record = {
"timestamp":
timestamp,
"inverter_id":
inverter.inverter_id,
"nominal_power_kW":
inverter.nominal_power_kW,
"fault_factor":
inverter_factor,
"enabled":
inverter.enabled
}
result.inverter.append(
inverter_record
)
# --------------------------------------------------
# Combiner level
# --------------------------------------------------
if config.store_combiner_data:
for combiner in (
inverter.combiners
):
combiner_factor = (
self._get_fault_factor(
combiner.combiner_id,
timestamp
)
)
combiner_record = {
"timestamp":
timestamp,
"inverter_id":
inverter.inverter_id,
"combiner_id":
combiner.combiner_id,
"fault_factor":
combiner_factor,
"panel_count":
len(
combiner.panels
)
}
result.combiner.append(
combiner_record
)
# ------------------------------------------
# Panel level
# ------------------------------------------
if config.store_panel_data:
for panel in (
combiner.panels
):
panel_factor = (
self._get_fault_factor(
panel.panel_id,
timestamp
)
)
panel_record = {
"timestamp":
timestamp,
"inverter_id":
inverter.inverter_id,
"combiner_id":
combiner.combiner_id,
"panel_id":
panel.panel_id,
"fault_factor":
panel_factor,
"enabled":
panel.enabled,
"nominal_power_W":
panel.nominal_power
}
result.panel.append(
panel_record
)
# ----------------------------------------------------------
# Fault level
# ----------------------------------------------------------
result.faults.extend(
timestep[
"faults"
]
)
# ----------------------------------------------------------
# Incremento temporale
# ----------------------------------------------------------
timestamp += timedelta(
minutes=
config.timestep_minutes
)
return result
# ------------------------------------------------------------------
# Simulazione rapida
# ------------------------------------------------------------------
def run_single_day(
self,
date: datetime,
timestep_minutes: int = 5
) -> SimulationResult:
"""
Esegue una simulazione di una singola giornata.
"""
start = datetime(
year=
date.year,
month=
date.month,
day=
date.day
)
end = (
start
+ timedelta(
days=1
)
)
config = SimulationConfig(
start=
start,
end=
end,
timestep_minutes=
timestep_minutes
)
return self.run(
config
)
# ------------------------------------------------------------------
# Simulazione annuale
# ------------------------------------------------------------------
def run_year(
self,
year: int,
timestep_minutes: int = 15
) -> SimulationResult:
"""
Esegue una simulazione annuale.
Nota:
per una simulazione annuale a 5 minuti
il numero di record a livello pannello
può diventare molto elevato.
Per questo motivo il default è 15 minuti.
"""
start = datetime(
year=
year,
month=
1,
day=
1
)
end = datetime(
year=
year + 1,
month=
1,
day=
1
)
config = SimulationConfig(
start=
start,
end=
end,
timestep_minutes=
timestep_minutes,
years_from_start=
0.0
)
return self.run(
config
)
# ------------------------------------------------------------------
# Riepilogo
# ------------------------------------------------------------------
def summary(
self,
result: SimulationResult
) -> Dict[str, Any]:
"""
Calcola un riepilogo della simulazione.
"""
summary = {
"plant_id":
self.plant.plant_id,
"inverters":
self.plant.get_inverter_count(),
"combiners":
self.plant.get_combiner_count(),
"panels":
self.plant.get_panel_count(),
"plant_records":
len(
result.plant
),
"inverter_records":
len(
result.inverter
),
"combiner_records":
len(
result.combiner
),
"panel_records":
len(
result.panel
),
"fault_records":
len(
result.faults
),
"total_ac_energy_Wh":
self.plant.total_ac_energy_Wh,
"total_dc_energy_Wh":
self.plant.total_dc_energy_Wh
}
return summary
# ------------------------------------------------------------------
# Rappresentazione
# ------------------------------------------------------------------
def __repr__(
self
) -> str:
"""
Rappresentazione del simulatore.
"""
return (
f"PVSimulator("
f"plant="
f"{self.plant.plant_id}, "
f"inverters="
f"{self.plant.get_inverter_count()}, "
f"combiners="
f"{self.plant.get_combiner_count()}, "
f"panels="
f"{self.plant.get_panel_count()})"
)
```