Air pollution forecasts

Author: csparpa

pyowm/airpollutionapi30/airpollution_client.py

@@ -1,7 +1,8 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-from pyowm.airpollutionapi30.uris import CO_INDEX_URL, OZONE_URL, NO2_INDEX_URL, SO2_INDEX_URL, AIR_POLLUTION_URL
+from pyowm.airpollutionapi30.uris import CO_INDEX_URL, OZONE_URL, NO2_INDEX_URL, SO2_INDEX_URL, AIR_POLLUTION_URL, \
+    AIR_POLLUTION_FORECAST_URL
 from pyowm.utils import formatting
 
 
@@ -164,6 +165,18 @@ def get_air_pollution(self, params_dict):
         _, json_data = self._client.get_json(AIR_POLLUTION_URL, params=params_dict)
         return json_data
 
+    def get_forecast_air_pollution(self, params_dict):
+        """
+        Invokes the new AirPollution API forecast endpoint
+
+        :param params_dict: dict of parameters
+        :returns: a string containing raw JSON data
+        :raises: *ValueError*, *APIRequestError*
+
+        """
+        _, json_data = self._client.get_json(AIR_POLLUTION_FORECAST_URL, params=params_dict)
+        return json_data
+
     def __repr__(self):
         return "<%s.%s - httpclient=%s>" % \
-               (__name__, self.__class__.__name__, str(self._client))
+               (__name__, self.__class__.__name__, str(self._client))

pyowm/airpollutionapi30/airpollution_manager.py

@@ -197,7 +197,7 @@ def so2index_around_coords(self, lat, lon, start=None, interval=None):
 
     def air_quality_at_coords(self, lat, lon):
         """
-        Queries the OWM AirPollution API for all available air quality indicators around the specified coordinates.
+        Queries the OWM AirPollution API for available air quality indicators around the specified coordinates.
 
         :param lat: the location's latitude, must be between -90.0 and 90.0
         :type lat: int/float
@@ -212,8 +212,32 @@ def air_quality_at_coords(self, lat, lon):
         geo.assert_is_lat(lat)
         params = {'lon': lon, 'lat': lat}
         json_data = self.new_ap_client.get_air_pollution(params)
-        air_status = airstatus.AirStatus.from_dict(json_data)
-        return air_status
+        try:
+            return airstatus.AirStatus.from_dict(json_data)
+        except:
+            return None
+
+    def air_quality_forecast_at_coords(self, lat, lon):
+        """
+        Queries the OWM AirPollution API for available forecasted air quality indicators around the specified coordinates.
+
+        :param lat: the location's latitude, must be between -90.0 and 90.0
+        :type lat: int/float
+        :param lon: the location's longitude, must be between -180.0 and 180.0
+        :type lon: int/float
+        :return: a `list` of *AirStatus* instances or an empty `list` if data is not available
+        :raises: *ParseResponseException* when OWM AirPollution API responses' data
+            cannot be parsed, *APICallException* when OWM AirPollution API can not be
+            reached, *ValueError* for wrong input values
+        """
+        geo.assert_is_lon(lon)
+        geo.assert_is_lat(lat)
+        params = {'lon': lon, 'lat': lat}
+        json_data = self.new_ap_client.get_forecast_air_pollution(params)
+        try:
+            return airstatus.AirStatus.from_dict(json_data)
+        except:
+            return []
 
     def __repr__(self):
         return '<%s.%s>' % (__name__, self.__class__.__name__)

pyowm/airpollutionapi30/airstatus.py

@@ -76,41 +76,48 @@ def reception_time(self, timeformat='unix'):
     @classmethod
     def from_dict(cls, the_dict):
         """
-        Parses a *AirStatus* instance out of a data dictionary.
+        Parses an *AirStatus* instance or `list` of instances out of a data dictionary.
 
         :param the_dict: the input dictionary
         :type the_dict: `dict`
