"""Hochgeladene Fotos abholen, bestimmen und das Ergebnis zurückmelden. Der zweite Weg, auf dem Bilder zur Erkennung kommen. Der erste ist die Kamera am Haus: Der Detector holt sich dort einen Stream und entscheidet selbst, wann ein Besuch beginnt und endet. Hier liegt das Bild schon in der Web-App — jemand hat unterwegs fotografiert und hochgeladen. 🔑 **Die Richtung bleibt dieselbe: Der Detector fragt, niemand ruft ihn.** Das ist keine Bequemlichkeit, sondern die Bedingung dafür, dass die Erkennung im Heimnetz stehen kann: kein offener Port, keine erreichbare Adresse, keine Portfreigabe im Router. Dieselbe Mechanik benutzt ``fernkonfig.py``, um alle fünf Minuten die Konfiguration zu holen. ⚠️ **Keine Bewegungserkennung, keine Mittelung über Bilder.** Ein Foto ist ein Foto. Die ganze Logik aus ``watcher.py`` — Bewegung als Auslöser, Besuch über viele Einzelbilder, Mittelwert der Vogel-Werte — hat hier keine Entsprechung und wäre bei einem Einzelbild sogar falsch. Benutzt wird deshalb direkt ``Pipeline.run()``, genau wie bei ``python -m detector test``. 🔴 **Das Intervall ist eine Wartezeit vor den Augen eines Menschen, keine Hintergrundaufgabe.** Es stand auf 60 Sekunden — gedacht als schonende Hintergrundabfrage, tatsächlich aber die Zeit, die jemand vor dem Bildschirm „Der Detektor schaut sich das an …" wartet, nachdem er ein Foto geschickt hat. Christian hat das zweimal bemängelt, bevor es jemand als Fehler gelesen hat. Jetzt 5 Sekunden, und ein Durchlauf, der etwas gefunden hat, fragt **sofort** noch einmal nach: Mehrere Fotos hintereinander sollen nicht je 5 Sekunden kosten. Die Abfrage ist ein kurzes JSON gegen die eigene Instanz — bei 5 Sekunden sind das 720 Anfragen in der Stunde, für einen Django-Server nichts. """ from __future__ import annotations import io import json import time from PIL import Image from .classify import Pipeline from .sink import _encode class Warteschlange: """Fragt die Web-App nach offenen Fotos und arbeitet sie ab.""" def __init__( self, api: str, lese_token: str, schreib_token: str, pipeline: Pipeline, intervall: float = 5.0, anzahl: int = 20, timeout: int = 20, ): self.basis = api.rstrip("/") # ⚠️ Zwei verschiedene Token, und das ist Absicht: Das Auflisten ist # ein Lesevorgang nach außen (derselbe Token wie für Home # Assistant), das Zurückmelden ein Schreibvorgang wie das normale # Einliefern. Wer nur zuschauen darf, soll keine Bestimmung # eintragen können. self.lese_token = lese_token self.schreib_token = schreib_token self.pipeline = pipeline self.intervall = intervall self.anzahl = anzahl self.timeout = timeout self._letzter_versuch: float | None = None self._fehler_gemeldet = False def faellig(self, jetzt: float) -> bool: return ( self._letzter_versuch is None or jetzt - self._letzter_versuch >= self.intervall ) # -- Abruf ------------------------------------------------------------ def _offene(self) -> list[dict]: import requests resp = requests.get( f"{self.basis}/api/offen/", params={"anzahl": self.anzahl}, headers={"Authorization": f"Bearer {self.lese_token}"}, timeout=self.timeout, ) resp.raise_for_status() return resp.json().get("fotos") or [] def _bild_holen(self, url: str) -> Image.Image: import requests resp = requests.get(url, timeout=self.timeout) resp.raise_for_status() return Image.open(io.BytesIO(resp.content)).convert("RGB") # -- Bearbeiten ------------------------------------------------------- def _melden(self, foto_id: int, ausschnitt, result, box=None) -> None: import requests # ⚠️ ``Result.best`` ist eine Property, kein Aufruf — ``result.best()`` # scheitert erst zur Laufzeit und genau dann, wenn ein Vogel # erkannt wurde. bester = result.best if result else None # 🔴 **„Vogel ja, Art nein" ist etwas anderes als „kein Vogel" — und # ging hier verloren.** # # ``Pipeline.run`` gibt in DREI Fällen ``result = None`` zurück: kein # Fund, Beinahe-Treffer unter der Schwelle, und — der interessante — # ein sicher erkannter Vogel, dessen Art keine heimische ist. Die Box # unterscheidet die Fälle, nur hat sie vorher niemand angesehen. # # Folge am 14.10.2026: Ein Foto mit einer Meise im Gegenlicht wurde # mit 93 % als Vogel erkannt und kam in der App als „Kein Vogel # erkannt. Auf dem Foto war nichts zu finden" an. Wer das liest, # schneidet das Bild zu — gegen ein Problem, das nie der Zuschnitt # war. **Eine falsche Auskunft ist teurer als keine.** # # ``box.score`` ist genau die Zahl, die „da ist ein Vogel" bedeutet; # sie gehört in ``mean_bird_score``, dessen Sinn der Server selbst so # beschreibt: Vogelwert 0 heißt „auf keinem Bild war ein Vogel". vogelwert = 0.0 if result is not None: vogelwert = result.bird_score elif box is not None and self.pipeline.zaehlt_als_vogel(box): vogelwert = box.score nutzlast = { "species": bester.species.slug if bester else None, "confidence": round(bester.confidence, 4) if bester else 0.0, "mean_bird_score": round(vogelwert, 4), "votes": {g.species.slug: round(g.confidence, 4) for g in (result.guesses[:5] if result else [])}, } # Wo der Vogel sitzt. Die Sammlung schneidet daraus selbst zu -- # siehe web/dex/zuschnitt.py. if box is not None: nutzlast["box"] = [round(box.x0, 5), round(box.y0, 5), round(box.x1, 5), round(box.y1, 5)] dateien = {} if ausschnitt is not None: dateien["crop"] = ("crop.jpg", _encode(ausschnitt, 640), "image/jpeg") resp = requests.post( f"{self.basis}/api/offen/{foto_id}/", data={"data": json.dumps(nutzlast)}, files=dateien or None, headers={"Authorization": f"Bearer {self.schreib_token}"}, timeout=self.timeout, ) if not resp.ok: print(f"[warteschlange] {foto_id}: Rückmeldung {resp.status_code} " f"— {resp.text[:160]}") return body = resp.json() art = body.get("art") or "nichts erkannt" teile = [f"Foto {foto_id}: {art}"] if body.get("xp"): teile.append(f"+{body['xp']} XP") if body.get("erstfang"): teile.append("ERSTFANG!") for name in body.get("abzeichen") or []: teile.append(f"Abzeichen: {name}") print("[warteschlange] " + " · ".join(teile)) def durchlauf(self, jetzt: float | None = None) -> int: """Holt die offenen Fotos und bestimmt sie. Gibt die Anzahl zurück. ⚠️ Ein Fehler bei einem Foto darf die anderen nicht aufhalten. Sonst blockiert ein einziges kaputtes Bild die ganze Warteschlange, und niemand sieht, warum nichts mehr vorangeht. """ import requests self._letzter_versuch = time.monotonic() if jetzt is None else jetzt try: offene = self._offene() except requests.RequestException as exc: if not self._fehler_gemeldet: print(f"[warteschlange] nicht erreicht: {exc}") self._fehler_gemeldet = True return 0 if self._fehler_gemeldet: print("[warteschlange] wieder erreichbar") self._fehler_gemeldet = False fertig = 0 for foto in offene: if not foto.get("bild"): continue try: bild = self._bild_holen(foto["bild"]) box, result = self.pipeline.run(bild) # Den Ausschnitt mitschicken, damit die Sammlung dasselbe Bild # zeigt wie bei einer Sichtung von der Kamera. # ⚠️ `anzeige_ausschnitt`, nicht `crop`: Der enge Schnitt ist # fuer den Klassifikator gedacht; in der fast quadratischen # Kachel der Sammlung bleibt davon nur die Bauchmitte uebrig. ausschnitt = box.anzeige_ausschnitt(bild) if box is not None else None self._melden(foto["id"], ausschnitt, result, box) fertig += 1 except Exception as exc: # absichtlich breit, siehe Docstring print(f"[warteschlange] Foto {foto['id']} übersprungen: {exc}") return fertig