API Reference

def as_pandas(self):
    """Return a plain pandas DataFrame view of this object.

    Returns
    -------
    pandas.DataFrame
        Equivalent dataframe without solarpandas subclass semantics.
    """
    return pd.DataFrame(self)

def describe(self):
    """Compute descriptive statistics as a plain pandas dataframe.

    Returns
    -------
    pandas.DataFrame
        Result of ``pandas.DataFrame.describe`` on this dataset.
    """
    return self.as_pandas().describe()

@classmethod
def read_csv(cls, path: str | Path, **kwargs):
    """Read a CSV file written by :meth:`to_csv`.

    Parameters
    ----------
    path : str or pathlib.Path
        Input file path.
    **kwargs : Any
        Additional keyword arguments passed to :func:`pandas.read_csv`.

    Returns
    -------
    SolarDataFrame
        Parsed dataset with restored site metadata.

    Examples
    --------
    >>> sdf.to_csv("data.csv")
    >>> restored = SolarDataFrame.read_csv("data.csv")
    >>> restored.latitude == sdf.latitude
    True
    """
    if not (p := Path(path)).exists():
        raise ValueError(f"missing file {path}")

    metadata_line = linecache.getline(p.as_posix(), 1)
    must_kwargs = {"header": 1, "index_col": 0, "parse_dates": True}
    data = pd.read_csv(path, **(kwargs | must_kwargs))
    metadata = json.loads(metadata_line)

    return cls(
        data=data,
        latitude=float(metadata.pop("latitude")),
        longitude=float(metadata.pop("longitude")),
        elevation=float(metadata.pop("elevation")),
        custom_metadata=metadata,
    )

@classmethod
def read_parquet(cls, path: str | Path):
    """Read a Parquet file written by :meth:`to_parquet`.

    Parameters
    ----------
    path : str or pathlib.Path
        Input file path.

    Returns
    -------
    SolarDataFrame
        Parsed dataset with restored site metadata.

    Examples
    --------
    >>> sdf.to_parquet("data.parquet")
    >>> restored = SolarDataFrame.read_parquet("data.parquet")
    >>> restored.longitude == sdf.longitude
    True
    """
    if not (p := Path(path)).exists():
        raise ValueError(f"missing file {path}")
    table = pq.read_table(p)
    df = table.to_pandas()
    metadata_json = table.schema.metadata["solarpandas".encode()]
    metadata = {}
    for key, value in json.loads(metadata_json).items():
        if key.endswith("/ndarray"):
            metadata[key.split("/", 1)[0]] = np.array(value)
        elif key.endswith("/dataframe"):
            metadata[key.split("/", 1)[0]] = pd.DataFrame(value)
        else:
            metadata[key] = value

    return cls(
        data=df,
        latitude=float(metadata.pop("latitude")),
        longitude=float(metadata.pop("longitude")),
        elevation=float(metadata.pop("elevation")),
        custom_metadata=metadata,
    )

def replace_data(
    self, other: pd.Series | pd.DataFrame | Sequence[Number] | Number
) -> Self:
    """Create a copy with identical metadata and replaced data.

    Parameters
    ----------
    other : pandas.Series, pandas.DataFrame, sequence, or scalar number
        New data used to build the cloned object.

    Returns
    -------
    Self
        A new :class:`SolarDataFrame` preserving metadata from the current
        object.
    """
    kwargs = {
        "latitude": self.latitude,
        "longitude": self.longitude,
        "elevation": self.elevation,
        "custom_metadata": copy.deepcopy(self.custom_metadata),
    }
    if isinstance(other, Number):
        return self.__class__(
            data=np.full((len(self),), other), index=self.index, **kwargs
        )
    if isinstance(other, (np.ndarray, list)):
        return self.__class__(data=copy.copy(other), index=self.index, **kwargs)
    return self.__class__(data=copy.copy(other), **kwargs)

def to_csv(self, path: str | Path, **kwargs):
    """Write dataframe and metadata to a CSV file.

    The first output line stores a JSON document with metadata. Data values
    are written from the second line onward using ``pandas.DataFrame.to_csv``.

    Parameters
    ----------
    path : str or pathlib.Path
        Destination path.
    **kwargs : Any
        Extra keyword arguments passed to ``DataFrame.to_csv``.
    """
    metadata = {
        "latitude": self.latitude,
        "longitude": self.longitude,
        "elevation": self.elevation,
    } | self.custom_metadata

    if not (p := Path(path)).parent.exists():
        p.parent.mkdir(parents=True, exist_ok=True)

    default_kwargs = {"header": True}
    with p.open("w") as f:
        f.write(json.dumps(metadata) + "\n")
        pd.DataFrame(self).to_csv(f, **(default_kwargs | kwargs))

