[FEAT] frame: --bird-weather draws the collage from a ZIP code, no mic (#12)
This commit is contained in:
Teddy Warner
@@ -0,0 +1,206 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Top recently-heard birds near a ZIP, from BirdWeather, for the frame's
|
||||
--bird-weather mode.
|
||||
|
||||
Returns the shape the collage already expects, [{"sci","com","n"}], so the
|
||||
existing renderer draws it unchanged. Standalone (Python stdlib only). Filtered
|
||||
to species the collage can actually draw, read from the slug list bundled in
|
||||
apt.js, so no network call and it tracks whatever illustrations the repo ships.
|
||||
BirdWeather's public GraphQL needs no key.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
BIRDWEATHER = "https://app.birdweather.com/graphql"
|
||||
GEOCODER = "https://api.zippopotam.us"
|
||||
EBIRD = "https://api.ebird.org/v2/data/obs/geo/recent"
|
||||
APT_JS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "avian", "frontend", "apt.js")
|
||||
|
||||
_drawable = None
|
||||
|
||||
|
||||
def _graphql(query, timeout):
|
||||
req = urllib.request.Request(
|
||||
BIRDWEATHER,
|
||||
data=json.dumps({"query": query}).encode(),
|
||||
headers={"Content-Type": "application/json", "User-Agent": "AvianVisitors-frame/1.0"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return json.loads(r.read(2_000_000))
|
||||
except Exception:
|
||||
return {} # a transient upstream failure degrades to the next radius or fallback
|
||||
|
||||
|
||||
def geocode(zip_code, country="us", timeout=20):
|
||||
"""ZIP / postal code to (lat, lon) via the keyless zippopotam.us gazetteer."""
|
||||
url = f"{GEOCODER}/{country}/{urllib.parse.quote(zip_code.strip())}"
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "AvianVisitors-frame/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
place = json.loads(r.read(200_000))["places"][0]
|
||||
return float(place["latitude"]), float(place["longitude"])
|
||||
|
||||
|
||||
def bbox(lat, lon, miles):
|
||||
"""Square box `miles` to each side of the point. Returns (ne, sw) corners."""
|
||||
dlat = miles / 69.0
|
||||
dlon = miles / (69.0 * max(0.2, math.cos(math.radians(lat))))
|
||||
return (lat + dlat, lon + dlon), (lat - dlat, lon - dlon)
|
||||
|
||||
|
||||
def top_species(lat, lon, miles, days=7, limit=60, timeout=20):
|
||||
"""BirdWeather's most-detected species in the box, as [{sci,com,n}]."""
|
||||
(ne_lat, ne_lon), (sw_lat, sw_lon) = bbox(lat, lon, miles)
|
||||
query = (
|
||||
'{ topSpecies(period: {count: %d, unit: "day"}, '
|
||||
'ne: {lat: %f, lon: %f}, sw: {lat: %f, lon: %f}, limit: %d) '
|
||||
'{ count species { commonName scientificName } } }'
|
||||
% (days, ne_lat, ne_lon, sw_lat, sw_lon, limit)
|
||||
)
|
||||
rows = (_graphql(query, timeout).get("data") or {}).get("topSpecies") or []
|
||||
out = []
|
||||
for row in rows:
|
||||
sp = row.get("species") or {}
|
||||
sci = sp.get("scientificName")
|
||||
if sci:
|
||||
out.append({"sci": sci, "com": sp.get("commonName") or sci, "n": row.get("count") or 1})
|
||||
return out
|
||||
|
||||
|
||||
def _haversine(lat1, lon1, lat2, lon2):
|
||||
"""Great-circle distance between two points, in miles."""
|
||||
r = 3958.8
|
||||
p1, p2 = math.radians(lat1), math.radians(lat2)
|
||||
dp, dl = math.radians(lat2 - lat1), math.radians(lon2 - lon1)
|
||||
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
|
||||
return 2 * r * math.asin(math.sqrt(a))
|
||||
|
||||
|
||||
def nearest_stations(lat, lon, n=3, boxes=(50, 150, 400), timeout=20):
|
||||
"""The n closest BirdWeather stations to the point, as [(miles, id)], found
|
||||
by growing the search box until at least n are in view."""
|
||||
nodes = []
|
||||
for miles in boxes:
|
||||
(ne_lat, ne_lon), (sw_lat, sw_lon) = bbox(lat, lon, miles)
|
||||
query = ('{ stations(ne: {lat: %f, lon: %f}, sw: {lat: %f, lon: %f}, first: 200) '
|
||||
'{ nodes { id coords { lat lon } } } }' % (ne_lat, ne_lon, sw_lat, sw_lon))
|
||||
nodes = (((_graphql(query, timeout).get("data") or {}).get("stations") or {}).get("nodes")) or []
|
||||
if len(nodes) >= n:
|
||||
break
|
||||
out = []
|
||||
for node in nodes:
|
||||
c = node.get("coords") or {}
|
||||
if c.get("lat") is not None and c.get("lon") is not None:
|
||||
out.append((_haversine(lat, lon, c["lat"], c["lon"]), node["id"]))
|
||||
out.sort()
|
||||
return out[:n]
|
||||
|
||||
|
||||
def triangulate(lat, lon, n=3, days=7, timeout=20):
|
||||
"""Estimate the birds near a point from the n closest stations, inverse-
|
||||
distance weighted so the nearest station counts most. Used where the local
|
||||
box has too few stations to fill a collage, so remote ZIPs still get birds."""
|
||||
weighted = {}
|
||||
for miles, sid in nearest_stations(lat, lon, n, timeout=timeout):
|
||||
weight = 1.0 / max(miles, 1.0)
|
||||
query = ('{ topSpecies(stationIds: [%s], period: {count: %d, unit: "day"}, limit: 40) '
|
||||
'{ count species { commonName scientificName } } }' % (sid, days))
|
||||
for row in (((_graphql(query, timeout).get("data") or {}).get("topSpecies")) or []):
|
||||
sp = row.get("species") or {}
|
||||
sci = sp.get("scientificName")
|
||||
if not sci:
|
||||
continue
|
||||
entry = weighted.setdefault(sci, {"com": sp.get("commonName") or sci, "score": 0.0})
|
||||
entry["score"] += (row.get("count") or 0) * weight
|
||||
out = [{"sci": sci, "com": e["com"], "n": max(1, round(e["score"]))} for sci, e in weighted.items()]
|
||||
out.sort(key=lambda s: -s["n"])
|
||||
return out
|
||||
|
||||
|
||||
def ebird_nearby(lat, lon, days=14, key=None, timeout=20):
|
||||
"""Recent eBird observations near the point, as [{sci,com,n}]. The deepest
|
||||
fallback, for spots with no station in range. Needs a free eBird API key in
|
||||
EBIRD_API_KEY and returns [] without one, so keyless installs just skip it."""
|
||||
key = key or os.environ.get("EBIRD_API_KEY")
|
||||
if not key:
|
||||
return []
|
||||
url = f"{EBIRD}?lat={lat:.4f}&lng={lon:.4f}&dist=50&back={min(days, 30)}"
|
||||
req = urllib.request.Request(url, headers={"X-eBirdApiToken": key, "User-Agent": "AvianVisitors-frame/1.0"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
observations = json.loads(r.read(5_000_000))
|
||||
except Exception:
|
||||
return []
|
||||
tally = {}
|
||||
for o in observations:
|
||||
sci = o.get("sciName")
|
||||
if not sci:
|
||||
continue
|
||||
entry = tally.setdefault(sci, {"com": o.get("comName") or sci, "n": 0})
|
||||
entry["n"] += o.get("howMany") or 1
|
||||
out = [{"sci": sci, "com": e["com"], "n": e["n"]} for sci, e in tally.items()]
|
||||
out.sort(key=lambda s: -s["n"])
|
||||
return out
|
||||
|
||||
|
||||
def slugify(sci):
|
||||
return re.sub(r"[^a-z0-9]+", "-", sci.lower()).strip("-")
|
||||
|
||||
|
||||
def drawable_slugs(apt_js=APT_JS):
|
||||
"""Base slugs we have a cutout for, parsed once from the collage's bundled
|
||||
DIMS table (perched and flight entries collapse to the base slug)."""
|
||||
global _drawable
|
||||
if _drawable is None:
|
||||
with open(apt_js, encoding="utf-8") as f:
|
||||
block = re.search(r"var DIMS = (\{.*?\});", f.read(), re.S)
|
||||
keys = re.findall(r'"([a-z0-9-]+)"\s*:', block.group(1)) if block else []
|
||||
_drawable = {re.sub(r"-2$", "", k) for k in keys}
|
||||
return _drawable
|
||||
|
||||
|
||||
def species_for_zip(zip_code, country="us", target=10, days=7, radii=(15, 30, 50),
|
||||
apt_js=APT_JS, timeout=20):
|
||||
"""Geocode the ZIP, pull BirdWeather top species, and grow the search radius
|
||||
only until `target` drawable species are found, so it stays as local as the
|
||||
data allows. Returns the top `target` by detection count, or fewer where
|
||||
birds or stations are sparse.
|
||||
"""
|
||||
lat, lon = geocode(zip_code, country, timeout)
|
||||
have = drawable_slugs(apt_js)
|
||||
found = []
|
||||
for miles in radii:
|
||||
found = [s for s in top_species(lat, lon, miles, days, 60, timeout)
|
||||
if slugify(s["sci"]) in have]
|
||||
if len(found) >= target:
|
||||
break
|
||||
if len(found) < target:
|
||||
# box too sparse: estimate from the nearest stations so remote ZIPs still fill
|
||||
tri = [s for s in triangulate(lat, lon, 3, days, timeout) if slugify(s["sci"]) in have]
|
||||
if len(tri) > len(found):
|
||||
found = tri
|
||||
if len(found) < target:
|
||||
# no station in range at all: fall back to eBird sightings, if a key is set
|
||||
eb = [s for s in ebird_nearby(lat, lon, days * 2, timeout=timeout) if slugify(s["sci"]) in have]
|
||||
if len(eb) > len(found):
|
||||
found = eb
|
||||
found.sort(key=lambda s: -(s["n"] or 0))
|
||||
return found[:target]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
ap = argparse.ArgumentParser(description="Print BirdWeather top drawable species for a ZIP.")
|
||||
ap.add_argument("zip")
|
||||
ap.add_argument("--country", default="us")
|
||||
ap.add_argument("--target", type=int, default=10)
|
||||
ap.add_argument("--days", type=int, default=7)
|
||||
a = ap.parse_args()
|
||||
for s in species_for_zip(a.zip, a.country, a.target, a.days):
|
||||
print(f'{s["n"]:>8} {s["com"]} ({s["sci"]})')
|
||||
Reference in New Issue
Block a user