air_quality_map.py

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"<br>Latest AQI: {aqi_value} ({aqi_category})"
            
            # Create popup content
            popup_content = f"""
            <b>{row['local_site_name']}</b><br>
            Parameter: {row['parameter_name']}<br>
            Site ID: {site_id}<br>
            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 = """
        <div style="padding: 10px; border: 1px solid #ccc; border-radius: 5px; background-color: white; margin-top: 10px;">
        <h4 style="margin-top: 0;">AQI Categories</h4>
        <div style="display: grid; grid-template-columns: auto 1fr; grid-gap: 5px; align-items: center;">
        """
        
        for category, color in self.aqi_categories.items():
            legend_html += f'<span style="background-color: {color}; width: 20px; height: 20px; display: inline-block;"></span>'
            legend_html += f'<span>{category}</span>'
        
        legend_html += """
        </div>
        </div>
        """
        return legend_html
    
    def get_aqi_category(self, aqi_value):
        """Determine AQI category based on value"""
        aqi = int(aqi_value)
        if aqi <= 50:
            return "Good"
        elif aqi <= 100:
            return "Moderate"
        elif aqi <= 150:
            return "Unhealthy for Sensitive Groups"
        elif aqi <= 200:
            return "Unhealthy"
        elif aqi <= 300:
            return "Very Unhealthy"
        else:
            return "Hazardous"
    
    def mock_get_counties(self, state_code):
        """Return mock county data for the specified state"""
        if state_code in self.mock_counties:
            counties = self.mock_counties[state_code]
            return [f"{c['code']}: {c['value']}" for c in counties]
        else:
            # Return generic counties for other states
            return [
                "001: County 1",
                "002: County 2",
                "003: County 3",
                "004: County 4"
            ]
    
    def mock_get_parameters(self):
        """Return mock parameter data"""
        return [f"{p['code']}: {p['value_represented']}" for p in self.mock_parameters]
    
    def mock_get_monitors(self, state_code, county_code=None, parameter_code=None):
        """Mock function to return sample data for development"""
        # Get state code in proper format
        if len(state_code) == 2:
            # Convert 2-letter state code to numeric format for mock data
            state_code_mapping = {
                "CA": "06",
                "NY": "36",
                "TX": "48"
            }
            numeric_state_code = state_code_mapping.get(state_code, "01")  # Default to "01" if not found
        else:
            numeric_state_code = state_code
        # Sample data for California
        if state_code == "CA" or numeric_state_code == "06":
            monitors = [
                {
                    "state_code": "06",
                    "county_code": "037",
                    "site_number": "0001",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 34.0667,
                    "longitude": -118.2275,
                    "local_site_name": "Los Angeles - North Main Street",
                    "address": "1630 North Main Street",
                    "city_name": "Los Angeles",
                    "cbsa_name": "Los Angeles-Long Beach-Anaheim",
                    "date_established": "1998-01-01",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": "06",
                    "county_code": "037",
                    "site_number": "0002",
                    "parameter_code": "44201",
                    "parameter_name": "Ozone",
                    "poc": 1,
                    "latitude": 34.0667,
                    "longitude": -118.2275,
                    "local_site_name": "Los Angeles - North Main Street",
                    "address": "1630 North Main Street",
                    "city_name": "Los Angeles",
                    "cbsa_name": "Los Angeles-Long Beach-Anaheim",
                    "date_established": "1998-01-01",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": "06",
                    "county_code": "067",
                    "site_number": "0010",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 38.5661,
                    "longitude": -121.4926,
                    "local_site_name": "Sacramento - T Street",
                    "address": "1309 T Street",
                    "city_name": "Sacramento",
                    "cbsa_name": "Sacramento-Roseville",
                    "date_established": "1999-03-01",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": "06",
                    "county_code": "073",
                    "site_number": "0005",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 32.7333,
                    "longitude": -117.1500,
                    "local_site_name": "San Diego - Beardsley Street",
                    "address": "1110 Beardsley Street",
                    "city_name": "San Diego",
                    "cbsa_name": "San Diego-Carlsbad",
                    "date_established": "1999-04-15",
                    "last_sample_date": "2024-04-10"
                }
            ]
        # Sample data for New York
        elif state_code == "NY" or numeric_state_code == "36":
            monitors = [
                {
                    "state_code": "36",
                    "county_code": "061",
                    "site_number": "0010",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 40.7159,
                    "longitude": -73.9876,
                    "local_site_name": "New York - PS 59",
                    "address": "228 East 57th Street",
                    "city_name": "New York",
                    "cbsa_name": "New York-Newark-Jersey City",
                    "date_established": "1999-07-15",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": "36",
                    "county_code": "061",
                    "site_number": "0079",
                    "parameter_code": "44201",
                    "parameter_name": "Ozone",
                    "poc": 1,
                    "latitude": 40.8160,
                    "longitude": -73.9510,
                    "local_site_name": "New York - IS 52",
                    "address": "681 Kelly Street",
                    "city_name": "Bronx",
                    "cbsa_name": "New York-Newark-Jersey City",
                    "date_established": "1998-01-01",
                    "last_sample_date": "2024-04-10"
                }
            ]
        # Sample data for Texas
        elif state_code == "TX" or numeric_state_code == "48":
            monitors = [
                {
                    "state_code": "48",
                    "county_code": "201",
                    "site_number": "0024",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 29.7349,
                    "longitude": -95.3063,
                    "local_site_name": "Houston - Clinton Drive",
                    "address": "9525 Clinton Drive",
                    "city_name": "Houston",
                    "cbsa_name": "Houston-The Woodlands-Sugar Land",
                    "date_established": "1997-09-01",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": "48",
                    "county_code": "113",
                    "site_number": "0050",
                    "parameter_code": "44201",
                    "parameter_name": "Ozone",
                    "poc": 1,
                    "latitude": 32.8198,
                    "longitude": -96.8602,
                    "local_site_name": "Dallas - Hinton Street",
                    "address": "1415 Hinton Street",
                    "city_name": "Dallas",
                    "cbsa_name": "Dallas-Fort Worth-Arlington",
                    "date_established": "1998-01-01",
                    "last_sample_date": "2024-04-10"
                }
            ]
        else:
            # Default data for other states - generate some random monitors
            monitors = [
                {
                    "state_code": state_code,
                    "county_code": "001",
                    "site_number": "0001",
                    "parameter_code": "88101",
                    "parameter_name": "PM2.5 - Local Conditions",
                    "poc": 1,
                    "latitude": 40.0 + float(ord(state_code[0])) / 10,
                    "longitude": -90.0 - float(ord(state_code[1])) / 10,
                    "local_site_name": f"{self.states.get(state_code, 'Unknown')} - Station 1",
                    "address": "123 Main Street",
                    "city_name": "City 1",
                    "cbsa_name": f"{self.states.get(state_code, 'Unknown')} Metro Area",
                    "date_established": "2000-01-01",
                    "last_sample_date": "2024-04-10"
                },
                {
                    "state_code": state_code,
                    "county_code": "002",
                    "site_number": "0002",
                    "parameter_code": "44201",
                    "parameter_name": "Ozone",
                    "poc": 1,
                    "latitude": 40.5 + float(ord(state_code[0])) / 10,
                    "longitude": -90.5 - float(ord(state_code[1])) / 10,
                    "local_site_name": f"{self.states.get(state_code, 'Unknown')} - Station 2",
                    "address": "456 Oak Street",
                    "city_name": "City 2",
                    "cbsa_name": f"{self.states.get(state_code, 'Unknown')} Metro Area",
                    "date_established": "2000-01-01",
                    "last_sample_date": "2024-04-10"
                }
            ]
        
        # Filter by county if provided
        if county_code:
            monitors = [m for m in monitors if m["county_code"] == county_code]
            
        # Filter by parameter if provided
        if parameter_code:
            monitors = [m for m in monitors if m["parameter_code"] == parameter_code]
            
        return monitors

def create_air_quality_map_ui():
    """Create the Gradio interface for the Air Quality Map application"""
    app = AirQualityApp()
    
    def update_counties(state_code):
        """Callback to update counties dropdown when state changes"""
        counties = app.get_counties(state_code)
        return counties
    
    def show_map(state, county=None, parameter=None):
        """Callback to generate and display the map"""
        # Extract code from county string if provided
        county_code = None
        if county and ":" in county:
            county_code = county.split(":")[0].strip()
            
        # Extract code from parameter string if provided
        parameter_code = None
        if parameter and ":" in parameter:
            parameter_code = parameter.split(":")[0].strip()
            
        # Generate the map
        result = app.create_map(state, county_code, parameter_code)
        
        if isinstance(result, dict):
            # Combine map and legend HTML
            html_content = f"""
            <div>
                {result["map"]}
                {result["legend"]}
            </div>
            """
            return html_content
        else:
            # Return error message or whatever was returned
            return result
    
    # Create the UI
    with gr.Blocks(title="Air Quality Monitoring Stations") as interface:
        gr.Markdown("# NOAA Air Quality Monitoring Stations Map")
        gr.Markdown("""
        This application displays air quality monitoring stations in the United States.
        
        **Note:** To use the actual EPA AQS API, you need to register for an API key at 
        [https://aqs.epa.gov/aqsweb/documents/data_api.html](https://aqs.epa.gov/aqsweb/documents/data_api.html) 
        and update the EMAIL and API_KEY constants in the code.
        
        For demonstration without an API key, the app shows sample data for California (CA), New York (NY), and Texas (TX).
        """)
        
        with gr.Row():
            with gr.Column(scale=1):
                # State dropdown with default value
                state_dropdown = gr.Dropdown(
                    choices=list(app.states.keys()),
                    label="Select State",
                    value="CA"
                )
                
                # County dropdown with mock counties for the default state
                county_dropdown = gr.Dropdown(
                    choices=app.mock_get_counties("CA"),
                    label="Select County (Optional)",
                    allow_custom_value=True
                )
                
                # Parameter dropdown (pollutant type)
                parameter_dropdown = gr.Dropdown(
                    choices=app.mock_get_parameters(),
                    label="Select Pollutant (Optional)",
                    allow_custom_value=True
                )
                
                # Button to generate map
                map_button = gr.Button("Show Map")
            
            # HTML component to display the map in a larger column
            with gr.Column(scale=3):
                map_html = gr.HTML(label="Air Quality Monitoring Stations Map")
        
        # Set up event handlers
        state_dropdown.change(
            fn=update_counties,
            inputs=state_dropdown,
            outputs=county_dropdown
        )
        
        map_button.click(
            fn=show_map,
            inputs=[state_dropdown, county_dropdown, parameter_dropdown],
            outputs=map_html
        )
    
    return interface

# Create and launch the app
if __name__ == "__main__":
    air_quality_map_ui = create_air_quality_map_ui()
    air_quality_map_ui.launch()