Source code for epa.aqs

"""
Python module to download EPA AQS Data hosted at https://www.epa.gov/aqs

The utility expects user to specify a list of years, a list of
EPA Parameter Codes (see https://www.epa.gov/aqs/aqs-code-list)
and some instructions how to format output

The data is downloaded from \
https://aqs.epa.gov/aqsweb/airdata/download_files.html

The tool also adds a column containing a uniquely generated Monitor Key

"""
#  Copyright (c) 2021. Harvard University
#
#  Developed by Research Software Engineering,
#  Faculty of Arts and Sciences, Research Computing (FAS RC)
#  Author: Michael A Bouzinier
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

import os

from epa.aqs_ds_def import AQSContext
from epa.aqs_tools import collect_aqs_download_tasks, download_data
from nsaph import init_logging


[docs]class AQS: """ Main class, describes the whole download job """ def __init__(self, context: AQSContext = None): """ Creates a new instance :param context: An optional AQSContext object, if not specified, then it is constructed from the command line arguments """ init_logging() if not context: context = AQSContext(__doc__) self.context = context if self.context.destination is None: self.context.destination = os.curdir self.download_tasks = None
[docs] def collect_downloads(self): """ Constructs a list of all download tasks necessary to execute to download all the data defined in the configuration () :return: Nothing """ self.download_tasks = collect_aqs_download_tasks(self.context)
[docs] def is_downloaded(self): """ Checks if there anything remains to be downloaded or if all files are up to date :return: True if everything is up to date, False otherwise """ if not self.download_tasks: self.collect_downloads() for task in self.download_tasks: if not task.is_up_to_date(): return False return True
[docs] def clear(self): """ Removes all previously downloaded data from the local storage :return: Nothing """ if not self.download_tasks: self.collect_downloads() for task in self.download_tasks: task.reset()
[docs] def download(self): """ Downloads all remaining data absent or not up to date) :return: Nothing """ if not self.download_tasks: self.collect_downloads() for task in self.download_tasks: if task.is_up_to_date(): continue download_data (task)
if __name__ == '__main__': downloader = AQS() print(downloader.context) print("Up to date: " + str(downloader.is_downloaded())) downloader.download() print("Up to date: " + str(downloader.is_downloaded()))