-        :returns: a *AirStatus* instance or ``None`` if no data is available
+        :returns: a *AirStatus* instance or ``list` of such instances
         :raises: *ParseAPIResponseError* if it is impossible to find or parse the data needed to build the result
 
         """
         if the_dict is None:
             raise exceptions.ParseAPIResponseError('Data is None')
         try:
-
-            item = the_dict['list'][0]
-
-            # -- reference time (strip away Z and T on ISO8601 format)
-            reference_time = item['dt']
-
-            # -- reception time (now)
-            reception_time = timestamps.now('unix')
-
             # -- location
             lon = float(the_dict['coord']['lat'])
             lat = float(the_dict['coord']['lon'])
             place = location.Location(None, lon, lat, None)
 
-            # -- air quality data
-            data = item['components']
-            data['aqi'] = item['main']['aqi']
+            # -- reception time (now)
+            rcp_time = timestamps.now('unix')
+
+            def build_air_status(item_dict, location, reception_time):
+                # -- reference time (strip away Z and T on ISO8601 format)
+                reference_time = item_dict['dt']
+
+                # -- air quality data
+                data = item_dict['components']
+                data['aqi'] = item_dict['main']['aqi']
+
+                return AirStatus(reference_time, location, data, reception_time)
+
+            items = the_dict['list']
+
+            # one datapoint
+            if len(items) == 1:
+                return build_air_status(items[0], place, rcp_time)
+            # multiple datapoints
+            else:
+                return [build_air_status(item, place, rcp_time) for item in items]
 
         except KeyError:
             raise exceptions.ParseAPIResponseError(
                       ''.join([__name__, ': impossible to parse AirStatus']))
 
-        return AirStatus(reference_time, place, data, reception_time)
-
     def to_dict(self):
         """Dumps object to a dictionary
 

pyowm/airpollutionapi30/uris.py

@@ -12,3 +12,4 @@
 # current API endpoint
 NEW_ROOT_POLLUTION_API_URL = 'openweathermap.org/data/2.5'
 AIR_POLLUTION_URL = 'air_pollution'
+AIR_POLLUTION_FORECAST_URL = 'air_pollution/forecast'

sphinx/v3/code-recipes.md

@@ -831,7 +831,7 @@ TBD
 
 Instead of getting a `weather_manager`, get from the main OWM object a `airpollution_manager` and use it
 
-### Getting air polluting concentrations and Air Quality Index on geographic coords 
+### Getting air pollution concentrations and Air Quality Index on geographic coords 
 Air polluting agents concentration can be queried in one shot:
 
 ```python
