denshooter
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
import httpx
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
|
|
# USGS Earthquake Hazards Program - completely free, no API key
|
|
# Significant earthquakes (M 4.5+) from the past 7 days
|
|
USGS_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/4.5_week.geojson"
|
|
|
|
|
|
async def fetch_earthquakes() -> list:
|
|
"""
|
|
Fetches real earthquake data from USGS.
|
|
Returns quakes with M >= 4.5 from the past week.
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(USGS_URL, timeout=12.0)
|
|
if response.status_code != 200:
|
|
print(f"[SEISMIC] USGS returned {response.status_code}")
|
|
return []
|
|
|
|
data = response.json()
|
|
quakes = []
|
|
|
|
for feature in data.get("features", []):
|
|
props = feature.get("properties", {})
|
|
geom = feature.get("geometry", {})
|
|
coords = geom.get("coordinates", [])
|
|
|
|
if len(coords) < 2:
|
|
continue
|
|
|
|
lon, lat = coords[0], coords[1]
|
|
depth_km = coords[2] if len(coords) > 2 else 0
|
|
mag = props.get("mag")
|
|
|
|
if mag is None:
|
|
continue
|
|
# Validate coordinates
|
|
if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
|
|
continue
|
|
|
|
# Severity classification
|
|
if mag >= 7.0:
|
|
severity = "MAJOR"
|
|
elif mag >= 6.0:
|
|
severity = "STRONG"
|
|
elif mag >= 5.0:
|
|
severity = "MODERATE"
|
|
else:
|
|
severity = "MINOR"
|
|
|
|
# Convert USGS epoch ms to ISO string
|
|
epoch_ms = props.get("time", 0)
|
|
try:
|
|
dt = datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc)
|
|
time_str = dt.strftime("%Y-%m-%d %H:%M UTC")
|
|
except Exception:
|
|
time_str = "Unknown"
|
|
|
|
quakes.append({
|
|
"id": feature.get("id", ""),
|
|
"title": props.get("title", "Earthquake"),
|
|
"place": props.get("place", "Unknown Location"),
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"depth_km": round(depth_km, 1),
|
|
"magnitude": mag,
|
|
"severity": severity,
|
|
"time": time_str,
|
|
"url": props.get("url", ""),
|
|
"type": "earthquake",
|
|
"felt": props.get("felt", 0),
|
|
"tsunami": props.get("tsunami", 0),
|
|
})
|
|
|
|
# Sort by magnitude descending
|
|
quakes.sort(key=lambda q: q["magnitude"], reverse=True)
|
|
print(f"[SEISMIC] {len(quakes)} earthquakes (M≥4.5) — strongest: M{quakes[0]['magnitude'] if quakes else 'N/A'}")
|
|
return quakes
|
|
|
|
except Exception as e:
|
|
print(f"[SEISMIC] Fetch error: {e}")
|
|
return []
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = asyncio.run(fetch_earthquakes())
|
|
print(f"Earthquakes: {len(result)}")
|
|
for q in result[:5]:
|
|
print(f" M{q['magnitude']} {q['severity']} — {q['place']}")
|