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.
959 lines
35 KiB
Python
959 lines
35 KiB
Python
"""Tests für Bewegungserkennung und Besuchslogik.
|
|
|
|
./venv/bin/python -m unittest discover tests
|
|
|
|
Läuft ohne die Modelle: der Klassifikator wird durch eine Attrappe ersetzt,
|
|
damit die Zustandsmaschine für sich prüfbar ist.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageChops, ImageDraw
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from detector import karte # noqa: E402
|
|
from detector.classify import Box # noqa: E402
|
|
from detector.motion import MotionDetector, sharpness # noqa: E402
|
|
from detector.species_de import ( # noqa: E402
|
|
BY_SCIENTIFIC, SPECIES, catalog, label_to_scientific,
|
|
)
|
|
from detector.species_extra import EXTRA # noqa: E402
|
|
from detector.watcher import Watcher, WatcherConfig # noqa: E402
|
|
|
|
|
|
def blank(color=(40, 60, 40), size=(320, 240)) -> Image.Image:
|
|
return Image.new("RGB", size, color)
|
|
|
|
|
|
def with_blob(x: int, y: int, r: int = 40, size=(320, 240)) -> Image.Image:
|
|
img = blank(size=size)
|
|
ImageDraw.Draw(img).ellipse([x - r, y - r, x + r, y + r], fill=(220, 200, 60))
|
|
return img
|
|
|
|
|
|
class MotionTests(unittest.TestCase):
|
|
def test_first_frame_reports_no_motion(self):
|
|
self.assertEqual(MotionDetector().update(blank()), 0.0)
|
|
|
|
def test_identical_frames_report_no_motion(self):
|
|
m = MotionDetector()
|
|
m.update(blank())
|
|
self.assertEqual(m.update(blank()), 0.0)
|
|
|
|
def test_blob_appearing_triggers_motion(self):
|
|
m = MotionDetector()
|
|
m.update(blank())
|
|
ratio = m.update(with_blob(160, 120))
|
|
self.assertGreater(ratio, m.min_ratio)
|
|
self.assertTrue(m.is_motion(ratio))
|
|
|
|
def test_background_adapts_to_a_permanent_change(self):
|
|
"""Ein umgestellter Ast darf nicht dauerhaft als Bewegung gelten."""
|
|
m = MotionDetector()
|
|
m.update(blank())
|
|
changed = blank(color=(90, 110, 90))
|
|
ratios = [m.update(changed) for _ in range(200)]
|
|
self.assertGreater(ratios[0], m.min_ratio)
|
|
self.assertLess(ratios[-1], m.min_ratio)
|
|
|
|
def test_reset_clears_background(self):
|
|
m = MotionDetector()
|
|
m.update(blank())
|
|
m.reset()
|
|
self.assertEqual(m.update(with_blob(160, 120)), 0.0)
|
|
|
|
|
|
class SharpnessTests(unittest.TestCase):
|
|
def test_flat_image_has_almost_no_sharpness(self):
|
|
self.assertLess(sharpness(blank()), 1.0)
|
|
|
|
def test_edges_score_higher_than_flat(self):
|
|
self.assertGreater(sharpness(with_blob(160, 120)), sharpness(blank()))
|
|
|
|
def test_blurred_scores_lower_than_crisp(self):
|
|
from PIL import ImageFilter
|
|
|
|
crisp = with_blob(160, 120)
|
|
blurred = crisp.filter(ImageFilter.GaussianBlur(6))
|
|
self.assertGreater(sharpness(crisp), sharpness(blurred))
|
|
|
|
def test_tiny_image_does_not_crash(self):
|
|
self.assertEqual(sharpness(Image.new("RGB", (2, 2)), width=2), 0.0)
|
|
|
|
|
|
class DaylightTests(unittest.TestCase):
|
|
"""Abschaltung bei Dunkelheit. Die Hysterese ist der interessante Teil:
|
|
ohne sie flattert die Entscheidung in der Dämmerung im Sekundentakt."""
|
|
|
|
def gate(self, **kw):
|
|
from detector.daylight import DaylightGate
|
|
|
|
return DaylightGate(**({"patience": 3} | kw))
|
|
|
|
def test_brightness_matches_the_image(self):
|
|
from detector.daylight import brightness
|
|
|
|
self.assertLess(brightness(Image.new("RGB", (64, 48), (0, 0, 0))), 1)
|
|
self.assertGreater(brightness(Image.new("RGB", (64, 48), (255, 255, 255))), 254)
|
|
|
|
def test_bright_scene_keeps_running(self):
|
|
g = self.gate()
|
|
hell = Image.new("RGB", (64, 48), (130, 130, 130))
|
|
for _ in range(20):
|
|
self.assertTrue(g.update(hell))
|
|
self.assertFalse(g.is_dark)
|
|
|
|
def test_darkness_pauses_after_patience(self):
|
|
g = self.gate()
|
|
dunkel = Image.new("RGB", (64, 48), (2, 2, 2))
|
|
# Erst nach 'patience' Bildern am Stück, nicht schon beim ersten.
|
|
self.assertTrue(g.update(dunkel))
|
|
self.assertTrue(g.update(dunkel))
|
|
self.assertFalse(g.update(dunkel))
|
|
self.assertTrue(g.is_dark)
|
|
|
|
def test_a_single_flash_does_not_wake_it(self):
|
|
"""Ein Autoscheinwerfer soll die Analyse nicht wecken."""
|
|
g = self.gate()
|
|
dunkel = Image.new("RGB", (64, 48), (2, 2, 2))
|
|
for _ in range(5):
|
|
g.update(dunkel)
|
|
self.assertTrue(g.is_dark)
|
|
|
|
g.update(Image.new("RGB", (64, 48), (200, 200, 200)))
|
|
self.assertTrue(g.is_dark)
|
|
|
|
def test_hysteresis_prevents_flapping_at_dusk(self):
|
|
"""Ein Wert zwischen den Schwellen ändert nichts — in beide Richtungen."""
|
|
g = self.gate(dark_below=15, light_above=30)
|
|
daemmerung = Image.new("RGB", (64, 48), (22, 22, 22))
|
|
for _ in range(50):
|
|
g.update(daemmerung)
|
|
self.assertFalse(g.is_dark)
|
|
|
|
for _ in range(10):
|
|
g.update(Image.new("RGB", (64, 48), (2, 2, 2)))
|
|
self.assertTrue(g.is_dark)
|
|
|
|
for _ in range(50):
|
|
g.update(daemmerung)
|
|
self.assertTrue(g.is_dark)
|
|
|
|
def test_morning_resumes(self):
|
|
g = self.gate()
|
|
for _ in range(10):
|
|
g.update(Image.new("RGB", (64, 48), (2, 2, 2)))
|
|
self.assertTrue(g.is_dark)
|
|
|
|
hell = Image.new("RGB", (64, 48), (130, 130, 130))
|
|
for _ in range(10):
|
|
g.update(hell)
|
|
self.assertFalse(g.is_dark)
|
|
|
|
def test_night_vision_never_pauses(self):
|
|
g = self.gate(enabled=False)
|
|
dunkel = Image.new("RGB", (64, 48), (0, 0, 0))
|
|
for _ in range(50):
|
|
self.assertTrue(g.update(dunkel))
|
|
self.assertFalse(g.is_dark)
|
|
|
|
def test_thresholds_must_leave_room_for_hysteresis(self):
|
|
with self.assertRaises(ValueError):
|
|
self.gate(dark_below=30, light_above=30)
|
|
|
|
|
|
class BoxTests(unittest.TestCase):
|
|
def test_crop_returns_the_marked_region(self):
|
|
img = blank(size=(400, 200))
|
|
box = Box(0.25, 0.0, 0.75, 1.0, 0.9)
|
|
crop = box.crop(img, pad=0.0)
|
|
self.assertEqual(crop.size, (200, 200))
|
|
|
|
def test_padding_widens_the_crop(self):
|
|
img = blank(size=(400, 400))
|
|
box = Box(0.4, 0.4, 0.6, 0.6, 0.9)
|
|
self.assertGreater(box.crop(img, pad=0.5).size[0], box.crop(img, pad=0.0).size[0])
|
|
|
|
def test_degenerate_box_falls_back_to_full_image(self):
|
|
img = blank(size=(400, 400))
|
|
self.assertEqual(Box(0.5, 0.5, 0.5, 0.5, 0.9).crop(img).size, img.size)
|
|
|
|
def test_area(self):
|
|
self.assertAlmostEqual(Box(0.0, 0.0, 0.5, 0.5, 1.0).area, 0.25)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Attrappe statt Modell: liefert eine feste Bestimmung, damit sich die
|
|
# Abstimmungs- und Besuchslogik isoliert prüfen lässt.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FakeGuess:
|
|
species: object
|
|
confidence: float
|
|
|
|
|
|
@dataclass
|
|
class FakeResult:
|
|
guesses: list
|
|
bird_score: float = 0.9
|
|
raw_top: list = None
|
|
ms: float = 1.0
|
|
|
|
|
|
class FakePipeline:
|
|
def __init__(self, plan, detect_threshold=0.4):
|
|
"""``plan`` ist eine Liste von (Art|None, Confidence) je Aufruf.
|
|
|
|
Ein Eintrag darf zusaetzlich einen Box-Score tragen —
|
|
``(Art, Confidence, 0.2)`` ist ein Beinahe-Treffer unter der
|
|
Sichtungsschwelle, wie ihn das Protokoll sehen soll.
|
|
"""
|
|
self.plan = list(plan)
|
|
self.calls = 0
|
|
self.detect_threshold = detect_threshold
|
|
|
|
def zaehlt_als_vogel(self, box):
|
|
return box is not None and box.score >= self.detect_threshold
|
|
|
|
def run(self, img):
|
|
entry = self.plan[min(self.calls, len(self.plan) - 1)]
|
|
self.calls += 1
|
|
if entry is None:
|
|
return None, None
|
|
species, confidence, *rest = entry
|
|
box = Box(0.3, 0.3, 0.7, 0.7, rest[0] if rest else 0.95)
|
|
if species is None or not self.zaehlt_als_vogel(box):
|
|
return box, None
|
|
return box, FakeResult([FakeGuess(species, confidence)])
|
|
|
|
|
|
class WatcherTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.kohlmeise = BY_SCIENTIFIC["Parus major"]
|
|
self.blaumeise = BY_SCIENTIFIC["Cyanistes caeruleus"]
|
|
|
|
def make(self, plan, **cfg) -> Watcher:
|
|
defaults = {"classify_every": 1, "cooldown_s": 0.0, "min_bird_frames": 2}
|
|
return Watcher(
|
|
pipeline=FakePipeline(plan),
|
|
config=WatcherConfig(**(defaults | cfg)),
|
|
)
|
|
|
|
def test_no_motion_produces_no_sighting(self):
|
|
w = self.make([(self.kohlmeise, 0.9)])
|
|
for _ in range(5):
|
|
self.assertIsNone(w.process(blank()))
|
|
self.assertIsNone(w.flush())
|
|
|
|
def test_a_visit_yields_one_sighting(self):
|
|
w = self.make([(self.kohlmeise, 0.9)])
|
|
w.process(blank())
|
|
for x in range(100, 200, 25):
|
|
w.process(with_blob(x, 120))
|
|
sighting = w.flush()
|
|
self.assertIsNotNone(sighting)
|
|
self.assertEqual(sighting.species, self.kohlmeise)
|
|
self.assertGreater(sighting.frames_with_bird, 0)
|
|
|
|
def test_majority_wins_the_vote(self):
|
|
plan = [(self.kohlmeise, 0.9)] * 3 + [(self.blaumeise, 0.9)]
|
|
w = self.make(plan)
|
|
w.process(blank())
|
|
for x in range(100, 200, 25):
|
|
w.process(with_blob(x, 120))
|
|
sighting = w.flush()
|
|
self.assertEqual(sighting.species, self.kohlmeise)
|
|
self.assertGreater(sighting.confidence, 0.5)
|
|
|
|
def test_split_vote_stays_undetermined(self):
|
|
"""Kippt die Abstimmung, wird nicht geraten."""
|
|
plan = [(self.kohlmeise, 0.9), (self.blaumeise, 0.9)] * 3
|
|
w = self.make(plan, min_confidence=0.8)
|
|
w.process(blank())
|
|
for x in range(100, 200, 16):
|
|
w.process(with_blob(x, 120))
|
|
sighting = w.flush()
|
|
self.assertIsNone(sighting.species)
|
|
|
|
def test_too_few_bird_frames_is_discarded(self):
|
|
w = self.make([(self.kohlmeise, 0.9), None, None], min_bird_frames=3)
|
|
w.process(blank())
|
|
w.process(with_blob(160, 120))
|
|
w.process(with_blob(180, 120))
|
|
self.assertIsNone(w.flush())
|
|
|
|
def beobachtet(self, plan, **cfg):
|
|
"""Watcher mit Mitschrift — gibt (Watcher, Liste der Meldungen)."""
|
|
gesehen: list = []
|
|
defaults = {
|
|
"classify_every": 1,
|
|
"cooldown_s": 0.0,
|
|
"min_bird_frames": 1,
|
|
"gate": "detector",
|
|
}
|
|
w = Watcher(
|
|
pipeline=FakePipeline(plan),
|
|
config=WatcherConfig(**(defaults | cfg)),
|
|
beobachter=lambda box, result: gesehen.append((box, result)),
|
|
)
|
|
return w, gesehen
|
|
|
|
def test_near_miss_is_logged_but_is_no_sighting(self):
|
|
"""Der entscheidende Fall fuers Protokoll: Vogel knapp verpasst.
|
|
|
|
Ohne diese Meldung sieht ein Tag, an dem die Voegel nur zu klein im
|
|
Bild waren, genauso aus wie ein Tag ohne einen einzigen Vogel.
|
|
"""
|
|
w, gesehen = self.beobachtet([(self.kohlmeise, 0.9, 0.2)])
|
|
for x in range(100, 200, 25):
|
|
self.assertIsNone(w.process(with_blob(x, 120)))
|
|
self.assertIsNone(w.flush())
|
|
|
|
self.assertTrue(gesehen, "Beobachter wurde gar nicht gerufen")
|
|
self.assertTrue(all(box is not None for box, _ in gesehen))
|
|
self.assertTrue(all(box.score == 0.2 for box, _ in gesehen))
|
|
|
|
def test_box_above_threshold_still_becomes_a_sighting(self):
|
|
"""Gegenprobe: an der normalen Schwelle aendert sich nichts."""
|
|
w, gesehen = self.beobachtet([(self.kohlmeise, 0.9, 0.95)])
|
|
for x in range(100, 200, 25):
|
|
w.process(with_blob(x, 120))
|
|
sighting = w.flush()
|
|
self.assertIsNotNone(sighting)
|
|
self.assertEqual(sighting.species, self.kohlmeise)
|
|
self.assertTrue(gesehen)
|
|
|
|
def test_best_shot_prefers_the_sharper_frame(self):
|
|
w = self.make([(self.kohlmeise, 0.9)])
|
|
w.process(blank())
|
|
w.process(with_blob(140, 120))
|
|
w.process(with_blob(170, 120))
|
|
sighting = w.flush()
|
|
self.assertIsNotNone(sighting.best_image)
|
|
self.assertGreater(sighting.best_sharpness, 0)
|
|
|
|
def test_watcher_resets_between_visits(self):
|
|
w = self.make([(self.kohlmeise, 0.9)])
|
|
w.process(blank())
|
|
w.process(with_blob(160, 120))
|
|
w.process(with_blob(180, 120))
|
|
first = w.flush()
|
|
|
|
w.process(blank())
|
|
w.process(with_blob(160, 120))
|
|
w.process(with_blob(180, 120))
|
|
second = w.flush()
|
|
|
|
self.assertIsNotNone(first)
|
|
self.assertIsNotNone(second)
|
|
self.assertGreater(second.started_at, first.started_at - _one_second())
|
|
|
|
|
|
def _one_second():
|
|
from datetime import timedelta
|
|
|
|
return timedelta(seconds=1)
|
|
|
|
|
|
class ReportTests(unittest.TestCase):
|
|
"""Die Auswertung der Rohdaten. Sie beantwortet die Frage nach den ersten
|
|
Tagen — sehen wir Vögel, und taugen die Bilder etwas."""
|
|
|
|
def setUp(self):
|
|
import json
|
|
import tempfile
|
|
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.ordner = Path(self.tmp.name)
|
|
|
|
voll = with_blob(160, 120, r=50, size=(400, 300))
|
|
crop = voll.crop((110, 70, 210, 170))
|
|
voll.save(self.ordner / "20260828-112200_amsel.jpg")
|
|
crop.save(self.ordner / "20260828-112200_amsel_crop.jpg")
|
|
(self.ordner / "20260828-112200_amsel.json").write_text(
|
|
json.dumps({
|
|
"species": "amsel", "confidence": 0.97,
|
|
"frames_with_bird": 25, "frames_total": 67,
|
|
}), encoding="utf-8")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_reads_a_saved_sighting(self):
|
|
from detector.report import lade
|
|
|
|
eintraege = lade(self.ordner)
|
|
self.assertEqual(len(eintraege), 1)
|
|
e = eintraege[0]
|
|
self.assertEqual(e.art, "amsel")
|
|
self.assertAlmostEqual(e.confidence, 0.97)
|
|
self.assertAlmostEqual(e.trefferquote, 25 / 67)
|
|
|
|
def test_measures_area_from_the_image_sizes(self):
|
|
from detector.report import lade
|
|
|
|
e = lade(self.ordner)[0]
|
|
# 100x100 Ausschnitt aus 400x300 sind 8,3 %.
|
|
self.assertAlmostEqual(e.flaeche, 100 * 100 / (400 * 300) * 100, places=2)
|
|
self.assertGreater(e.schaerfe, 0)
|
|
self.assertGreater(e.helligkeit, 0)
|
|
|
|
def test_empty_folder_is_not_an_error(self):
|
|
import tempfile
|
|
|
|
from detector.report import lade
|
|
|
|
with tempfile.TemporaryDirectory() as leer:
|
|
self.assertEqual(lade(Path(leer)), [])
|
|
|
|
def test_broken_json_is_skipped(self):
|
|
from detector.report import lade
|
|
|
|
(self.ordner / "kaputt.json").write_text("{nicht valide", encoding="utf-8")
|
|
self.assertEqual(len(lade(self.ordner)), 1)
|
|
|
|
def test_contact_sheet_is_written(self):
|
|
from detector.report import kontaktbogen, lade
|
|
|
|
ziel = self.ordner / "bogen.jpg"
|
|
n = kontaktbogen(lade(self.ordner), ziel)
|
|
self.assertEqual(n, 1)
|
|
self.assertTrue(ziel.exists())
|
|
self.assertGreater(Image.open(ziel).width, 0)
|
|
|
|
def test_verdict_flags_a_bird_that_is_too_small(self):
|
|
from detector.report import Eintrag, _urteil
|
|
|
|
winzig = Eintrag("x", "amsel", 0.9, 5, 20, 400, 1.2, 120, None, None)
|
|
self.assertEqual(_urteil(winzig), "Vogel zu klein")
|
|
|
|
verwackelt = Eintrag("x", "amsel", 0.9, 5, 20, 12, 20, 120, None, None)
|
|
self.assertEqual(_urteil(verwackelt), "verwackelt")
|
|
|
|
gut = Eintrag("x", "amsel", 0.9, 5, 20, 300, 15, 120, None, None)
|
|
self.assertEqual(_urteil(gut), "sehr gut")
|
|
|
|
|
|
class ProtokollTests(unittest.TestCase):
|
|
"""Das Dauerprotokoll. Der wichtigste Teil sind die Lebenszeichen: ohne
|
|
sie ist eine Lücke nicht von einem Ausfall zu unterscheiden."""
|
|
|
|
def setUp(self):
|
|
import tempfile
|
|
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.pfad = Path(self.tmp.name) / "unter" / "protokoll.csv"
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def lade(self):
|
|
import csv
|
|
|
|
with self.pfad.open(encoding="utf-8") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
def test_creates_file_and_header(self):
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad)
|
|
p.close()
|
|
self.assertTrue(self.pfad.exists())
|
|
self.assertIn("zeit", self.pfad.read_text(encoding="utf-8"))
|
|
|
|
def test_weak_detection_is_recorded(self):
|
|
"""Genau dafür ist das Protokoll da: knapp verpasste Vögel sichtbar
|
|
machen, die es nie in eine Sichtung schaffen."""
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, grundschwelle=0.08)
|
|
p.erkennung(Box(0.4, 0.4, 0.5, 0.5, 0.22), None, 120)
|
|
p.close()
|
|
|
|
zeilen = self.lade()
|
|
self.assertEqual(len(zeilen), 1)
|
|
self.assertEqual(zeilen[0]["art"], "sicht")
|
|
self.assertAlmostEqual(float(zeilen[0]["score"]), 0.22)
|
|
self.assertAlmostEqual(float(zeilen[0]["flaeche_pct"]), 1.0, places=1)
|
|
|
|
def test_below_the_floor_is_ignored(self):
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, grundschwelle=0.08)
|
|
p.erkennung(Box(0.4, 0.4, 0.5, 0.5, 0.03), None, 120)
|
|
p.erkennung(None, None, 120)
|
|
p.close()
|
|
self.assertEqual(self.lade(), [])
|
|
|
|
def test_pulse_respects_its_interval(self):
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, puls_sekunden=100)
|
|
p.puls(0.0, 120) # erster zaehlt
|
|
p.puls(50.0, 120) # zu frueh
|
|
p.puls(150.0, 120) # wieder faellig
|
|
p.close()
|
|
self.assertEqual(len([z for z in self.lade() if z["art"] == "puls"]), 2)
|
|
|
|
def test_pulse_can_be_forced(self):
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, puls_sekunden=9999)
|
|
p.puls(0.0, 120)
|
|
p.puls(1.0, 120, erzwingen=True)
|
|
p.close()
|
|
self.assertEqual(len(self.lade()), 2)
|
|
|
|
def test_frame_counts_reset_after_a_pulse(self):
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, puls_sekunden=10)
|
|
for _ in range(7):
|
|
p.bild()
|
|
for _ in range(3):
|
|
p.bild(dunkel=True)
|
|
p.puls(20.0, 120)
|
|
p.bild()
|
|
p.puls(40.0, 120)
|
|
p.close()
|
|
|
|
pulse = [z for z in self.lade() if z["art"] == "puls"]
|
|
self.assertEqual(pulse[0]["bilder"], "10")
|
|
self.assertEqual(pulse[0]["dunkel"], "3")
|
|
self.assertEqual(pulse[1]["bilder"], "1")
|
|
|
|
def test_reopening_appends_instead_of_truncating(self):
|
|
"""Ein Neustart des Dienstes darf die bisherigen Tage nicht wegwerfen."""
|
|
from detector.protokoll import Protokoll
|
|
|
|
p = Protokoll(self.pfad, puls_sekunden=1)
|
|
p.puls(0.0, 120)
|
|
p.close()
|
|
|
|
p = Protokoll(self.pfad, puls_sekunden=1)
|
|
p.puls(0.0, 120)
|
|
p.close()
|
|
|
|
self.assertEqual(len(self.lade()), 2)
|
|
self.assertEqual(self.pfad.read_text(encoding="utf-8").count("zeit,art"), 1)
|
|
|
|
|
|
class WatcherObserverTests(unittest.TestCase):
|
|
"""Der Beobachter meldet jede Analyse — auch die ohne Fund."""
|
|
|
|
def test_observer_sees_hits_and_misses(self):
|
|
gesehen = []
|
|
w = Watcher(
|
|
pipeline=FakePipeline([(BY_SCIENTIFIC["Parus major"], 0.9), None]),
|
|
config=WatcherConfig(classify_every=1, cooldown_s=0.0),
|
|
beobachter=lambda box, res: gesehen.append(box),
|
|
)
|
|
w.process(blank())
|
|
w.process(with_blob(160, 120))
|
|
w.process(with_blob(180, 120))
|
|
w.flush()
|
|
|
|
self.assertGreaterEqual(len(gesehen), 2)
|
|
self.assertIsNotNone(gesehen[0])
|
|
self.assertIsNone(gesehen[1])
|
|
|
|
|
|
class CatalogTests(unittest.TestCase):
|
|
def test_slugs_are_unique(self):
|
|
slugs = [s.slug for s in SPECIES]
|
|
self.assertEqual(len(slugs), len(set(slugs)))
|
|
|
|
def test_scientific_names_are_unique(self):
|
|
names = [s.scientific for s in SPECIES]
|
|
self.assertEqual(len(names), len(set(names)))
|
|
|
|
def test_slugs_have_no_umlauts(self):
|
|
for s in SPECIES:
|
|
with self.subTest(species=s.german):
|
|
self.assertTrue(s.slug.isascii(), f"{s.german} -> {s.slug}")
|
|
|
|
def test_every_species_has_a_fact(self):
|
|
for s in SPECIES:
|
|
with self.subTest(species=s.german):
|
|
self.assertTrue(s.fact.strip())
|
|
|
|
def test_rarer_species_are_worth_more(self):
|
|
order = ["haeufig", "gelegentlich", "selten", "episch", "legendaer"]
|
|
xp = [
|
|
next(s.xp for s in SPECIES if s.rarity == r)
|
|
for r in order
|
|
if any(s.rarity == r for s in SPECIES)
|
|
]
|
|
self.assertEqual(xp, sorted(xp))
|
|
|
|
def test_label_parsing(self):
|
|
self.assertEqual(label_to_scientific("Parus major (Great Tit)"), "Parus major")
|
|
self.assertEqual(label_to_scientific("Parus major"), "Parus major")
|
|
|
|
def test_catalog_matches_the_model_labels(self):
|
|
"""Jede Art im Katalog muss im Label-File des Modells vorkommen —
|
|
sonst wird sie nie erkannt und die Karte bleibt für immer leer.
|
|
|
|
Gilt für den ganzen Katalog, nicht nur die Kernliste: eine Art ohne
|
|
Label wäre ein Eintrag, den niemand je freischalten kann.
|
|
"""
|
|
labels = Path(__file__).resolve().parents[1] / "models" / "inat_bird_labels.txt"
|
|
if not labels.exists():
|
|
self.skipTest("Modelle nicht geladen (scripts/fetch_models.sh)")
|
|
|
|
known = {
|
|
label_to_scientific(line)
|
|
for line in labels.read_text(encoding="utf-8").splitlines()
|
|
}
|
|
missing = [s.german for s in catalog() if s.scientific not in known]
|
|
self.assertEqual(missing, [], f"Nicht im Modell: {missing}")
|
|
|
|
|
|
class ExtendedCatalogTests(unittest.TestCase):
|
|
"""Der erweiterte Katalog — 104 Arten ohne Erzähltext, nur Strukturdaten."""
|
|
|
|
def test_no_slug_collisions_with_the_core(self):
|
|
kern = {s.slug for s in SPECIES}
|
|
doppelt = [e.slug for e in EXTRA if e.slug in kern]
|
|
self.assertEqual(doppelt, [])
|
|
|
|
def test_slugs_are_unique_and_ascii(self):
|
|
slugs = [e.slug for e in EXTRA]
|
|
self.assertEqual(len(slugs), len(set(slugs)))
|
|
for slug in slugs:
|
|
with self.subTest(slug=slug):
|
|
self.assertTrue(slug.isascii(), slug)
|
|
|
|
def test_scientific_names_do_not_repeat(self):
|
|
namen = [s.scientific for s in catalog()]
|
|
self.assertEqual(len(namen), len(set(namen)))
|
|
|
|
def test_habitat_filter_narrows_the_catalog(self):
|
|
alle = catalog()
|
|
garten = catalog({"garten"})
|
|
self.assertLess(len(garten), len(alle))
|
|
self.assertTrue(all(s.lebensraum == "garten" for s in garten))
|
|
# Die Kernarten müssen im Gartenfilter vollständig enthalten sein —
|
|
# sonst verschwände ausgerechnet das, was die Kamera wirklich sieht.
|
|
self.assertTrue({s.slug for s in SPECIES} <= {s.slug for s in garten})
|
|
|
|
def test_every_extra_has_plausible_measurements(self):
|
|
for e in EXTRA:
|
|
with self.subTest(art=e.german):
|
|
self.assertGreater(e.gewicht_g, 0)
|
|
self.assertGreater(e.spannweite_cm, 0)
|
|
self.assertIn(e.zug, {"stand", "teil", "zug"})
|
|
self.assertIn(
|
|
e.lebensraum, {"garten", "wald", "offenland", "wasser", "gebirge"}
|
|
)
|
|
|
|
def test_wingspan_exceeds_body_length(self):
|
|
"""Ein Vogel mit kürzerer Spannweite als Körperlänge wäre ein Tippfehler."""
|
|
for e in EXTRA:
|
|
laenge = float(e.size_cm.split()[0])
|
|
with self.subTest(art=e.german):
|
|
self.assertGreater(e.spannweite_cm, laenge * 0.7)
|
|
|
|
|
|
class KarteTests(unittest.TestCase):
|
|
"""Die Erkennungskarte laeuft ohne Modelle: die Pipeline ist eine Attrappe,
|
|
die nach Position entscheidet. Geprueft wird die Karte, nicht der Detektor."""
|
|
|
|
def setUp(self):
|
|
self.vogel = with_blob(60, 60, r=50, size=(120, 120))
|
|
self.hg = [blank(size=(640, 480)) for _ in range(2)]
|
|
|
|
def test_montage_hits_the_wanted_area(self):
|
|
gross = karte.montiere(self.hg[0], self.vogel, 0.25)
|
|
klein = karte.montiere(self.hg[0], self.vogel, 0.04)
|
|
# Der Vogel ist gelb, der Hintergrund gruen — gezaehlt wird, wie viele
|
|
# Pixel sich geaendert haben.
|
|
def anteil(bild):
|
|
diff = ImageChops.difference(bild, self.hg[0]).convert("L")
|
|
geaendert = sum(n for wert, n in enumerate(diff.histogram()) if wert > 0)
|
|
return geaendert / (bild.width * bild.height)
|
|
|
|
# Die weiche Ellipse frisst Flaeche gegenueber dem Rechteck, deshalb
|
|
# grosszuegige Grenzen — es geht um die Groessenordnung.
|
|
self.assertGreater(anteil(gross), anteil(klein) * 3)
|
|
self.assertLess(anteil(gross), 0.30)
|
|
|
|
def test_bird_stands_on_the_given_line(self):
|
|
"""Der Fusspunkt ist der Sinn der Uebung: ein Vogel steht auf dem
|
|
Brett, er schwebt nicht um seinen Mittelpunkt."""
|
|
bild = karte.montiere(self.hg[0], self.vogel, 0.09, mitte_x=0.5, fuss_y=0.5)
|
|
H = bild.height
|
|
veraendert = [
|
|
y for y in range(H)
|
|
if any(bild.getpixel((x, y)) != (40, 60, 40) for x in range(0, bild.width, 4))
|
|
]
|
|
self.assertTrue(veraendert)
|
|
self.assertLessEqual(max(veraendert), H * 0.5 + 2)
|
|
self.assertGreater(max(veraendert), H * 0.5 - 20)
|
|
|
|
def test_map_marks_the_dead_zone(self):
|
|
"""Der Fall, fuer den es die Karte gibt: unten faellt alles durch."""
|
|
|
|
class NurObenPipeline:
|
|
detect_threshold = 0.4
|
|
|
|
def zaehlt_als_vogel(self, box):
|
|
return box is not None and box.score >= self.detect_threshold
|
|
|
|
def run(self, img):
|
|
# Untere Bildhaelfte des montierten Vogels -> schlechter Score.
|
|
unten = img.getpixel((img.width // 2, int(img.height * 0.97)))
|
|
score = 0.2 if unten != (40, 60, 40) else 0.9
|
|
return Box(0.4, 0.4, 0.6, 0.6, score), None
|
|
|
|
zellen = karte.vermessen(
|
|
NurObenPipeline(), self.hg, self.vogel, anteil=0.09,
|
|
spalten=(0.5,), zeilen=(0.6, 0.99),
|
|
)
|
|
oben = next(z for z in zellen if z.fuss_y == 0.6)
|
|
unten = next(z for z in zellen if z.fuss_y == 0.99)
|
|
self.assertTrue(oben.erkannt)
|
|
self.assertFalse(unten.erkannt)
|
|
self.assertIn("tote Zone", karte.rat(zellen))
|
|
|
|
def test_a_single_failure_disqualifies_a_position(self):
|
|
z = karte.Zelle(0.5, 0.5, treffer=2, gesamt=3, bester_score=0.9)
|
|
self.assertFalse(z.erkannt)
|
|
|
|
def test_drawing_keeps_the_image_size(self):
|
|
zellen = [karte.Zelle(0.5, 0.5, 1, 1, 0.9), karte.Zelle(0.2, 0.9, 0, 1, 0.1)]
|
|
bild = karte.zeichne(self.hg[0], zellen)
|
|
self.assertEqual(bild.size, self.hg[0].size)
|
|
self.assertEqual(bild.mode, "RGB")
|
|
|
|
def test_table_has_a_row_per_line(self):
|
|
zellen = karte.vermessen(
|
|
FakePipeline([(None, 0.0)]),
|
|
self.hg, self.vogel, anteil=0.09,
|
|
spalten=(0.3, 0.7), zeilen=(0.5, 0.9),
|
|
)
|
|
text = karte.tabelle(zellen)
|
|
self.assertEqual(len(text.splitlines()), 3) # Kopf + zwei Zeilen
|
|
|
|
|
|
class StichprobeTests(unittest.TestCase):
|
|
"""Die Stichproben beantworten, was das Protokoll nicht kann: kommt
|
|
überhaupt ein Vogel ins Bild? Dafür müssen sie regelmäßig fallen und
|
|
dürfen die Platte nicht volllaufen lassen."""
|
|
|
|
def setUp(self):
|
|
import tempfile
|
|
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.ordner = Path(self.tmp.name) / "stichproben"
|
|
self.frame = Image.new("RGB", (320, 180), (90, 120, 80))
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_first_frame_is_kept(self):
|
|
from detector.stichprobe import Stichprobe
|
|
|
|
s = Stichprobe(self.ordner, alle_sekunden=300)
|
|
self.assertIsNotNone(s.bild(self.frame, 1000.0))
|
|
self.assertEqual(len(list(self.ordner.glob("*.jpg"))), 1)
|
|
|
|
def test_interval_is_respected(self):
|
|
from detector.stichprobe import Stichprobe
|
|
|
|
s = Stichprobe(self.ordner, alle_sekunden=300)
|
|
s.bild(self.frame, 1000.0)
|
|
self.assertIsNone(s.bild(self.frame, 1299.0))
|
|
self.assertIsNotNone(s.bild(self.frame, 1300.0))
|
|
|
|
def test_oldest_is_dropped_at_the_limit(self):
|
|
"""Ein Dauerläufer ohne Obergrenze füllt irgendwann die Platte."""
|
|
from detector.stichprobe import Stichprobe
|
|
|
|
s = Stichprobe(self.ordner, max_bilder=3)
|
|
for tag in ("01", "02", "03", "04", "05"):
|
|
self.frame.save(self.ordner / f"202609{tag}-120000.jpg")
|
|
|
|
s.bild(self.frame, 1000.0) # raeumt nach dem Speichern auf
|
|
|
|
uebrig = sorted(p.name for p in self.ordner.glob("*.jpg"))
|
|
self.assertEqual(len(uebrig), 3)
|
|
# Die aeltesten sind weg, das frisch gespeicherte ist da.
|
|
self.assertNotIn("20260901-120000.jpg", uebrig)
|
|
self.assertIn("20260905-120000.jpg", uebrig)
|
|
|
|
def test_contact_sheet_covers_the_newest_first(self):
|
|
from detector.stichprobe import kontaktbogen
|
|
|
|
self.ordner.mkdir(parents=True)
|
|
for tag in ("01", "02", "03", "04"):
|
|
self.frame.save(self.ordner / f"202609{tag}-120000.jpg")
|
|
ziel = Path(self.tmp.name) / "bogen.jpg"
|
|
|
|
pfad, n, seiten = kontaktbogen(self.ordner, ziel, spalten=2, zeilen=1)
|
|
self.assertTrue(pfad.exists())
|
|
self.assertEqual(n, 2)
|
|
self.assertEqual(seiten, 2)
|
|
|
|
def test_contact_sheet_without_images_says_so(self):
|
|
from detector.stichprobe import kontaktbogen
|
|
|
|
self.ordner.mkdir(parents=True)
|
|
with self.assertRaises(FileNotFoundError):
|
|
kontaktbogen(self.ordner, Path(self.tmp.name) / "bogen.jpg")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class UebernimmTests(unittest.TestCase):
|
|
"""Einstellungen im laufenden Betrieb ändern.
|
|
|
|
Läuft gegen die echte Pipeline, weil genau das die Zusage ist: das Modell
|
|
bleibt geladen, nur die Zuordnung wird neu berechnet. Mit einem Doppel
|
|
wäre der Test wertlos.
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
from detector.classify import CLASSIFIER_MODEL, DETECTOR_MODEL
|
|
|
|
if not CLASSIFIER_MODEL.exists() or not DETECTOR_MODEL.exists():
|
|
raise unittest.SkipTest("Modelle fehlen — scripts/fetch_models.sh")
|
|
|
|
from detector.classify import Pipeline
|
|
|
|
# Mit Gartenfilter, wie im Betrieb: sonst stehen 148 Arten zur
|
|
# Wahl und das Zurücksetzen im Test trifft die falsche Zahl.
|
|
cls.pipeline = Pipeline(detect_threshold=0.4, min_bird_score=0.05,
|
|
report_threshold=0.08, habitats={"garten"})
|
|
|
|
def test_ohne_aenderung_passiert_nichts(self):
|
|
self.assertEqual(
|
|
self.pipeline.uebernimm(detect_threshold=0.4, min_bird_score=0.05), []
|
|
)
|
|
|
|
def test_schwelle_zieht_den_finder_mit(self):
|
|
"""Sonst sucht der Finder weiter nach der alten Schwelle — der Fehler,
|
|
der das Protokoll schon einmal blind gemacht hat."""
|
|
p = self.pipeline
|
|
try:
|
|
p.uebernimm(detect_threshold=0.05)
|
|
self.assertEqual(p.detect_threshold, 0.05)
|
|
# Der Finder sucht ab der niedrigeren von Wunsch (0.08) und
|
|
# Sichtungsschwelle (0.05).
|
|
self.assertEqual(p.report_threshold, 0.05)
|
|
self.assertEqual(p.finder.threshold, 0.05)
|
|
finally:
|
|
p.uebernimm(detect_threshold=0.4)
|
|
self.assertEqual(p.report_threshold, 0.08)
|
|
|
|
def test_artenliste_wird_getauscht_und_das_modell_bleibt(self):
|
|
from detector.species_de import Species
|
|
|
|
p = self.pipeline
|
|
interp_vorher = p.classifier.interp
|
|
vorher = len(p.classifier.native_idx)
|
|
|
|
nur_zwei = [
|
|
Species("Parus major", "Kohlmeise", "Great Tit", "haeufig", ""),
|
|
Species("Turdus merula", "Amsel", "Blackbird", "haeufig", ""),
|
|
]
|
|
try:
|
|
geaendert = p.uebernimm(arten=nur_zwei)
|
|
self.assertTrue(any("Katalog" in z for z in geaendert))
|
|
self.assertEqual(len(p.classifier.native_idx), 2)
|
|
# Dasselbe Modell, nur eine neue Zuordnung.
|
|
self.assertIs(p.classifier.interp, interp_vorher)
|
|
finally:
|
|
from detector.species_de import catalog
|
|
|
|
p.uebernimm(arten=catalog({"garten"}))
|
|
self.assertEqual(len(p.classifier.native_idx), vorher)
|
|
|
|
def test_gleiche_artenliste_meldet_keine_aenderung(self):
|
|
from detector.species_de import catalog
|
|
|
|
self.assertEqual(self.pipeline.uebernimm(arten=catalog({"garten"})), [])
|
|
|
|
|
|
class AnzeigeAusschnittTests(unittest.TestCase):
|
|
"""🔑 Der Klassifikator und das Auge wollen Verschiedenes.
|
|
|
|
``crop`` liegt eng am Vogel, weil Hintergrund die Bestimmung verwaessert.
|
|
In der fast quadratischen Kachel der Sammlung schneidet genau das links
|
|
und rechts den Vogel weg — bei einem Vogel im Hochformat bleibt die
|
|
Bauchmitte uebrig. Gemeldet als „der Zuschnitt sieht merkwuerdig aus".
|
|
"""
|
|
|
|
def bild(self, groesse=(1800, 2400)):
|
|
from PIL import Image
|
|
|
|
return Image.new("RGB", groesse, (120, 160, 200))
|
|
|
|
def test_anzeige_ist_quadratisch(self):
|
|
from detector.classify import Box
|
|
|
|
# Ein hochkant sitzender Vogel, schmal und hoch.
|
|
box = Box(x0=0.40, y0=0.30, x1=0.52, y1=0.62, score=0.9)
|
|
im = self.bild()
|
|
eng = box.crop(im)
|
|
anzeige = box.anzeige_ausschnitt(im)
|
|
|
|
self.assertLess(eng.width / eng.height, 0.7, "Testbox ist nicht hochkant")
|
|
self.assertAlmostEqual(anzeige.width / anzeige.height, 1.0, places=1)
|
|
|
|
def test_anzeige_enthaelt_den_ganzen_vogel(self):
|
|
from detector.classify import Box
|
|
|
|
box = Box(x0=0.40, y0=0.30, x1=0.52, y1=0.62, score=0.9)
|
|
im = self.bild()
|
|
anzeige = box.anzeige_ausschnitt(im)
|
|
# Mindestens so hoch wie der Vogel plus Rand.
|
|
vogel_hoehe = (0.62 - 0.30) * im.height
|
|
self.assertGreater(anzeige.height, vogel_hoehe)
|
|
|
|
def test_crop_bleibt_unveraendert_eng(self):
|
|
"""Die Artbestimmung ist gemessen und darf sich nicht mitaendern."""
|
|
from detector.classify import Box
|
|
|
|
box = Box(x0=0.40, y0=0.30, x1=0.52, y1=0.62, score=0.9)
|
|
im = self.bild()
|
|
eng = box.crop(im)
|
|
erwartet_breite = (0.52 - 0.40) * im.width * 1.24 # pad 0.12 beidseits
|
|
self.assertAlmostEqual(eng.width, erwartet_breite, delta=3)
|
|
|
|
def test_am_bildrand_bleibt_der_ausschnitt_im_bild(self):
|
|
"""⚠️ Sonst sitzt ein Vogel am Rand ausserhalb seiner eigenen Kachel."""
|
|
from detector.classify import Box
|
|
|
|
im = self.bild()
|
|
for box in (
|
|
Box(x0=0.00, y0=0.00, x1=0.08, y1=0.10, score=0.9),
|
|
Box(x0=0.92, y0=0.90, x1=1.00, y1=1.00, score=0.9),
|
|
):
|
|
with self.subTest(box=box):
|
|
a = box.anzeige_ausschnitt(im)
|
|
self.assertGreater(a.width, 0)
|
|
self.assertLessEqual(a.width, im.width)
|
|
self.assertLessEqual(a.height, im.height)
|
|
self.assertAlmostEqual(a.width / a.height, 1.0, places=1)
|
|
|
|
def test_winziges_bild_sprengt_nichts(self):
|
|
from detector.classify import Box
|
|
|
|
im = self.bild((60, 40))
|
|
a = Box(x0=0.3, y0=0.3, x1=0.7, y1=0.7, score=0.9).anzeige_ausschnitt(im)
|
|
self.assertLessEqual(a.width, 60)
|
|
self.assertLessEqual(a.height, 40)
|