Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add get_observation and get_observation_for_place in client #520

Merged
merged 5 commits into from
Feb 22, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 53 additions & 1 deletion src/meteofrance_api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,14 @@
from .const import METEOFRANCE_API_TOKEN
from .const import METEOFRANCE_API_URL
from .model import CurrentPhenomenons
from .model import Observation
from .model import Forecast
from .model import Full
from .model import PictureOfTheDay
from .model import Place
from .model import Rain
from .session import MeteoFranceSession

# TODO: http://webservice.meteofrance.com/observation
# TODO: investigate bulletincote, montagne, etc...
# http://ws.meteofrance.com/ws//getDetail/france/330630.json
# TODO: add protection for warning if domain not valid
Expand Down Expand Up @@ -74,6 +74,58 @@ def search_places(
resp = self.session.request("get", "places", params=params)
return [Place(place_data) for place_data in resp.json()]


#
# Observation
#
def get_observation(
self,
latitude: float,
longitude: float,
language: str = "fr",
) -> Observation:
"""Retrieve the weather observation for a given GPS location.

Results can be fetched in french or english according to the language parameter.

Args:
latitude: Latitude in degree of the GPS point corresponding to the weather
forecast.
longitude: Longitude in degree of the GPS point corresponding to the weather
forecast.
language: Optional; If language is equal "fr" (default value) results will
be in French. All other value will give results in English.

Returns:
An Observation instance.
"""
resp = self.session.request(
bbesset marked this conversation as resolved.
Show resolved Hide resolved
"get",
"v2/observation",
params={"lat": latitude, "lon": longitude, "lang": language})
return Observation(resp.json())


def get_observation_for_place(
self,
place: Place,
language: str = "fr",
) -> Observation:
"""Retrieve the weather observation for a given Place instance.

Results can be fetched in french or english according to the language parameter.

Args:
place: Place class instance corresponding to a location.
language: Optional; If language is equal "fr" (default value) results will
be in French. All other value will give results in English.

Returns:
An Observation intance.
"""
return self.get_observation(place.latitude, place.longitude, language)
bbesset marked this conversation as resolved.
Show resolved Hide resolved


#
# Forecast
#
Expand Down
3 changes: 2 additions & 1 deletion src/meteofrance_api/model/__init__.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
"""Météo-France models for the REST API."""
from .forecast import Forecast
from .observation import Observation
from .picture_of_the_day import PictureOfTheDay
from .place import Place
from .rain import Rain
from .warning import CurrentPhenomenons
from .warning import Full

__all__ = ["Forecast", "Place", "PictureOfTheDay", "Rain", "CurrentPhenomenons", "Full"]
__all__ = ["Forecast", "Observation", "Place", "PictureOfTheDay", "Rain", "CurrentPhenomenons", "Full"]
111 changes: 111 additions & 0 deletions src/meteofrance_api/model/observation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
# -*- coding: utf-8 -*-
"""Weather observation Python model for the Météo-France REST API."""
import sys
from datetime import datetime

if sys.version_info >= (3, 8):
from typing import TypedDict # pylint: disable=no-name-in-module
else:
from typing_extensions import TypedDict


class ObservationDataPropertiesGridded(TypedDict, total=True):
time: str
T: float
wind_speed: float
wind_direction: int
wind_icon: str
weather_icon: str
weather_description: str


class ObservationDataProperties(TypedDict, total=False):
timezone: str
gridded: ObservationDataPropertiesGridded


class ObservationData(TypedDict, total=False):
"""Describing the data structure of the observation object returned by the REST API."""

properties: ObservationDataProperties


class Observation:
"""Class to access the results of an `observation` API request.

Attributes:
timezone: The observation timezone
time: The time at which the observation was made
temperature: The observed temperature (°C)
wind_speed: The observed wind speed (km/h)
wind_direction: The observed wind direction (°)
wind_icon: An icon ID illustrating the observed wind direction
weather_icon: An icon ID illustrating the observed weather condition
weather_description: A description of the observed weather condition
"""

def __init__(self, raw_data: ObservationData) -> None:
"""Initialize an Observation object.

Args:
raw_data: A dictionary representing the JSON response from 'observation' REST
API request. The structure is described by the ObservationData class.
"""
self.properties = raw_data.get('properties', dict())

@property
def timezone(self) -> str:
"""Returns the observation timezone"""
return self.properties.get('timezone')

@property
def _gridded(self) -> ObservationDataPropertiesGridded:
"""Returns the observation gridded properties"""
return self.properties.get('gridded', dict())

@property
def time_as_string(self) -> str:
"""Returns the time at which the observation was made"""
return self._gridded.get('time')

@property
def time_as_datetime(self) -> str:
"""Returns the time at which the observation was made"""
time = self.time_as_string
return None if time is None else datetime.strptime(time, '%Y-%m-%dT%H:%M:%S.%f%z')

@property
def temperature(self) -> float:
"""Returns the observed temperature (°C)"""
return self._gridded.get('T')

@property
def wind_speed(self) -> float:
"""Returns the observed wind speed (km/h)"""
return self._gridded.get('wind_speed')

@property
def wind_direction(self) -> int:
"""Returns the observed wind direction (°)"""
return self._gridded.get('wind_direction')

@property
def wind_icon(self) -> str:
"""Returns an icon ID illustrating the observed wind direction"""
return self._gridded.get('wind_icon')

@property
def weather_icon(self) -> str:
"""Returns an icon ID illustrating the observed weather condition"""
return self._gridded.get('weather_icon')

@property
def weather_description(self) -> str:
"""Returns a description of the observed weather condition"""
return self._gridded.get('weather_description')

def __repr__(self) -> str:
"""Returns a stringified version of the object"""
return f"Observation(timezone={self.timezone}, time={self.time}, temperature={self.temperature}°C, "\
f"wind_speed={self.wind_speed} km/h, wind_direction={self.wind_direction}°, wind_icon={self.wind_icon}, "\
f"weather_icon={self.weather_icon}, weather_description={self.weather_description}"
61 changes: 61 additions & 0 deletions tests/test_observation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# coding: utf-8
"""Tests Météo-France module. Observation class."""
import time
from datetime import datetime, timezone, timedelta

from .const import MOUNTAIN_CITY
from meteofrance_api import MeteoFranceClient
from meteofrance_api.model import Place


def assert_types(observation) -> None:
"""Check observation types"""
assert type(observation.timezone) == str
assert type(observation.time_as_string) == str
assert type(observation.time_as_datetime) == datetime
assert type(observation.temperature) == float
assert type(observation.wind_speed) == float
assert type(observation.wind_direction) == int
assert type(observation.wind_icon) == str
assert type(observation.weather_icon) == str
assert type(observation.weather_description) == str


def assert_datetime(observation) -> None:
"""Check observation time is before now but after now - 1h."""
now = datetime.now(timezone.utc)
assert now - timedelta(hours=1) < observation.time_as_datetime < now


def test_observation_france() -> None:
"""Test weather observation results from API (valid result, from lat/lon)."""
client = MeteoFranceClient()
observation = client.get_observation(latitude=48.8075, longitude=2.24028)

assert_types(observation)
assert_datetime(observation)


def test_observation_world() -> None:
"""Test weather observation results from API (null result)."""
client = MeteoFranceClient()
observation = client.get_observation(latitude=45.5016889, longitude=73.567256)

assert observation.timezone is None
assert observation.time_as_string is None
assert observation.time_as_datetime is None
assert observation.temperature is None
assert observation.wind_speed is None
assert observation.wind_direction is None
assert observation.wind_icon is None
assert observation.weather_icon is None
assert observation.weather_description is None


def test_observation_place() -> None:
"""Test weather observation results from API (valid result, from place)."""
client = MeteoFranceClient()
observation = client.get_observation_for_place(place=Place(MOUNTAIN_CITY))

assert_types(observation)
assert_datetime(observation)