"""
Tools for retrieving data from the National Atmospheric Deposition Program
(NADP) including the National Trends Network (NTN), the Mercury Deposition
Network (MDN).
National Trends Network
-----------------------
The NTN provides long-term records of precipitation chemistry across the
United States. See https://nadp.slh.wisc.edu/ntn for more info.
Mercury Deposition Network
--------------------------
The MDN provides long-term records of total mercury (Hg) concentration and
deposition in precipitation in the United States and Canada. For more
information visit https://nadp.slh.wisc.edu/networks/mercury-deposition-network/
Notes
-----
Gridded data on NADP is served as zipped tif files. Functions in this module
will either download and extract the data, when a path is specified, or open
the data as a GDAL memory-mapped file when no path is specified.
.. todo::
- include AIRMoN, AMNet, and AMoN
- flexible handling of strings for parameters and measurement types
- add errorchecking
- add tests
"""
import io
import os
import re
import zipfile
from os.path import basename
import requests
NADP_URL = "https://nadp.slh.wisc.edu"
NADP_MAP_EXT = "filelib/maps"
NTN_CONC_PARAMS = ["pH", "So4", "NO3", "NH4", "Ca", "Mg", "K", "Na", "Cl", "Br"]
NTN_DEP_PARAMS = [
"H",
"So4",
"NO3",
"NH4",
"Ca",
"Mg",
"K",
"Na",
"Cl",
"Br",
"N",
"SPlusN",
]
NTN_MEAS_TYPE = ["conc", "dep", "precip"] # concentration or deposition
[docs]
class NADP_ZipFile(zipfile.ZipFile):
"""Extend zipfile.ZipFile for working on data from NADP"""
[docs]
def tif_name(self):
"""Get the name of the tif file in the zip file."""
filenames = self.namelist()
r = re.compile(".*tif$")
tif_list = list(filter(r.match, filenames))
return tif_list[0]
[docs]
def tif(self):
"""Read the tif file in the zip file."""
return self.read(self.tif_name())
[docs]
def get_annual_MDN_map(measurement_type, year, path):
"""Download a MDN map from NDAP.
This function looks for a zip file containing gridded information at:
https://nadp.slh.wisc.edu/maps-data/mdn-gradient-maps/.
The function will download the zip file and extract it, exposing the tif
file if a path is provided.
Parameters
----------
measurement_type: string
The type of measurement (concentration or deposition) as a string,
either 'conc' or 'dep' respectively.
year: string
Year as a string 'YYYY'
path: string
Download directory.
Returns
-------
path: string
Path that zip file was extracted into if path was specified.
Examples
--------
.. code::
>>> # get map of mercury concentration in 2010 and extract it to a path
>>> data_path = dataretrieval.nadp.get_annual_MDN_map(
... measurement_type="conc", year="2010", path="somepath"
... )
"""
url = f"{NADP_URL}/{NADP_MAP_EXT}/MDN/grids/"
filename = f"Hg_{measurement_type}_{year}.zip"
z = get_zip(url, filename)
if path:
z.extractall(path)
return f"{path}{os.sep}{basename(filename)}"
[docs]
def get_annual_NTN_map(measurement_type, measurement=None, year=None, path="."):
"""Download a NTN map from NDAP.
This function looks for a zip file containing gridded information at:
https://nadp.slh.wisc.edu/maps-data/ntn-gradient-maps/.
The function will download the zip file and extract it, exposing the tif
file at the provided path.
.. note::
Measurement type abbreviations for concentration and deposition are
all lower-case, but for precipitation data, the first letter must be
capitalized!
Parameters
----------
measurement : string
The measured constituent to return.
measurement_type : string
The type of measurement, 'conc', 'dep', or 'Precip', which represent
concentration, deposition, or precipitation respectively.
year : string
Year as a string 'YYYY'
path : string
Download directory, defaults to current directory if not specified.
Returns
-------
path: string
Path that zip file was extracted into if path was specified.
Examples
--------
.. code::
>>> # get a map of precipitation in 2015 and extract it to a path
>>> data_path = dataretrieval.nadp.get_annual_NTN_map(
... measurement_type="Precip", year="2015", path="somepath"
... )
"""
url = f"{NADP_URL}/{NADP_MAP_EXT}/NTN/grids/{year}/"
filename = f"{measurement_type}_{year}.zip"
if measurement:
filename = f"{measurement}_{filename}"
z = get_zip(url, filename)
if path:
z.extractall(path)
return f"{path}{os.sep}{basename(filename)}"
[docs]
def get_zip(url, filename):
"""Gets a ZipFile at url and returns it
Parameters
----------
url : string
URL to zip file
filename : string
Name of zip file
Returns
-------
ZipFile
.. todo::
finish docstring
"""
req = requests.get(url + filename)
req.raise_for_status()
# z = zipfile.ZipFile(io.BytesIO(req.content))
z = NADP_ZipFile(io.BytesIO(req.content))
return z