Create troubleshooting.py

troubleshooting.py

@@ -0,0 +1,186 @@
+import requests
+import json
+import os
+
+# Get API credentials from environment variables
+EPA_AQS_API_BASE_URL = "https://aqs.epa.gov/data/api"
+EMAIL = os.environ.get("EPA_AQS_EMAIL", "")
+API_KEY = os.environ.get("EPA_AQS_API_KEY", "")
+
+def print_separator():
+    print("=" * 50)
+
+def test_api_connection():
+    """Test the basic API connectivity and credentials"""
+    print_separator()
+    print("TESTING API CONNECTION")
+    print(f"Using email: {EMAIL}")
+    print(f"API key provided: {'Yes' if API_KEY else 'No'}")
+    
+    if not EMAIL or not API_KEY:
+        print("WARNING: Email or API key is missing. Real API calls will fail.")
+        return False
+    
+    # Test with a simple API call
+    endpoint = f"{EPA_AQS_API_BASE_URL}/list/states"
+    params = {
+        "email": EMAIL,
+        "key": API_KEY
+    }
+    
+    try:
+        response = requests.get(endpoint, params=params)
+        print(f"Response status code: {response.status_code}")
+        data = response.json()
+        print(f"Response header: {data.get('Header', {})}")
+        
+        if data.get("Header", {}).get("status") == "Success":
+            print("API connection successful!")
+            return True
+        else:
+            print(f"API connection failed: {data.get('Header', {}).get('error', 'Unknown error')}")
+            return False
+    except Exception as e:
+        print(f"Exception during API connection test: {e}")
+        return False
+
+def test_state_lookup(state_code):
+    """Test looking up a specific state"""
+    print_separator()
+    print(f"TESTING STATE LOOKUP FOR: {state_code}")
+    
+    endpoint = f"{EPA_AQS_API_BASE_URL}/list/states"
+    params = {
+        "email": EMAIL,
+        "key": API_KEY
+    }
+    
+    try:
+        response = requests.get(endpoint, params=params)
+        data = response.json()
+        
+        if data.get("Header", {}).get("status") == "Success":
+            states = data.get("Data", [])
+            print(f"Found {len(states)} states in the API")
+            
+            # Look for the specific state
+            matching_states = [s for s in states if s.get("state_code") == state_code or s.get("value") == state_code]
+            if matching_states:
+                print(f"Found matching state: {matching_states[0]}")
+                return matching_states[0]
+            else:
+                print(f"No matching state found for code: {state_code}")
+                print("Available state codes:")
+                for s in states[:10]:  # Print first 10 for brevity
+                    print(f"  {s.get('state_code')} - {s.get('value')}")
+                if len(states) > 10:
+                    print(f"  ... and {len(states)-10} more")
+                return None
+        else:
+            print(f"API lookup failed: {data.get('Header', {}).get('error', 'Unknown error')}")
+            return None
+    except Exception as e:
+        print(f"Exception during state lookup: {e}")
+        return None
+
+def test_counties_lookup(state_code):
+    """Test looking up counties for a state"""
+    print_separator()
+    print(f"TESTING COUNTIES LOOKUP FOR STATE: {state_code}")
+    
+    endpoint = f"{EPA_AQS_API_BASE_URL}/list/countiesByState"
+    params = {
+        "email": EMAIL,
+        "key": API_KEY,
+        "state": state_code
+    }
+    
+    try:
+        response = requests.get(endpoint, params=params)
+        data = response.json()
+        
+        if data.get("Header", {}).get("status") == "Success":
+            counties = data.get("Data", [])
+            print(f"Found {len(counties)} counties for state {state_code}")
+            
+            if counties:
+                print("Sample counties:")
+                for c in counties[:5]:  # Print first 5 for brevity
+                    print(f"  {c}")
+                if len(counties) > 5:
+                    print(f"  ... and {len(counties)-5} more")
+            else:
+                print("No counties found for this state")
+            
+            return counties
+        else:
+            print(f"API lookup failed: {data.get('Header', {}).get('error', 'Unknown error')}")
+            return None
+    except Exception as e:
+        print(f"Exception during counties lookup: {e}")
+        return None
+
+def test_monitors_lookup(state_code):
+    """Test looking up monitors for a state"""
+    print_separator()
+    print(f"TESTING MONITORS LOOKUP FOR STATE: {state_code}")
+    
+    endpoint = f"{EPA_AQS_API_BASE_URL}/monitors/byState"
+    params = {
+        "email": EMAIL,
+        "key": API_KEY,
+        "state": state_code,
+        "bdate": "20240101",
+        "edate": "20240414"
+    }
+    
+    try:
+        response = requests.get(endpoint, params=params)
+        data = response.json()
+        
+        if data.get("Header", {}).get("status") == "Success":
+            monitors = data.get("Data", [])
+            print(f"Found {len(monitors)} monitors for state {state_code}")
+            
+            if monitors:
+                print("Sample monitor:")
+                sample_monitor = monitors[0]
+                print(json.dumps(sample_monitor, indent=2))
+                
+                # Check keys that might be causing issues
+                print("\nChecking critical keys for indexing issues:")
+                for key in ["state_code", "county_code", "site_number", "parameter_code", "parameter_name", "latitude", "longitude", "local_site_name"]:
+                    if key in sample_monitor:
+                        print(f"  ✓ '{key}' found: {sample_monitor[key]}")
+                    else:
+                        print(f"  ✗ '{key}' missing!")
+            else:
+                print("No monitors found for this state")
+            
+            return monitors
+        else:
+            print(f"API lookup failed: {data.get('Header', {}).get('error', 'Unknown error')}")
+            return None
+    except Exception as e:
+        print(f"Exception during monitors lookup: {e}")
+        return None
+
+def run_tests():
+    """Run all tests"""
+    print("Starting API Tests")
+    api_working = test_api_connection()
+    
+    if api_working:
+        # Test some states with different formats
+        for state_code in ["06", "36", "CA", "NY"]:
+            state_info = test_state_lookup(state_code)
+            if state_info:
+                state_code_to_use = state_info.get("state_code")
+                test_counties_lookup(state_code_to_use)
+                test_monitors_lookup(state_code_to_use)
+    
+    print_separator()
+    print("Tests completed.")
+
+if __name__ == "__main__":
+    run_tests()