Clients

AEMET

class mango.clients.aemet.AEMETClient(*, api_key: str, wait_time: float | int | None = None)

Bases: RESTClient

This class will handle the connection to the AEMET API. It will allow to get the meteorological data from the meteorological stations in Spain. To initialize the class, an API key is needed. This key can be obtained in the AEMET website: https://opendata.aemet.es/centrodedescargas/altaUsuario

Parameters:

  • api_key – API key to connect to the AEMET API

  • wait_time – Wait time between requests in seconds. Default is 1.25 seconds

Doc-author:

baobab soluciones

Usage

>>> import os
>>> from mango.clients.aemet import AEMETClient
>>> client = AEMETClient(api_key=os.environ["AEMET_API_KEY"])
connect() → None

This method will connect to the AEMET API and cache the stations and municipios. Checks if the API key is valid. :doc-author: baobab soluciones

property all_stations: List[dict]

This property will return all the meteorological stations in Spain.

Returns:

List of dictionaries with the meteorological stations

Doc-author:

baobab soluciones

property municipios: List[dict]

This property will return all the municipios in Spain.

Returns:

List of dictionaries with the municipios

Doc-author:

baobab soluciones

get_meteo_data(station_code: str | None = None, lat: float | None = None, long: float | None = None, province: str | None = None, start_date: datetime | None = None, end_date: datetime | None = None, output_format: Literal['df', 'raw'] = 'raw')

Main method of the class. This method will return the meteorological data from the meteorological stations in Spain.

Parameters:

  • station_code (str) – meteorological station code

  • lat (float) – latitude

  • long (float) – longitude

  • province (str) – province

  • start_date (datetime) – start date

  • end_date (datetime) – end date

Returns:

a list of dictionaries with the meteorological data

Doc-author:

baobab soluciones

Usage

>>> import pandas as pd
>>> from datetime import datetime
>>> from mango.clients.aemet import AEMETClient
>>> import os
>>>
>>> client = AEMETClient(api_key=os.environ["AEMET_API_KEY"])
>>> client.connect()

If lat and long are provided this method will search for the closest meteorological station to that point. Province is not necessary however it will speed up the search.

>>> data = client.get_meteo_data(lat=40.4165, long=-3.70256, province="Madrid", start_date=datetime(2021, 1, 1), end_date=datetime(2021, 1, 31), output_format="df")

If lat and long are not provided but province is provided, this method will search for all the meteorological stations in that province.

>>> data = client.get_meteo_data(province="Madrid", start_date=datetime(2021, 1, 1), end_date=datetime(2021, 1, 31), output_format="df")

If lat and long are not provided and province is not provided, this method will search for all the meteorological stations in Spain.

>>> data = client.get_meteo_data(start_date=datetime(2021, 1, 1), end_date=datetime(2021, 1, 31), output_format="df")

If neither start_date nor end_date are provided, this method will search for live data following the rules described above. end_date needs start_date to be provided.

>>> data = client.get_meteo_data(lat=40.4165, long=-3.70256, province="Madrid", output_format="df")
get_forecast_data(postal_code: str | None = None)

This method will return the forecast data for the given postal code for the next days provided by the AEMET API. If postal_code is not provided, it will return the forecast data for all the municipios in Spain, doing so takes a long time due to API limitations.

In this version it returns the raw data from the API. In future versions it will return a dataframe with the forecast data properly formatted.

Parameters:

postal_code – Postal code of the municipio to get the forecast data from

Returns:

List of dictionaries with the forecast data

custom_endpoint(endpoint: str)
static expect_status(func, expected_status=None, response_type: Literal['json', 'raw'] = 'json')

Decorator for functions that return a response from the server using requests library. It will check the status code of the response and raise an exception if the status of the response is not the expected and raise an exception if the status of the response is not the expected

static request_handler(url: str, params: dict, wait_time: float | int = 0.5, if_error: Literal['raise', 'warn', 'ignore'] = 'raise', expected_schema: Type[BaseModel] | None = None) → dict | List[dict]

This function will handle the request to a URL, implements the wait time and checks for errors. :param url: URL to make the request to :param params: Parameters to pass to the request :param wait_time: Wait time in seconds. Default: 0.5 seconds :param if_error: What to do if an error is found. Options: raise, warn, ignore :param expected_schema: Pydantic schema to validate the response or generate a default response :return: Dictionary with the response :doc-author: baobab soluciones