#!/usr/bin/env python3
"""
Poll weather stations for Santa Clara County, CA and Middlesex County, MA.
Generates individual county maps plus a combined map with layer controls.
"""
import requests
import json
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
# API endpoints
ALL_STATIONS_URL = "https://www.weatherlink.com/map/data"
STATION_DETAIL_URL = "https://www.weatherlink.com/map/data/station/{id}"
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
}
MAX_WORKERS = 5
# County definitions
COUNTIES = {
'santa_clara': {
'name': 'Santa Clara County, CA',
'short_name': 'Santa Clara County',
'state': 'California',
'bounds': {
'min_lat': 36.89,
'max_lat': 37.48,
'min_lng': -122.2,
'max_lng': -121.21
},
'center': [37.23, -121.7],
'zoom': 10,
'html_file': 'santa_clara_county_weather.html',
'data_file': 'santa_clara_county_data.json',
'btn_color': '#5cb85c' # green
},
'middlesex': {
'name': 'Middlesex County, MA',
'short_name': 'Middlesex County',
'state': 'Massachusetts',
'bounds': {
'min_lat': 42.24,
'max_lat': 42.73,
'min_lng': -71.66,
'max_lng': -71.02
},
'center': [42.48, -71.34],
'zoom': 10,
'html_file': 'middlesex_county_weather.html',
'data_file': 'middlesex_county_data.json',
'btn_color': '#9b59b6' # purple
}
}
def is_in_bounds(lat, lng, bounds):
return (bounds['min_lat'] <= lat <= bounds['max_lat'] and
bounds['min_lng'] <= lng <= bounds['max_lng'])
def get_temp_color(temp):
"""Get color for temperature value."""
try:
t = float(temp)
if t < 0: return '#29333b'
elif t < 10: return '#535f79'
elif t < 20: return '#2f6695'
elif t < 32: return '#1b7dbf'
elif t < 40: return '#27a5af'
elif t < 50: return '#00b58d'
elif t < 60: return '#009247'
elif t < 70: return '#fdd03b'
elif t < 80: return '#ed9922'
else: return '#dd5626'
except:
return '#888888'
def fetch_station_detail(station):
"""Fetch full details for a single station."""
station_id = station.get('id')
if not station_id:
return None
try:
url = STATION_DETAIL_URL.format(id=station_id)
resp = requests.get(url, headers=HEADERS, timeout=10)
resp.raise_for_status()
detail = resp.json()
temp = detail.get('temperature', '--')
# Filter out bad readings (outside -40 to 130 range)
try:
t = float(temp)
if t < -40 or t > 130:
temp = '--'
except:
pass
return {
'lat': station.get('lat'),
'lng': station.get('lng'),
'name': detail.get('sStation', 'Unknown'),
'temp': temp,
'humidity': detail.get('humidity', '--'),
'wind': detail.get('windSpeed', '--'),
'wind_dir': detail.get('windDirection', '--'),
'barometer': detail.get('barometer', '--'),
'updated': detail.get('lastUpdatedAt', '--')
}
except:
return None
def poll_county(county_key, all_stations):
"""Poll all stations in a county."""
county = COUNTIES[county_key]
bounds = county['bounds']
print(f"\n--- {county['name']} ---")
# Filter stations to this county
county_stations = [s for s in all_stations
if is_in_bounds(s.get('lat', 0), s.get('lng', 0), bounds)]
print(f" Found {len(county_stations)} stations")
if not county_stations:
return [], 0, 0
# Poll stations in parallel
stations = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {executor.submit(fetch_station_detail, s): s for s in county_stations}
for future in as_completed(futures):
result = future.result()
if result:
stations.append(result)
print(f" Got data for {len(stations)} stations")
# Calculate temp stats (excluding bad readings)
temps = []
for s in stations:
try:
t = float(s['temp'])
temps.append(t)
except:
pass
if temps:
min_temp = min(temps)
max_temp = max(temps)
print(f" Temp range: {min_temp:.0f}°F to {max_temp:.0f}°F")
else:
min_temp = max_temp = 0
# Add colors and county tag
for s in stations:
s['color'] = get_temp_color(s['temp'])
s['county'] = county_key
return stations, min_temp, max_temp
def generate_single_county_html(county_key, stations, min_temp, max_temp):
"""Generate HTML map for a single county."""
county = COUNTIES[county_key]
now = datetime.now()
html = f"""
{county['short_name']} Weather
"""
return html
def generate_combined_html(all_county_data):
"""Generate combined HTML map with both counties and layer controls."""
now = datetime.now()
# Combine all stations
all_stations = []
county_stats = {}
for county_key, data in all_county_data.items():
stations, min_temp, max_temp = data
all_stations.extend(stations)
county_stats[county_key] = {
'count': len(stations),
'min': min_temp,
'max': max_temp
}
html = f"""
Weather Map - Santa Clara & Middlesex Counties
"""
return html
def main():
print("=" * 60)
print("County Weather Map Poller")
print("Santa Clara County, CA & Middlesex County, MA")
print(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
# Fetch all stations once
print("\nFetching station list...")
resp = requests.get(ALL_STATIONS_URL, headers=HEADERS, timeout=30)
all_stations = resp.json()
print(f" Total worldwide: {len(all_stations):,}")
# Poll each county and store results
all_county_data = {}
for county_key in COUNTIES:
county = COUNTIES[county_key]
stations, min_temp, max_temp = poll_county(county_key, all_stations)
all_county_data[county_key] = (stations, min_temp, max_temp)
if stations:
# Generate and save individual county HTML
html = generate_single_county_html(county_key, stations, min_temp, max_temp)
with open(county['html_file'], 'w') as f:
f.write(html)
print(f" ✓ Saved {county['html_file']}")
# Save data JSON
with open(county['data_file'], 'w') as f:
json.dump(stations, f, indent=2)
print(f" ✓ Saved {county['data_file']}")
# Generate combined map
print("\n--- Combined Map ---")
combined_html = generate_combined_html(all_county_data)
with open('weather_map_sc_mx.html', 'w') as f:
f.write(combined_html)
print(" ✓ Saved weather_map_sc_mx.html")
# Save combined data
all_stations_combined = []
for county_key, data in all_county_data.items():
all_stations_combined.extend(data[0])
with open('sc_mx_stations_data.json', 'w') as f:
json.dump(all_stations_combined, f, indent=2)
print(" ✓ Saved sc_mx_stations_data.json")
print("\n" + "=" * 60)
print(f"Completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
if __name__ == "__main__":
main()