Funzioni di utilità condivise dal progetto PV Simulator.

2026-07-19 20:23:00 +00:00
parent a424fad26b
commit 4beb2ba677
1 changed files with 590 additions and 0 deletions
--- a/pvsim/utils.py
+++ 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",
 ]
 ```