airbase.csv_api package

class airbase.csv_api.CSVData(country, pollutant_id, source, year, city=None, output=Output.TEXT)[source]

Bases: NamedTuple

info needed for requesting the URLs for country, source, year and pollutant_id the request can be further restricted with the city param

Create new instance of CSVData(country, pollutant_id, source, year, city, output)

Parameters:

  • country (str) –

  • pollutant_id (int | Literal['']) –

  • source (Source) –

  • year (int) –

  • city (str | None) –

  • output (Output) –

city: str | None

Alias for field number 4

country: str

Alias for field number 0

output: Output

Alias for field number 5

param()[source]
Return type:

CSVDataJSON

pollutant_id: int | Literal['']

Alias for field number 1

source: Source

Alias for field number 2

year: int

Alias for field number 3

class airbase.csv_api.Client(*, timeout=None, max_concurrent=10)[source]

Bases: AbstractAsyncContextManager

Handle for requests to Legacy AirQualityExport https://discomap.eea.europa.eu/map/fme/AirQualityExport.htm

Parameters:

  • timeout (float | None) –

  • max_concurrent (int) –

async download_binary(url, path)[source]

get request to url, write response body content (in binary form) into a a binary file, and return path (exactly as the input)

Parameters:

  • url (str) –

  • path (Path) –

Return type:

Path

async download_metadata(path)[source]

download metadata file and return path (exactly as the input)

Parameters:

path (Path) –

Return type:

Path

async download_urls(params)[source]

get request to AirQualityExport

Parameters:

params (CSVDataJSON) –

Return type:

str

class airbase.csv_api.Session(*, progress=False, raise_for_status=True)[source]

Bases: AbstractAsyncContextManager

Parameters:

  • progress (bool) – (optional, default False) Show progress bars

  • raise_for_status (bool) – (optional, default True) Raise exceptions if any request from summary, url_to_files or download_to_directory methods returns “bad” HTTP status codes. If False, a warnings.warn() will be issued instead. Default True.

add_urls(more_urls)[source]

add to the unique URLs ready for download

Parameters:

more_urls (Iterable[str]) –

Return type:

None

clear()[source]

reset URLs

Return type:

None

client: Client = <airbase.csv_api.client.Client object>
async download_metadata(path, skip_existing=True)[source]

download station metadata into the given path.

Parameters:

  • path (Path) – pathlib.Path to the station metadata (parent directory must exist)

  • skip_existing (bool) – (optional, default True) Don’t re-download metadata if path already exists. If False, path may be overwritten.

Return type:

None

async download_to_directory(root_path, *, country_subdir=True, skip_existing=True)[source]

download into a directory

Parameters:

  • root_path (Path) – The directory to save files in (must exist)

  • country_subdir (bool) – (optional, default True) Download files for different counties to different root_path sub directories. If False, download all files to root_path

  • skip_existing (bool) – (optional, default True) Don’t re-download files if they exist in root_path. If False, existing files in root_path may be overwritten. Empty files will be re-downloaded regardless of this option.

Return type:

None

NOTE need to call url_to_files first, in order to retrieve the URLs to download, or add the urls directly with add_urls

property number_of_urls: int

number of unique URLs ready for download

remove_url(url)[source]

remove URL from unique URLs ready for download

Parameters:

url (str) –

Return type:

None

async url_to_files(*download_infos)[source]

multiple request for file URLs and return only unique URLs from each responses

Parameters:

download_infos (CSVData) – info about requested urls

Return type:

None

property urls: Iterable[str]

unique URLs ready for download

class airbase.csv_api.Source(value)[source]

Bases: str, Enum

E1a: Verified data from 2013 to 2023 reported by countries by 30 September each year for the previous year. E2a: Unverified data transmitted continuously data from the beginning of 2024. ALL: E1a and E2a

NOTE - only 2024 data available -> no E2a data

ALL = 'ALL'
Unverified = 'E2a'
Verified = 'E1a'
async airbase.csv_api.download(source, year, root_path, *, countries, pollutants=None, cities=None, metadata=False, country_subdir=True, overwrite=False, quiet=True, raise_for_status=False, session=<airbase.csv_api.session.Session object>)[source]

request file urls by country|city/pollutant and download unique files

