jdejaegh/python-irceline
1
0
Fork 0
mirror of https://github.com/jdejaegh/python-irceline.git synced 2025-06-26 19:35:40 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Re-export in __init__

Browse source
This commit is contained in:
Jules 2024-06-23 16:05:42 +02:00
parent 8e8333690d
commit db938d83d6
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
5 changed files with 31 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
src/open_irceline/__init__.py
View file

@ -1,10 +1,14 @@
from pyproj import Transformer
from pyproj import Transformer as _Transformer
from .data import RioFeature, ForecastFeature, FeatureValue, BelAqiIndex
from .api import IrcelineRioClient, IrcelineForecastClient
from .belaqi import belaqi_index, belaqi_index_actual, belaqi_index_forecast
__version__ = '0.0.3'
project_transform = Transformer.from_crs('EPSG:4326', 'EPSG:31370', always_xy=False)
rio_wfs_base_url = 'https://geo.irceline.be/wfs'
_project_transform = _Transformer.from_crs('EPSG:4326', 'EPSG:31370', always_xy=False)
_rio_wfs_base_url = 'https://geo.irceline.be/wfs'
# noinspection HttpUrlsUsage
# There is not HTTPS version of this endpoint
forecast_base_url = 'http://ftp.irceline.be/forecast'
user_agent = 'github.com/jdejaegh/python-irceline'
_forecast_base_url = 'http://ftp.irceline.be/forecast'
_user_agent = 'github.com/jdejaegh/python-irceline'

12
src/open_irceline/api.py
View file

@ -12,7 +12,7 @@ import aiohttp
import async_timeout
from aiohttp import ClientResponse
from . import rio_wfs_base_url, user_agent, forecast_base_url
from . import _rio_wfs_base_url, _user_agent, _forecast_base_url
from .data import RioFeature, FeatureValue, ForecastFeature, IrcelineFeature
from .utils import SizedDict, epsg_transform, round_coordinates
@ -43,7 +43,7 @@ class IrcelineBaseClient(ABC):
if headers is None:
headers = dict()
if 'User-Agent' not in headers:
headers |= {'User-Agent': user_agent}
headers |= {'User-Agent': _user_agent}
try:
async with async_timeout.timeout(60):
@ -122,7 +122,7 @@ class IrcelineRioClient(IrcelineBaseClient):
f"{key}>='{timestamp}'"
f" AND "
f"INTERSECTS(the_geom, POINT ({lat} {lon}))"}
r: ClientResponse = await self._api_wrapper(rio_wfs_base_url, querystring)
r: ClientResponse = await self._api_wrapper(_rio_wfs_base_url, querystring)
return self._format_result('rio', await r.json(), features)
async def get_rio_capabilities(self) -> Set[str]:
@ -133,7 +133,7 @@ class IrcelineRioClient(IrcelineBaseClient):
querystring = {"service": "WFS",
"version": "1.3.0",
"request": "GetCapabilities"}
r: ClientResponse = await self._api_wrapper(rio_wfs_base_url, querystring)
r: ClientResponse = await self._api_wrapper(_rio_wfs_base_url, querystring)
return self._parse_capabilities(await r.text())
@ -220,14 +220,14 @@ class IrcelineForecastClient(IrcelineBaseClient):
result = dict()
for feature, d in product(features, range(5)):
url = f"{forecast_base_url}/BE_{feature}_{timestamp.strftime('%Y%m%d')}_d{d}.csv"
url = f"{_forecast_base_url}/BE_{feature}_{timestamp.strftime('%Y%m%d')}_d{d}.csv"
try:
r: ClientResponse = await self._api_cached_wrapper(url)
ts = timestamp
except IrcelineApiError:
# retry for the day before
yesterday = timestamp - timedelta(days=1)
url = f"{forecast_base_url}/BE_{feature}_{yesterday.strftime('%Y%m%d')}_d{d}.csv"
url = f"{_forecast_base_url}/BE_{feature}_{yesterday.strftime('%Y%m%d')}_d{d}.csv"
try:
r: ClientResponse = await self._api_cached_wrapper(url)
ts = yesterday

