Giovanni/PV_Simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a424fad26b537267d8362c8963c8420d9b74bb9c
PV_Simulator/pvsim/storage.py

1445 lines
26 KiB
Python
Raw Blame History

"""
pvsim/storage.py
Layer di persistenza dei dati prodotti dal simulatore fotovoltaico.
Il modulo permette di salvare SimulationResult in:
- CSV
- Parquet
- SQLite
Struttura logica dei dati:
simulation
|
+-- weather
|
+-- panel
|
+-- combiner
|
+-- inverter
|
+-- plant
|
+-- faults
Il modulo è indipendente dal motore di simulazione.
Pipeline:
PVSimulator
|
v
SimulationResult
|
v
SimulationStorage
|
+---- CSV
|
+---- Parquet
|
+---- SQLite
Esempio:
result = simulator.run(config)
storage = SimulationStorage(
output_dir="output"
)
storage.save_csv(result)
storage.save_parquet(result)
storage.save_sqlite(
result,
"output/pv_simulation.db"
)
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional
import pandas as pd
from .simulator import SimulationResult
# ======================================================================
# STORAGE CONFIGURATION
# ======================================================================
@dataclass
class StorageConfig:
"""
Configurazione del sistema di storage.
Parameters
----------
output_dir:
Directory principale di output.
csv_dir:
Sottodirectory CSV.
parquet_dir:
Sottodirectory Parquet.
sqlite_filename:
Nome del database SQLite.
overwrite:
Se True, i file esistenti vengono sovrascritti.
compression:
Compressione Parquet.
Valori tipici:
"snappy"
"gzip"
"brotli"
None
"""
output_dir: str = "output"
csv_dir: str = "csv"
parquet_dir: str = "parquet"
sqlite_filename: str = "pv_simulation.db"
overwrite: bool = True
compression: Optional[str] = "snappy"
def output_path(
self,
) -> Path:
"""
Restituisce il path principale di output.
"""
return Path(
self.output_dir
)
def csv_path(
self,
) -> Path:
"""
Restituisce il path della directory CSV.
"""
return (
self.output_path()
/ self.csv_dir
)
def parquet_path(
self,
) -> Path:
"""
Restituisce il path della directory Parquet.
"""
return (
self.output_path()
/ self.parquet_dir
)
def sqlite_path(
self,
) -> Path:
"""
Restituisce il path del database SQLite.
"""
return (
self.output_path()
/ self.sqlite_filename
)
# ======================================================================
# STORAGE CLASS
# ======================================================================
class SimulationStorage:
"""
Gestisce la persistenza dei risultati della simulazione.
Parameters
----------
config:
Configurazione dello storage.
Esempio
-------
storage = SimulationStorage()
storage.save_csv(
result
)
storage.save_parquet(
result
)
storage.save_sqlite(
result
)
"""
def __init__(
self,
config: Optional[
StorageConfig
] = None,
output_dir: Optional[
str
] = None,
) -> None:
if config is not None:
self.config = config
elif output_dir is not None:
self.config = StorageConfig(
output_dir=
output_dir
)
else:
self.config = StorageConfig()
self._create_directories()
# ==================================================================
# DIRECTORIES
# ==================================================================
def _create_directories(
self,
) -> None:
"""
Crea le directory necessarie.
"""
self.config.output_path().mkdir(
parents=True,
exist_ok=True
)
self.config.csv_path().mkdir(
parents=True,
exist_ok=True
)
self.config.parquet_path().mkdir(
parents=True,
exist_ok=True
)
# ==================================================================
# DATASET EXTRACTION
# ==================================================================
def _get_datasets(
self,
result: SimulationResult,
) -> Dict[
str,
pd.DataFrame
]:
"""
Estrae tutti i dataset dal SimulationResult.
Returns
-------
Dict[str, DataFrame]
Keys:
weather
panel
combiner
inverter
plant
faults
"""
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(),
}
# ==================================================================
# DATAFRAME NORMALIZATION
# ==================================================================
@staticmethod
def _normalize_dataframe(
dataframe: pd.DataFrame,
) -> pd.DataFrame:
"""
Normalizza un DataFrame prima del salvataggio.
In particolare:
- converte timestamp in datetime;
- ordina temporalmente;
- resetta l'indice.
"""
if dataframe.empty:
return dataframe
dataframe = dataframe.copy()
if "timestamp" in dataframe.columns:
dataframe[
"timestamp"
] = pd.to_datetime(
dataframe[
"timestamp"
]
)
dataframe = dataframe.sort_values(
by="timestamp"
)
dataframe = dataframe.reset_index(
drop=True
)
return dataframe
# ==================================================================
# CSV
# ==================================================================
def save_csv(
self,
result: SimulationResult,
prefix: str = "",
) -> Dict[
str,
Path
]:
"""
Salva i dati in formato CSV.
Crea:
weather.csv
panel.csv
combiner.csv
inverter.csv
plant.csv
faults.csv
Parameters
----------
result:
Risultato della simulazione.
prefix:
Prefisso opzionale per i file.
Returns
-------
Dict[str, Path]
Path dei file creati.
"""
datasets = (
self._get_datasets(
result
)
)
created_files = {}
for name, dataframe in datasets.items():
dataframe = (
self._normalize_dataframe(
dataframe
)
)
if dataframe.empty:
continue
filename = (
f"{prefix}{name}.csv"
)
output_file = (
self.config.csv_path()
/ filename
)
if (
output_file.exists()
and not self.config.overwrite
):
raise FileExistsError(
f"Il file esiste già: "
f"{output_file}"
)
dataframe.to_csv(
output_file,
index=False
)
created_files[
name
] = output_file
return created_files
# ==================================================================
# PARQUET
# ==================================================================
def save_parquet(
self,
result: SimulationResult,
prefix: str = "",
) -> Dict[
str,
Path
]:
"""
Salva i dati in formato Parquet.
Parquet è consigliato per simulazioni
di grandi dimensioni perché:
- occupa meno spazio;
- è più veloce da leggere;
- mantiene i tipi delle colonne;
- è adatto a Data Lake e pipeline analytics.
Returns
-------
Dict[str, Path]
Path dei file creati.
"""
datasets = (
self._get_datasets(
result
)
)
created_files = {}
for name, dataframe in datasets.items():
dataframe = (
self._normalize_dataframe(
dataframe
)
)
if dataframe.empty:
continue
filename = (
f"{prefix}{name}.parquet"
)
output_file = (
self.config.parquet_path()
/ filename
)
if (
output_file.exists()
and not self.config.overwrite
):
raise FileExistsError(
f"Il file esiste già: "
f"{output_file}"
)
dataframe.to_parquet(
output_file,
index=False,
compression=
self.config.compression
)
created_files[
name
] = output_file
return created_files
# ==================================================================
# SQLITE CONNECTION
# ==================================================================
def _connect_sqlite(
self,
database_path: Optional[
str
] = None,
) -> sqlite3.Connection:
"""
Apre una connessione SQLite.
"""
if database_path is None:
database_path = (
str(
self.config
.sqlite_path()
)
)
else:
database_path = str(
database_path
)
Path(
database_path
).parent.mkdir(
parents=True,
exist_ok=True
)
connection = sqlite3.connect(
database_path
)
connection.execute(
"""
PRAGMA journal_mode=WAL;
"""
)
connection.execute(
"""
PRAGMA foreign_keys=ON;
"""
)
return connection
# ==================================================================
# SQLITE SCHEMA
# ==================================================================
def _create_sqlite_schema(
self,
connection: sqlite3.Connection,
) -> None:
"""
Crea gli indici principali del database.
Le tabelle vengono create dinamicamente
da pandas.to_sql.
Gli indici vengono poi aggiunti manualmente.
"""
# --------------------------------------------------------------
# Panel
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_panel_timestamp
ON panel(timestamp);
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_panel_id_timestamp
ON panel(
panel_id,
timestamp
);
"""
)
# --------------------------------------------------------------
# Combiner
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_combiner_timestamp
ON combiner(timestamp);
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_combiner_id_timestamp
ON combiner(
combiner_id,
timestamp
);
"""
)
# --------------------------------------------------------------
# Inverter
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_inverter_timestamp
ON inverter(timestamp);
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_inverter_id_timestamp
ON inverter(
inverter_id,
timestamp
);
"""
)
# --------------------------------------------------------------
# Plant
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_plant_timestamp
ON plant(timestamp);
"""
)
# --------------------------------------------------------------
# Weather
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_weather_timestamp
ON weather(timestamp);
"""
)
# --------------------------------------------------------------
# Faults
# --------------------------------------------------------------
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_fault_timestamp
ON faults(timestamp);
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_fault_component
ON faults(
component_id,
timestamp
);
"""
)
connection.commit()
# ==================================================================
# SQLITE
# ==================================================================
def save_sqlite(
self,
result: SimulationResult,
database_path: Optional[
str
] = None,
if_exists: str = "append",
) -> Path:
"""
Salva il risultato in SQLite.
Tabelle create:
weather
panel
combiner
inverter
plant
faults
Parameters
----------
result:
Risultato della simulazione.
database_path:
Path opzionale del database.
if_exists:
Strategia pandas.to_sql:
append
replace
fail
Returns
-------
Path
Path del database.
"""
if database_path is None:
database_path = (
self.config.sqlite_path()
)
database_path = Path(
database_path
)
if (
database_path.exists()
and not self.config.overwrite
and if_exists == "replace"
):
raise FileExistsError(
f"Il database esiste già: "
f"{database_path}"
)
connection = (
self._connect_sqlite(
str(
database_path
)
)
)
try:
datasets = (
self._get_datasets(
result
)
)
for name, dataframe in datasets.items():
dataframe = (
self._normalize_dataframe(
dataframe
)
)
if dataframe.empty:
continue
dataframe.to_sql(
name,
connection,
if_exists=
if_exists,
index=False
)
self._create_sqlite_schema(
connection
)
finally:
connection.close()
return database_path
# ==================================================================
# QUERY SQLITE
# ==================================================================
def query_sqlite(
self,
query: str,
database_path: Optional[
str
] = None,
) -> pd.DataFrame:
"""
Esegue una query SQL e restituisce un DataFrame.
Esempio:
df = storage.query_sqlite(
'''
SELECT
timestamp,
ac_power_W
FROM plant
ORDER BY timestamp
'''
)
"""
if database_path is None:
database_path = (
self.config.sqlite_path()
)
connection = (
self._connect_sqlite(
str(
database_path
)
)
)
try:
dataframe = pd.read_sql_query(
query,
connection
)
finally:
connection.close()
return dataframe
# ==================================================================
# READ TABLE
# ==================================================================
def read_table(
self,
table_name: str,
database_path: Optional[
str
] = None,
) -> pd.DataFrame:
"""
Legge una tabella SQLite.
Tabelle valide:
weather
panel
combiner
inverter
plant
faults
"""
allowed_tables = {
"weather",
"panel",
"combiner",
"inverter",
"plant",
"faults",
}
if table_name not in allowed_tables:
raise ValueError(
"Tabella non valida. "
f"Valori consentiti: "
f"{sorted(allowed_tables)}"
)
query = (
f"SELECT * "
f"FROM {table_name}"
)
return self.query_sqlite(
query,
database_path
)
# ==================================================================
# READ TIME RANGE
# ==================================================================
def read_time_range(
self,
table_name: str,
start: str,
end: str,
database_path: Optional[
str
] = None,
) -> pd.DataFrame:
"""
Legge i dati di una tabella in un intervallo temporale.
"""
allowed_tables = {
"weather",
"panel",
"combiner",
"inverter",
"plant",
"faults",
}
if table_name not in allowed_tables:
raise ValueError(
"Tabella non valida."
)
query = f"""
SELECT *
FROM {table_name}
WHERE timestamp >= ?
AND timestamp < ?
ORDER BY timestamp
"""
if database_path is None:
database_path = (
self.config.sqlite_path()
)
connection = (
self._connect_sqlite(
str(
database_path
)
)
)
try:
dataframe = pd.read_sql_query(
query,
connection,
params=(
start,
end
)
)
finally:
connection.close()
return dataframe
# ==================================================================
# READ COMPONENT
# ==================================================================
def read_component(
self,
table_name: str,
component_id: str,
database_path: Optional[
str
] = None,
) -> pd.DataFrame:
"""
Legge la serie temporale di un componente.
Esempio:
storage.read_component(
"panel",
"PANEL_001"
)
"""
component_columns = {
"panel":
"panel_id",
"combiner":
"combiner_id",
"inverter":
"inverter_id",
}
if table_name not in component_columns:
raise ValueError(
"La lettura per componente "
"è supportata per: "
"panel, combiner, inverter"
)
column = (
component_columns[
table_name
]
)
query = f"""
SELECT *
FROM {table_name}
WHERE {column} = ?
ORDER BY timestamp
"""
if database_path is None:
database_path = (
self.config.sqlite_path()
)
connection = (
self._connect_sqlite(
str(
database_path
)
)
)
try:
dataframe = pd.read_sql_query(
query,
connection,
params=(
component_id,
)
)
finally:
connection.close()
return dataframe
# ==================================================================
# PLANT POWER QUERY
# ==================================================================
def read_plant_power(
self,
start: Optional[
str
] = None,
end: Optional[
str
] = None,
database_path: Optional[
str
] = None,
) -> pd.DataFrame:
"""
Restituisce la produzione AC dell'impianto.
Se start/end sono specificati,
filtra l'intervallo temporale.
"""
query = """
SELECT
timestamp,
ac_power_W,
effective_ac_power_W,
energy_Wh,
cumulative_energy_Wh,
performance_ratio,
availability,
fault_active
FROM plant
"""
params = []
conditions = []
if start is not None:
conditions.append(
"timestamp >= ?"
)
params.append(
start
)
if end is not None:
conditions.append(
"timestamp < ?"
)
params.append(
end
)
if conditions:
query += (
" WHERE "
+ " AND ".join(
conditions
)
)
query += (
" ORDER BY timestamp"
)
if database_path is None:
database_path = (
self.config.sqlite_path()
)
connection = (
self._connect_sqlite(
str(
database_path
)
)
)
try:
dataframe = pd.read_sql_query(
query,
connection,
params=params
)
finally:
connection.close()
return dataframe
# ==================================================================
# FULL SAVE
# ==================================================================
def save_all(
self,
result: SimulationResult,
save_csv: bool = True,
save_parquet: bool = True,
save_sqlite: bool = True,
) -> Dict[
str,
object
]:
"""
Salva il risultato in tutti i formati selezionati.
Returns
-------
Dict
Dizionario contenente i file generati.
"""
output = {}
if save_csv:
output[
"csv"
] = self.save_csv(
result
)
if save_parquet:
output[
"parquet"
] = self.save_parquet(
result
)
if save_sqlite:
output[
"sqlite"
] = self.save_sqlite(
result
)
return output
# ======================================================================
# UTILITY FUNCTIONS
# ======================================================================
def save_simulation(
result: SimulationResult,
output_dir: str = "output",
) -> Dict[
str,
object
]:
"""
Funzione helper per salvare una simulazione
in tutti i formati.
Esempio:
from pvsim.storage import save_simulation
files = save_simulation(
result,
"output"
)
"""
storage = SimulationStorage(
output_dir=
output_dir
)
return storage.save_all(
result
)
# ======================================================================
# EXPORT
# ======================================================================
__all__ = [
"StorageConfig",
"SimulationStorage",
"save_simulation",
]"""
Reference in New Issue View Git Blame Copy Permalink