Parameters:

  • source (Source) – Source.Verified or Source.Unverified.

  • year (int) – Observations year.

  • root_path (Path) – The directory to save files in (must exist).

  • countries (frozenset[str] | set[str]) – Request observations for these countries.

  • pollutants (frozenset[str] | set[str] | None) – (optional, default None) Limit requests to these specific pollutants.

  • cities (frozenset[str] | set[str] | None) – (optional, default None) Limit requests to these specific cities.

  • metadata (bool) – (optional, default False) Download station metadata into root_path/”metadata.csv”.

  • country_subdir (bool) – (optional, default True) Download files for different counties to different root_path sub directories. If False, download all files to root_path.

  • quiet (bool) – (optional, default True) Disable progress bars.

  • raise_for_status (bool) – (optional, default False) Raise exceptions if any request return “bad” HTTP status codes. If False, a warnings.warn() will be issued instead.

  • overwrite (bool) –

  • session (Session) –

airbase.csv_api.request_info_by_city(source, year, *cities, pollutants=None)[source]

download info one city at the time

Parameters:

  • source (Source) –

  • year (int) –

  • cities (str) –

  • pollutants (frozenset[str] | set[str] | None) –

Return type:

set[airbase.csv_api.dataset.CSVData]

airbase.csv_api.request_info_by_country(source, year, *countries, pollutants=None)[source]

download info one country at the time

Parameters:

  • source (Source) –

  • year (int) –

  • countries (str) –

  • pollutants (frozenset[str] | set[str] | None) –

Return type:

set[airbase.csv_api.dataset.CSVData]

Submodules

airbase.csv_api.client module

Client for Legacy AirQualityExport https://discomap.eea.europa.eu/map/fme/AirQualityExport.htm

class airbase.csv_api.client.Client(*, timeout=None, max_concurrent=10)[source]

Bases: AbstractAsyncContextManager

Handle for requests to Legacy AirQualityExport https://discomap.eea.europa.eu/map/fme/AirQualityExport.htm

Parameters:

  • timeout (float | None) –

  • max_concurrent (int) –

async download_binary(url, path)[source]

get request to url, write response body content (in binary form) into a a binary file, and return path (exactly as the input)

Parameters:

  • url (str) –

  • path (Path) –

Return type:

Path

async download_metadata(path)[source]

download metadata file and return path (exactly as the input)

Parameters:

path (Path) –

Return type:

Path

async download_urls(params)[source]

get request to AirQualityExport

Parameters:

params (CSVDataJSON) –

Return type:

str

airbase.csv_api.dataset module

class airbase.csv_api.dataset.CSVData(country, pollutant_id, source, year, city=None, output=Output.TEXT)[source]

Bases: NamedTuple

info needed for requesting the URLs for country, source, year and pollutant_id the request can be further restricted with the city param

Create new instance of CSVData(country, pollutant_id, source, year, city, output)

Parameters:

  • country (str) –

  • pollutant_id (int | Literal['']) –

  • source (Source) –

  • year (int) –

  • city (str | None) –

  • output (Output) –

city: str | None

Alias for field number 4

country: str

Alias for field number 0

output: Output

Alias for field number 5

param()[source]
Return type:

CSVDataJSON

pollutant_id: int | Literal['']

Alias for field number 1

source: Source

Alias for field number 2

year: int

Alias for field number 3

class airbase.csv_api.dataset.Output(value)[source]

Bases: str, Enum

An enumeration.

HTML = 'HTML'
TEXT = 'TEXT'
class airbase.csv_api.dataset.Source(value)[source]

Bases: str, Enum

E1a: Verified data from 2013 to 2023 reported by countries by 30 September each year for the previous year. E2a: Unverified data transmitted continuously data from the beginning of 2024. ALL: E1a and E2a

NOTE - only 2024 data available -> no E2a data

ALL = 'ALL'
Unverified = 'E2a'
Verified = 'E1a'
airbase.csv_api.dataset.request_info_by_city(source, year, *cities, pollutants=None)[source]

download info one city at the time

Parameters:

  • source (Source) –

  • year (int) –

  • cities (str) –

  • pollutants (frozenset[str] | set[str] | None) –

Return type:

set[airbase.csv_api.dataset.CSVData]

airbase.csv_api.dataset.request_info_by_country(source, year, *countries, pollutants=None)[source]

download info one country at the time

Parameters:

  • source (Source) –

  • year (int) –

  • countries (str) –

  • pollutants (frozenset[str] | set[str] | None) –

Return type:

set[airbase.csv_api.dataset.CSVData]

