"""Chemical observation module for AEMet data.
This module provides functions to download chemical composition data from the Spanish
national weather service (AEMet). The data comes from the Spanish EMEP/VAG/CAMP network
dedicated to the observation of atmospheric chemical composition at regional scale,
away from contaminant sources.
Data is updated hourly. The available products correspond to:
* ``'o'``: Ozono superficial (Ozone surface)
* ``'n'``: Dióxido y monóxido de nitrógeno (Nitrogen oxides)
* ``'s'``: Dióxido de azufre (Sulfur dioxide)
* ``'r'``: Radiación solar (Solar radiation)
Locations and their corresponding monitoring stations:
* ``"Badajoz"``: Barcarrota (Badajoz)
* ``"Girona"``: Cabo de Creus (Girona)
* ``"Guadalajara"``: Campisábalos (Guadalajara)
* ``"Huelva"``: Doñana (Huelva)
* ``"Lleida"``: Els Torms (Lleida)
* ``"Mahon"``: Mahón (Illes Balears)
* ``"Asturias"``: Niembro-Llanes (Asturias)
* ``"Coruna"``: Noia (A Coruña)
* ``"Lugo"``: O Saviñao (Lugo)
* ``"Zamora"``: Peñausende (Zamora)
* ``"Toledo"``: San Pablo de los Montes (Toledo)
* ``"Granada"``: Víznar (Granada)
* ``"Valencia"``: Zarra (Valencia)
See also: https://www.aemet.es/es/eltiempo/observacion/contaminacionfondo/ayuda
"""
import pathlib
import urllib.request
from aemetxfb.utils import CONFIG
# Base URL for chemical observation images
_BASE_URL = "https://www.aemet.es/imagenes_d/eltiempo/observacion/contaminacionfondo/"
# Shared location mapping (same for all functions)
_LOCATION_MAP = {
"Badajoz": "w1ges11",
"Girona": "w1ges10",
"Guadalajara": "w1ges09",
"Huelva": "w1ges17",
"Lleida": "w1ges14",
"Mahon": "w1ges06",
"Asturias": "w1ges08",
"Coruna": "w1ges05",
"Lugo": "w1ges16",
"Zamora": "w1ges13",
"Toledo": "w1ges01",
"Granada": "w1ges07",
"Valencia": "w1ges12",
}
# Location mapping for previous day (derived from base)
_LOCATION_MAP_PREV_DAY = {
loc: code.replace("w1", "w2") for loc, code in _LOCATION_MAP.items()
}
# Location mapping for previous month (derived from base)
_LOCATION_MAP_PREV_MONTH = {
loc: code.replace("w1", "mw3") for loc, code in _LOCATION_MAP.items()
}
# Product mapping (shared)
_PRODUCT_MAP = {
"o": "Ozono superficial",
"n": "Dióxido y monóxido de nitrógeno",
"s": "Dióxido de azufre",
"r": "Radiación solar",
}
def _validate_output_path(output_path: str | pathlib.Path) -> str:
"""Validate and normalize the output path.
Parameters
----------
output_path : str | pathlib.Path
Path where the GIF file will be saved.
Returns
-------
str
String representation of the path.
Raises
------
ValueError
If ``output_path`` is empty, None, or doesn't have .gif extension.
"""
if not output_path:
raise ValueError("output_path cannot be empty")
output_path_str = str(output_path)
if not output_path_str.endswith(".gif"):
raise ValueError("output_path must have a .gif extension")
return output_path_str
def _validate_loc(loc: str, location_map: dict[str, str]) -> None:
"""Validate the location parameter.
Parameters
----------
loc : str
Location name to validate.
location_map : dict[str, str]
Dictionary of valid locations.
Raises
------
ValueError
If ``loc`` is empty or not in the location map.
"""
if not loc:
raise ValueError("loc cannot be empty")
if loc not in location_map:
raise ValueError(
f"Invalid location: {loc}. "
f"Valid locations are: {', '.join(sorted(location_map.keys()))}"
)
def _validate_prod(prod: str) -> None:
"""Validate the product parameter.
Parameters
----------
prod : str
Product code to validate.
Raises
------
ValueError
If ``prod`` is empty or not in the product map.
"""
if not prod:
raise ValueError("prod cannot be empty")
if prod not in _PRODUCT_MAP:
raise ValueError(
f"Invalid product: {prod}. "
f"Valid products are: {', '.join(sorted(_PRODUCT_MAP.keys()))}"
)
def _download_image(url: str, output_path_str: str) -> None:
"""Download an image from a URL and save it to the specified path.
Parameters
----------
url : str
URL to download the image from.
output_path_str : str
Path where the image will be saved.
Raises
------
RuntimeError
If the download fails.
"""
parent_dir = pathlib.Path(output_path_str).parent
if str(parent_dir) != ".":
parent_dir.mkdir(exist_ok=True)
try:
with urllib.request.urlopen(url, timeout=CONFIG.download_timeout) as response:
pathlib.Path(output_path_str).write_bytes(response.read())
except Exception as e:
raise RuntimeError(f"Failed to download image: {e}")
def _download_chem_image(
*,
output_path: str | pathlib.Path,
loc: str,
prod: str | None,
location_map: dict[str, str],
) -> str:
"""Download a chemical observation image from AEMet.
Shared engine used by the public chemical observation download functions.
Each wrapper passes the correct location map and optionally a product code.
Parameters
----------
output_path : str | pathlib.Path
Path where the GIF file will be saved. Must have ``.gif`` extension.
loc : str
Location name to validate against ``location_map``.
prod : str | None
Product code to validate (``"o"``, ``"n"``, ``"s"``, ``"r"``), or
``None`` if the function does not use products.
location_map : dict[str, str]
Dictionary mapping location names to their station codes.
Returns
-------
str
Path to the downloaded file (same as ``output_path``).
Raises
------
ValueError
If parameters are invalid.
RuntimeError
If the download fails.
"""
output_path_str = _validate_output_path(output_path)
_validate_loc(loc, location_map)
if prod is not None:
_validate_prod(prod)
loc_code = location_map[loc]
if prod is not None:
filename = f"{prod}{loc_code}.gif"
else:
filename = f"{loc_code}.gif"
url = f"{_BASE_URL}{filename}"
_download_image(url, output_path_str)
return output_path_str
[docs]
def get_chem_today(
output_path: str | pathlib.Path,
loc: str,
prod: str,
) -> str:
"""Download a chemical observation image from AEMet.
Downloads a GIF image showing chemical composition data from the Spanish
EMEP/VAG/CAMP network dedicated to the observation of atmospheric chemical
composition at regional scale, away from contaminant sources.
Parameters
----------
output_path : str | pathlib.Path
Path where the GIF file will be saved. Must have ``.gif`` extension.
loc : str
Location name. Valid values: ``Badajoz``, ``Girona``, ``Guadalajara``,
``Huelva``, ``Lleida``, ``Mahon``, ``Asturias``, ``Coruna``, ``Lugo``,
``Zamora``, ``Toledo``, ``Granada``, ``Valencia``.
prod : str
Product type. Valid values:
* ``'o'``: Ozono superficial (Ozone surface)
* ``'n'``: Dióxido y monóxido de nitrógeno (Nitrogen oxides)
* ``'s'``: Dióxido de azufre (Sulfur dioxide)
* ``'r'``: Radiación solar (Solar radiation)
Returns
-------
str
Path to the downloaded file (same as ``output_path``).
Raises
------
ValueError
If ``output_path`` doesn't have ``.gif`` extension, or if ``loc`` or
``prod`` are invalid.
RuntimeError
If the download fails.
See Also
--------
get_chem_previous_day : Download chemical observation for the previous day.
get_chem_previous_month : Download chemical observation for the previous month.
External reference : https://www.aemet.es/es/eltiempo/observacion/contaminacionfondo/ayuda
Examples
--------
>>> get_chem_today("kaka/test.gif", "Lleida", "o") # doctest: +SKIP
'kaka/test.gif'
>>> get_chem_today("kaka/test.gif", "Invalid", "o") # doctest: +SKIP
Traceback (most recent call last):
...
ValueError: Invalid location: Invalid...
"""
# Validate prod before delegating (today requires a product)
_validate_prod(prod)
return _download_chem_image(
output_path=output_path,
loc=loc,
prod=prod,
location_map=_LOCATION_MAP,
)
[docs]
def get_chem_previous_day(
output_path: str | pathlib.Path,
loc: str,
prod: str,
) -> str:
"""Download a chemical observation image from AEMet for the previous day.
Downloads a GIF image showing chemical composition data from the Spanish
EMEP/VAG/CAMP network dedicated to the observation of atmospheric chemical
composition at regional scale, away from contaminant sources.
The data corresponds to the previous day's measurements.
Parameters
----------
output_path : str | pathlib.Path
Path where the GIF file will be saved. Must have ``.gif`` extension.
loc : str
Location name. Valid values: ``Badajoz``, ``Girona``, ``Guadalajara``,
``Huelva``, ``Lleida``, ``Mahon``, ``Asturias``, ``Coruna``, ``Lugo``,
``Zamora``, ``Toledo``, ``Granada``, ``Valencia``.
prod : str
Product type. Valid values:
* ``'o'``: Ozono superficial (Ozone surface)
* ``'n'``: Dióxido y monóxido de nitrógeno (Nitrogen oxides)
* ``'s'``: Dióxido de azufre (Sulfur dioxide)
* ``'r'``: Radiación solar (Solar radiation)
Returns
-------
str
Path to the downloaded file (same as ``output_path``).
Raises
------
ValueError
If ``output_path`` doesn't have ``.gif`` extension, or if ``loc`` or
``prod`` are invalid.
RuntimeError
If the download fails.
See Also
--------
get_chem_today : Download current chemical observation.
get_chem_previous_month : Download chemical observation for the previous month.
External reference : https://www.aemet.es/es/eltiempo/observacion/contaminacionfondo/ayuda
Examples
--------
>>> get_chem_previous_day("kaka/test.gif", "Lleida", "o") # doctest: +SKIP
'kaka/test.gif'
"""
# Validate prod before delegating (previous day requires a product)
_validate_prod(prod)
return _download_chem_image(
output_path=output_path,
loc=loc,
prod=prod,
location_map=_LOCATION_MAP_PREV_DAY,
)
[docs]
def get_chem_previous_month(
output_path: str | pathlib.Path,
loc: str,
) -> str:
"""Download a chemical observation image from AEMet for the previous month.
Downloads a GIF image showing chemical composition data from the Spanish
EMEP/VAG/CAMP network dedicated to the observation of atmospheric chemical
composition at regional scale, away from contaminant sources.
The data corresponds to the previous month's measurements and shows the
evolution of ozone and nitrogen dioxide.
Parameters
----------
output_path : str | pathlib.Path
Path where the GIF file will be saved. Must have ``.gif`` extension.
loc : str
Location name. Valid values: ``Badajoz``, ``Girona``, ``Guadalajara``,
``Huelva``, ``Lleida``, ``Mahon``, ``Asturias``, ``Coruna``, ``Lugo``,
``Zamora``, ``Toledo``, ``Granada``, ``Valencia``.
Returns
-------
str
Path to the downloaded file (same as ``output_path``).
Raises
------
ValueError
If ``output_path`` doesn't have ``.gif`` extension, or if ``loc`` is
invalid.
RuntimeError
If the download fails.
See Also
--------
get_chem_today : Download current chemical observation.
get_chem_previous_day : Download chemical observation for the previous day.
External reference : https://www.aemet.es/es/eltiempo/observacion/contaminacionfondo/ayuda
Examples
--------
>>> get_chem_previous_month("kaka/test.gif", "Lleida") # doctest: +SKIP
'kaka/test.gif'
"""
return _download_chem_image(
output_path=output_path,
loc=loc,
prod=None,
location_map=_LOCATION_MAP_PREV_MONTH,
)