Esportazione dei risultati della simulazione.

pvsim/exporter.py

@@ -0,0 +1,600 @@
+"""
+pvsim/exporter.py
+
+Esportazione dei risultati della simulazione.
+
+Formati supportati:
+
+    CSV
+    JSON
+    Parquet
+
+Dataset esportati:
+
+    weather
+    panel
+    combiner
+    inverter
+    plant
+    faults
+
+Il modulo non contiene logica di simulazione.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from typing import Dict, Optional
+
+import pandas as pd
+
+from .simulator import SimulationResult
+
+from .statistics import SimulationStatistics
+
+from .utils import (
+    ensure_directory,
+    normalize_dataframe,
+    save_json,
+)
+
+
+# ======================================================================
+# EXPORTER
+# ======================================================================
+
+
+class SimulationExporter:
+    """
+    Esporta SimulationResult su filesystem.
+    """
+
+    def __init__(
+        self,
+        output_dir: str = "output",
+    ) -> None:
+
+        self.output_dir = ensure_directory(
+
+            output_dir
+        )
+
+    # ==================================================================
+    # DATASETS
+    # ==================================================================
+
+    def get_datasets(
+        self,
+        result: SimulationResult,
+    ) -> Dict[
+        str,
+        pd.DataFrame
+    ]:
+        """
+        Restituisce tutti i dataset.
+        """
+
+        return {
+
+            "weather":
+                result.weather_dataframe(),
+
+            "panel":
+                result.panel_dataframe(),
+
+            "combiner":
+                result.combiner_dataframe(),
+
+            "inverter":
+                result.inverter_dataframe(),
+
+            "plant":
+                result.plant_dataframe(),
+
+            "faults":
+                result.faults_dataframe(),
+        }
+
+    # ==================================================================
+    # CSV
+    # ==================================================================
+
+    def export_csv(
+        self,
+        result: SimulationResult,
+        directory: Optional[
+            str
+        ] = None,
+    ) -> Dict[
+        str,
+        Path
+    ]:
+        """
+        Esporta tutti i dataset CSV.
+        """
+
+        if directory is None:
+
+            directory = (
+
+                self.output_dir
+
+                / "csv"
+            )
+
+        directory = ensure_directory(
+
+            directory
+        )
+
+        created = {}
+
+        datasets = self.get_datasets(
+
+            result
+        )
+
+        for name, dataframe in datasets.items():
+
+            dataframe = normalize_dataframe(
+
+                dataframe
+            )
+
+            if dataframe.empty:
+
+                continue
+
+            path = (
+
+                directory
+
+                / f"{name}.csv"
+            )
+
+            dataframe.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                name
+            ] = path
+
+        return created
+
+    # ==================================================================
+    # PARQUET
+    # ==================================================================
+
+    def export_parquet(
+        self,
+        result: SimulationResult,
+        directory: Optional[
+            str
+        ] = None,
+        compression: str = "snappy",
+    ) -> Dict[
+        str,
+        Path
+    ]:
+        """
+        Esporta tutti i dataset Parquet.
+        """
+
+        if directory is None:
+
+            directory = (
+
+                self.output_dir
+
+                / "parquet"
+            )
+
+        directory = ensure_directory(
+
+            directory
+        )
+
+        created = {}
+
+        datasets = self.get_datasets(
+
+            result
+        )
+
+        for name, dataframe in datasets.items():
+
+            dataframe = normalize_dataframe(
+
+                dataframe
+            )
+
+            if dataframe.empty:
+
+                continue
+
+            path = (
+
+                directory
+
+                / f"{name}.parquet"
+            )
+
+            dataframe.to_parquet(
+
+                path,
+
+                index=False,
+
+                compression=
+
+                    compression
+            )
+
+            created[
+                name
+            ] = path
+
+        return created
+
+    # ==================================================================
+    # JSON
+    # ==================================================================
+
+    def export_json(
+        self,
+        result: SimulationResult,
+        filename: str = "simulation.json",
+    ) -> Path:
+        """
+        Esporta un riepilogo JSON della simulazione.
+
+        Per dataset molto grandi si raccomanda CSV o Parquet.
+        """
+
+        path = (
+
+            self.output_dir
+
+            / filename
+        )
+
+        data = {
+
+            "summary":
+
+                result.summary(),
+
+            "plant":
+
+                [
+
+                    item.to_dict()
+
+                    for item
+                    in result.plant_records
+                ],
+
+            "faults":
+
+                [
+
+                    item.to_dict()
+
+                    for item
+                    in result.fault_records
+                ],
+        }
+
+        return save_json(
+
+            data,
+
+            path
+        )
+
+    # ==================================================================
+    # KPI
+    # ==================================================================
+
+    def export_statistics(
+        self,
+        result: SimulationResult,
+    ) -> Dict[
+        str,
+        Path
+    ]:
+        """
+        Esporta i KPI e le statistiche.
+        """
+
+        statistics = SimulationStatistics(
+
+            result
+        )
+
+        created = {}
+
+        # --------------------------------------------------------------
+        # Plant KPI
+        # --------------------------------------------------------------
+
+        plant_kpi = statistics.plant_kpi()
+
+        path = (
+
+            self.output_dir
+
+            / "plant_kpi.json"
+        )
+
+        save_json(
+
+            plant_kpi.to_dict(),
+
+            path
+        )
+
+        created[
+            "plant_kpi"
+        ] = path
+
+        # --------------------------------------------------------------
+        # Panel
+        # --------------------------------------------------------------
+
+        panel = statistics.panel_kpi()
+
+        if not panel.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "panel_kpi.csv"
+            )
+
+            panel.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "panel_kpi"
+            ] = path
+
+        # --------------------------------------------------------------
+        # Combiner
+        # --------------------------------------------------------------
+
+        combiner = statistics.combiner_kpi()
+
+        if not combiner.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "combiner_kpi.csv"
+            )
+
+            combiner.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "combiner_kpi"
+            ] = path
+
+        # --------------------------------------------------------------
+        # Inverter
+        # --------------------------------------------------------------
+
+        inverter = statistics.inverter_kpi()
+
+        if not inverter.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "inverter_kpi.csv"
+            )
+
+            inverter.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "inverter_kpi"
+            ] = path
+
+        # --------------------------------------------------------------
+        # Daily
+        # --------------------------------------------------------------
+
+        daily = statistics.daily_statistics()
+
+        if not daily.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "daily_statistics.csv"
+            )
+
+            daily.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "daily"
+            ] = path
+
+        # --------------------------------------------------------------
+        # Monthly
+        # --------------------------------------------------------------
+
+        monthly = statistics.monthly_statistics()
+
+        if not monthly.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "monthly_statistics.csv"
+            )
+
+            monthly.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "monthly"
+            ] = path
+
+        # --------------------------------------------------------------
+        # Faults
+        # --------------------------------------------------------------
+
+        faults = statistics.fault_summary()
+
+        if not faults.empty:
+
+            path = (
+
+                self.output_dir
+
+                / "fault_summary.csv"
+            )
+
+            faults.to_csv(
+
+                path,
+
+                index=False
+            )
+
+            created[
+                "faults"
+            ] = path
+
+        return created
+
+    # ==================================================================
+    # EXPORT EVERYTHING
+    # ==================================================================
+
+    def export_all(
+        self,
+        result: SimulationResult,
+        csv: bool = True,
+        parquet: bool = True,
+        json: bool = True,
+        statistics: bool = True,
+    ) -> Dict[
+        str,
+        object
+    ]:
+        """
+        Esegue tutte le esportazioni.
+        """
+
+        output = {}
+
+        if csv:
+
+            output[
+                "csv"
+            ] = self.export_csv(
+
+                result
+            )
+
+        if parquet:
+
+            output[
+                "parquet"
+            ] = self.export_parquet(
+
+                result
+            )
+
+        if json:
+
+            output[
+                "json"
+            ] = self.export_json(
+
+                result
+            )
+
+        if statistics:
+
+            output[
+                "statistics"
+            ] = self.export_statistics(
+
+                result
+            )
+
+        return output
+
+
+# ======================================================================
+# HELPER
+# ======================================================================
+
+
+def export_simulation(
+    result: SimulationResult,
+    output_dir: str = "output",
+) -> Dict[
+    str,
+    object
+]:
+    """
+    Helper function.
+    """
+
+    exporter = SimulationExporter(
+
+        output_dir
+    )
+
+    return exporter.export_all(
+
+        result
+    )
+
+
+# ======================================================================
+
+
+__all__ = [
+
+    "SimulationExporter",
+
+    "export_simulation",
+]
+```