diff --git a/pvsim/storage.py b/pvsim/storage.py new file mode 100644 index 0000000..2094d16 --- /dev/null +++ b/pvsim/storage.py @@ -0,0 +1,1444 @@ +""" +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", +]"""