def to_parquet(self, path: str | Path, **kwargs):
    """Write dataframe and metadata to a Parquet file.

    Parameters
    ----------
    path : str or pathlib.Path
        Destination path.
    **kwargs : Any
        Extra keyword arguments passed to ``pyarrow.parquet.write_table``.

    Notes
    -----
    Metadata is stored in the Parquet schema metadata.
    """
    metadata = {
        "latitude": self.latitude,
        "longitude": self.longitude,
        "elevation": self.elevation,
    } | self.custom_metadata

    # convert the dataframe to a Arrow Table
    table = pa.Table.from_pandas(self)

    # copy the metadata to a dictionary and convert the numpy
    # arrays and dataframes to a json-serializable format
    df_metadata = {}
    for key, value in metadata.items():
        if isinstance(value, np.ndarray):
            df_metadata[f"{key}/ndarray"] = value.tolist()
        elif isinstance(value, pd.DataFrame):
            df_metadata[f"{key}/dataframe"] = value.to_dict()
        else:
            df_metadata[key] = value

    # add the dictionary to the schema metadata of table. Note
    # that I am using the keyword "syngena" and encode to bytes
    combined_metadata = {
        "solarpandas".encode(): json.dumps(df_metadata).encode(),
        **table.schema.metadata,
    }  # DataFrame"s metadata
    table = table.replace_schema_metadata(combined_metadata)

    # serialize to a parquet file
    p = Path(path)
    if not p.parent.exists():
        p.parent.mkdir(parents=True, exist_ok=True)
    pa.parquet.write_table(table, p, **kwargs)

def replace_data(
    self, other: pd.Series | pd.DataFrame | Sequence[Number] | Number
) -> Self:
    """Create a copy with identical site metadata and new data values.

    Parameters
    ----------
    other : pandas.Series, pandas.DataFrame, sequence, or scalar number
        New data values.

    Returns
    -------
    Self
        A new :class:`SolarSeries` preserving index (when applicable) and
        metadata from the source object.
    """
    kwargs = {
        "latitude": self.latitude,
        "longitude": self.longitude,
        "elevation": self.elevation,
        "custom_metadata": copy.deepcopy(self.custom_metadata),
    }
    if isinstance(other, Number):
        return self.__class__(
            data=np.full((len(self),), other), index=self.index, **kwargs
        )
    if isinstance(other, (np.ndarray, list)):
        return self.__class__(data=copy.copy(other), index=self.index, **kwargs)
    return self.__class__(data=copy.copy(other), **kwargs)

@staticmethod
def clear_cache() -> None:
    """Clear the in-memory solar-position cache.

    Notes
    -----
    Use this when changing model options and forcing a full recomputation.
    """
    _compute_cached_solpos.cache_clear()
    logger.debug("solpos cache cleared")

def compute(
    self, algorithm: str = "psa", refraction: bool = True, engine: str = "numexpr"
) -> sunwhere._base.Sunpos:
    """Compute solar position without using the accessor cache.

    Parameters
    ----------
    algorithm : str, default "psa"
        Solar-position algorithm accepted by ``sunwhere``.
    refraction : bool, default True
        Whether to include atmospheric refraction corrections.
    engine : str, default "numexpr"
        Backend engine used by ``sunwhere``.

    Returns
    -------
    sunwhere._base.Sunpos
        ``sunwhere`` result object containing angular and temporal variables.
    """
    logger.debug(
        f"evaluating solar position with `{algorithm}` algorithm, "
        f"refraction={refraction}, engine=`{engine}`..."
    )
    args = (self._sdf.index, self._sdf.latitude, self._sdf.longitude)
    kwargs = {"algorithm": algorithm, "refraction": refraction, "engine": engine}
    return sunwhere.sites(*args, **kwargs)

@staticmethod
def get_cache_info():
    """Return cache statistics for solar-position computations.

    Returns
    -------
    dict[str, int | None]
        Dictionary with ``hits``, ``misses``, ``current_size`` and ``max_size``.
    """
    info = _compute_cached_solpos.cache_info()
    return {
        "hits": info.hits,
        "misses": info.misses,
        "current_size": info.currsize,
        "max_size": info.maxsize,
    }

def sunrise(self, units: Literal["rad", "deg", "tst", "lst", "utc"] = "utc"):
    """Return sunrise in the selected coordinate system.

    Parameters
    ----------
    units : {"rad", "deg", "tst", "lst", "utc"}, default "utc"
        Units or time reference used by ``sunwhere``.

    Returns
    -------
    SolarSeries
        Sunrise values aligned to the dataframe index.
    """
    sr = self._get_cached_solpos("sunrise", units={"lst": "utc"}.get(units, units))
    if units == "lst":
        # convert from UTC to LST
        deltat = pd.Timedelta(self._sdf.longitude * 4, "min")
        sr = sr + deltat
    return sr

