Added Station Information

pyweatherflow_forecast/__init__.py

@@ -1,7 +1,7 @@
 """Python Wrapper for WeatherFlow Forecast API."""
 from __future__ import annotations
 from pyweatherflow_forecast.wffcst_lib import WeatherFlow, WeatherFlowAPIBase
-from pyweatherflow_forecast.data import WeatherFlowForecastData, WeatherFlowForecastDaily, WeatherFlowForecastHourly
+from pyweatherflow_forecast.data import WeatherFlowForecastData, WeatherFlowForecastDaily, WeatherFlowForecastHourly, WeatherFlowStationData
 
 __title__ = "PYWEATHERFLOWFORECAST"
 __version__ = "0.0.2"

pyweatherflow_forecast/const.py

@@ -8,6 +8,7 @@
 
 WEATHERFLOW_BASE_URL = "https://swd.weatherflow.com/swd/rest"
 WEATHERFLOW_FORECAST_URL = f"{WEATHERFLOW_BASE_URL}/better_forecast?station_id="
+WEATHERFLOW_STATION_URL = f"{WEATHERFLOW_BASE_URL}/stations/"
 
 ICON_LIST = {
     "clear-day": "sunny",

pyweatherflow_forecast/data.py

@@ -275,3 +275,38 @@ def valid_time(self) -> datetime:
         """Valid time"""
         return self._valid_time
 
+class WeatherFlowStationData:
+    """Class to hold station data."""
+        # pylint: disable=R0913, R0902, R0914
+    def __init__(
+            self,
+            station_name: str,
+            latitude: float,
+            longitude: float,
+            timezone: str,
+    ) -> None:
+        """Constructor."""
+        self._station_name = station_name
+        self._latitude = latitude
+        self._longitude = longitude
+        self._timezone = timezone
+
+    @property
+    def station_name(self) -> str:
+        """Name of the Station"""
+        return self._station_name
+
+    @property
+    def latitude(self) -> float:
+        """Latitude of station."""
+        return self._latitude
+
+    @property
+    def longitude(self) -> float:
+        """Longitude of station."""
+        return self._longitude
+
+    @property
+    def timezone(self) -> str:
+        """Timezone of station."""
+        return self._timezone

pyweatherflow_forecast/wffcst_lib.py

@@ -19,12 +19,14 @@
     FORECAST_TYPE_DAILY,
     FORECAST_TYPE_HOURLY,
     ICON_LIST,
-    WEATHERFLOW_FORECAST_URL
+    WEATHERFLOW_FORECAST_URL,
+    WEATHERFLOW_STATION_URL,
 )
 from .data import (
     WeatherFlowForecastData,
     WeatherFlowForecastDaily,
-    WeatherFlowForecastHourly
+    WeatherFlowForecastHourly,
+    WeatherFlowStationData,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -46,6 +48,13 @@ def get_forecast_api(self, station_id: int, api_token: str) -> Dict[str, Any]:
             "users must define get_forecast to use this base class"
         )
 
