"""Misst, wie gut die Erkennung wirklich ist. Holt für jede Art mehrere echte Fotos von Wikimedia Commons, jagt sie durch dieselbe Pipeline, die auch am Vogelhaus läuft, und vergleicht die Bestimmung mit der Kategorie, aus der das Bild stammt. ./venv/bin/python scripts/benchmark.py --hole # Bilder laden (einmalig) ./venv/bin/python scripts/benchmark.py # messen Wichtige Einschränkung: Commons-Fotos sind in der Regel gut belichtete Nahaufnahmen von Fotografen, die den Vogel im Sucher hatten. Eine feste Kamera im Vogelhaus liefert schlechteres Material — Gegenlicht, Bewegungsunschärfe, halb verdeckte Vögel. Die Werte hier sind eine Obergrenze, keine Prognose für den Alltag. Die echte Zahl liefert erst die eigene Kamera. Die Kategorien enthalten außerdem vereinzelt falsch einsortierte Bilder. Ein paar Prozent Fehler gehen also auf Commons, nicht auf das Modell. """ from __future__ import annotations import argparse import json import sys import time import urllib.error import urllib.parse import urllib.request from collections import Counter, defaultdict from pathlib import Path from PIL import Image sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from detector.classify import Pipeline # noqa: E402 from detector.species_de import SPECIES # noqa: E402 UA = "VogelalbumBot/1.0 (https://gaertner.onl; webmaster@gaertner.onl)" COMMONS_API = "https://commons.wikimedia.org/w/api.php" DATA = Path(__file__).resolve().parents[1] / "benchmark_data" class RateLimited(Exception): pass def _open(url: str, timeout: int = 45) -> bytes: """Ein Abruf mit Rücksicht auf Wikimedias Ratenbegrenzung. Bei HTTP 429 wird gewartet statt weitergefeuert — erst ``Retry-After``, sonst exponentiell. Wer trotz 429 weitermacht, wird zu Recht gesperrt. """ delay = 5.0 for attempt in range(5): req = urllib.request.Request(url, headers={"User-Agent": UA}) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read() except urllib.error.HTTPError as exc: if exc.code != 429: raise wait = float(exc.headers.get("Retry-After") or delay) print(f" 429 — warte {wait:.0f}s", flush=True) time.sleep(min(wait, 120)) delay *= 2 raise RateLimited(f"nach 5 Versuchen weiterhin 429: {url[:80]}") def api(params: dict) -> dict: return json.loads(_open(f"{COMMONS_API}?{urllib.parse.urlencode(params)}", 30)) def image_urls(scientific: str, limit: int) -> list[str]: """Bilder aus der Commons-Kategorie zur Art.""" try: data = api( { "action": "query", "generator": "categorymembers", "gcmtitle": f"Category:{scientific}", "gcmtype": "file", # Großzügig anfragen: ein Teil sind Zeichnungen, Eier, Federn # oder Videos, die als Testbild nichts taugen. "gcmlimit": str(limit * 4), "prop": "imageinfo", "iiprop": "url|mime", "iiurlwidth": "900", "format": "json", "formatversion": "2", } ) except OSError: return [] urls = [] for page in data.get("query", {}).get("pages", []): info = (page.get("imageinfo") or [{}])[0] if info.get("mime") not in ("image/jpeg", "image/png"): continue title = page.get("title", "").lower() # Eier, Federn, Skelette und Verbreitungskarten sind keine Vogelfotos. if any(w in title for w in ("egg", "nest", "skull", "map", "range", "feather", "distribution", "skeleton", "illustration", "drawing", "plate", "stamp", "coin")): continue url = info.get("thumburl") or info.get("url") if url: urls.append(url) if len(urls) >= limit: break return urls def fetch(args) -> None: DATA.mkdir(exist_ok=True) for i, species in enumerate(SPECIES, 1): folder = DATA / species.slug folder.mkdir(exist_ok=True) have = len(list(folder.glob("*.jpg"))) if have >= args.pro_art: print(f"[{i:>2}/{len(SPECIES)}] {species.german}: {have} vorhanden") continue urls = image_urls(species.scientific, args.pro_art) if not urls: print(f"[{i:>2}/{len(SPECIES)}] {species.german}: keine Bilder in der " f"Kategorie '{species.scientific}'", flush=True) continue got, failed = 0, 0 for n, url in enumerate(urls): target = folder / f"{n:02d}.jpg" if target.exists(): got += 1 continue try: target.write_bytes(_open(url)) # Kaputte Downloads sofort aussortieren. Image.open(target).verify() got += 1 except Exception as exc: target.unlink(missing_ok=True) failed += 1 if failed == 1: # nicht pro Bild jammern, einmal genügt print(f" Fehler: {type(exc).__name__}: {exc}", flush=True) time.sleep(args.delay) note = f", {failed} fehlgeschlagen" if failed else "" print(f"[{i:>2}/{len(SPECIES)}] {species.german}: {got} Bilder{note}", flush=True) time.sleep(args.delay) def measure(args) -> int: habitats = set(args.habitats.split(",")) if args.habitats else None pipeline = Pipeline(min_bird_score=args.min_bird_score, habitats=habitats) by_slug = {s.slug: s for s in SPECIES} total = correct = top3 = no_bird = undetermined = 0 per_species: dict[str, list[bool]] = defaultdict(list) confusion: Counter = Counter() # Zwei Quellen: die Referenzfotos (ein Bild je Art, Dateiname = Slug) oder # die Commons-Kategorien (ein Ordner je Art, mehrere Bilder). if args.referenz: ref = Path(args.referenz) groups = [ (by_slug[p.stem], [p]) for p in sorted(ref.glob("*.jpg")) if p.stem in by_slug ] else: folders = sorted(p for p in DATA.iterdir() if p.is_dir()) if DATA.exists() else [] groups = [ (by_slug[f.name], sorted(f.glob("*.jpg"))) for f in folders if f.name in by_slug ] if not groups: print("Keine Bilder. Erst mit --hole laden (oder --referenz angeben).") return 1 for truth, paths in groups: for path in paths: try: img = Image.open(path).convert("RGB") except OSError: continue total += 1 box, result = pipeline.run(img) if box is None: no_bird += 1 per_species[truth.slug].append(False) continue if result is None or not result.guesses: undetermined += 1 per_species[truth.slug].append(False) continue names = [g.species.slug for g in result.guesses[:3]] hit = names[0] == truth.slug correct += hit top3 += truth.slug in names per_species[truth.slug].append(hit) if not hit: confusion[(truth.german, result.guesses[0].species.german)] += 1 if not total: print("Keine auswertbaren Bilder gefunden.") return 1 print(f"\n{'='*62}\nERGEBNIS ({total} Bilder, {len(groups)} Arten)\n{'='*62}") print(f" Richtig bestimmt (Top 1) : {correct:>4} / {total} {correct/total:>6.1%}") print(f" Unter den Top 3 : {top3:>4} / {total} {top3/total:>6.1%}") print(f" Kein Vogel gefunden : {no_bird:>4} / {total} {no_bird/total:>6.1%}") print(f" Als unbestimmt verworfen : {undetermined:>4} / {total} {undetermined/total:>6.1%}") rows = sorted( ((slug, sum(v), len(v)) for slug, v in per_species.items()), key=lambda r: (r[1] / r[2] if r[2] else 0, r[2]), ) print(f"\n{'-'*62}\nSchwächste Arten\n{'-'*62}") for slug, hits, n in rows[:12]: bar = "█" * round(12 * hits / n) if n else "" print(f" {by_slug[slug].german:<22} {hits:>2}/{n:<3} {hits/n:>6.0%} {bar}") print(f"\n{'-'*62}\nStärkste Arten\n{'-'*62}") for slug, hits, n in rows[-8:][::-1]: bar = "█" * round(12 * hits / n) if n else "" print(f" {by_slug[slug].german:<22} {hits:>2}/{n:<3} {hits/n:>6.0%} {bar}") if confusion: print(f"\n{'-'*62}\nHäufigste Verwechslungen\n{'-'*62}") for (truth, guess), n in confusion.most_common(12): print(f" {truth:<22} → {guess:<22} {n:>3}×") return 0 def main() -> int: p = argparse.ArgumentParser(description="Genauigkeit der Arterkennung messen") p.add_argument("--hole", action="store_true", help="Testbilder von Commons laden") p.add_argument("--pro-art", type=int, default=8) p.add_argument("--delay", type=float, default=1.5, help="Pause zwischen Abrufen. Wikimedia drosselt sonst.") p.add_argument("--min-bird-score", type=float, default=0.05) p.add_argument("--referenz", help="Ordner mit .jpg statt der Commons-Kategorien") p.add_argument("--habitats", help="Lebensräume kommagetrennt, z.B. 'garten'") args = p.parse_args() if args.hole: fetch(args) return 0 return measure(args) if __name__ == "__main__": sys.exit(main())