4
src/open_irceline/utils.py
View file

@ -1,7 +1,7 @@
from collections import OrderedDict
from typing import Tuple
from src.open_irceline import project_transform
from src.open_irceline import _project_transform
class SizedDict(OrderedDict):
@ -35,7 +35,7 @@ def epsg_transform(position: Tuple[float, float]) -> Tuple[int, int]:
:param position: (x, y) coordinates
:return: tuple of int in the EPSG:31370 system
"""
result = project_transform.transform(position[0], position[1])
result = _project_transform.transform(position[0], position[1])
return round(result[0]), round(result[1])

18
tests/test_api_forecasts.py
View file

@ -1,7 +1,7 @@
from datetime import date
from unittest.mock import call
from src.open_irceline import forecast_base_url, user_agent
from src.open_irceline import _forecast_base_url, _user_agent
from src.open_irceline.api import IrcelineForecastClient
from src.open_irceline.data import ForecastFeature
from tests.conftest import get_api_data, get_mock_session_many_csv
@ -30,9 +30,9 @@ async def test_cached_calls():
calls = [
call(method='GET',
url=f"{forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240619_d{i}.csv",
url=f"{_forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240619_d{i}.csv",
params=None,
headers={'User-Agent': user_agent}
headers={'User-Agent': _user_agent}
) for i in range(5)
]
@ -47,9 +47,9 @@ async def test_cached_calls():
calls += [
call(method='GET',
url=f"{forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240619_d{i}.csv",
url=f"{_forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240619_d{i}.csv",
params=None,
headers={'User-Agent': user_agent, 'If-None-Match': 'my-etag-here'}
headers={'User-Agent': _user_agent, 'If-None-Match': 'my-etag-here'}
) for i in range(5)
]
@ -72,14 +72,14 @@ async def test_missed_cached_calls():
for i in range(5):
calls += [
call(method='GET',
url=f"{forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240621_d{i}.csv",
url=f"{_forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240621_d{i}.csv",
params=None,
headers={'User-Agent': user_agent}
headers={'User-Agent': _user_agent}
),
call(method='GET',
url=f"{forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240620_d{i}.csv",
url=f"{_forecast_base_url}/BE_{ForecastFeature.NO2_MAXHMEAN}_20240620_d{i}.csv",
params=None,
headers={'User-Agent': user_agent}
headers={'User-Agent': _user_agent}
)
]

10
tests/test_api_rio.py
View file

@ -2,7 +2,7 @@ from datetime import datetime, date
from freezegun import freeze_time
from src.open_irceline import rio_wfs_base_url, user_agent
from src.open_irceline import _rio_wfs_base_url, _user_agent
from src.open_irceline.api import IrcelineRioClient
from src.open_irceline.data import RioFeature, FeatureValue
from src.open_irceline.utils import epsg_transform
@ -102,7 +102,7 @@ async def test_api_rio():
_ = await client.get_data(d, features, pos)
session.request.assert_called_once_with(
method='GET',
url=rio_wfs_base_url,
url=_rio_wfs_base_url,
params={"service": "WFS",
"version": "1.3.0",
"request": "GetFeature",
@ -112,7 +112,7 @@ async def test_api_rio():
f"date>='2024-06-17'"
f" AND "
f"INTERSECTS(the_geom, POINT ({x} {y}))"},
headers={'User-Agent': user_agent}
headers={'User-Agent': _user_agent}
)
@ -124,9 +124,9 @@ async def test_api_rio_get_capabilities():
session.request.assert_called_once_with(
method='GET',
url=rio_wfs_base_url,
url=_rio_wfs_base_url,
params={"service": "WFS",
"version": "1.3.0",
"request": "GetCapabilities"},
headers={'User-Agent': user_agent}
headers={'User-Agent': _user_agent}
)