Vogelalbum 1.0.0
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.
This commit is contained in:
christian committed
commit
dbec78aa7a
186 files changed
+29117
No files matched your search
@@ -0,0 +1,323 @@
|
||||
"""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))
|
||||
|
||||
@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
|
||||
Reference in new issue
Block a user