diff --git a/pvsim/utils.py b/pvsim/utils.py new file mode 100644 index 0000000..30e3ced --- /dev/null +++ b/pvsim/utils.py @@ -0,0 +1,590 @@ +""" +pvsim/utils.py + +Funzioni di utilità condivise dal progetto PV Simulator. + +Contiene funzioni per: + +- gestione timestamp; +- generazione intervalli temporali; +- conversione oggetti -> dizionari; +- conversione oggetti -> DataFrame; +- aggregazione temporale; +- gestione directory; +- conversione sicura dei valori numerici; +- calcolo di energia; +- serializzazione JSON. + +Questo modulo non contiene logica specifica del modello fotovoltaico. +""" + +from __future__ import annotations + +import json + +from dataclasses import asdict, is_dataclass + +from datetime import datetime, timedelta + +from pathlib import Path + +from typing import Any, Dict, Iterable, List, Optional + +import pandas as pd + + +# ====================================================================== +# TIME +# ====================================================================== + + +def generate_time_range( + start: datetime, + end: datetime, + timestep_minutes: int, +) -> Iterable[datetime]: + """ + Genera una sequenza temporale uniforme. + + Parameters + ---------- + start: + Timestamp iniziale. + + end: + Timestamp finale, escluso. + + timestep_minutes: + Durata del timestep in minuti. + + Yields + ------ + datetime + """ + + if end <= start: + + raise ValueError( + "end deve essere maggiore di start" + ) + + if timestep_minutes <= 0: + + raise ValueError( + "timestep_minutes deve essere > 0" + ) + + current = start + + delta = timedelta( + + minutes=timestep_minutes + ) + + while current < end: + + yield current + + current += delta + + +# ====================================================================== + + +def datetime_to_string( + value: datetime, +) -> str: + """ + Converte datetime in stringa ISO. + """ + + return value.isoformat() + + +# ====================================================================== + + +def ensure_datetime( + value: Any, +) -> datetime: + """ + Converte un valore in datetime. + """ + + if isinstance( + value, + datetime, + ): + + return value + + return pd.to_datetime( + + value + ).to_pydatetime() + + +# ====================================================================== +# OBJECT CONVERSION +# ====================================================================== + + +def object_to_dict( + obj: Any, +) -> Dict[str, Any]: + """ + Converte un oggetto in dizionario. + + Supporta: + + - dataclass; + - oggetti con to_dict(); + - oggetti con __dict__; + - dizionari. + """ + + if isinstance( + obj, + dict, + ): + + return dict( + obj + ) + + if hasattr( + obj, + "to_dict", + ): + + return obj.to_dict() + + if is_dataclass( + obj + ): + + return asdict( + obj + ) + + if hasattr( + obj, + "__dict__", + ): + + return dict( + obj.__dict__ + ) + + raise TypeError( + f"Tipo non supportato: " + f"{type(obj)}" + ) + + +# ====================================================================== + + +def objects_to_dataframe( + objects: Iterable[Any], +) -> pd.DataFrame: + """ + Converte una sequenza di oggetti in DataFrame. + """ + + records = [ + + object_to_dict( + obj + ) + + for obj in objects + ] + + if not records: + + return pd.DataFrame() + + return pd.DataFrame( + records + ) + + +# ====================================================================== +# DATAFRAME +# ====================================================================== + + +def normalize_dataframe( + dataframe: pd.DataFrame, +) -> pd.DataFrame: + """ + Normalizza un DataFrame. + + Operazioni: + + - copia del DataFrame; + - conversione timestamp; + - ordinamento temporale; + - reset dell'indice. + """ + + if dataframe.empty: + + return dataframe.copy() + + dataframe = dataframe.copy() + + if "timestamp" in dataframe.columns: + + dataframe[ + "timestamp" + ] = pd.to_datetime( + + dataframe[ + "timestamp" + ] + ) + + dataframe = dataframe.sort_values( + + "timestamp" + ) + + return dataframe.reset_index( + + drop=True + ) + + +# ====================================================================== + + +def aggregate_dataframe( + dataframe: pd.DataFrame, + frequency: str = "1D", + numeric_only: bool = True, +) -> pd.DataFrame: + """ + Aggrega un DataFrame temporalmente. + + Esempio: + + daily = aggregate_dataframe( + df, + "1D" + ) + + Frequenze comuni: + + 5min + 15min + 1H + 1D + 1W + 1ME + """ + + if dataframe.empty: + + return dataframe.copy() + + if "timestamp" not in dataframe.columns: + + raise ValueError( + "Il DataFrame deve contenere " + "la colonna timestamp" + ) + + df = normalize_dataframe( + + dataframe + ) + + df = df.set_index( + + "timestamp" + ) + + result = ( + + df.resample( + + frequency + ).mean( + + numeric_only=numeric_only + ) + ) + + return result.reset_index() + + +# ====================================================================== + + +def ensure_directory( + path: str | Path, +) -> Path: + """ + Crea una directory se non esiste. + """ + + path = Path( + path + ) + + path.mkdir( + + parents=True, + + exist_ok=True + ) + + return path + + +# ====================================================================== +# NUMBERS +# ====================================================================== + + +def safe_float( + value: Any, + default: float = 0.0, +) -> float: + """ + Converte un valore in float in modo sicuro. + """ + + try: + + if value is None: + + return default + + return float( + value + ) + + except ( + TypeError, + ValueError, + ): + + return default + + +# ====================================================================== + + +def safe_divide( + numerator: float, + denominator: float, + default: float = 0.0, +) -> float: + """ + Divisione sicura. + """ + + if denominator == 0: + + return default + + return numerator / denominator + + +# ====================================================================== + + +def power_to_energy( + power_W: float, + timestep_minutes: float, +) -> float: + """ + Converte potenza istantanea in energia. + + W * h = Wh + """ + + timestep_hours = ( + + timestep_minutes + + / 60.0 + ) + + return ( + + power_W + + * timestep_hours + ) + + +# ====================================================================== +# JSON +# ====================================================================== + + +def make_json_serializable( + value: Any, +) -> Any: + """ + Converte ricorsivamente oggetti Python + in strutture compatibili con JSON. + """ + + if isinstance( + value, + datetime, + ): + + return value.isoformat() + + if is_dataclass( + value + ): + + return make_json_serializable( + + asdict( + value + ) + ) + + if isinstance( + value, + dict, + ): + + return { + + str(key): + + make_json_serializable( + item + ) + + for key, item + in value.items() + } + + if isinstance( + value, + ( + list, + tuple, + set, + ), + ): + + return [ + + make_json_serializable( + item + ) + + for item in value + ] + + if hasattr( + value, + "item", + ): + + try: + + return value.item() + + except ( + ValueError, + TypeError, + ): + + pass + + return value + + +# ====================================================================== + + +def save_json( + data: Any, + path: str | Path, + indent: int = 2, +) -> Path: + """ + Salva dati in JSON. + """ + + path = Path( + path + ) + + ensure_directory( + + path.parent + ) + + serializable = ( + + make_json_serializable( + data + ) + ) + + with path.open( + + "w", + + encoding="utf-8", + ) as file: + + json.dump( + + serializable, + + file, + + indent=indent, + + ensure_ascii=False, + ) + + return path + + +# ====================================================================== + + +__all__ = [ + + "generate_time_range", + + "datetime_to_string", + + "ensure_datetime", + + "object_to_dict", + + "objects_to_dataframe", + + "normalize_dataframe", + + "aggregate_dataframe", + + "ensure_directory", + + "safe_float", + + "safe_divide", + + "power_to_energy", + + "make_json_serializable", + + "save_json", +] +```