Merge pull request #6 from jdejaegh/rio_ifdm

Implement RIO IFDM client

src/open_irceline/__init__.py

@@ -1,6 +1,6 @@
 from .api import IrcelineApiError
-from .rio import IrcelineRioClient
+from .data import RioFeature, ForecastFeature, FeatureValue, RioIfdmFeature
 from .forecast import IrcelineForecastClient
-from .data import RioFeature, ForecastFeature, FeatureValue, BelAqiIndex
+from .rio import IrcelineRioClient, IrcelineRioIfdmClient
 
 __version__ = '2.0.0'

src/open_irceline/api.py

@@ -2,17 +2,17 @@
 import socket
 from abc import ABC, abstractmethod
 from typing import Tuple, List, Set
+from xml.etree import ElementTree
 
 import aiohttp
 import async_timeout
+from aiohttp import ClientResponse
 
 from .data import IrcelineFeature
-from .utils import SizedDict
 
 _rio_wfs_base_url = 'https://geo.irceline.be/wfs'
 _forecast_wms_base_url = 'https://geo.irceline.be/forecast/wms'
-# noinspection HttpUrlsUsage
-# There is not HTTPS version of this endpoint
+_rio_ifdm_wms_base_url = 'https://geobelair.irceline.be/rioifdm/wms'
 _user_agent = 'github.com/jdejaegh/python-irceline'
 
 
@@ -21,9 +21,8 @@ class IrcelineApiError(Exception):
 
 
 class IrcelineBaseClient(ABC):
-    def __init__(self, session: aiohttp.ClientSession, cache_size: int = 20) -> None:
+    def __init__(self, session: aiohttp.ClientSession) -> None:
         self._session = session
