PV_Simulator/pvsim/utils.py

"""
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",
]