giocaizzi / pysurfline Goto Github PK

"""api functions and classes"""

import requests

from .models.spots import Wave, Wind, Weather, SunlightTimes, Tides, Details


class ApiResponseObject:
    _data: dict = None  # spot/forecasts response
    _associated: dict = None
    _spot: dict = None  # spot/details response
    # permissions : dict = None TODO: add permissions
    _url: str = None
    model_class = None

    def __init__(self, response: requests.Response, model_class=None):
        if "data" in response.json():
            self._data = response.json()["data"]
        if "associated" in response.json():
            self._associated = response.json()["associated"]
        if "spot" in response.json():
            self._spot = response.json()["spot"]
        # urse
        self._url = response.url

        # parse data

        # type(model_class) is Details returns False as it is a class
        # type(model_class) == Details returns True when model_class is an istance
        # of a class
        if model_class is not None:
            self.model_class = model_class
            if self._data is not None:
                self._parse_data(model_class)

    @property
    def data(self):
        return self._data

    @property
    def associated(self):
        return self._associated

    @property
    def spot(self):
        return self._spot

    @property
    def url(self):
        return self._url

    def _parse_data(self, model_class) -> None:
        """parse data into model class"""
        # make keys lowercase
        self._data = {key.lower(): value for key, value in self._data.items()}

        self._data = [
            model_class(**item) for item in self._data[model_class.__name__.lower()]
        ]

    def __str__(self):
        if self.model_class is None:
            return f"ApiObject({self.url})"
        else:
            return f"{self.model_class.__name__}({self.url})"

    def __repr__(self):
        return str(self)


class SpotForecastsWave(ApiResponseObject):
    """spots/forecasts/wave endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=Wave)


class SpotForecastsWind(ApiResponseObject):
    """spots/forecasts/wind endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=Wind)


class SpotForecastsWeather(ApiResponseObject):
    """spots/forecasts/weather endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=Weather)


class SpotForecastsSunlightTimes(ApiResponseObject):
    """spots/forecasts/sunlightTimes endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=SunlightTimes)


class SpotForecastsTides(ApiResponseObject):
    """spots/forecasts/tides endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=Tides)


class SpotDetails(ApiResponseObject):
    """spots/details endpoint"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs, model_class=Details)

from pysurfline.api.services import ApiService
from pysurfline.core import SpotForecasts


class SurflineClient:
    """surfline client

    Surfline API client.
    At the moment, does not require authentication.

    TODO: Login and authentication
    """

    _baseurl: str = "https://services.surfline.com/kbyg/"

    def get_spot_forecasts(
        self, spotId: str, intervalHours: int = None, days: int = None
    ) -> SpotForecasts:
        """create a SpotForecast object from API responses

        Arguments:
            spotId (str): spot id
            intervalHours (int, optional): interval hours. Defaults to None.
            days (int, optional): days. Defaults to None.

        Returns:
            SpotForecast: SpotForecast object
        """
        params = {}
        params["spotId"] = spotId

        # add optional parameters
        if intervalHours:
            params["intervalHours"] = intervalHours
        if days:
            params["days"] = days

        return SpotForecasts(
            spotId,
            # spot datails are integrated with forecast,
            # hence the different GET args
            details=ApiService(self, "spots/details").get({"spotId": spotId}),
            waves=ApiService(self, "spots/forecasts/wave").get(params=params),
            winds=ApiService(self, "spots/forecasts/wind").get(params=params),
            tides=ApiService(self, "spots/forecasts/tides").get(params=params),
            # weather and sunlight times are in the same response
            weather=ApiService(self, "spots/forecasts/weather").get(params=params)[0],
            sunlightTimes=ApiService(self, "spots/forecasts/weather").get(
                params=params
            )[1],
        )

"""core module: SpotForecast"""

import pandas as pd

from .utils import flatten
from .api.objects import (
    SpotDetails,
    SpotForecastsWave,
    SpotForecastsWind,
    SpotForecastsTides,
    SpotForecastsSunlightTimes,
    SpotForecastsWeather,
)


