Giovanni/PV_Simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Modulo per il calcolo delle statistiche e dei KPI della simulazione fotovoltaica.

Browse Source
This commit is contained in:
Giovanni
2026-07-19 20:24:05 +00:00
parent 4beb2ba677
commit a73ad9dbce
1 changed files with 840 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

840
pvsim/statistics.py Normal file
View File

@@ -0,0 +1,840 @@
"""
pvsim/statistics.py
Modulo per il calcolo delle statistiche e dei KPI
della simulazione fotovoltaica.
Livelli supportati:
Plant
|
+-- Inverter
|
+-- Combiner
|
+-- Panel
KPI principali:
- energia prodotta;
- potenza media;
- potenza massima;
- potenza minima;
- efficienza;
- performance ratio;
- availability;
- fault count;
- fault duration;
- clipping;
- yield specifico.
Il modulo riceve un SimulationResult e non modifica
i dati della simulazione.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Optional
import pandas as pd
from .simulator import SimulationResult
from .utils import (
normalize_dataframe,
safe_divide,
)
# ======================================================================
# KPI RESULT
# ======================================================================
@dataclass
class KPIResult:
"""
Contenitore dei KPI principali.
"""
energy_Wh: float = 0.0
energy_kWh: float = 0.0
peak_power_W: float = 0.0
average_power_W: float = 0.0
minimum_power_W: float = 0.0
nominal_power_W: float = 0.0
efficiency: float = 0.0
performance_ratio: float = 0.0
availability: float = 0.0
fault_count: int = 0
clipping_loss_Wh: float = 0.0
specific_yield_kWh_kWp: float = 0.0
def to_dict(
self,
) -> Dict[str, Any]:
return {
"energy_Wh":
self.energy_Wh,
"energy_kWh":
self.energy_kWh,
"peak_power_W":
self.peak_power_W,
"average_power_W":
self.average_power_W,
"minimum_power_W":
self.minimum_power_W,
"nominal_power_W":
self.nominal_power_W,
"efficiency":
self.efficiency,
"performance_ratio":
self.performance_ratio,
"availability":
self.availability,
"fault_count":
self.fault_count,
"clipping_loss_Wh":
self.clipping_loss_Wh,
"specific_yield_kWh_kWp":
self.specific_yield_kWh_kWp,
}
# ======================================================================
# STATISTICS
# ======================================================================
class SimulationStatistics:
"""
Calcola statistiche e KPI su SimulationResult.
"""
def __init__(
self,
result: SimulationResult,
) -> None:
self.result = result
# ==================================================================
# DATAFRAME
# ==================================================================
def plant_dataframe(
self,
) -> pd.DataFrame:
return normalize_dataframe(
self.result.plant_dataframe()
)
# ------------------------------------------------------------------
def inverter_dataframe(
self,
) -> pd.DataFrame:
return normalize_dataframe(
self.result.inverter_dataframe()
)
# ------------------------------------------------------------------
def combiner_dataframe(
self,
) -> pd.DataFrame:
return normalize_dataframe(
self.result.combiner_dataframe()
)
# ------------------------------------------------------------------
def panel_dataframe(
self,
) -> pd.DataFrame:
return normalize_dataframe(
self.result.panel_dataframe()
)
# ------------------------------------------------------------------
def faults_dataframe(
self,
) -> pd.DataFrame:
return normalize_dataframe(
self.result.faults_dataframe()
)
# ==================================================================
# PLANT KPI
# ==================================================================
def plant_kpi(
self,
) -> KPIResult:
"""
Calcola i KPI dell'intero impianto.
"""
df = self.plant_dataframe()
if df.empty:
return KPIResult()
# --------------------------------------------------------------
# Energia
# --------------------------------------------------------------
if "energy_Wh" in df.columns:
energy_Wh = (
df[
"energy_Wh"
].sum()
)
else:
energy_Wh = 0.0
# --------------------------------------------------------------
# Potenza
# --------------------------------------------------------------
power_column = (
"effective_ac_power_W"
if "effective_ac_power_W"
in df.columns
else "ac_power_W"
)
power = df[
power_column
]
peak_power = (
power.max()
)
average_power = (
power.mean()
)
minimum_power = (
power.min()
)
# --------------------------------------------------------------
# Nominale
# --------------------------------------------------------------
nominal_power = 0.0
if "nominal_power_W" in df.columns:
nominal_power = (
df[
"nominal_power_W"
].max()
)
# --------------------------------------------------------------
# Efficienza
# --------------------------------------------------------------
efficiency = 0.0
if (
"efficiency"
in df.columns
):
efficiency = (
df[
"efficiency"
].mean()
)
# --------------------------------------------------------------
# Performance Ratio
# --------------------------------------------------------------
performance_ratio = 0.0
if (
"performance_ratio"
in df.columns
):
performance_ratio = (
df[
"performance_ratio"
].mean()
)
# --------------------------------------------------------------
# Availability
# --------------------------------------------------------------
availability = 0.0
if (
"availability"
in df.columns
):
availability = (
df[
"availability"
].mean()
)
# --------------------------------------------------------------
# Faults
# --------------------------------------------------------------
faults = self.faults_dataframe()
fault_count = len(
faults
)
# --------------------------------------------------------------
# Clipping
# --------------------------------------------------------------
clipping_loss_Wh = 0.0
if (
"clipping_loss_W"
in df.columns
):
clipping_loss_Wh = (
df[
"clipping_loss_W"
].sum()
)
# --------------------------------------------------------------
# Specific Yield
# --------------------------------------------------------------
energy_kWh = (
energy_Wh
/ 1000.0
)
nominal_kWp = (
nominal_power
/ 1000.0
)
specific_yield = (
safe_divide(
energy_kWh,
nominal_kWp
)
)
return KPIResult(
energy_Wh=
energy_Wh,
energy_kWh=
energy_kWh,
peak_power_W=
peak_power,
average_power_W=
average_power,
minimum_power_W=
minimum_power,
nominal_power_W=
nominal_power,
efficiency=
efficiency,
performance_ratio=
performance_ratio,
availability=
availability,
fault_count=
fault_count,
clipping_loss_Wh=
clipping_loss_Wh,
specific_yield_kWh_kWp=
specific_yield,
)
# ==================================================================
# GENERIC GROUP KPI
# ==================================================================
def _group_kpi(
self,
dataframe: pd.DataFrame,
group_column: str,
power_column: str,
) -> pd.DataFrame:
"""
Calcola KPI aggregati per componente.
"""
if dataframe.empty:
return pd.DataFrame()
if group_column not in dataframe.columns:
return pd.DataFrame()
if power_column not in dataframe.columns:
return pd.DataFrame()
grouped = (
dataframe
.groupby(
group_column
)
.agg(
power_mean_W=(
power_column,
"mean"
),
power_peak_W=(
power_column,
"max"
),
power_min_W=(
power_column,
"min"
),
records=(
power_column,
"count"
),
)
.reset_index()
)
return grouped
# ==================================================================
# PANEL KPI
# ==================================================================
def panel_kpi(
self,
) -> pd.DataFrame:
"""
KPI per ogni pannello.
"""
df = self.panel_dataframe()
return self._group_kpi(
df,
"panel_id",
"dc_power_W",
)
# ==================================================================
# COMBINER KPI
# ==================================================================
def combiner_kpi(
self,
) -> pd.DataFrame:
"""
KPI per ogni Combiner Box.
"""
df = self.combiner_dataframe()
return self._group_kpi(
df,
"combiner_id",
"dc_power_W",
)
# ==================================================================
# INVERTER KPI
# ==================================================================
def inverter_kpi(
self,
) -> pd.DataFrame:
"""
KPI per ogni inverter.
"""
df = self.inverter_dataframe()
return self._group_kpi(
df,
"inverter_id",
"ac_power_W",
)
# ==================================================================
# DAILY
# ==================================================================
def daily_statistics(
self,
) -> pd.DataFrame:
"""
Calcola statistiche giornaliere.
"""
df = self.plant_dataframe()
if df.empty:
return pd.DataFrame()
df = df.copy()
df[
"date"
] = df[
"timestamp"
].dt.date
aggregation = {
"ac_power_W":
[
"mean",
"max",
],
"energy_Wh":
"sum",
}
valid_columns = {
key: value
for key, value
in aggregation.items()
if key in df.columns
}
if not valid_columns:
return pd.DataFrame()
return (
df
.groupby(
"date"
)
.agg(
valid_columns
)
.reset_index()
)
# ==================================================================
# MONTHLY
# ==================================================================
def monthly_statistics(
self,
) -> pd.DataFrame:
"""
Calcola statistiche mensili.
"""
df = self.plant_dataframe()
if df.empty:
return pd.DataFrame()
df = df.set_index(
"timestamp"
)
result = (
df.resample(
"ME"
)
.agg(
{
"ac_power_W":
[
"mean",
"max",
],
"energy_Wh":
"sum",
}
)
)
return result.reset_index()
# ==================================================================
# FAULT SUMMARY
# ==================================================================
def fault_summary(
self,
) -> pd.DataFrame:
"""
Aggrega i fault per livello e componente.
"""
df = self.faults_dataframe()
if df.empty:
return pd.DataFrame()
columns = [
"component_level",
"component_id",
"fault_type",
]
columns = [
column
for column
in columns
if column in df.columns
]
if not columns:
return pd.DataFrame()
return (
df
.groupby(
columns
)
.size()
.reset_index(
name="fault_count"
)
)
# ==================================================================
# FULL REPORT
# ==================================================================
def report(
self,
) -> Dict[str, Any]:
"""
Genera un report completo.
"""
plant = self.plant_kpi()
return {
"plant": plant.to_dict(),
"panel_kpi":
self.panel_kpi(),
"combiner_kpi":
self.combiner_kpi(),
"inverter_kpi":
self.inverter_kpi(),
"daily":
self.daily_statistics(),
"monthly":
self.monthly_statistics(),
"faults":
self.fault_summary(),
}
# ======================================================================
# HELPER
# ======================================================================
def calculate_statistics(
result: SimulationResult,
) -> Dict[str, Any]:
"""
Helper function.
"""
statistics = SimulationStatistics(
result
)
return statistics.report()
# ======================================================================
__all__ = [
"KPIResult",
"SimulationStatistics",
"calculate_statistics",
]
```