@@ -855,6 +855,28 @@ air_status.nh3
 air_status.aqi 
 ```
 
+### Getting forecasts for air pollution on geographic coords 
+We can get also get forecasts for air pollution agents concentration and air quality index:
+
+```python
+from pyowm.owm import OWM
+owm = OWM('your-api-key')
+mgr = owm.airpollution_manager()
+
+list_of_forecasts = mgr.air_quality_forecast_at_coords(51.507351, -0.127758)  # London, GB
+
+# Each item in the list_of_forecasts is an AirStatus object 
+for air_status in list_of_forecasts:
+    air_status.co
+    air_status.no
+    air_status.no2
+    air_status.o3
+    air_status.so2
+    air_status.pm2_5
+    air_status.pm10
+    air_status.nh3
+    air_status.aqi  # air quality index
+```
 
 
 <div id="station_measurements"/>

tests/integration/pollutionapi30/__init__.py renamed to tests/integration/airpollutionapi30/__init__.py

@@ -12,7 +12,7 @@ class IntegrationTestsPollutionAPI30(unittest.TestCase):
 
     def test_air_quality_at_coords(self):
         """
-        Test feature: get all air quality data  around geo-coordinates.
+        Test feature: get all air quality data around geo-coordinates.
         """
         airstatus = self.__owm.air_quality_at_coords(45, 9)
         self.assertIsNotNone(airstatus)
@@ -21,6 +21,18 @@ def test_air_quality_at_coords(self):
         self.assertIsNotNone(airstatus.reference_time())
         self.assertIsNotNone(airstatus.location)
 
+    def test_air_quality_forecast_at_coords(self):
+        """
+        Test feature: get all forecasted air quality data around geo-coordinates.
+        """
+        list_of_airstatuses = self.__owm.air_quality_forecast_at_coords(45, 9)
+        self.assertTrue(list_of_airstatuses)
+        for airstatus in list_of_airstatuses:
+            self.assertIsNotNone(airstatus.air_quality_data)
+            self.assertIsNotNone(airstatus.reception_time())
+            self.assertIsNotNone(airstatus.reference_time())
+            self.assertIsNotNone(airstatus.location)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/integration/pollutionapi30/test_integration_pollutionapi30.py renamed to tests/integration/airpollutionapi30/test_integration_pollutionapi30.py

@@ -10,7 +10,7 @@
 from tests.unit.airpollutionapi30.test_coindex import COINDEX_JSON
 from tests.unit.airpollutionapi30.test_no2index import NO2INDEX_JSON
 from tests.unit.airpollutionapi30.test_so2index import SO2INDEX_JSON
-from tests.unit.airpollutionapi30.test_airstatus import AIRSTATUS_JSON
+from tests.unit.airpollutionapi30.test_airstatus import AIRSTATUS_JSON, AIRSTATUS_MULTIPLE_JSON
 
 
 class TestAirPollutionManager(unittest.TestCase):
@@ -29,6 +29,9 @@ def mock_get_no2_returning_no2index_around_coords(self, params_dict):
     def mock_get_air_pollution(self, params_dict):
         return json.loads(AIRSTATUS_JSON)
 
+    def mock_get_forecast_air_pollution(self, params_dict):
+        return json.loads(AIRSTATUS_MULTIPLE_JSON)
+
     def mock_get_so2_returning_so2index_around_coords(self, params_dict):
         return json.loads(SO2INDEX_JSON)
 
@@ -199,5 +202,32 @@ def test_air_quality_at_coords_fails_with_wrong_parameters(self):
         self.assertRaises(ValueError, airpollution_manager.AirPollutionManager.air_quality_at_coords, \
                           self.__test_instance, 200, 2.5)
 
+    def test_air_quality_forecast_at_coords(self):
+        ref_to_original = airpollution_client.AirPollutionHttpClient.get_forecast_air_pollution
+        airpollution_client.AirPollutionHttpClient.get_forecast_air_pollution = \
+            self.mock_get_forecast_air_pollution
+        result = self.__test_instance.air_quality_forecast_at_coords(45, 9)
+        airpollution_client.AirPollutionHttpClient.get_forecast_air_pollution = ref_to_original
+        self.assertTrue(isinstance(result, list))
+        for item in result:
+            self.assertIsInstance(item, airstatus.AirStatus)
+            self.assertIsNotNone(item.reference_time)
+            self.assertIsNotNone(item.reception_time())
+            loc = item.location
+            self.assertIsNotNone(loc)
+            self.assertIsNotNone(loc.lat)
+            self.assertIsNotNone(loc.lon)
+            self.assertIsNotNone(item.air_quality_data)
+
+    def test_air_quality_forecast_at_coords_fails_with_wrong_parameters(self):
+        self.assertRaises(ValueError, airpollution_manager.AirPollutionManager.air_quality_forecast_at_coords, \
+                          self.__test_instance, 43.7, -200.0)
+        self.assertRaises(ValueError, airpollution_manager.AirPollutionManager.air_quality_forecast_at_coords, \
+                          self.__test_instance, 43.7, 200.0)
+        self.assertRaises(ValueError, airpollution_manager.AirPollutionManager.air_quality_forecast_at_coords, \
+                          self.__test_instance, -200, 2.5)
+        self.assertRaises(ValueError, airpollution_manager.AirPollutionManager.air_quality_forecast_at_coords, \
+                          self.__test_instance, 200, 2.5)
+
     def test_repr(self):
         print(self.__test_instance)

tests/unit/airpollutionapi30/test_airpollution_manager.py

@@ -10,6 +10,7 @@
 
 
 AIRSTATUS_JSON = '{"coord":{"lon":-0.1278,"lat":51.5074},"list":[{"main":{"aqi":1},"components":{"co":250.34,"no":0.19,"no2":35.99,"o3":30.76,"so2":8.11,"pm2_5":3.15,"pm10":3.81,"nh3":0.74},"dt":1611597600}]}'
+AIRSTATUS_MULTIPLE_JSON = '{"coord":{"lon":50,"lat":50},"list":[{"main":{"aqi":1},"components":{"co":240.33,"no":0,"no2":1.07,"o3":79.39,"so2":0.97,"pm2_5":1.84,"pm10":1.9,"nh3":1.25},"dt":1613606400},{"main":{"aqi":1},"components":{"co":240.33,"no":0,"no2":0.98,"o3":79.39,"so2":0.69,"pm2_5":1.92,"pm10":1.97,"nh3":1.36},"dt":1613610000}]}'
 AIRSTATUS_MALFORMED_JSON = '{"time":"2016-10-01T13:07:01Z","xyz":[]}'
 AIRSTATUS_JSON_DUMP = '{"reference_time": 1234567, "location": {"name": "test", "coordinates": {"lon": 12.3, "lat": 43.7}, "ID": 987, "country": "UK"}, "air_quality_data": {"aqi": 1, "co": 250.34, "no": 0.19, "no2": 35.99, "o3": 30.76, "so2": 8.11, "pm2_5": 3.15, "pm10": 3.81, "nh3": 0.74}, "reception_time": 1475283600}'
 
@@ -64,11 +65,11 @@ def test_returning_different_formats_for_reception_time(self):
                          self.__test_date_reception_time)
 
     def test_from_dict(self):
+        # one item
         d = json.loads(AIRSTATUS_JSON)
         result = AirStatus.from_dict(d)
         self.assertIsNotNone(result)
         self.assertIsNotNone(result.reference_time())
-        self.assertIsNotNone(result.reference_time())
         loc = result.location
         self.assertIsNotNone(loc)
         self.assertIsNone(loc.name)
@@ -78,6 +79,22 @@ def test_from_dict(self):
         for key in self.__test_air_quality_data:
             getattr(result, key)
 
+        # multiple items
+        d = json.loads(AIRSTATUS_MULTIPLE_JSON)
+        result = AirStatus.from_dict(d)
+        self.assertIsInstance(result, list)
+        for item in result:
+            self.assertIsInstance(item, AirStatus)
+            self.assertIsNotNone(item.reference_time())
+            loc = item.location
+            self.assertIsNotNone(loc)
+            self.assertIsNone(loc.name)
+            self.assertIsNone(loc.id)
+            self.assertIsNotNone(loc.lon)
+            self.assertIsNotNone(loc.lat)
+            for key in self.__test_air_quality_data:
+                getattr(item, key)
+
     def test_from_dict_fails_when_JSON_data_is_None(self):
         self.assertRaises(pyowm.commons.exceptions.ParseAPIResponseError, AirStatus.from_dict, None)