Giovanni/PV_Simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2059354e744733d2ba0f61de23cfb74d6ecb5b36
PV_Simulator/pvsim/models.py
Giovanni e656d78bb1 Update pvsim/models.py
2026-07-19 20:36:13 +00:00

1127 lines
20 KiB
Python
Raw Blame History

"""
pvsim/models.py
Modelli dati standardizzati per la telemetria del simulatore
di campo fotovoltaico.
Gerarchia:
PVPlant
|
+-- Inverter
|
+-- CombinerBox
|
+-- PVPanel
I modelli definiti in questo modulo rappresentano i dati prodotti
dalla simulazione a ogni timestep.
Struttura:
PanelData
|
v
CombinerData
|
v
InverterData
|
v
PlantData
Sono inoltre definiti:
WeatherData
FaultData
SimulationStep
Obiettivo:
Separare la logica di simulazione dalla rappresentazione
e serializzazione dei dati.
Questo permette in futuro di salvare gli stessi modelli in:
- CSV
- Parquet
- SQLite
- PostgreSQL
- TimescaleDB
- MQTT
- API REST
"""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
# ======================================================================
# WEATHER DATA
# ======================================================================
@dataclass
class WeatherData:
"""
Condizioni meteorologiche e solari del timestep.
Tutti i valori sono riferiti allo stesso timestamp.
Parameters
----------
timestamp:
Istante della misura.
ambient_temperature_C:
Temperatura ambiente [°C].
solar_elevation_deg:
Elevazione solare [°].
solar_azimuth_deg:
Azimuth solare [°].
solar_zenith_deg:
Angolo zenitale solare [°].
angle_of_incidence_deg:
Angolo di incidenza sul piano del pannello [°].
ghi_Wm2:
Global Horizontal Irradiance [W/m²].
dni_Wm2:
Direct Normal Irradiance [W/m²].
dhi_Wm2:
Diffuse Horizontal Irradiance [W/m²].
poa_global_Wm2:
Irraggiamento globale sul piano del pannello [W/m²].
poa_direct_Wm2:
Componente diretta POA [W/m²].
poa_diffuse_Wm2:
Componente diffusa POA [W/m²].
cloud_factor:
Fattore di attenuazione delle nuvole [0..1].
rain_factor:
Fattore di attenuazione della pioggia [0..1].
rain_active:
Indica se è attiva la pioggia.
daylight:
Indica se il sole è sopra l'orizzonte.
"""
timestamp: datetime
ambient_temperature_C: float
solar_elevation_deg: float = 0.0
solar_azimuth_deg: float = 0.0
solar_zenith_deg: float = 180.0
angle_of_incidence_deg: float = 90.0
ghi_Wm2: float = 0.0
dni_Wm2: float = 0.0
dhi_Wm2: float = 0.0
poa_global_Wm2: float = 0.0
poa_direct_Wm2: float = 0.0
poa_diffuse_Wm2: float = 0.0
cloud_factor: float = 1.0
rain_factor: float = 1.0
rain_active: bool = False
daylight: bool = False
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in un dizionario.
"""
return asdict(
self
)
# ======================================================================
# PANEL DATA
# ======================================================================
@dataclass
class PanelData:
"""
Telemetria di un singolo pannello fotovoltaico.
Questo è il livello più basso della gerarchia.
Ogni record rappresenta lo stato di un pannello
in un determinato timestep.
Parameters
----------
timestamp:
Istante della misura.
panel_id:
Identificativo univoco del pannello.
combiner_id:
Combiner Box a cui appartiene il pannello.
inverter_id:
Inverter a cui appartiene il pannello.
dc_power_W:
Potenza DC prodotta [W].
dc_voltage_V:
Tensione DC [V].
dc_current_A:
Corrente DC [A].
nominal_power_W:
Potenza nominale del pannello [W].
irradiance_Wm2:
Irraggiamento incidente [W/m²].
panel_temperature_C:
Temperatura stimata del pannello [°C].
efficiency:
Efficienza istantanea [0..1].
performance_ratio:
Rapporto tra produzione effettiva e produzione attesa.
fault_factor:
Fattore di riduzione dovuto ai fault [0..1].
enabled:
Indica se il pannello è operativo.
fault_active:
Indica se il pannello presenta almeno un fault.
status:
Stato testuale del pannello.
"""
timestamp: datetime
panel_id: str
combiner_id: str
inverter_id: str
dc_power_W: float = 0.0
dc_voltage_V: float = 0.0
dc_current_A: float = 0.0
nominal_power_W: float = 0.0
irradiance_Wm2: float = 0.0
panel_temperature_C: float = 25.0
efficiency: float = 0.0
performance_ratio: float = 0.0
fault_factor: float = 1.0
enabled: bool = True
fault_active: bool = False
status: str = "normal"
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in dizionario.
"""
return asdict(
self
)
# ======================================================================
# COMBINER DATA
# ======================================================================
@dataclass
class CombinerData:
"""
Telemetria di una Combiner Box.
La Combiner Box aggrega la produzione DC
di più stringhe o gruppi di pannelli.
Parameters
----------
timestamp:
Istante della misura.
combiner_id:
Identificativo della Combiner Box.
inverter_id:
Inverter associato.
dc_power_W:
Potenza DC totale [W].
dc_voltage_V:
Tensione DC [V].
dc_current_A:
Corrente DC totale [A].
nominal_power_W:
Potenza nominale aggregata [W].
panel_count:
Numero totale di pannelli collegati.
active_panel_count:
Numero di pannelli operativi.
fault_factor:
Fattore di riduzione dovuto ai fault.
fault_active:
Indica se è presente almeno un'anomalia.
availability:
Disponibilità del combiner [0..1].
status:
Stato del combiner.
"""
timestamp: datetime
combiner_id: str
inverter_id: str
dc_power_W: float = 0.0
dc_voltage_V: float = 0.0
dc_current_A: float = 0.0
nominal_power_W: float = 0.0
panel_count: int = 0
active_panel_count: int = 0
fault_factor: float = 1.0
fault_active: bool = False
availability: float = 1.0
status: str = "normal"
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in dizionario.
"""
return asdict(
self
)
# ======================================================================
# INVERTER DATA
# ======================================================================
@dataclass
class InverterData:
"""
Telemetria di un inverter.
L'inverter riceve la produzione DC aggregata
dalle Combiner Box e la converte in AC.
Parameters
----------
timestamp:
Istante della misura.
inverter_id:
Identificativo dell'inverter.
dc_power_W:
Potenza DC in ingresso [W].
ac_power_W:
Potenza AC in uscita [W].
dc_voltage_V:
Tensione DC [V].
dc_current_A:
Corrente DC [A].
ac_voltage_V:
Tensione AC [V].
ac_current_A:
Corrente AC [A].
efficiency:
Efficienza di conversione [0..1].
nominal_power_W:
Potenza nominale dell'inverter [W].
clipping_loss_W:
Perdita dovuta al clipping [W].
temperature_C:
Temperatura dell'inverter [°C].
fault_factor:
Fattore di riduzione dovuto ai fault.
combiner_count:
Numero di Combiner Box collegate.
active_combiner_count:
Numero di Combiner Box operative.
fault_active:
Indica la presenza di un fault.
availability:
Disponibilità dell'inverter [0..1].
status:
Stato operativo.
"""
timestamp: datetime
inverter_id: str
dc_power_W: float = 0.0
ac_power_W: float = 0.0
dc_voltage_V: float = 0.0
dc_current_A: float = 0.0
ac_voltage_V: float = 0.0
ac_current_A: float = 0.0
efficiency: float = 0.0
nominal_power_W: float = 0.0
clipping_loss_W: float = 0.0
temperature_C: float = 25.0
fault_factor: float = 1.0
combiner_count: int = 0
active_combiner_count: int = 0
fault_active: bool = False
availability: float = 1.0
status: str = "normal"
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in dizionario.
"""
return asdict(
self
)
# ======================================================================
# PLANT DATA
# ======================================================================
@dataclass
class PlantData:
"""
Telemetria aggregata dell'intero campo fotovoltaico.
Parameters
----------
timestamp:
Istante della misura.
plant_id:
Identificativo dell'impianto.
dc_power_W:
Potenza DC totale [W].
ac_power_W:
Potenza AC totale [W].
effective_ac_power_W:
Potenza AC dopo l'applicazione dei fault [W].
nominal_power_W:
Potenza nominale installata [W].
energy_Wh:
Energia prodotta nel timestep [Wh].
cumulative_energy_Wh:
Energia cumulativa [Wh].
inverter_count:
Numero di inverter.
active_inverter_count:
Numero di inverter operativi.
combiner_count:
Numero di Combiner Box.
active_combiner_count:
Numero di Combiner Box operative.
panel_count:
Numero totale di pannelli.
active_panel_count:
Numero di pannelli operativi.
efficiency:
Efficienza globale dell'impianto.
performance_ratio:
Performance Ratio globale.
availability:
Disponibilità globale.
fault_factor:
Fattore globale di riduzione.
fault_active:
Presenza di anomalie.
status:
Stato dell'impianto.
"""
timestamp: datetime
plant_id: str
dc_power_W: float = 0.0
ac_power_W: float = 0.0
effective_ac_power_W: float = 0.0
nominal_power_W: float = 0.0
energy_Wh: float = 0.0
cumulative_energy_Wh: float = 0.0
inverter_count: int = 0
active_inverter_count: int = 0
combiner_count: int = 0
active_combiner_count: int = 0
panel_count: int = 0
active_panel_count: int = 0
efficiency: float = 0.0
performance_ratio: float = 0.0
availability: float = 1.0
fault_factor: float = 1.0
fault_active: bool = False
status: str = "normal"
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in dizionario.
"""
return asdict(
self
)
# ======================================================================
# FAULT DATA
# ======================================================================
@dataclass
class FaultData:
"""
Rappresentazione serializzabile di un fault.
Questo modello è separato da FaultEvent perché
FaultEvent rappresenta la logica interna del FaultManager,
mentre FaultData rappresenta il dato destinato
allo storage o alla telemetria.
Parameters
----------
timestamp:
Timestamp del rilevamento.
fault_id:
ID del fault.
fault_type:
Tipo di fault.
component_level:
Livello gerarchico.
component_id:
ID del componente.
severity:
Severità [0..1].
reduction_factor:
Fattore di produzione residua [0..1].
description:
Descrizione.
"""
timestamp: datetime
fault_id: str
fault_type: str
component_level: str
component_id: str
severity: float = 0.0
reduction_factor: float = 1.0
description: str = ""
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte il modello in dizionario.
"""
return asdict(
self
)
# ======================================================================
# SIMULATION STEP
# ======================================================================
@dataclass
class SimulationStep:
"""
Contenitore di tutti i dati prodotti durante
un singolo timestep.
Questo modello rappresenta una fotografia completa
dell'impianto in un determinato istante.
Gerarchia:
SimulationStep
|
+-- WeatherData
|
+-- PanelData[]
|
+-- CombinerData[]
|
+-- InverterData[]
|
+-- PlantData
|
+-- FaultData[]
"""
timestamp: datetime
weather: Optional[
WeatherData
] = None
panels: List[
PanelData
] = field(
default_factory=list
)
combiners: List[
CombinerData
] = field(
default_factory=list
)
inverters: List[
InverterData
] = field(
default_factory=list
)
plant: Optional[
PlantData
] = None
faults: List[
FaultData
] = field(
default_factory=list
)
# ------------------------------------------------------------------
# Conversione dizionario
# ------------------------------------------------------------------
def to_dict(
self
) -> Dict[str, Any]:
"""
Converte l'intero timestep in un dizionario.
"""
return {
"timestamp":
self.timestamp,
"weather":
(
self.weather.to_dict()
if self.weather
else None
),
"panels":
[
panel.to_dict()
for panel
in self.panels
],
"combiners":
[
combiner.to_dict()
for combiner
in self.combiners
],
"inverters":
[
inverter.to_dict()
for inverter
in self.inverters
],
"plant":
(
self.plant.to_dict()
if self.plant
else None
),
"faults":
[
fault.to_dict()
for fault
in self.faults
]
}
# ======================================================================
# UTILITY FUNCTIONS
# ======================================================================
def flatten_simulation_step(
step: SimulationStep
) -> Dict[str, List[Dict[str, Any]]]:
"""
Trasforma un SimulationStep in quattro liste
facilmente convertibili in DataFrame.
Returns
-------
Dict
Dizionario con:
weather
panel
combiner
inverter
plant
faults
"""
return {
"weather":
(
[
step.weather.to_dict()
]
if step.weather is not None
else []
),
"panel":
[
panel.to_dict()
for panel
in step.panels
],
"combiner":
[
combiner.to_dict()
for combiner
in step.combiners
],
"inverter":
[
inverter.to_dict()
for inverter
in step.inverters
],
"plant":
(
[
step.plant.to_dict()
]
if step.plant is not None
else []
),
"faults":
[
fault.to_dict()
for fault
in step.faults
]
}
# ======================================================================
# FACTORY FUNCTIONS
# ======================================================================
def create_panel_data(
timestamp: datetime,
panel_id: str,
combiner_id: str,
inverter_id: str,
dc_power_W: float,
nominal_power_W: float,
irradiance_Wm2: float,
panel_temperature_C: float,
fault_factor: float = 1.0,
enabled: bool = True
) -> PanelData:
"""
Factory function per creare un record PanelData.
La funzione calcola automaticamente:
- corrente;
- efficienza;
- performance ratio;
- stato.
"""
# --------------------------------------------------------------
# Tensione semplificata
# --------------------------------------------------------------
dc_voltage_V = 40.0
# --------------------------------------------------------------
# Corrente
# --------------------------------------------------------------
if dc_voltage_V > 0:
dc_current_A = (
dc_power_W
/ dc_voltage_V
)
else:
dc_current_A = 0.0
# --------------------------------------------------------------
# Efficienza
# --------------------------------------------------------------
if irradiance_Wm2 > 0:
efficiency = (
dc_power_W
/ (
irradiance_Wm2
* 2.0
)
)
else:
efficiency = 0.0
# --------------------------------------------------------------
# Performance Ratio
# --------------------------------------------------------------
if nominal_power_W > 0:
performance_ratio = (
dc_power_W
/ nominal_power_W
)
else:
performance_ratio = 0.0
# --------------------------------------------------------------
# Stato
# --------------------------------------------------------------
if not enabled:
status = "disabled"
elif fault_factor < 1.0:
status = "fault"
elif dc_power_W <= 0:
status = "idle"
else:
status = "normal"
return PanelData(
timestamp=
timestamp,
panel_id=
panel_id,
combiner_id=
combiner_id,
inverter_id=
inverter_id,
dc_power_W=
dc_power_W,
dc_voltage_V=
dc_voltage_V,
dc_current_A=
dc_current_A,
nominal_power_W=
nominal_power_W,
irradiance_Wm2=
irradiance_Wm2,
panel_temperature_C=
panel_temperature_C,
efficiency=
efficiency,
performance_ratio=
performance_ratio,
fault_factor=
fault_factor,
enabled=
enabled,
fault_active=
fault_factor < 1.0,
status=
status
)
# ======================================================================
# DATAFRAME HELPERS
# ======================================================================
def records_to_dataframe(
records: List[Any]
):
"""
Converte una lista di modelli dati in DataFrame pandas.
Esempio:
df = records_to_dataframe(
simulation_result.panel
)
"""
import pandas as pd
if not records:
return pd.DataFrame()
return pd.DataFrame(
[
record.to_dict()
for record
in records
]
)
# ======================================================================
# EXPORT
# ======================================================================
__all__ = [
"WeatherData",
"PanelData",
"CombinerData",
"InverterData",
"PlantData",
"FaultData",
"SimulationStep",
"flatten_simulation_step",
"create_panel_data",
"records_to_dataframe"
]
```
Reference in New Issue View Git Blame Copy Permalink