From 3daaacf03b0c0f984894f81547a93236e3359164 Mon Sep 17 00:00:00 2001 From: Jules Dejaeghere Date: Sun, 16 Jun 2024 18:08:55 +0200 Subject: [PATCH] Add docstring --- src/open_irceline/api.py | 23 ++++++++++++++++++++++- src/open_irceline/data.py | 1 + src/open_irceline/utils.py | 11 +++++++++-- tests/test_utils.py | 1 + 4 files changed, 33 insertions(+), 3 deletions(-) diff --git a/src/open_irceline/api.py b/src/open_irceline/api.py index 3f62e7b..0abb628 100644 --- a/src/open_irceline/api.py +++ b/src/open_irceline/api.py @@ -197,7 +197,20 @@ class IrcelineRioClient(IrcelineBaseClient): class IrcelineForecastClient(IrcelineBaseClient): """API client for forecast IRCEL - CELINE open data""" - async def get_forecast(self, day: date, features: List[ForecastFeature], position: Tuple[float, float]) -> dict: + async def get_forecasts(self, + day: date, + features: List[ForecastFeature], + position: Tuple[float, float] + ) -> Dict[Tuple[ForecastFeature, date], FeatureValue]: + """ + Get forecasted concentrations for the given features at the given position. The forecasts are downloaded for + the specified day and the 4 next days as well + :param day: date at which the forecast are computed (generally today). If unavailable, the day before will be + tried as well + :param features: pollutants to get the forecasts for + :param position: (lat, long) + :return: dict where key is (ForecastFeature, date of the forecast) and value is a FeatureValue + """ x, y = round_coordinates(position[0], position[1]) result = dict() @@ -227,6 +240,14 @@ class IrcelineForecastClient(IrcelineBaseClient): @staticmethod def extract_result_from_csv(x: float, y: float, csv_text: str) -> float | None: + """ + Find the value of the forecast for the given (x, y) position in the csv text. + x, y should already be rounded to match the positions found in the csv + :param x: latitude (rounded) + :param y: longitude (rounded) + :param csv_text: text of the CSV file + :return: value matching the position if found, else None + """ f = StringIO(csv_text) for row in csv.reader(f, delimiter=';'): try: diff --git a/src/open_irceline/data.py b/src/open_irceline/data.py index 9045151..2972f25 100644 --- a/src/open_irceline/data.py +++ b/src/open_irceline/data.py @@ -34,5 +34,6 @@ class ForecastFeature(StrEnum): class FeatureValue(TypedDict): + # Timestamp at which the value was computed timestamp: datetime | date value: int | float | None diff --git a/src/open_irceline/utils.py b/src/open_irceline/utils.py index e81cb63..65480a6 100644 --- a/src/open_irceline/utils.py +++ b/src/open_irceline/utils.py @@ -29,7 +29,7 @@ class SizedDict(OrderedDict): raise NotImplementedError() -def epsg_transform(position: Tuple[float, float]) -> tuple: +def epsg_transform(position: Tuple[float, float]) -> Tuple[int, int]: """ Convert 'EPSG:4326' coordinates to 'EPSG:31370' coordinates :param position: (x, y) coordinates @@ -39,6 +39,13 @@ def epsg_transform(position: Tuple[float, float]) -> tuple: return round(result[0]), round(result[1]) -def round_coordinates(x: float, y: float, step=.05): +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 diff --git a/tests/test_utils.py b/tests/test_utils.py index b8fac2f..e2a9b1a 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,6 +1,7 @@ from src.open_irceline.utils import SizedDict, round_coordinates, epsg_transform import pytest + def test_size_dict(): s_dict = SizedDict(5) assert len(s_dict) == 0