def sunset(self, units: Literal["rad", "deg", "tst", "lst", "utc"] = "utc"):
    """Return sunset in the selected coordinate system.

    Parameters
    ----------
    units : {"rad", "deg", "tst", "lst", "utc"}, default "utc"
        Units or time reference used by ``sunwhere``.

    Returns
    -------
    SolarSeries
        Sunset values aligned to the dataframe index.
    """
    ss = self._get_cached_solpos("sunset", units={"lst": "utc"}.get(units, units))
    if units == "lst":
        # convert from UTC to LST
        deltat = pd.Timedelta(self._sdf.longitude * 4, "min")
        ss = ss + deltat
    return ss

@staticmethod
def clear_cache() -> None:
    """Clear the in-memory clear-sky irradiance cache.
    """
    _compute_cached_clearsky.cache_clear()
    logger.debug("clearsky cache cleared")

@staticmethod
def get_cache_info():
    """Return cache statistics for clear-sky computations.

    Returns
    -------
    dict[str, int | None]
        Dictionary with ``hits``, ``misses``, ``current_size`` and ``max_size``.
    """
    info = _compute_cached_clearsky.cache_info()
    return {
        "hits": info.hits,
        "misses": info.misses,
        "current_size": info.currsize,
        "max_size": info.maxsize,
    }

@staticmethod
def clear_cache() -> None:
    """Clear the in-memory clear-sky irradiance cache.
    """
    _compute_cached_clearsky.cache_clear()
    logger.debug("clearsky cache cleared")

@staticmethod
def get_cache_info():
    """Return cache statistics for clear-sky computations.

    Returns
    -------
    dict[str, int | None]
        Dictionary with ``hits``, ``misses``, ``current_size`` and ``max_size``.
    """
    info = _compute_cached_clearsky.cache_info()
    return {
        "hits": info.hits,
        "misses": info.misses,
        "current_size": info.currsize,
        "max_size": info.maxsize,
    }

@staticmethod
def clear_cache() -> None:
    """Clear the in-memory clear-sky irradiance cache.
    """
    _compute_cached_clearsky.cache_clear()
    logger.debug("clearsky cache cleared")

def compute(
    self,
    atmosphere: Literal[
        "merra2_daily", "merra2_gee", "merra2_lta", "crs_soda", "custom"
    ],
    model: str = "SPARTA",
) -> SolarDataFrame:
    """Compute clear-sky irradiance once without using cache.

    Parameters
    ----------
    atmosphere : {"merra2_daily", "merra2_gee", "merra2_lta", "crs_soda", "custom"}
        Atmosphere dataset source.
    model : str, default "SPARTA"
        Irradiance model name accepted by the selected atmosphere backend.

    Returns
    -------
    SolarDataFrame
        Dataframe with columns ``ghi``, ``dni``, ``dif`` and ``csi``
        computed for the requested atmosphere and model."""
    logger.debug(
        f"evaluating clearsky with `{model}` model and `{atmosphere}` atmosphere..."
    )
    if not hasattr(spartasolar.atmosphere, atmosphere):
        raise ValueError(f"invalid clearsky atmosphere `{atmosphere}`")
    if atmosphere == "custom":
        raise NotImplementedError(
            "TODO: implement support for user-provided custom "
            "atmosphere datasets from dataframe columns"
        )
    atmos_obj = getattr(spartasolar.atmosphere, atmosphere)
    args = (
        self._sdf.index.tz_convert("UTC").tz_localize(None),
        self._sdf.latitude,
        self._sdf.longitude,
    )
    try:
        xa_result = atmos_obj.at_site(*args).compute(model)
    except AttributeError:
        xa_result = atmos_obj.at_sites(*args).compute(model)
    df_result = (
        xa_result.isel(site=0).drop_vars(["lat", "lon", "site"]).to_dataframe()
    )
    return self._sdf.replace_data(df_result)

@staticmethod
def get_cache_info():
    """Return cache statistics for clear-sky computations.

    Returns
    -------
    dict[str, int | None]
        Dictionary with ``hits``, ``misses``, ``current_size`` and ``max_size``.
    """
    info = _compute_cached_clearsky.cache_info()
    return {
        "hits": info.hits,
        "misses": info.misses,
        "current_size": info.currsize,
        "max_size": info.maxsize,
    }