class SpotForecasts:
    """spot forecasts data model

    Composite data model of all the spot forecasts data,
    - wave
    - condition (TODO)
    - wind
    - tides
    - weather and sunrise and sunset times

    TODO: add associated data and improve utcOffset
    """

    name: str = None

    def __init__(
        self,
        spotId: str,
        details: SpotDetails,
        waves: SpotForecastsWave,
        winds: SpotForecastsWind,
        tides: SpotForecastsTides,
        sunlightTimes: SpotForecastsSunlightTimes,
        weather: SpotForecastsWeather,
    ):
        self.spotId = spotId
        self.name = details.spot["name"]
        self.waves = waves.data
        self.wind = winds.data
        self.tides = tides.data
        self.weather = weather.data
        self.sunlightTimes = sunlightTimes.data

    def get_dataframe(self, attr="surf") -> pd.DataFrame:
        """pandas dataframe of selected attribute

        Use default to get the pandas dataframe of surf data,
        or of the selected attribute `attr`:
        - waves
        - wind
        - tides
        - weather
        - sunlightTimes

        Args:
            attr (str, optional): attribute to get dataframe from.
                Defaults to "surf".

        Raises:
            ValueError: if attr is not a valid attribute
        """
        if attr == "surf":
            # concat all dataframes
            data = []
            for attr in ["waves", "wind", "weather"]:
                # excluding "sunlightTimes" and "tides" due to different timestamps
                # TODO: include "surf" in `surf` output
                data.append(
                    pd.DataFrame(_flatten_objects(getattr(self, attr)))
                    .set_index("timestamp_dt")
                    .reset_index()
                )
            df = pd.concat(data, axis=1)
            # remove duplicated columns
            df = df.loc[:, ~df.columns.duplicated()]
            return df
        elif attr in ["waves", "wind", "tides", "weather", "sunlightTimes"]:
            # return single
            return pd.DataFrame(_flatten_objects(getattr(self, attr))).reset_index()
        else:
            raise ValueError(f"Attribute {attr} not supported. Use a valid attribute.")


def _flatten_objects(list_of_objects) -> list:
    """return list of flattened objects"""
    return [flatten(item.__dict__) for item in list_of_objects]

import pytest

from pysurfline.core import SpotForecasts

# TODO: add tests for SpotForecasts

"""core module: SpotForecast"""

import pandas as pd

from .utils import flatten
from .api.objects import (
    SpotDetails,
    SpotForecastsWave,
    SpotForecastsWind,
    SpotForecastsTides,
    SpotForecastsSunlightTimes,
    SpotForecastsWeather,
)


class SpotForecasts:
    """spot forecasts data model

    Composite data model of all the spot forecasts data,
    - wave
    - condition (TODO)
    - wind
    - tides
    - weather and sunrise and sunset times

    TODO: add associated data and improve utcOffset
    """

    name: str = None

    def __init__(
        self,
        spotId: str,
        details: SpotDetails,
        waves: SpotForecastsWave,
        winds: SpotForecastsWind,
        tides: SpotForecastsTides,
        sunlightTimes: SpotForecastsSunlightTimes,
        weather: SpotForecastsWeather,
    ):
        self.spotId = spotId
        self.name = details.spot["name"]
        self.waves = waves.data
        self.wind = winds.data
        self.tides = tides.data
        self.weather = weather.data
        self.sunlightTimes = sunlightTimes.data

    def get_dataframe(self, attr="surf") -> pd.DataFrame:
        """pandas dataframe of selected attribute

        Use default to get the pandas dataframe of surf data,
        or of the selected attribute `attr`:
        - waves
        - wind
        - tides
        - weather
        - sunlightTimes

        Args:
            attr (str, optional): attribute to get dataframe from.
                Defaults to "surf".

        Raises:
            ValueError: if attr is not a valid attribute
        """
        if attr == "surf":
            # concat all dataframes
            data = []
            for attr in ["waves", "wind", "weather"]:
                # excluding "sunlightTimes" and "tides" due to different timestamps
                # TODO: include "surf" in `surf` output
                data.append(
                    pd.DataFrame(_flatten_objects(getattr(self, attr)))
                    .set_index("timestamp_dt")
                    .reset_index()
                )
            df = pd.concat(data, axis=1)
            # remove duplicated columns
            df = df.loc[:, ~df.columns.duplicated()]
            return df
        elif attr in ["waves", "wind", "tides", "weather", "sunlightTimes"]:
            # return single
            return pd.DataFrame(_flatten_objects(getattr(self, attr))).reset_index()
        else:
            raise ValueError(f"Attribute {attr} not supported. Use a valid attribute.")


def _flatten_objects(list_of_objects) -> list:
    """return list of flattened objects"""
    return [flatten(item.__dict__) for item in list_of_objects]