-        self._cache = SizedDict(cache_size)
 
     @abstractmethod
     async def get_data(self,
@@ -65,3 +64,38 @@ async def _api_wrapper(self, url: str, querystring: dict = None, headers: dict =
             raise IrcelineApiError(f"Something really wrong happened! {exception}") from exception
 
 
+class IrcelineBaseWmsClient(IrcelineBaseClient, ABC):
+    _default_querystring = {"service": "WMS",
+                            "version": "1.1.1",
+                            "request": "GetFeatureInfo",
+                            "info_format": "application/json",
+                            "width": "1",
+                            "height": "1",
+                            "srs": "EPSG:4326",
+                            "X": "1",
+                            "Y": "1"}
+    _epsilon = 0.00001
+    _base_url = None
+
+    @staticmethod
+    def _parse_capabilities(xml_string: str) -> Set[str]:
+        try:
+            root = ElementTree.fromstring(xml_string)
+        except ElementTree.ParseError:
+            return set()
+
+        path = './/Capability/Layer/Layer/Name'
+        feature_type_names = {t.text for t in root.findall(path)}
+        return feature_type_names
+
+    async def get_capabilities(self) -> Set[str]:
+        """
+        Fetch the list of possible features from the WMS server
+        :return: set of features available on the WMS server
+        """
+        querystring = {"service": "WMS",
+                       "version": "1.1.1",
+                       "request": "GetCapabilities"}
+        r: ClientResponse = await self._api_wrapper(self._base_url, querystring)
+
+        return self._parse_capabilities(await r.text())

src/open_irceline/data.py

@@ -1,5 +1,5 @@
 from datetime import datetime, date
-from enum import StrEnum, Enum
+from enum import StrEnum
 from typing import TypedDict
 
 
@@ -30,6 +30,14 @@ class RioFeature(IrcelineFeature):
     SO2_HMEAN = 'rio:so2_hmean'
 
 
+class RioIfdmFeature(IrcelineFeature):
+    PM25_HMEAN = 'rioifdm:pm25_hmean'
+    NO2_HMEAN = 'rioifdm:no2_hmean'
+    PM10_HMEAN = 'rioifdm:pm10_hmean'
+    O3_HMEAN = 'rioifdm:o3_hmean'
+    BELAQI = 'rioifdm:belaqi'
+
+
 class ForecastFeature(IrcelineFeature):
     NO2_MAXHMEAN = 'forecast:no2_maxhmean'
     NO2_DMEAN = 'forecast:no2_dmean'
@@ -40,20 +48,7 @@ class ForecastFeature(IrcelineFeature):
     BELAQI = 'forecast:belaqi'
 
 
-class BelAqiIndex(Enum):
-    EXCELLENT = 1
-    VERY_GOOD = 2
-    GOOD = 3
-    FAIRLY_GOOD = 4
-    MODERATE = 5
-    POOR = 6
-    VERY_POOR = 7
-    BAD = 8
-    VERY_BAD = 9
-    HORRIBLE = 10
-
-
 class FeatureValue(TypedDict):
     # Timestamp at which the value was computed
     timestamp: datetime | date | None
-    value: int | float | BelAqiIndex | None
+    value: int | float | None

src/open_irceline/forecast.py

@@ -1,16 +1,16 @@
+import asyncio
 from datetime import date, timedelta, datetime
 from itertools import product
-from typing import List, Tuple, Dict, Set
-from xml.etree import ElementTree
+from typing import List, Tuple, Dict
 
 from aiohttp import ClientResponse, ClientResponseError
 
-from .api import IrcelineBaseClient, _forecast_wms_base_url, IrcelineApiError
+from .api import IrcelineApiError, IrcelineBaseWmsClient, _forecast_wms_base_url
 from .data import ForecastFeature, FeatureValue
 
 
-class IrcelineForecastClient(IrcelineBaseClient):
-    _epsilon = 0.00001
+class IrcelineForecastClient(IrcelineBaseWmsClient):
+    _base_url = _forecast_wms_base_url
 
     async def get_data(self,
                        features: List[ForecastFeature],
@@ -26,50 +26,30 @@ async def get_data(self,
         timestamp = date.today()
         result = dict()
         lat, lon = position
-        base_querystring = {"service": "WMS",
-                            "version": "1.1.1",
-                            "request": "GetFeatureInfo",
-                            "info_format": "application/json",
-                            "width": "1",
-                            "height": "1",
-                            "srs": "EPSG:4326",
-                            "bbox": f"{lon},{lat},{lon + self._epsilon},{lat + self._epsilon}",
-                            "X": "1",
-                            "Y": "1"}
+        base_querystring = (self._default_querystring |
+                            {"bbox": f"{lon},{lat},{lon + self._epsilon},{lat + self._epsilon}"})
 
-        for feature, d in product(features, range(4)):
-            querystring = base_querystring | {"layers": f"{feature}_d{d}",
-                                              "query_layers": f"{feature}_d{d}"}
-            try:
-                r: ClientResponse = await self._api_wrapper(_forecast_wms_base_url, querystring)
-                r: dict = await r.json()
-                result[(feature, timestamp + timedelta(days=d))] = FeatureValue(
-                    value=r.get('features', [{}])[0].get('properties', {}).get('GRAY_INDEX'),
-                    timestamp=datetime.fromisoformat(r.get('timeStamp')) if 'timeStamp' in r else None)
-            except (IrcelineApiError, ClientResponseError, IndexError):
-                result[(feature, timestamp + timedelta(days=d))] = FeatureValue(value=None, timestamp=None)
+        tasks = [asyncio.create_task(self._get_single_feature(base_querystring, d, feature, timestamp))
+                 for feature, d in product(features, range(4))]
+        results = await asyncio.gather(*tasks)
 
-        return result
+        for r in results:
+            result |= r
 
-    async def get_capabilities(self) -> Set[str]:
-        """
-        Fetch the list of possible features from the WMS server
-        :return: set of features available on the WMS server
-        """
-        querystring = {"service": "WMS",
-                       "version": "1.1.1",
-                       "request": "GetCapabilities"}
-        r: ClientResponse = await self._api_wrapper(_forecast_wms_base_url, querystring)
+        return result
 
-        return self._parse_capabilities(await r.text())
+    async def _get_single_feature(self, base_querystring: dict, d: int, feature: ForecastFeature,
+                                  timestamp: date) -> dict:
+        result = dict()
 
-    @staticmethod
-    def _parse_capabilities(xml_string: str) -> Set[str]:
+        querystring = base_querystring | {"layers": f"{feature}_d{d}",
+                                          "query_layers": f"{feature}_d{d}"}
         try:
-            root = ElementTree.fromstring(xml_string)
-        except ElementTree.ParseError:
-            return set()
-
-        path = './/Capability/Layer/Layer/Name'
-        feature_type_names = {t.text for t in root.findall(path)}
-        return feature_type_names
+            r: ClientResponse = await self._api_wrapper(self._base_url, querystring)
+            r: dict = await r.json()
+            result[(feature, timestamp + timedelta(days=d))] = FeatureValue(
+                value=r.get('features', [{}])[0].get('properties', {}).get('GRAY_INDEX'),
+                timestamp=datetime.fromisoformat(r.get('timeStamp')) if 'timeStamp' in r else None)
+        except (IrcelineApiError, ClientResponseError, IndexError):
+            result[(feature, timestamp + timedelta(days=d))] = FeatureValue(value=None, timestamp=None)
+        return result

src/open_irceline/rio.py

@@ -1,16 +1,21 @@
+import asyncio
 from datetime import datetime, date, UTC, timedelta
 from typing import List, Tuple, Dict, Set
 from xml.etree import ElementTree
 
-from aiohttp import ClientResponse
+from aiohttp import ClientResponse, ClientResponseError
 
-from .api import IrcelineBaseClient, _rio_wfs_base_url, IrcelineApiError
-from .data import RioFeature, FeatureValue
+from .api import IrcelineBaseClient, _rio_wfs_base_url, IrcelineApiError, _rio_ifdm_wms_base_url, IrcelineBaseWmsClient
+from .data import RioFeature, FeatureValue, RioIfdmFeature
 from .utils import epsg_transform
 
 
 class IrcelineRioClient(IrcelineBaseClient):
-    """API client for RIO interpolated IRCEL - CELINE open data"""
+    """
+    API client for RIO interpolated IRCEL - CELINE open data
+    RIO is more coarse grained for interpolation than RIO IFDM and allows to request multiple features in the same
+        request, which may be faster.
+    """
 
     async def get_data(self,
                        features: List[RioFeature],
@@ -124,3 +129,48 @@ def _format_result(prefix: str, data: dict, features: List[RioFeature]) -> dict:
                 result[name] = FeatureValue(timestamp=timestamp, value=value)
 
         return result
+
+
+class IrcelineRioIfdmClient(IrcelineBaseWmsClient):
+    """
+    API client for RIO IFDM interpolated IRCEL - CELINE open data
+    RIO IFDM is more fine-grained for interpolation than RIO but only allows one feature to be request at a time, which
+    may be slower
+    """
+    _base_url = _rio_ifdm_wms_base_url
+
+    async def get_data(self,
+                       features: List[RioIfdmFeature],
+                       position: Tuple[float, float]
+                       ) -> Dict[RioIfdmFeature, FeatureValue]:
+        """
+        Get interpolated concentrations for the given features at the given position.
+        :param features: pollutants to get the forecasts for
+        :param position: (lat, long)
+        :return: dict where key is RioIfdmFeature and value is a FeatureValue
+        """
+        result = dict()
+        lat, lon = position
+        base_querystring = (self._default_querystring |
+                            {"bbox": f"{lon},{lat},{lon + self._epsilon},{lat + self._epsilon}"})
+
+        tasks = [asyncio.create_task(self._get_single_feature(base_querystring, feature)) for feature in features]
+        results = await asyncio.gather(*tasks)
+
+        for r in results:
+            result |= r
+
+        return result
+
+    async def _get_single_feature(self, base_querystring: dict, feature: RioIfdmFeature) -> dict:
+        result = dict()
+        querystring = base_querystring | {"layers": f"{feature}", "query_layers": f"{feature}"}
+        try:
+            r: ClientResponse = await self._api_wrapper(self._base_url, querystring)
+            r: dict = await r.json()
+            result[feature] = FeatureValue(
+                value=r.get('features', [{}])[0].get('properties', {}).get('GRAY_INDEX'),
+                timestamp=datetime.fromisoformat(r.get('timeStamp')) if 'timeStamp' in r else None)
+        except (IrcelineApiError, ClientResponseError, IndexError):
+            result[feature] = FeatureValue(value=None, timestamp=None)
+        return result

src/open_irceline/utils.py

@@ -1,36 +1,10 @@
-from collections import OrderedDict
 from typing import Tuple
 
 from pyproj import Transformer
 
 _project_transform = Transformer.from_crs('EPSG:4326', 'EPSG:31370', always_xy=False)
 
 
-class SizedDict(OrderedDict):
-    """Dictionary with a maximum size.  When more items are added, the least recently accessed element is evicted"""
-
-    def __init__(self, size: int):
-        super().__init__()
-        self._size = size
-
-    def __setitem__(self, key, value):
-        super().__setitem__(key, value)
-        self.move_to_end(key)
-        if len(self) > self._size:
-            self.popitem(False)
-
-    def __getitem__(self, key):
-        self.move_to_end(key)
-        return super().__getitem__(key)
-
-    def get(self, __key, __default=None):
-        self.move_to_end(__key)
-        return super().get(__key, __default)
-
-    def update(self, __m, **kwargs):
-        raise NotImplementedError()
-
-
 def epsg_transform(position: Tuple[float, float]) -> Tuple[int, int]:
     """
     Convert 'EPSG:4326' coordinates to 'EPSG:31370' coordinates
@@ -39,15 +13,3 @@ def epsg_transform(position: Tuple[float, float]) -> Tuple[int, int]:
     """
     result = _project_transform.transform(position[0], position[1])
     return round(result[0]), round(result[1])
-
-
-def round_coordinates(x: float, y: float, step=.05) -> Tuple[float, float]:
-    """
-    Round the coordinate to the precision given by step
-    :param x: latitude
-    :param y: longitude
-    :param step: precision of the rounding
-    :return: x and y round to the closest step increment
-    """
-    n = 1 / step
-    return round(x * n) / n, round(y * n) / n

tests/fixtures/ifdm_interpolation_feature_info.json

@@ -0,0 +1,17 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "id": "",
+      "geometry": null,
+      "properties": {
+        "GRAY_INDEX": 84.33950805664062
+      }
+    }
+  ],
+  "totalFeatures": "unknown",
+  "numberReturned": 1,
+  "timeStamp": "2024-06-30T15:43:07.222Z",
+  "crs": null
+}