+    @abc.abstractmethod
+    def get_station_api(self, station_id: int, api_token: str) -> Dict[str, Any]:
+        """Override this"""
+        raise NotImplementedError(
+            "users must define get_station to use this base class"
+        )
+
     @abc.abstractmethod
     async def async_get_forecast_api(
         self, station_id: int, api_token: str
@@ -55,9 +64,18 @@ async def async_get_forecast_api(
             "users must define get_forecast to use this base class"
         )
 
+    @abc.abstractmethod
+    async def async_get_station_api(
+        self, station_id: int, api_token: str
+    ) -> Dict[str, Any]:
+        """Override this"""
+        raise NotImplementedError(
+            "users must define get_station to use this base class"
+        )
+
 
 class WeatherFlowAPI(WeatherFlowAPIBase):
-    """Default implementation for WeatherFlow Forecast api"""
+    """Default implementation for WeatherFlow api"""
 
     def __init__(self) -> None:
         """Init the API with or without session"""
@@ -73,6 +91,18 @@ def get_forecast_api(self, station_id: int, api_token: str) -> Dict[str, Any]:
 
         return json_data
 
+    def get_station_api(self, station_id: int, api_token: str) -> Dict[str, Any]:
+        """gets data from API"""
+        api_url = f"{WEATHERFLOW_STATION_URL}{station_id}?token={api_token}"
+        _LOGGER.debug("URL: %s", api_url)
+
+        response = urlopen(api_url)
+        data = response.read().decode("utf-8")
+        json_data = json.loads(data)
+
+        return json_data
+
+
     async def async_get_forecast_api(
         self, station_id: int, api_token: str
     ) -> Dict[str, Any]:
@@ -89,7 +119,31 @@ async def async_get_forecast_api(
                 if is_new_session:
                     await self.session.close()
                 raise WeatherFlowForecastException(
-                    f"Failed to access weather API with status code {response.status}"
+                    f"Failed to access weatherFlow API with status code {response.status}"
+                )
+            data = await response.text()
+            if is_new_session:
+                await self.session.close()
+
+            return json.loads(data)
+
+    async def async_get_station_api(
+        self, station_id: int, api_token: str
+    ) -> Dict[str, Any]:
+        """gets data from API asynchronous"""
+        api_url = f"{WEATHERFLOW_STATION_URL}{station_id}?token={api_token}"
+
+        is_new_session = False
+        if self.session is None:
+            self.session = aiohttp.ClientSession()
+            is_new_session = True
+
+        async with self.session.get(api_url) as response:
+            if response.status != 200:
+                if is_new_session:
+                    await self.session.close()
+                raise WeatherFlowForecastException(
+                    f"Failed to access weatherFlow API with status code {response.status}"
                 )
             data = await response.text()
             if is_new_session:
@@ -141,6 +195,14 @@ def get_forecast_hour(self) -> List[WeatherFlowForecastData]:
             self._json_data = self._api.get_forecast_api(self._station_id, self._api_token)
         return _get_forecast_hour(self._json_data)
 
+    def get_station(self) -> List[WeatherFlowStationData]:
+        """
+        Returns a list of station information.
+        """
+        json_data = self._api.get_station_api(self._station_id, self._api_token)
+
+        return _get_station(json_data)
+
     async def async_get_forecast(self) -> List[WeatherFlowForecastData]:
         """
         Returns a list of forecasts. The first in list are the current one
@@ -169,6 +231,15 @@ async def async_get_forecast_hour(self) -> List[WeatherFlowForecastData]:
             )
         return _get_forecast_hour(json_data)
 
+    async def async_get_station(self) -> List[WeatherFlowStationData]:
+        """
+        Returns a list with Station information.
+        """
+        json_data = await self._api.async_get_station_api(
+                self._station_id, self._api_token
+            )
+        return _get_station(json_data)
+
 def validate_data(json_data) -> bool:
     """Returns true if data is valid else false."""
     data_time = json_data["current_conditions"]["time"]
@@ -298,3 +369,25 @@ def _get_forecast_current(api_result: dict, forecast_type: int) -> List[WeatherF
     )
 
     return current_condition
+
+
+# pylint: disable=R0914, R0912, W0212, R0915
+def _get_station(api_result: dict) -> List[WeatherFlowStationData]:
+    """Converts results from API to WeatherFlowForecast list"""
+
+    item = api_result["stations"][0]
+
+    station_name = item.get("name", None)
+    latitude = item.get("latitude", None)
+    longitude = item.get("longitude", None)
+    timezone = item.get("timezone", None)
+
+    station_data = WeatherFlowStationData(
+        station_name,
+        latitude,
+        longitude,
+        timezone
+    )
+
+    return station_data
+

test_module.py

@@ -12,6 +12,7 @@
 from pyweatherflow_forecast import (
     WeatherFlow,
     WeatherFlowForecastData,
+    WeatherFlowStationData,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -23,12 +24,15 @@
 
 weatherflow = WeatherFlow(station_id, api_token)
 
-data: WeatherFlowForecastData = weatherflow.get_forecast()
-print("TEMPERATURE: ", data.temperature)
-for item in data.forecast:
-    print(item.temperature, item.temp_low, item.icon, item.condition, item.precipitation_probability)
+data: WeatherFlowStationData = weatherflow.get_station()
+print("STATION NAME: ", data.station_name)
 
-hourly: WeatherFlowForecastData = weatherflow.get_forecast_hour()
-print("FEELS LIKE: ", hourly.apparent_temperature)
-for item in hourly.forecast:
-    print(item. valid_time, item.temperature, item.apparent_temperature, item.icon, item.condition, item.precipitation, item.precipitation_probability)
+# data: WeatherFlowForecastData = weatherflow.get_forecast()
+# print("TEMPERATURE: ", data.temperature)
+# for item in data.forecast:
+#     print(item.temperature, item.temp_low, item.icon, item.condition, item.precipitation_probability)
+
+# hourly: WeatherFlowForecastData = weatherflow.get_forecast_hour()
+# print("FEELS LIKE: ", hourly.apparent_temperature)
+# for item in hourly.forecast:
+#     print(item. valid_time, item.temperature, item.apparent_temperature, item.icon, item.condition, item.precipitation, item.precipitation_probability)