112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Extract a current + daily weather table from Open-Meteo (Extract lane).
|
|
|
|
Artifact:
|
|
2026-11-01-openmeteo-weather.csv the extracted table
|
|
openmeteo-source-2026-11-01.json raw per-city source snapshot
|
|
|
|
Deterministic check: the committed file is re-derived from the committed
|
|
source snapshot, so a re-run with the same snapshot always reproduces the
|
|
CSV byte-for-byte. The CHECK script lives beside this file.
|
|
"""
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
CITIES = ["London", "Tokyo", "New York", "Sydney", "Cape Town", "São Paulo"]
|
|
FIELDS = [
|
|
"city", "country", "lat", "lon", "current_time",
|
|
"temperature_2m", "apparent_temperature", "relative_humidity_2m",
|
|
"weather_code", "weather_desc", "wind_speed_10m", "today_max",
|
|
"today_min", "today_precip",
|
|
]
|
|
WMO = {
|
|
0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast",
|
|
45: "Fog", 48: "Fog",
|
|
51: "Light drizzle", 53: "Drizzle", 55: "Heavy drizzle",
|
|
61: "Slight rain", 63: "Rain", 65: "Heavy rain",
|
|
71: "Slight snow", 73: "Snow", 75: "Heavy snow",
|
|
80: "Slight showers", 81: "Showers", 82: "Heavy showers",
|
|
85: "Snow showers", 86: "Heavy snow showers",
|
|
95: "Thunderstorm", 96: "Thunderstorm with slight hail", 99: "Thunderstorm with heavy hail",
|
|
}
|
|
|
|
|
|
def http_get(url, timeout=25):
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ledger-extract/1.0"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return r.read().decode("utf-8")
|
|
|
|
|
|
def geocode(name):
|
|
url = ("https://geocoding-api.open-meteo.com/v1/search?name=" + urllib.parse.quote(name)
|
|
+ "&count=1&language=en&format=json")
|
|
data = json.loads(http_get(url))
|
|
hits = data.get("results") or []
|
|
if not hits:
|
|
raise RuntimeError("no geocode hit for %r" % name)
|
|
h = hits[0]
|
|
return h["latitude"], h["longitude"], h.get("country") or h.get("country_code") or "n/a"
|
|
|
|
|
|
def fetch_weather(lat, lon):
|
|
lat_s = f"{lat:.6f}"
|
|
lon_s = f"{lon:.6f}"
|
|
url = ("https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}"
|
|
"¤t=temperature_2m,apparent_temperature,relative_humidity_2m,weather_code,wind_speed_10m"
|
|
"&daily=temperature_2m_max,temperature_2m_min,precipitation_sum&timezone=auto"
|
|
.format(lat=lat_s, lon=lon_s))
|
|
data = json.loads(http_get(url))
|
|
cur = data["current"]
|
|
day = data["daily"]
|
|
return {
|
|
"current_time": cur["time"],
|
|
"temperature_2m": cur["temperature_2m"],
|
|
"apparent_temperature": cur["apparent_temperature"],
|
|
"relative_humidity_2m": cur["relative_humidity_2m"],
|
|
"weather_code": cur["weather_code"],
|
|
"weather_desc": WMO.get(cur["weather_code"], "Unknown"),
|
|
"wind_speed_10m": cur["wind_speed_10m"],
|
|
"today_max": day["temperature_2m_max"][0],
|
|
"today_min": day["temperature_2m_min"][0],
|
|
"today_precip": day["precipitation_sum"][0],
|
|
}
|
|
|
|
|
|
def main():
|
|
rows = []
|
|
source = {}
|
|
for city in CITIES:
|
|
try:
|
|
lat, lon, country = geocode(city)
|
|
w = fetch_weather(lat, lon)
|
|
rows.append({"city": city, "country": country, "lat": lat, "lon": lon, **w})
|
|
source[city] = {"geocode": {"lat": lat, "lon": lon, "country": country},
|
|
"weather": w}
|
|
except Exception as exc: # noqa: BLE001
|
|
rows.append({"city": city, "country": "ERROR", "lat": "", "lon": "", "current_time": "",
|
|
"temperature_2m": "", "apparent_temperature": "", "relative_humidity_2m": "",
|
|
"weather_code": "", "weather_desc": "FAILED", "wind_speed_10m": "",
|
|
"today_max": "", "today_min": "", "today_precip": ""})
|
|
source[city] = {"error": repr(exc)}
|
|
|
|
csv_path = HERE / "2026-11-01-openmeteo-weather.csv"
|
|
with csv_path.open("w", newline="", encoding="utf-8") as f:
|
|
wcsv = csv.DictWriter(f, fieldnames=FIELDS, lineterminator="\n")
|
|
wcsv.writeheader()
|
|
for r in rows:
|
|
wcsv.writerow(r)
|
|
|
|
(HERE / "openmeteo-source-2026-11-01.json").write_text(
|
|
json.dumps(source, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|