601 lines
11 KiB
Python
601 lines
11 KiB
Python
"""
|
|
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",
|
|
]
|
|
```
|