christian
c11fe9ab04
Eine Kamera am Vogelhaus, ein Rechner der erkennt wer da war — und eine Sammlung, die sich von allein fuellt. Zwei neuronale Netze: eines findet den Vogel im Bild, eines bestimmt die Art. 148 heimische Arten, auf Deutschland eingeschraenkt und neu normiert — ohne diese Einschraenkung wird aus einer Kohlmeise eine nordamerikanische Schwesterart. Daraus entsteht ein Sammelalbum: nummerierte Seiten, gefundene Arten in Farbe, fehlende als Schatten, dazu Erstfang, Punkte, Stufen und 15 Abzeichen. Laeuft vollstaendig auf eigener Hardware. Kein Cloud-Dienst, kein Konto, keine Telemetrie; die Bilder verlassen das eigene Netz nicht. Zwei getrennt deploybare Teile: die Sammlung als Django-Anwendung, die Erkennung als eigener Dienst. Der Erkennungs-Dienst fragt selbst an und braucht deshalb weder eine erreichbare Adresse noch einen offenen Port — er laeuft hinter einem gewoehnlichen Router. Dabei: Firmware fuer Raspberry Pi und ESP32-S3, Anbindung an Home Assistant ueber MQTT, Oberflaeche auf Deutsch und Englisch, Einstellungen je Kamera aus dem Browser, und ein Weg fuer Fotos von unterwegs. AGPL-3.0.
367 lines
14 KiB
Python
367 lines
14 KiB
Python
"""Artbestimmung per LiteRT (ehemals TFLite).
|
||
|
||
Zwei Stufen:
|
||
|
||
1. ``BirdFinder`` — SSD-MobileNet auf COCO. Beantwortet nur: "ist da überhaupt
|
||
ein Vogel, und wo?". Verhindert, dass ein wackelnder Ast als Kohlmeise
|
||
in die Sammlung wandert, und liefert den Ausschnitt fürs Zuschneiden.
|
||
|
||
2. ``SpeciesClassifier`` — MobileNetV2 auf iNaturalist-Vögeln (965 Arten).
|
||
Läuft auf dem zugeschnittenen Vogel, nicht auf dem ganzen Bild.
|
||
|
||
Beide Modelle sind int8-quantisiert und laufen auf dem N100 in wenigen
|
||
Dutzend Millisekunden — reicht dicke, weil nur bei Bewegung klassifiziert wird.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
from PIL import Image
|
||
|
||
from .species_de import BY_SCIENTIFIC, Species, catalog, label_to_scientific
|
||
|
||
MODELS = Path(__file__).resolve().parent.parent / "models"
|
||
|
||
CLASSIFIER_MODEL = MODELS / "mobilenet_v2_1.0_224_inat_bird_quant.tflite"
|
||
CLASSIFIER_LABELS = MODELS / "inat_bird_labels.txt"
|
||
DETECTOR_MODEL = MODELS / "ssd_mobilenet_v2_coco_quant_postprocess.tflite"
|
||
|
||
COCO_BIRD_CLASS = 15 # in diesem Modell ist 15 = "bird"
|
||
|
||
|
||
def _interpreter(path: Path):
|
||
from ai_edge_litert.interpreter import Interpreter
|
||
|
||
if not path.exists():
|
||
raise FileNotFoundError(
|
||
f"Modell fehlt: {path}\nHol es mit: ./scripts/fetch_models.sh"
|
||
)
|
||
interp = Interpreter(model_path=str(path), num_threads=4)
|
||
interp.allocate_tensors()
|
||
return interp
|
||
|
||
|
||
@dataclass
|
||
class Box:
|
||
"""Bounding Box in relativen Koordinaten (0..1)."""
|
||
|
||
x0: float
|
||
y0: float
|
||
x1: float
|
||
y1: float
|
||
score: float
|
||
|
||
def crop(self, img: Image.Image, pad: float = 0.12) -> Image.Image:
|
||
"""Schneidet den Vogel aus — mit etwas Rand, sonst fehlen Schwanz und
|
||
Schnabelspitze, und genau daran hängt oft die Artbestimmung."""
|
||
w, h = img.size
|
||
bw, bh = (self.x1 - self.x0) * w, (self.y1 - self.y0) * h
|
||
x0 = max(0, int(self.x0 * w - bw * pad))
|
||
y0 = max(0, int(self.y0 * h - bh * pad))
|
||
x1 = min(w, int(self.x1 * w + bw * pad))
|
||
y1 = min(h, int(self.y1 * h + bh * pad))
|
||
if x1 <= x0 or y1 <= y0:
|
||
return img
|
||
return img.crop((x0, y0, x1, y1))
|
||
|
||
def anzeige_ausschnitt(self, img: Image.Image, pad: float = 0.35) -> Image.Image:
|
||
"""Derselbe Vogel, aber als Bild zum Ansehen statt zum Bestimmen.
|
||
|
||
🔑 **Der Klassifikator und das Auge wollen Verschiedenes.** ``crop``
|
||
liegt eng am Vogel, weil jedes Stück Hintergrund die Bestimmung
|
||
verwässert. Für die Sammlung ist genau das falsch: Dort steht der
|
||
Ausschnitt in einer fast quadratischen Kachel, und die schneidet links
|
||
und rechts weg, was über das Quadrat hinausragt.
|
||
|
||
Bei einem Vogel im Hochformat — und so sitzt fast jeder — bleibt davon
|
||
die Bauchmitte übrig, formatfüllend und unscharf. Gemeldet wurde es als
|
||
*„der Zuschnitt sieht merkwürdig aus"*, und das traf es: Vom Vogel war nur Gefieder zu
|
||
sehen, Kopf und Schwanz lagen außerhalb.
|
||
|
||
Deshalb hier mehr Rand **und** ein Ausschnitt in Richtung Quadrat: Was
|
||
die Kachel abschneidet, ist dann Umgebung statt Vogel.
|
||
|
||
⚠️ Absichtlich eine eigene Methode. ``crop`` hier zu ändern hieße, die
|
||
Artbestimmung zu ändern — und die ist gemessen und soll so bleiben.
|
||
"""
|
||
w, h = img.size
|
||
bx0, by0 = self.x0 * w, self.y0 * h
|
||
bx1, by1 = self.x1 * w, self.y1 * h
|
||
bw, bh = bx1 - bx0, by1 - by0
|
||
if bw <= 0 or bh <= 0:
|
||
return img
|
||
|
||
# Erst Rand, dann die kürzere Seite auf die längere ziehen.
|
||
seite = max(bw, bh) * (1 + 2 * pad)
|
||
mx, my = (bx0 + bx1) / 2, (by0 + by1) / 2
|
||
|
||
# ⚠️ Nicht über das Bild hinaus, und die Mitte nachführen statt den
|
||
# Ausschnitt zu stauchen — sonst sitzt ein Vogel am Bildrand
|
||
# außermittig in seiner eigenen Kachel.
|
||
seite = min(seite, float(w), float(h))
|
||
mx = min(max(mx, seite / 2), w - seite / 2)
|
||
my = min(max(my, seite / 2), h - seite / 2)
|
||
|
||
return img.crop((
|
||
int(mx - seite / 2), int(my - seite / 2),
|
||
int(mx + seite / 2), int(my + seite / 2),
|
||
))
|
||
|
||
@property
|
||
def area(self) -> float:
|
||
return max(0.0, self.x1 - self.x0) * max(0.0, self.y1 - self.y0)
|
||
|
||
|
||
class BirdFinder:
|
||
"""Findet Vögel im Bild. Sagt nichts über die Art."""
|
||
|
||
def __init__(self, model: Path = DETECTOR_MODEL, threshold: float = 0.4):
|
||
self.interp = _interpreter(model)
|
||
self.threshold = threshold
|
||
inp = self.interp.get_input_details()[0]
|
||
self.index = inp["index"]
|
||
self.size = (inp["shape"][2], inp["shape"][1])
|
||
self.out = self.interp.get_output_details()
|
||
|
||
def find(self, img: Image.Image) -> list[Box]:
|
||
arr = np.asarray(
|
||
img.convert("RGB").resize(self.size, Image.BILINEAR), dtype=np.uint8
|
||
)
|
||
self.interp.set_tensor(self.index, arr[None, ...])
|
||
self.interp.invoke()
|
||
|
||
boxes = self.interp.get_tensor(self.out[0]["index"])[0]
|
||
classes = self.interp.get_tensor(self.out[1]["index"])[0]
|
||
scores = self.interp.get_tensor(self.out[2]["index"])[0]
|
||
count = int(self.interp.get_tensor(self.out[3]["index"])[0])
|
||
|
||
found = []
|
||
for i in range(min(count, len(scores))):
|
||
if int(classes[i]) != COCO_BIRD_CLASS or scores[i] < self.threshold:
|
||
continue
|
||
y0, x0, y1, x1 = boxes[i]
|
||
found.append(
|
||
Box(float(x0), float(y0), float(x1), float(y1), float(scores[i]))
|
||
)
|
||
found.sort(key=lambda b: b.score, reverse=True)
|
||
return found
|
||
|
||
|
||
@dataclass
|
||
class Guess:
|
||
species: Species
|
||
confidence: float # nach Neunormierung auf die deutsche Artenliste
|
||
|
||
def __repr__(self) -> str: # praktisch beim Debuggen im REPL
|
||
return f"<{self.species.german} {self.confidence:.1%}>"
|
||
|
||
|
||
@dataclass
|
||
class Result:
|
||
guesses: list[Guess]
|
||
raw_top: list[tuple[str, float]] # ungefiltert, alle 965 Arten
|
||
ms: float
|
||
bird_score: float = 0.0
|
||
"""Wie viel der Gesamtwahrscheinlichkeit fällt auf heimische Arten?
|
||
|
||
Ein niedriger Wert heißt: das Modell ist sich sicher, einen Vogel zu sehen
|
||
— nur eben einen, den es hier nicht gibt. Das ist das Signal für eine
|
||
Fehlbestimmung, nicht die Confidence des Top-Treffers, denn die ist nach
|
||
der Neunormierung per Konstruktion hoch.
|
||
"""
|
||
|
||
@property
|
||
def best(self) -> Guess | None:
|
||
return self.guesses[0] if self.guesses else None
|
||
|
||
|
||
class SpeciesClassifier:
|
||
def __init__(
|
||
self,
|
||
model: Path = CLASSIFIER_MODEL,
|
||
labels: Path = CLASSIFIER_LABELS,
|
||
restrict_to_germany: bool = True,
|
||
habitats: set[str] | None = None,
|
||
):
|
||
"""``habitats`` schränkt den Katalog ein, z.B. ``{"garten"}``.
|
||
|
||
Je enger die Liste, desto weniger kann das Modell danebengreifen —
|
||
ohne Einschränkung stehen 148 Arten zur Wahl, mit Gartenfilter 47.
|
||
"""
|
||
self.interp = _interpreter(model)
|
||
inp = self.interp.get_input_details()[0]
|
||
self.index = inp["index"]
|
||
self.size = (inp["shape"][2], inp["shape"][1])
|
||
out = self.interp.get_output_details()[0]
|
||
self.out_index = out["index"]
|
||
self.scale, self.zero = out["quantization"]
|
||
|
||
self.labels = labels.read_text(encoding="utf-8").splitlines()
|
||
self.restrict = restrict_to_germany
|
||
self.setze_arten(catalog(habitats))
|
||
|
||
def setze_arten(self, arten: list[Species]) -> None:
|
||
"""Den gültigen Katalog austauschen.
|
||
|
||
Getrennt vom Konstruktor, weil die Web-App die Artenliste vorgibt und
|
||
sie sich im Betrieb ändern darf. Das Modell bleibt dabei geladen — neu
|
||
berechnet wird nur die Zuordnung der 965 Ausgaben auf den Katalog, und
|
||
das kostet Millisekunden statt Sekunden.
|
||
"""
|
||
nach_name = {a.scientific: a for a in arten}
|
||
|
||
# Welche der 965 Ausgaben entsprechen Arten aus unserem Katalog?
|
||
self.native_idx: dict[int, Species] = {}
|
||
for i, label in enumerate(self.labels):
|
||
sci = label_to_scientific(label)
|
||
if sci in nach_name:
|
||
self.native_idx[i] = nach_name[sci]
|
||
|
||
def classify(self, img: Image.Image, top_k: int = 5) -> Result:
|
||
t0 = time.perf_counter()
|
||
arr = np.asarray(
|
||
img.convert("RGB").resize(self.size, Image.BILINEAR), dtype=np.uint8
|
||
)
|
||
self.interp.set_tensor(self.index, arr[None, ...])
|
||
self.interp.invoke()
|
||
raw = self.interp.get_tensor(self.out_index)[0].astype(np.float32)
|
||
probs = (raw - self.zero) * self.scale
|
||
ms = (time.perf_counter() - t0) * 1000
|
||
|
||
order = np.argsort(probs)[::-1][:top_k]
|
||
raw_top = [(self.labels[i], float(probs[i])) for i in order]
|
||
|
||
native = [(i, float(probs[i])) for i in self.native_idx]
|
||
native.sort(key=lambda t: t[1], reverse=True)
|
||
native_mass = sum(p for _, p in native)
|
||
|
||
if not self.restrict:
|
||
guesses = [Guess(self.native_idx[i], p) for i, p in native[:top_k] if p > 0]
|
||
return Result(guesses, raw_top, ms, native_mass)
|
||
|
||
# Neunormierung: die Wahrscheinlichkeitsmasse der ~45 heimischen Arten
|
||
# auf 1.0 skalieren. Ohne das liegt jede Confidence im einstelligen
|
||
# Prozentbereich, weil sich 965 Klassen die Masse teilen.
|
||
guesses = []
|
||
if native_mass > 0:
|
||
for i, p in native[:top_k]:
|
||
if p <= 0:
|
||
continue
|
||
guesses.append(Guess(self.native_idx[i], p / native_mass))
|
||
|
||
return Result(guesses, raw_top, ms, native_mass)
|
||
|
||
|
||
class Pipeline:
|
||
"""Detektor + Klassifikator als eine Einheit."""
|
||
|
||
def __init__(
|
||
self,
|
||
detect_threshold: float = 0.4,
|
||
min_bird_score: float = 0.05,
|
||
restrict_to_germany: bool = True,
|
||
habitats: set[str] | None = None,
|
||
report_threshold: float | None = None,
|
||
):
|
||
# 0.05 ist gemessen, nicht geraten: scripts/benchmark.py über die
|
||
# 44 Referenzfotos ergibt 43/44 richtig bei 0.02–0.05, aber nur 40/44
|
||
# bei 0.10. Der Grund ist die Nordamerika-Schlagseite des Modells —
|
||
# beim Mäusebussard gewinnt global der amerikanische Buteo lineatus,
|
||
# die heimische Masse fällt auf 5 %, obwohl Buteo buteo innerhalb der
|
||
# deutschen Arten klar vorn liegt. Ein zu hoher Schwellwert wirft also
|
||
# richtige Bestimmungen weg.
|
||
self.detect_threshold = detect_threshold
|
||
# Der Detektor sucht ab der *niedrigeren* der beiden Schwellen, damit
|
||
# das Protokoll die Beinahe-Treffer ueberhaupt zu sehen bekommt. Mit
|
||
# nur einer Schwelle war `--protokoll-ab 0.08` wirkungslos: was der
|
||
# Finder bei 0.4 verwarf, kam nie beim Beobachter an — man konnte
|
||
# danach nicht unterscheiden zwischen "kein Vogel da" und "Vogel zu
|
||
# klein im Bild". Genau diese Frage sollte das Protokoll beantworten.
|
||
# Der Wunschwert wird gemerkt, weil 'uebernimm' die Schwelle spaeter
|
||
# neu mischen muss: aendert sich detect_threshold, gilt weiterhin die
|
||
# niedrigere von beiden.
|
||
self._report_wunsch = (
|
||
detect_threshold if report_threshold is None else report_threshold
|
||
)
|
||
self.report_threshold = min(detect_threshold, self._report_wunsch)
|
||
self.finder = BirdFinder(threshold=self.report_threshold)
|
||
self.classifier = SpeciesClassifier(
|
||
restrict_to_germany=restrict_to_germany, habitats=habitats
|
||
)
|
||
self.min_bird_score = min_bird_score
|
||
|
||
def zaehlt_als_vogel(self, box: Box | None) -> bool:
|
||
"""Reicht diese Box fuer eine Sichtung, oder ist sie nur Protokoll?"""
|
||
return box is not None and box.score >= self.detect_threshold
|
||
|
||
def uebernimm(
|
||
self,
|
||
detect_threshold: float | None = None,
|
||
min_bird_score: float | None = None,
|
||
arten: list[Species] | None = None,
|
||
) -> list[str]:
|
||
"""Einstellungen im laufenden Betrieb aendern.
|
||
|
||
Gibt zurueck, was sich geaendert hat — leer, wenn nichts. Gebraucht
|
||
wird das, weil die Einstellungen aus der Web-App kommen: wer dort eine
|
||
Schwelle verstellt, soll nicht den Detector neu starten muessen.
|
||
|
||
Das Modell bleibt geladen. Die Schwelle des Finders wird mitgezogen,
|
||
sonst sucht er weiter nach der alten.
|
||
"""
|
||
geaendert: list[str] = []
|
||
|
||
if detect_threshold is not None and detect_threshold != self.detect_threshold:
|
||
geaendert.append(
|
||
f"Vogelschwelle {self.detect_threshold:.2f} -> {detect_threshold:.2f}"
|
||
)
|
||
self.detect_threshold = detect_threshold
|
||
# Der Finder sucht ab der niedrigeren der beiden Schwellen; das
|
||
# Protokoll soll die Beinahe-Treffer weiter sehen.
|
||
self.report_threshold = min(self._report_wunsch, detect_threshold)
|
||
self.finder.threshold = self.report_threshold
|
||
|
||
if min_bird_score is not None and min_bird_score != self.min_bird_score:
|
||
geaendert.append(
|
||
f"Artenschwelle {self.min_bird_score:.2f} -> {min_bird_score:.2f}"
|
||
)
|
||
self.min_bird_score = min_bird_score
|
||
|
||
if arten is not None:
|
||
alt = {a.scientific for a in self.classifier.native_idx.values()}
|
||
neu = {a.scientific for a in arten}
|
||
if alt != neu:
|
||
self.classifier.setze_arten(arten)
|
||
erkannt = len(self.classifier.native_idx)
|
||
geaendert.append(
|
||
f"Katalog {len(alt)} -> {len(neu)} Arten "
|
||
f"({erkannt} davon kennt das Modell)"
|
||
)
|
||
|
||
return geaendert
|
||
|
||
def run(self, img: Image.Image) -> tuple[Box | None, Result | None]:
|
||
"""Beste Box im Bild, ggf. mit Artbestimmung.
|
||
|
||
Die Box kann unter ``detect_threshold`` liegen — dann ist sie fuers
|
||
Protokoll gedacht und nicht fuer eine Sichtung; ``result`` bleibt in
|
||
dem Fall ``None``, weil sich die Klassifikation fuer einen
|
||
Beinahe-Treffer nicht lohnt. ``zaehlt_als_vogel`` entscheidet.
|
||
"""
|
||
boxes = self.finder.find(img)
|
||
if not boxes:
|
||
return None, None
|
||
box = boxes[0]
|
||
if not self.zaehlt_als_vogel(box):
|
||
return box, None
|
||
result = self.classifier.classify(box.crop(img))
|
||
if result.bird_score < self.min_bird_score:
|
||
# Vogel ja, heimische Art nein — lieber gar keine Bestimmung als
|
||
# eine falsche im Vogelalbum.
|
||
return box, None
|
||
return box, result
|