airbase.csv_api.session module

class airbase.csv_api.session.Session(*, progress=False, raise_for_status=True)[source]

Bases: AbstractAsyncContextManager

Parameters:

  • progress (bool) – (optional, default False) Show progress bars

  • raise_for_status (bool) – (optional, default True) Raise exceptions if any request from summary, url_to_files or download_to_directory methods returns “bad” HTTP status codes. If False, a warnings.warn() will be issued instead. Default True.

add_urls(more_urls)[source]

add to the unique URLs ready for download

Parameters:

more_urls (Iterable[str]) –

Return type:

None

clear()[source]

reset URLs

Return type:

None

client: Client = <airbase.csv_api.client.Client object>
async download_metadata(path, skip_existing=True)[source]

download station metadata into the given path.

Parameters:

  • path (Path) – pathlib.Path to the station metadata (parent directory must exist)

  • skip_existing (bool) – (optional, default True) Don’t re-download metadata if path already exists. If False, path may be overwritten.

Return type:

None

async download_to_directory(root_path, *, country_subdir=True, skip_existing=True)[source]

download into a directory

Parameters:

  • root_path (Path) – The directory to save files in (must exist)

  • country_subdir (bool) – (optional, default True) Download files for different counties to different root_path sub directories. If False, download all files to root_path

  • skip_existing (bool) – (optional, default True) Don’t re-download files if they exist in root_path. If False, existing files in root_path may be overwritten. Empty files will be re-downloaded regardless of this option.

Return type:

None

NOTE need to call url_to_files first, in order to retrieve the URLs to download, or add the urls directly with add_urls

property number_of_urls: int

number of unique URLs ready for download

remove_url(url)[source]

remove URL from unique URLs ready for download

Parameters:

url (str) –

Return type:

None

async url_to_files(*download_infos)[source]

multiple request for file URLs and return only unique URLs from each responses

Parameters:

download_infos (CSVData) – info about requested urls

Return type:

None

property urls: Iterable[str]

unique URLs ready for download

async airbase.csv_api.session.download(source, year, root_path, *, countries, pollutants=None, cities=None, metadata=False, country_subdir=True, overwrite=False, quiet=True, raise_for_status=False, session=<airbase.csv_api.session.Session object>)[source]

request file urls by country|city/pollutant and download unique files

Parameters:

  • source (Source) – Source.Verified or Source.Unverified.

  • year (int) – Observations year.

  • root_path (Path) – The directory to save files in (must exist).

  • countries (frozenset[str] | set[str]) – Request observations for these countries.

  • pollutants (frozenset[str] | set[str] | None) – (optional, default None) Limit requests to these specific pollutants.

  • cities (frozenset[str] | set[str] | None) – (optional, default None) Limit requests to these specific cities.

  • metadata (bool) – (optional, default False) Download station metadata into root_path/”metadata.csv”.

  • country_subdir (bool) – (optional, default True) Download files for different counties to different root_path sub directories. If False, download all files to root_path.

  • quiet (bool) – (optional, default True) Disable progress bars.

  • raise_for_status (bool) – (optional, default False) Raise exceptions if any request return “bad” HTTP status codes. If False, a warnings.warn() will be issued instead.

  • overwrite (bool) –

  • session (Session) –

airbase.csv_api.types module

type annotations from https://discomap.eea.europa.eu/map/fme/AirQualityExport.htm

class airbase.csv_api.types.CSVDataJSON[source]

Bases: TypedDict

query params to https://fme.discomap.eea.europa.eu/fmedatastreaming/AirQualityDownload/AQData_Extract.fmw

NOTE - “&CountryCode=” –> all available countries - “&Pollutant=” –> all available pollutants - “&Source=” –> all available sources - no “&CountryCode=” –> no results - no “&Pollutant=” –> only SO2 (ID=1) - no “&Source=” –> error

CityName: NotRequired[str]
CountryCode: str
EoICode: NotRequired[str]
Output: Literal['HTML', 'TEXT'] | Output
Pollutant: int | Literal['']
Samplingpoint: NotRequired[str]
Source: Literal['E1a', 'E2a', 'ALL'] | Source
Station: NotRequired[str]
TimeCoverage: NotRequired[str]
Undelivered: NotRequired[str]
UpdateDate: NotRequired[str]
Year_from: NotRequired[int | Literal['']]
Year_to: NotRequired[int | Literal['']]