add weather alert endpoint

Author: jain-aayush1123

here_location_services/config/autosuggest_config.py

@@ -13,7 +13,7 @@ class SearchCircle:
         defined by its center and radius(in meters).
     """
 
-    def __init__(self, lat: str, lng: str, radius: int):
+    def __init__(self, lat: float, lng: float, radius: int):
         self.lat = lat
         self.lng = lng
         self.radius = radius

here_location_services/config/dest_weather_config.py

@@ -66,3 +66,81 @@ class DestWeatherUnits(Bunch):
 #: Use this config for ``units``` of Destination Weather API.
 #: Example: for ``metric`` units use ``DEST_WEATHER_UNITS.metric``.
 DEST_WEATHER_UNITS = DestWeatherUnits(**units)
+
+
+class WeatherSeverity(Bunch):
+    """A class to define the severity of the weather event
+
+    ``insignificant``:
+    Event doesn't have significance by nature
+
+    ``no_alerts``:
+    There are no alerts for the location
+
+    ``minor``:
+    Minor Severity, the event is potentially dangerous but not usual
+
+    ``medium``:
+    Medium Severity, the event is dangerous
+
+    ``high``:
+    High Severity, The event is very dangerous
+
+    ``emergency``:
+    Emergency. Take immediate action to protect life.
+    """
+
+
+weather_severity = {
+    "insignificant": 0,
+    "no_alerts": 1,
+    "minor": 2,
+    "medium": 3,
+    "high": 4,
+    "emergency": 5,
+}
+
+#: Use this config for ``weather_severity``` of get weather alerts endpoint.
+#: Example: for ``high severity`` events use ``WEATHER_SEVERITY.high``.
+WEATHER_SEVERITY = WeatherSeverity(**weather_severity)
+
+
+class WeatherType(Bunch):
+    """A class to define the type of the weather event"""
+
+
+weather_type = {
+    "extremely_high_temperature": 1,
+    "extremely_low_temperature": 2,
+    "fog": 3,
+    "ice": 4,
+    "rain": 5,
+    "snow": 6,
+    "thunderstorm": 7,
+    "wind": 8,
+    "air_quality": 9,
+    "volcanic_ashfall": 10,
+    "avalanche": 11,
+    "tsunami": 12,
+    "dust_storm": 13,
+    "earthquake": 14,
+    "fire_danger": 15,
+    "flood": 16,
+    "high_waves": 17,
+    "gigh_uv_index": 18,
+    "low_water": 19,
+    "smoke": 20,
+    "volcano": 21,
+    "ice_in_waterway": 22,
+    "coastal_event": 23,
+    "civil_danger": 24,
+    "evacuation": 25,
+    "hazardous_material": 26,
+    "radiological_hazard": 27,
+    "shelter_in_place": 28,
+    "warning": 29,
+}
+
+#: Use this config for ``weather_type``` of get weather alerts endpoint of Destination Weather API.
+#: Example: for ``fog`` weather type use ``WEATHER_TYPE.fog``.
+WEATHER_TYPE = WeatherType(**weather_type)

here_location_services/destination_weather_api.py

@@ -4,8 +4,9 @@
 """This module contains classes for accessing `HERE Destination Weather API <https://developer.here.com/documentation/destination-weather/dev_guide/topics/overview.html>`_.
 """  # noqa E501
 
+import time
 from datetime import datetime
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from here_location_services.platform.auth import Auth
 
@@ -32,7 +33,7 @@ def get_dest_weather(
         at: Optional[List] = None,
         query: Optional[str] = None,
         zipcode: Optional[str] = None,
-        hourly_date: Optional[datetime] = None,
+        hourly_date: Optional[Any] = None,
         one_observation: Optional[bool] = None,
         language: Optional[str] = None,
         units: Optional[str] = None,
@@ -83,3 +84,78 @@ def get_dest_weather(
             return resp
         else:
             raise ApiError(resp)
+
+    def get_weather_alerts(
+        self,
+        feature_type: str,
+        geometry_type: str,
+        geometry_coordinates: List,
+        start_time: datetime,
+        id: Optional[str] = None,
+        weather_severity: Optional[int] = None,
+        weather_type: Optional[str] = None,
+        country: Optional[str] = None,
+        end_time: Optional[datetime] = None,
+        width: Optional[int] = 50000,
+    ):
+        """Retrieves weather reports, weather forecasts, severe weather alerts and moon and sun rise and set information.
+
+        See further information `Here Destination Weather API <https://developer.here.com/documentation/destination-weather/dev_guide/topics/overview.html>_`.
+
+        :param feature_type: String to define feature type
+        :param geometry_type: Point or LineString or Polygon or MultiPolygon
+        :param geometry_coordinates: Array of coordinates corressponding to type provided
+        :param start_time: Start time of the event
+        :param id: Unique weather alert id.
+        :param weather_severity: Defines the severity of the weather event as defined
+            in :class:`WeatherSeverity`.
+        :param weather_type: Defines the type of the weather event as defined
+            in :class:`WeatherType`.
+        :param country: String for ISO-3166-1 2-letter country code.
+        :param end_time: End time of the event. If not present, warning is valid until
+            it is not removed from the feed by national weather institutes
+            (valid until warning is present in the response)
+        :param width: int. default 50000
+        :return: :class:`requests.Response` object.
+        :raises ApiError: If ``status_code`` of API response is not 200.
+        """  # noqa E501
+
+        path = "v3/alerts"
+        url = f"{self._base_url}/{path}"
+        data: Dict[str, Any] = {
+            "type": "FeatureCollection",
+        }
+        features: Dict[str, Any] = {
+            "type": feature_type,
+        }
+
+        if id:
+            features["id"] = id
+
+        geometry: Dict[str, Any] = {"type": geometry_type, "coordinates": geometry_coordinates}
+
+        properties: Dict[str, Any] = {
+            "width": width,
+        }
+        weather_warnings: Dict[str, Any] = {
+            "startTime": time.mktime(start_time.timetuple()),
+        }
+
+        if weather_severity:
+            weather_warnings["severity"] = weather_severity
+        if weather_type:
+            weather_warnings["type"] = weather_type
+        if country:
+            weather_warnings["country"] = country
+        if end_time:
+            weather_warnings["endTime"] = time.mktime(end_time.timetuple())
+
+        properties["warnings"] = [weather_warnings]
+        features["properties"] = properties
+        features["geometry"] = geometry
+        data["features"] = [features]
+        resp = self.post(url, data=data)
+        if resp.status_code == 200:
+            return resp
+        else:
+            raise ApiError(resp)

here_location_services/ls.py

@@ -8,7 +8,7 @@
 import urllib.request
 from datetime import datetime
 from time import sleep
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from here_location_services.config.routing_config import Scooter, Via
 from here_location_services.platform.apis.aaa_oauth2_api import AAAOauth2Api
@@ -320,7 +320,7 @@ def get_dest_weather(
         at: Optional[List] = None,
         query: Optional[str] = None,
         zipcode: Optional[str] = None,
-        hourly_date: Optional[datetime] = None,
+        hourly_date: Optional[Any] = None,
         one_observation: Optional[bool] = None,
         language: Optional[str] = None,
         units: Optional[str] = None,
@@ -368,6 +368,54 @@ def get_dest_weather(
         response = DestinationWeatherResponse.new(resp.json())
         return response
 
+    def get_weather_alerts(
+        self,
+        feature_type: str,
+        geometry_type: str,
+        geometry_coordinates: List,
+        start_time: datetime,
+        id: Optional[str] = None,
+        weather_severity: Optional[int] = None,
+        weather_type: Optional[str] = None,
+        country: Optional[str] = None,
+        end_time: Optional[datetime] = None,
+        width: Optional[int] = 50000,
+    ) -> DestinationWeatherResponse:
+        """Retrieves weather reports, weather forecasts, severe weather alerts
+            and moon and sun rise and set information.
+
+        :param feature_type: String to define feature type
+        :param geometry_type: Point or LineString or Polygon or MultiPolygon
+        :param geometry_coordinates: Array of coordinates corressponding to type provided
+        :param start_time: Start time of the event
+        :param id: Unique weather alert id.
+        :param weather_severity: Defines the severity of the weather event as defined
+            in :class:`WeatherSeverity`.
+        :param weather_type: Defines the type of the weather event as defined
+            in :class:`WeatherType`.
+        :param country: String for ISO-3166-1 2-letter country code.
+        :param end_time: End time of the event. If not present, warning is valid until
+            it is not removed from the feed by national weather institutes
+            (valid until warning is present in the response)
+        :param width: int. default 50000
+        :return: :class:`DestinationWeatherResponse` object.
+        """
+
+        resp = self.destination_weather_api.get_weather_alerts(
+            feature_type=feature_type,
+            geometry_type=geometry_type,
+            geometry_coordinates=geometry_coordinates,
+            id=id,
+            weather_severity=weather_severity,
+            weather_type=weather_type,
+            country=country,
+            start_time=start_time,
+            end_time=end_time,
+            width=width,
+        )
+        response = DestinationWeatherResponse.new(resp.json())
+        return response
+
     def discover(
         self,
         query: str,

here_location_services/responses.py

@@ -200,8 +200,6 @@ class DestinationWeatherResponse(ApiResponse):
 
     def __init__(self, **kwargs):
         super().__init__()
-        self._filters = {
-            "places": None,
-        }
+        self._filters = {"places": None, "features": None}
         for param, default in self._filters.items():
             setattr(self, param, kwargs.get(param, default))

tests/test_ls.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import json
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import pandas as pd
 import pytest
@@ -21,6 +21,8 @@
 from here_location_services.config.dest_weather_config import (
     DEST_WEATHER_PRODUCT,
     DEST_WEATHER_UNITS,
+    WEATHER_SEVERITY,
+    WEATHER_TYPE,
 )
 from here_location_services.config.isoline_routing_config import (
     ISOLINE_ROUTING_AVOID_FEATURES,
@@ -56,6 +58,41 @@
 LS_API_KEY = get_apikey()
 
 
+@pytest.mark.skipif(not LS_API_KEY, reason="No api key found.")
+def test_ls_weather_alerts():
+    """Test weather alerts endpoint of destination weather api."""
+    ls = LS(api_key=LS_API_KEY)
+    resp = ls.get_weather_alerts(
+        feature_type="Feature",
+        geometry_type="Point",
+        geometry_coordinates=[15.256, 23.456],
+        start_time=datetime.now(),
+        width=3000,
+    )
+    assert resp
+
+    resp2 = ls.get_weather_alerts(
+        feature_type="Feature",
+        geometry_type="Point",
+        geometry_coordinates=[15.256, 23.456],
+        start_time=datetime.now(),
+        weather_type=WEATHER_TYPE.ice,
+        weather_severity=WEATHER_SEVERITY.high,
+        country="US",
+        end_time=datetime.now() + timedelta(days=7),
+    )
+    assert resp2
+
+    with pytest.raises(ApiError):
+        ls2 = LS(api_key="dummy")
+        ls2.get_weather_alerts(
+            feature_type="Feature",
+            geometry_type="Point",
+            geometry_coordinates=[15.256, 23.456],
+            start_time=datetime.now(),
+        )
+
+
 @pytest.mark.skipif(not LS_API_KEY, reason="No api key found.")
 def test_ls_dest_weather():
     """Test destination weather api."""
@@ -121,7 +158,7 @@ def test_ls_autosuggest():
     assert resp.items
     assert len(resp.items) <= 5
 
-    search_in_circle1 = SearchCircle(lat="52.53", lng="13.38", radius="10000")
+    search_in_circle1 = SearchCircle(lat=52.53, lng=13.38, radius="10000")
     search_in_bbox1 = ("13.08836", "52.33812", "13.761", "52.6755")
 
     resp3 = ls.autosuggest(query="bar", limit=5, search_in_circle=search_in_circle1, lang=["en"])

tests/test_ls_apis.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from argparse import Namespace
+from datetime import datetime
 
 import pytest
 import requests
@@ -20,7 +21,7 @@
 
 
 @pytest.mark.skipif(not LS_API_KEY, reason="No api key found.")
-def test_destination(destination_weather_api):
+def test_destination_weather(destination_weather_api):
     """Test Destination Weather api."""
     resp = destination_weather_api.get_dest_weather(
         products=[DEST_WEATHER_PRODUCT.observation], query="Chicago"
@@ -29,6 +30,19 @@ def test_destination(destination_weather_api):
     assert resp.status_code == 200
 
 
+@pytest.mark.skipif(not LS_API_KEY, reason="No api key found.")
+def test_weather_alerts(destination_weather_api):
+    """Test Destination Weather api."""
+    resp = destination_weather_api.get_weather_alerts(
+        feature_type="Feature",
+        geometry_type="Point",
+        geometry_coordinates=[15.256, 23.456],
+        start_time=datetime.now(),
+    )
+    assert type(resp) == requests.Response
+    assert resp.status_code == 200
+
+
 @pytest.mark.skipif(not LS_API_KEY, reason="No api key found.")
 def test_autosuggest(autosuggest_api):
     """Test autosuggest api."""