import gradio as gr
import requests
import pandas as pd
import folium
from folium.plugins import MarkerCluster
import tempfile
import os
import json
# Get API credentials from environment variables
EPA_AQS_API_BASE_URL = "https://aqs.epa.gov/data/api"
EMAIL = os.environ.get("EPA_AQS_EMAIL", "") # Get from environment variable
API_KEY = os.environ.get("EPA_AQS_API_KEY", "") # Get from environment variable
class AirQualityApp:
def __init__(self):
self.states = {
"AL": "Alabama", "AK": "Alaska", "AZ": "Arizona", "AR": "Arkansas",
"CA": "California", "CO": "Colorado", "CT": "Connecticut", "DE": "Delaware",
"FL": "Florida", "GA": "Georgia", "HI": "Hawaii", "ID": "Idaho",
"IL": "Illinois", "IN": "Indiana", "IA": "Iowa", "KS": "Kansas",
"KY": "Kentucky", "LA": "Louisiana", "ME": "Maine", "MD": "Maryland",
"MA": "Massachusetts", "MI": "Michigan", "MN": "Minnesota", "MS": "Mississippi",
"MO": "Missouri", "MT": "Montana", "NE": "Nebraska", "NV": "Nevada",
"NH": "New Hampshire", "NJ": "New Jersey", "NM": "New Mexico", "NY": "New York",
"NC": "North Carolina", "ND": "North Dakota", "OH": "Ohio", "OK": "Oklahoma",
"OR": "Oregon", "PA": "Pennsylvania", "RI": "Rhode Island", "SC": "South Carolina",
"SD": "South Dakota", "TN": "Tennessee", "TX": "Texas", "UT": "Utah",
"VT": "Vermont", "VA": "Virginia", "WA": "Washington", "WV": "West Virginia",
"WI": "Wisconsin", "WY": "Wyoming", "DC": "District of Columbia"
}
# Mapping from two-letter state codes to numeric state codes for API
self.state_code_mapping = {
"AL": "01", "AK": "02", "AZ": "04", "AR": "05",
"CA": "06", "CO": "08", "CT": "09", "DE": "10",
"FL": "12", "GA": "13", "HI": "15", "ID": "16",
"IL": "17", "IN": "18", "IA": "19", "KS": "20",
"KY": "21", "LA": "22", "ME": "23", "MD": "24",
"MA": "25", "MI": "26", "MN": "27", "MS": "28",
"MO": "29", "MT": "30", "NE": "31", "NV": "32",
"NH": "33", "NJ": "34", "NM": "35", "NY": "36",
"NC": "37", "ND": "38", "OH": "39", "OK": "40",
"OR": "41", "PA": "42", "RI": "44", "SC": "45",
"SD": "46", "TN": "47", "TX": "48", "UT": "49",
"VT": "50", "VA": "51", "WA": "53", "WV": "54",
"WI": "55", "WY": "56", "DC": "11"
}
# AQI categories with their corresponding colors
self.aqi_categories = {
"Good": "#00e400", # Green
"Moderate": "#ffff00", # Yellow
"Unhealthy for Sensitive Groups": "#ff7e00", # Orange
"Unhealthy": "#ff0000", # Red
"Very Unhealthy": "#99004c", # Purple
"Hazardous": "#7e0023" # Maroon
}
# Sample county data for demo
self.mock_counties = {
"CA": [
{"code": "037", "value": "Los Angeles"},
{"code": "067", "value": "Sacramento"},
{"code": "073", "value": "San Diego"},
{"code": "075", "value": "San Francisco"}
],
"NY": [
{"code": "061", "value": "New York"},
{"code": "047", "value": "Kings (Brooklyn)"},
{"code": "081", "value": "Queens"},
{"code": "005", "value": "Bronx"}
],
"TX": [
{"code": "201", "value": "Harris (Houston)"},
{"code": "113", "value": "Dallas"},
{"code": "029", "value": "Bexar (San Antonio)"},
{"code": "453", "value": "Travis (Austin)"}
]
}
# Sample parameters for demo
self.mock_parameters = [
{"code": "88101", "value_represented": "PM2.5 - Local Conditions"},
{"code": "44201", "value_represented": "Ozone"},
{"code": "42401", "value_represented": "Sulfur dioxide"},
{"code": "42101", "value_represented": "Carbon monoxide"},
{"code": "42602", "value_represented": "Nitrogen dioxide"},
{"code": "81102", "value_represented": "PM10 - Local Conditions"}
]
def get_monitors(self, state_code, county_code=None, parameter_code=None):
"""Fetch monitoring stations for a given state and optional county"""
# If we don't have API credentials, use mock data
if not EMAIL or not API_KEY:
return self.mock_get_monitors(state_code, county_code, parameter_code)
# Convert state code to numeric format for API
api_state_code = state_code
if len(state_code) == 2 and state_code in self.state_code_mapping:
api_state_code = self.state_code_mapping[state_code]
# API endpoint for monitoring sites
endpoint = f"{EPA_AQS_API_BASE_URL}/monitors/byState"
params = {
"email": EMAIL,
"key": API_KEY,
"state": api_state_code,
"bdate": "20240101", # Beginning date (YYYYMMDD)
"edate": "20240414", # End date (YYYYMMDD)
}
if county_code:
params["county"] = county_code
if parameter_code:
params["param"] = parameter_code
try:
response = requests.get(endpoint, params=params)
data = response.json()
# Handle the specific response structure we observed
if isinstance(data, dict):
if "Data" in data and isinstance(data["Data"], list):
return data["Data"]
elif "Header" in data and isinstance(data["Header"], list):
if data["Header"][0].get("status") == "Success":
return data.get("Data", [])
# If we couldn't parse the response format, return empty list
print(f"Unexpected response format for monitors: {type(data)}")
return []
except Exception as e:
print(f"Error fetching monitors: {e}")
return []
def get_counties(self, state_code):
"""Fetch counties for a given state"""
# If we don't have API credentials, use mock data
if not EMAIL or not API_KEY:
return self.mock_get_counties(state_code)
# Convert state code to numeric format for API
api_state_code = state_code
if len(state_code) == 2 and state_code in self.state_code_mapping:
api_state_code = self.state_code_mapping[state_code]
endpoint = f"{EPA_AQS_API_BASE_URL}/list/countiesByState"
params = {
"email": EMAIL,
"key": API_KEY,
"state": api_state_code
}
try:
response = requests.get(endpoint, params=params)
data = response.json()
# Handle the specific response structure we observed
counties = []
if isinstance(data, dict) and "Data" in data and isinstance(data["Data"], list):
counties = data["Data"]
# Format as "code: name" for dropdown
result = []
for c in counties:
code = c.get("code")
value = c.get("value_represented")
if code and value:
result.append(f"{code}: {value}")
return result
except Exception as e:
print(f"Error fetching counties: {e}")
return []
def get_parameters(self):
"""Fetch available parameter codes (pollutants)"""
# If we don't have API credentials, use mock data
if not EMAIL or not API_KEY:
return self.mock_get_parameters()
endpoint = f"{EPA_AQS_API_BASE_URL}/list/parametersByClass"
params = {
"email": EMAIL,
"key": API_KEY,
"pc": "CRITERIA" # Filter to criteria pollutants
}
try:
response = requests.get(endpoint, params=params)
data = response.json()
# Handle the specific response structure we observed
parameters = []
if isinstance(data, dict) and "Data" in data and isinstance(data["Data"], list):
parameters = data["Data"]
# Format as "code: name" for dropdown
result = []
for p in parameters:
code = p.get("code")
value = p.get("value_represented")
if not code:
code = p.get("parameter_code")
if not value:
value = p.get("parameter_name")
if code and value:
result.append(f"{code}: {value}")
return result
except Exception as e:
print(f"Error fetching parameters: {e}")
return []
def get_latest_aqi(self, state_code, county_code=None, parameter_code=None):
"""Fetch the latest AQI data for monitors"""
# If we don't have API credentials, use mock data
if not EMAIL or not API_KEY:
return [] # We don't have mock AQI data for simplicity
# Convert state code to numeric format for API
api_state_code = state_code
if len(state_code) == 2 and state_code in self.state_code_mapping:
api_state_code = self.state_code_mapping[state_code]
endpoint = f"{EPA_AQS_API_BASE_URL}/dailyData/byState"
params = {
"email": EMAIL,
"key": API_KEY,
"state": api_state_code,
"bdate": "20240314", # Beginning date (YYYYMMDD) - last 30 days
"edate": "20240414", # End date (YYYYMMDD) - current date
}
if county_code:
params["county"] = county_code
if parameter_code:
params["param"] = parameter_code
try:
response = requests.get(endpoint, params=params)
data = response.json()
# Handle the specific response structure we observed
if isinstance(data, dict) and "Data" in data and isinstance(data["Data"], list):
return data["Data"]
else:
print(f"Unexpected response format for AQI data: {type(data)}")
return []
except Exception as e:
print(f"Error fetching AQI data: {e}")
return []
def create_map(self, state_code, county_code=None, parameter_code=None):
"""Create a map with air quality monitoring stations"""
monitors = self.get_monitors(state_code, county_code, parameter_code)
if not monitors:
return "No monitoring stations found for the selected criteria."
# Convert to DataFrame for easier manipulation
df = pd.DataFrame(monitors)
# Create a map centered on the mean latitude and longitude
center_lat = df["latitude"].mean()
center_lon = df["longitude"].mean()
# Create a map with a specific width and height - make it bigger
m = folium.Map(location=[center_lat, center_lon], zoom_start=7, width='100%', height=700)
# Add a marker cluster
marker_cluster = MarkerCluster().add_to(m)
# Get latest AQI data if credentials are provided
aqi_data = {}
if EMAIL and API_KEY:
aqi_results = self.get_latest_aqi(state_code, county_code, parameter_code)
# Create a lookup dictionary by site ID
for item in aqi_results:
site_id = f"{item['state_code']}-{item['county_code']}-{item['site_number']}"
if site_id not in aqi_data or item['date_local'] > aqi_data[site_id]['date_local']:
aqi_data[site_id] = item
# Add markers for each monitoring station
for _, row in df.iterrows():
site_id = f"{row['state_code']}-{row['county_code']}-{row['site_number']}"
# Default marker color is blue
color = "blue"
aqi_info = ""
# If we have AQI data for this monitor, set the color based on AQI category
if site_id in aqi_data:
aqi_value = aqi_data[site_id].get('aqi', 0)
if aqi_value:
aqi_category = self.get_aqi_category(aqi_value)
color = self.aqi_categories.get(aqi_category, "blue")
aqi_info = f"
Latest AQI: {aqi_value} ({aqi_category})"
# Create popup content
popup_content = f"""
{row['local_site_name']}
Parameter: {row['parameter_name']}
Site ID: {site_id}
Latitude: {row['latitude']}, Longitude: {row['longitude']}{aqi_info}
"""
# Add marker to cluster
folium.Marker(
location=[row["latitude"], row["longitude"]],
popup=folium.Popup(popup_content, max_width=300),
icon=folium.Icon(color=color, icon="cloud"),
).add_to(marker_cluster)
# Return map HTML and legend HTML separately
map_html = m._repr_html_()
# Create legend HTML outside the map
legend_html = self.create_legend_html()
return {"map": map_html, "legend": legend_html}
def create_legend_html(self):
"""Create the HTML for the AQI legend"""
legend_html = """