Refactor and add tests

This commit is contained in:
Jules 2024-06-16 17:24:03 +02:00
parent 0fd34a6581
commit 59b584d943
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
4 changed files with 42 additions and 22 deletions

View file

@ -11,9 +11,9 @@ import aiohttp
import async_timeout
from aiohttp import ClientResponse
from . import project_transform, 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
from .utils import SizedDict
from .utils import SizedDict, epsg_transform, round_coordinates
class IrcelineApiError(Exception):
@ -80,16 +80,6 @@ class IrcelineBaseClient:
class IrcelineRioClient(IrcelineBaseClient):
"""API client for RIO interpolated IRCEL - CELINE open data"""
@staticmethod
def _epsg_transform(position: Tuple[float, float]) -> tuple:
"""
Convert 'EPSG:4326' coordinates to 'EPSG:31370' coordinates
:param position: (x, y) coordinates
:return: tuple of int in the EPSG:31370 system
"""
result = project_transform.transform(position[0], position[1])
return round(result[0]), round(result[1])
async def get_rio_value(self,
timestamp: datetime | date,
features: List[RioFeature],
@ -116,7 +106,7 @@ class IrcelineRioClient(IrcelineBaseClient):
else:
raise IrcelineApiError(f"Wrong parameter type for timestamp: {type(timestamp)}")
coord = self._epsg_transform(position)
coord = epsg_transform(position)
querystring = {"service": "WFS",
"version": "1.3.0",
"request": "GetFeature",
@ -207,13 +197,8 @@ class IrcelineRioClient(IrcelineBaseClient):
class IrcelineForecastClient(IrcelineBaseClient):
"""API client for forecast IRCEL - CELINE open data"""
@staticmethod
def _round_coordinates(x: float, y: float, step=.05):
n = 1 / step
return round(x * n) / n, round(y * n) / n
async def get_forecast(self, day: date, features: List[ForecastFeature], position: Tuple[float, float]) -> dict:
x, y = self._round_coordinates(position[0], position[1])
x, y = round_coordinates(position[0], position[1])
result = dict()

View file

@ -1,4 +1,7 @@
from collections import OrderedDict
from typing import Tuple
from src.open_irceline import project_transform
class SizedDict(OrderedDict):
@ -24,3 +27,18 @@ class SizedDict(OrderedDict):
def update(self, __m, **kwargs):
raise NotImplementedError()
def epsg_transform(position: Tuple[float, float]) -> tuple:
"""
Convert 'EPSG:4326' coordinates to 'EPSG:31370' coordinates
:param position: (x, y) coordinates
:return: tuple of int in the EPSG:31370 system
"""
result = project_transform.transform(position[0], position[1])
return round(result[0]), round(result[1])
def round_coordinates(x: float, y: float, step=.05):
n = 1 / step
return round(x * n) / n, round(y * n) / n

View file

@ -61,3 +61,5 @@ def test_parse_capabilities():
'rio:bc_dmean'}
assert result == expected
# TODO add new tests for IrcelineForecastClient

View file

@ -1,5 +1,5 @@
from src.open_irceline.utils import SizedDict
from src.open_irceline.utils import SizedDict, round_coordinates, epsg_transform
import pytest
def test_size_dict():
s_dict = SizedDict(5)
@ -19,7 +19,7 @@ def test_size_dict():
s_dict['b'] = 42
s_dict['g'] = 7
assert s_dict['f'] == 6
assert s_dict.get('f') == 6
assert s_dict['g'] == 7
assert s_dict['b'] == 42
assert 'c' not in s_dict
@ -28,3 +28,18 @@ def test_size_dict():
del s_dict['b']
assert len(s_dict) == 4
assert 'b' not in s_dict
with pytest.raises(NotImplementedError):
s_dict.update({'a': 1})
def test_round_coord():
x, y = round_coordinates(50.4657, 4.8647)
assert x == 50.45
assert y == 4.85
def test_epsg_transform():
x, y = epsg_transform((50.4657, 4.8647))
assert x == 185211
assert y == 128437