-
Notifications
You must be signed in to change notification settings - Fork 0
/
get_historic_pm25.py
62 lines (52 loc) · 2.38 KB
/
get_historic_pm25.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
#
# OSU CS361: Intro to Software Development 1
# Historical PM2.5 Data
# Winter 2022
#
# Author: Maggie Liu
# Description: Provides historical PM2.5 data from the AQICN database on dbnomics for the requested city, as long as that city has data in the 57-city database.
#
import json
import pandas
import dbnomics
from geopy import geocoders
def fetch_data(city: str):
"""
Takes a string city name and fetches the median PM2.5 air quality data associated
with that city from AQICN. Writes the data to csv.
"""
fetch_code = 'AQICN/AQI/US.' + db_cities[city] + '.pm25.median'
df = dbnomics.fetch_series(fetch_code)
df = df[["original_period", "original_value"]] # remove unnecessary columns
# Convert df to csv and write
df = pandas.DataFrame.to_csv(df, line_terminator='\r')
with open('pm25py.csv','w') as outfile:
outfile.write(df)
outfile.close()
if __name__ == '__main__':
# Get location request from a text file. Could be a zip code or US city.
with open('historic_aqi.txt', 'r', encoding='UTF8') as infile:
location = infile.readline().rstrip()
infile.close()
locator = geocoders.Nominatim(user_agent="historic_aqi_app")
try:
search = locator.geocode(location, country_codes="us")
city = search.address.split(", ")
with open('cities.json', 'r', encoding='UTF8') as infile:
db_cities = json.load(infile) # needed to translate verified city name to AQICN city code
infile.close()
# Fetch and write data if the city is in the database
if city[0] in db_cities:
fetch_data(city[0])
elif city[1] in db_cities: # First result could be a neighborhood so check first two
fetch_data(city[1])
else:
# Write an an error message to the request file if the request was not in the database
with open('historic_aqi.txt', 'w') as outfile:
outfile.write("Location not found." + "\n")
outfile.close()
except:
# Write an an error message to the request file if the request was not a city
with open('historic_aqi.txt', 'w') as outfile:
outfile.write("Entry not recognized." + "\n")
outfile.close()