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.
124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
"""Dauerprotokoll der Erkennung.
|
|
|
|
Sechs Minuten zuschauen beantwortet die Frage „übersehen wir Vögel?" nur für
|
|
diese sechs Minuten. Über Tage beantwortet sie ein Protokoll.
|
|
|
|
Geschrieben wird als CSV, damit man es in eine Tabelle ziehen kann:
|
|
|
|
* **jede Erkennung ab einer niedrigen Grundschwelle** — auch die, die für eine
|
|
Sichtung nicht gereicht hat. Genau daran erkennt man, ob die Schwelle zu
|
|
hoch steht oder die Vögel zu klein im Bild sind.
|
|
* **eine Zeile je Viertelstunde**, immer, auch wenn nichts passiert ist. Ohne
|
|
diese Zeilen weiß man später nicht, ob eine Lücke bedeutet „nichts gesehen"
|
|
oder „lief gar nicht".
|
|
|
|
Die zweite Sorte ist die wichtigere. Ein Protokoll, das nur bei Erfolg
|
|
schreibt, sieht bei einem Ausfall genauso aus wie bei einem ruhigen Tag.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
KOPF = [
|
|
"zeit",
|
|
"art", # "puls" = Lebenszeichen, "sicht" = Erkennung
|
|
"score", # Sicherheit des Vogeldetektors
|
|
"flaeche_pct", # wie viel des Bildes der Vogel einnimmt
|
|
"helligkeit",
|
|
"top_art", # beste Artbestimmung, falls vorhanden
|
|
"confidence",
|
|
"bird_score", # Anteil heimischer Arten
|
|
"bilder", # verarbeitete Bilder seit dem letzten Puls
|
|
"dunkel", # davon im Dunkeln uebersprungen
|
|
]
|
|
|
|
|
|
class Protokoll:
|
|
"""Schreibt Erkennungen und regelmäßige Lebenszeichen in eine CSV."""
|
|
|
|
def __init__(
|
|
self,
|
|
pfad: str | Path,
|
|
grundschwelle: float = 0.08,
|
|
puls_sekunden: float = 900.0,
|
|
):
|
|
self.pfad = Path(pfad)
|
|
self.grundschwelle = grundschwelle
|
|
self.puls_sekunden = puls_sekunden
|
|
|
|
self.pfad.parent.mkdir(parents=True, exist_ok=True)
|
|
neu = not self.pfad.exists() or self.pfad.stat().st_size == 0
|
|
# Anhängen, nicht überschreiben: ein Neustart des Dienstes darf die
|
|
# bisherigen Tage nicht wegwerfen.
|
|
self._datei = self.pfad.open("a", newline="", encoding="utf-8")
|
|
self._csv = csv.writer(self._datei)
|
|
if neu:
|
|
self._csv.writerow(KOPF)
|
|
self._datei.flush()
|
|
|
|
self._letzter_puls: float | None = None
|
|
self._bilder = 0
|
|
self._dunkel = 0
|
|
|
|
# -- Zählen ------------------------------------------------------------
|
|
|
|
def bild(self, dunkel: bool = False) -> None:
|
|
self._bilder += 1
|
|
if dunkel:
|
|
self._dunkel += 1
|
|
|
|
# -- Schreiben ---------------------------------------------------------
|
|
|
|
def erkennung(self, box, result, helligkeit: float) -> None:
|
|
"""Eine Detektion festhalten — auch eine, die für nichts gereicht hat."""
|
|
if box is None or box.score < self.grundschwelle:
|
|
return
|
|
|
|
top, conf, bird = "", "", ""
|
|
if result is not None and result.guesses:
|
|
top = result.guesses[0].species.slug
|
|
conf = f"{result.guesses[0].confidence:.3f}"
|
|
bird = f"{result.bird_score:.3f}"
|
|
|
|
self._schreibe(
|
|
"sicht",
|
|
f"{box.score:.3f}",
|
|
f"{box.area * 100:.2f}",
|
|
f"{helligkeit:.0f}",
|
|
top,
|
|
conf,
|
|
bird,
|
|
)
|
|
|
|
def puls(self, jetzt: float, helligkeit: float, erzwingen: bool = False) -> None:
|
|
"""Lebenszeichen. Ohne diese Zeilen ist eine Lücke im Protokoll
|
|
nicht von einem Ausfall zu unterscheiden."""
|
|
if (
|
|
not erzwingen
|
|
and self._letzter_puls is not None
|
|
and jetzt - self._letzter_puls < self.puls_sekunden
|
|
):
|
|
return
|
|
self._letzter_puls = jetzt
|
|
self._schreibe("puls", "", "", f"{helligkeit:.0f}", "", "", "")
|
|
self._bilder = 0
|
|
self._dunkel = 0
|
|
|
|
def _schreibe(self, art: str, *felder: str) -> None:
|
|
self._csv.writerow(
|
|
[datetime.now(timezone.utc).isoformat(timespec="seconds"), art, *felder,
|
|
self._bilder, self._dunkel]
|
|
)
|
|
# Sofort auf Platte: Wer nach einem Absturz ins Protokoll schaut, will
|
|
# die letzten Minuten sehen und nicht einen halbleeren Puffer.
|
|
self._datei.flush()
|
|
|
|
def close(self) -> None:
|
|
try:
|
|
self._datei.close()
|
|
except OSError:
|
|
pass
|