From a73ad9dbce5b5e319c92d61af60ab2f3dd4ab865 Mon Sep 17 00:00:00 2001 From: Giovanni Date: Sun, 19 Jul 2026 20:24:05 +0000 Subject: [PATCH] Modulo per il calcolo delle statistiche e dei KPI della simulazione fotovoltaica. --- pvsim/statistics.py | 840 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 840 insertions(+) create mode 100644 pvsim/statistics.py diff --git a/pvsim/statistics.py b/pvsim/statistics.py new file mode 100644 index 0000000..dfda67b --- /dev/null +++ b/pvsim/statistics.py @@ -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", +] +```