113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
import csv
|
|
import json
|
|
import time
|
|
import requests
|
|
from pathlib import Path
|
|
|
|
PLACE_ID = 2143
|
|
BASE_URL = "https://api.inaturalist.org/v1/observations"
|
|
INPUT_FILE = Path(__file__).parent.parent / "wildlife" / "portland_wildlife.csv"
|
|
OUTPUT_FILE = Path(__file__).parent.parent / "wildlife" / "portland_wildlife_sightings.csv"
|
|
OUTPUT_FIELDS = ["taxon_id", "scientific_name", "common_name", "sighting_coordinates"]
|
|
|
|
|
|
def load_already_fetched(path: Path) -> set[int]:
|
|
"""Return set of taxon_ids already written to the output CSV."""
|
|
if not path.exists():
|
|
return set()
|
|
done = set()
|
|
try:
|
|
with open(path, newline="", encoding="utf-8") as f:
|
|
for row in csv.DictReader(f):
|
|
done.add(int(row["taxon_id"]))
|
|
except Exception:
|
|
pass
|
|
return done
|
|
|
|
|
|
def fetch_observations(taxon_id: int) -> list[list[float]]:
|
|
"""Fetch up to 600 random research-grade observations (3 pages x 200) and return [lat, lng] pairs."""
|
|
coords = []
|
|
for page in range(1, 4):
|
|
response = requests.get(
|
|
BASE_URL,
|
|
headers={"User-Agent": "WildDex (wildlife dex app; contact via colemanswenson@gmail.com)"},
|
|
params={
|
|
"taxon_id": taxon_id,
|
|
"quality_grade": "research",
|
|
"per_page": 200,
|
|
"page": page,
|
|
"order_by": "random",
|
|
"swlat": 14.5,
|
|
"swlng": -168.0,
|
|
"nelat": 72.0,
|
|
"nelng": -52.0,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
results = response.json().get("results", [])
|
|
if not results:
|
|
break
|
|
for obs in results:
|
|
location = obs.get("location")
|
|
if not location:
|
|
continue
|
|
try:
|
|
lat, lng = location.split(",")
|
|
coords.append([float(lat), float(lng)])
|
|
except (ValueError, AttributeError):
|
|
continue
|
|
time.sleep(1)
|
|
return coords
|
|
|
|
|
|
def read_species(path: Path) -> list[dict]:
|
|
"""Read portland_wildlife.csv and return all rows with taxon_id cast to int."""
|
|
with open(path, newline="", encoding="utf-8") as f:
|
|
rows = list(csv.DictReader(f))
|
|
for row in rows:
|
|
row["taxon_id"] = int(row["taxon_id"])
|
|
return rows
|
|
|
|
|
|
def main() -> None:
|
|
species = read_species(INPUT_FILE)
|
|
done = load_already_fetched(OUTPUT_FILE)
|
|
is_new = not OUTPUT_FILE.exists() or OUTPUT_FILE.stat().st_size == 0
|
|
total = len(species)
|
|
|
|
with open(OUTPUT_FILE, "a", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=OUTPUT_FIELDS)
|
|
if is_new:
|
|
writer.writeheader()
|
|
|
|
for i, sp in enumerate(species, start=1):
|
|
taxon_id = sp["taxon_id"]
|
|
|
|
if taxon_id in done:
|
|
print(f"[{i}/{total}] Skipping {sp['common_name']} (already done)")
|
|
continue
|
|
|
|
try:
|
|
coords = fetch_observations(taxon_id)
|
|
except requests.HTTPError as e:
|
|
print(f"[{i}/{total}] {sp['common_name']} — HTTP error {e}, skipping")
|
|
coords = []
|
|
|
|
writer.writerow({
|
|
"taxon_id": taxon_id,
|
|
"scientific_name": sp["scientific_name"],
|
|
"common_name": sp["common_name"],
|
|
"sighting_coordinates": json.dumps(coords),
|
|
})
|
|
f.flush()
|
|
|
|
print(f"[{i}/{total}] {sp['common_name']} ({taxon_id}) — {len(coords)} coords")
|
|
time.sleep(1)
|
|
|
|
print(f"\nDone. Output: {OUTPUT_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|