christian/vogelalbum
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Activity

Vogelalbum 1.0.0

Browse Source 
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 2026-10-15 14:30:02 +00:00
commit c11fe9ab04
189 files changed
+29806
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_bruecke.py
+160
View File
@@ -0,0 +1,160 @@
"""Woher die Bruecke das Livebild holt.
./venv/bin/python -m unittest discover tests
🔑 Diese Datei gibt es wegen eines Ausfalls, der acht Tage lang im Betrieb
stand, ohne dass irgendwo ein Fehler auftauchte: Seit dem Umzug nach Coolify
(03.10.2026) trug die oeffentliche Instanz ``VOGELALBUM_LIVE_ERLAUBT=0`` --
richtig so, ein Bild aus dem eigenen Garten gehoert nicht nach draussen.
Damit lieferte ``/api/zustand/`` aber ``bild_url: null``, und die Bruecke
veroeffentlichte kein Livebild mehr. In Home Assistant blieb das letzte Bild
stehen und **sah aus wie ein Livebild**.
⚠️ Der Riegel war gegen die Oeffentlichkeit gedacht und sperrte das eigene
Haus mit aus -- obwohl die Bruecke im Heimnetz laeuft und den Verteiler
direkt erreicht. Ein Schalter, zwei Zwecke.
Gegengeprueft: Ohne ``live_adresse`` sind die Faelle "Vorgabe schlaegt API"
und "Vorgabe trotz bild_url null" beide rot.
"""
from __future__ import annotations
import os
import unittest
from homeassistant import bruecke
class LiveAdresseTests(unittest.TestCase):
SCHLUESSEL = (
"VOGELALBUM_LIVE_VOGELHAUS",
"VOGELALBUM_LIVE_VOGELHAUS_2",
)
def setUp(self):
vorher = {k: os.environ.get(k) for k in self.SCHLUESSEL}
for k in self.SCHLUESSEL:
os.environ.pop(k, None)
def zuruecksetzen():
for k, wert in vorher.items():
if wert is None:
os.environ.pop(k, None)
else:
os.environ[k] = wert
self.addCleanup(zuruecksetzen)
@staticmethod
def kamera(kennung="vogelhaus", bild_url=None):
return {"kennung": kennung, "bild_url": bild_url}
# --- ohne eigene Vorgabe: alles bleibt wie bisher ----------------
def test_ohne_alles_leer(self):
self.assertEqual(bruecke.live_adresse(self.kamera()), "")
def test_bild_url_aus_der_schnittstelle(self):
kamera = self.kamera(bild_url="http://erkennung.local:8131/bild.jpg")
self.assertEqual(bruecke.live_adresse(kamera),
"http://erkennung.local:8131/bild.jpg")
def test_stromadresse_wird_zum_einzelbild(self):
"""``/live`` ist der endlose MJPEG-Strom -- den darf sie nicht lesen."""
kamera = self.kamera(bild_url="http://erkennung.local:8131/live")
self.assertEqual(bruecke.live_adresse(kamera),
"http://erkennung.local:8131/bild.jpg")
def test_schraegstrich_am_ende_stoert_nicht(self):
kamera = self.kamera(bild_url="http://erkennung.local:8131/live/")
self.assertEqual(bruecke.live_adresse(kamera),
"http://erkennung.local:8131/bild.jpg")
# --- mit eigener Vorgabe: der eigentliche Zweck ------------------
def test_vorgabe_auch_wenn_die_schnittstelle_nichts_liefert(self):
"""Der Ausfall vom 03.-11.10.2026: LIVE_ERLAUBT=0, also bild_url null."""
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = \
"http://erkennung.local:8131/bild.jpg"
self.assertEqual(bruecke.live_adresse(self.kamera()),
"http://erkennung.local:8131/bild.jpg")
def test_vorgabe_schlaegt_die_schnittstelle(self):
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = "http://127.0.0.1:8131/bild.jpg"
kamera = self.kamera(bild_url="http://erkennung.local:8131/bild.jpg")
self.assertEqual(bruecke.live_adresse(kamera),
"http://127.0.0.1:8131/bild.jpg")
def test_vorgabe_als_stromadresse(self):
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = "http://erkennung.local:8131/live"
self.assertEqual(bruecke.live_adresse(self.kamera()),
"http://erkennung.local:8131/bild.jpg")
# --- Randfaelle --------------------------------------------------
def test_kennung_mit_sonderzeichen(self):
""""vogelhaus-2" wird zu VOGELALBUM_LIVE_VOGELHAUS_2."""
os.environ["VOGELALBUM_LIVE_VOGELHAUS_2"] = \
"http://erkennung.local:8133/bild.jpg"
kamera = self.kamera(kennung="vogelhaus-2")
self.assertEqual(bruecke.live_adresse(kamera),
"http://erkennung.local:8133/bild.jpg")
def test_ohne_kennung_keine_vorgabe(self):
"""Ohne Kennung gibt es keinen Variablennamen -- aber auch keinen Absturz."""
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = "http://127.0.0.1:8131/bild.jpg"
kamera = {"bild_url": "http://erkennung.local:8131/bild.jpg"}
self.assertEqual(bruecke.live_adresse(kamera),
"http://erkennung.local:8131/bild.jpg")
class EntwuerfeTests(unittest.TestCase):
"""Die Kamera-Entitaet darf nur entstehen, wenn es ein Bild dazu gibt.
⚠️ Sonst legt Home Assistant eine Kamera an, die nie etwas zeigt -- und
eine leere Kachel ist schlimmer als keine.
"""
def setUp(self):
self.vorher = os.environ.get("VOGELALBUM_LIVE_VOGELHAUS")
os.environ.pop("VOGELALBUM_LIVE_VOGELHAUS", None)
def zuruecksetzen():
if self.vorher is None:
os.environ.pop("VOGELALBUM_LIVE_VOGELHAUS", None)
else:
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = self.vorher
self.addCleanup(zuruecksetzen)
@staticmethod
def zustand():
return {
"kameras": [{
"name": "Futterhaus Garten",
"kennung": "vogelhaus",
"aktiv": True,
"online": True,
"bild_url": None,
}],
"sammlung": {},
}
def _hat_livebild(self):
return any(
inhalt.get("name") == "Livebild"
for _, inhalt in bruecke.entwuerfe(self.zustand())
)
def test_ohne_adresse_keine_kamera_entitaet(self):
self.assertFalse(self._hat_livebild())
def test_mit_vorgabe_entsteht_die_kamera_entitaet(self):
os.environ["VOGELALBUM_LIVE_VOGELHAUS"] = \
"http://erkennung.local:8131/bild.jpg"
self.assertTrue(self._hat_livebild())
if __name__ == "__main__":
unittest.main()
tests/test_detector.py
+958
View File
@@ -0,0 +1,958 @@
"""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)
tests/test_fernkonfig.py
+119
View File
@@ -0,0 +1,119 @@
"""Einstellungen aus der Web-App.
Der Kern ist nicht der Abruf, sondern das Verhalten im Fehlerfall: eine
unerreichbare Web-App darf die Erkennung nicht anhalten und nicht auf
Vorgaben zurückwerfen. Wer zwei Wochen auf den ersten Vogel wartet, will ihn
nicht verpassen, weil ein Webserver neu gestartet wurde.
"""
from __future__ import annotations
import json
import threading
import unittest
from http import server
from detector.fernkonfig import Fernkonfig, arten_aus_json
ANTWORT = {
"kamera": "vogelhaus",
"stream_url": "http://10.0.0.9:8080/stream",
"habitats": ["garten"],
"detect_threshold": 0.35,
"min_bird_score": 0.07,
"region": "de",
"arten": [
{"scientific": "Parus major", "german": "Kohlmeise", "english": "Great Tit",
"rarity": "haeufig", "lebensraum": "garten"},
{"scientific": "Turdus merula", "german": "Amsel", "english": "Blackbird",
"rarity": "haeufig", "lebensraum": "garten"},
],
}
class _Handler(server.BaseHTTPRequestHandler):
antwort = ANTWORT
status = 200
gesehene_token: list[str] = []
def log_message(self, fmt, *args): # noqa: A003
pass
def do_GET(self): # noqa: N802
_Handler.gesehene_token.append(self.headers.get("Authorization", ""))
koerper = json.dumps(self.antwort).encode("utf-8")
self.send_response(self.status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(koerper)))
self.end_headers()
self.wfile.write(koerper)
class ArtenTests(unittest.TestCase):
def test_uebersetzt_die_felder(self):
arten = arten_aus_json(ANTWORT["arten"])
self.assertEqual([a.scientific for a in arten], ["Parus major", "Turdus merula"])
self.assertEqual(arten[0].german, "Kohlmeise")
self.assertEqual(arten[0].lebensraum, "garten")
def test_ohne_wissenschaftlichen_namen_faellt_weg(self):
"""Der wissenschaftliche Name ist der Schlüssel ins Modell — ohne ihn
trifft der Eintrag später leise nichts."""
arten = arten_aus_json([{"german": "Namenlos"}, {"scientific": " "}])
self.assertEqual(arten, [])
def test_fehlende_beschriftung_wird_ersetzt(self):
arten = arten_aus_json([{"scientific": "Pica pica"}])
self.assertEqual(arten[0].german, "Pica pica")
self.assertEqual(arten[0].rarity, "haeufig")
class AbrufTests(unittest.TestCase):
def setUp(self):
_Handler.antwort = ANTWORT
_Handler.status = 200
_Handler.gesehene_token = []
self.srv = server.HTTPServer(("127.0.0.1", 0), _Handler)
threading.Thread(target=self.srv.serve_forever, daemon=True).start()
self.addCleanup(self.srv.server_close)
self.addCleanup(self.srv.shutdown)
self.api = f"http://127.0.0.1:{self.srv.server_address[1]}"
def test_holt_und_merkt_sich(self):
f = Fernkonfig(self.api, "geheim")
daten = f.hole()
self.assertEqual(daten["kamera"], "vogelhaus")
self.assertEqual(f.letzte, daten)
self.assertEqual(_Handler.gesehene_token, ["Bearer geheim"])
def test_faellig_erst_nach_dem_intervall(self):
f = Fernkonfig(self.api, "geheim", intervall=300)
self.assertTrue(f.faellig(1000.0))
f.hole(1000.0)
self.assertFalse(f.faellig(1299.0))
self.assertTrue(f.faellig(1300.0))
def test_abgelehnter_token_gibt_none(self):
_Handler.status = 401
f = Fernkonfig(self.api, "falsch")
self.assertIsNone(f.hole())
def test_unerreichbar_gibt_none_und_behaelt_den_stand(self):
"""Das eigentliche Versprechen: ein Ausfall ändert nichts."""
f = Fernkonfig(self.api, "geheim")
vorher = f.hole()
self.assertIsNotNone(vorher)
self.srv.shutdown()
self.srv.server_close()
self.assertIsNone(f.hole())
self.assertEqual(f.letzte, vorher)
def test_unlesbare_antwort_gibt_none(self):
_Handler.antwort = ["kein", "objekt"]
f = Fernkonfig(self.api, "geheim")
self.assertIsNone(f.hole())
if __name__ == "__main__":
unittest.main()
tests/test_verteiler.py
+204
View File
@@ -0,0 +1,204 @@
"""Der Live-Verteiler.
Der Kern ist nicht der HTTP-Teil, sondern das Versprechen: die Erkennung darf
davon nichts merken. Deshalb pruefen die Tests vor allem, dass ohne Zuschauer
nicht kodiert wird und dass fertige JPEG-Bytes unveraendert durchlaufen.
"""
from __future__ import annotations
import json # noqa: F401 - Symmetrie zu den anderen Testdateien
import threading
import unittest
import urllib.error
import urllib.request
from io import BytesIO
from PIL import Image
from detector.verteiler import Bildhalter, Handler, Verteiler
BILD = Image.new("RGB", (320, 180), (60, 110, 70))
def jpeg_bytes(bild: Image.Image = BILD) -> bytes:
puffer = BytesIO()
bild.save(puffer, format="JPEG")
return puffer.getvalue()
class BildhalterTests(unittest.TestCase):
def test_ohne_bild_kommt_nichts(self):
self.assertIsNone(Bildhalter().jpeg())
def test_fertige_bytes_laufen_unveraendert_durch(self):
"""Der haeufige Fall: die Quelle liefert MJPEG. Wird hier umkodiert,
zahlt der Detector Rechenzeit fuer nichts."""
roh = jpeg_bytes()
h = Bildhalter()
h.zeige(BILD, roh)
nummer, daten = h.jpeg()
self.assertEqual(nummer, 1)
self.assertIs(daten, roh)
def test_ohne_fertige_bytes_wird_kodiert(self):
"""RTSP und Bilderordner liefern kein fertiges JPEG — dann muss der
Verteiler selbst eines machen."""
h = Bildhalter()
h.zeige(BILD, None)
nummer, daten = h.jpeg()
self.assertEqual(nummer, 1)
self.assertEqual(daten[:2], b"\xff\xd8")
self.assertEqual(Image.open(BytesIO(daten)).size, BILD.size)
def test_kodiert_nur_einmal_je_bild(self):
"""Sonst zahlt jeder Zuschauer seine eigene Kodierung."""
h = Bildhalter()
h.zeige(BILD, None)
erst = h.jpeg()
nochmal = h.jpeg()
self.assertIs(erst[1], nochmal[1])
def test_neues_bild_wird_neu_kodiert(self):
h = Bildhalter()
h.zeige(BILD, None)
erst = h.jpeg()
h.zeige(Image.new("RGB", (320, 180), (200, 30, 30)), None)
neu = h.jpeg()
self.assertEqual(neu[0], erst[0] + 1)
self.assertIsNot(neu[1], erst[1])
def test_zeige_kodiert_nicht(self):
"""Das Versprechen an die Erkennung: anmelden kostet Zeigerarbeit.
Solange niemand abholt, entsteht kein JPEG."""
h = Bildhalter()
for _ in range(50):
h.zeige(BILD, None)
self.assertIsNone(h._kodiert)
self.assertEqual(h.nummer, 50)
def test_grosse_bilder_werden_verkleinert(self):
h = Bildhalter(max_breite=160)
h.zeige(BILD, None)
self.assertEqual(Image.open(BytesIO(h.jpeg()[1])).width, 160)
def test_warte_laeuft_ohne_bild_ab(self):
h = Bildhalter()
self.assertEqual(h.warte(0, timeout=0.05), 0)
def test_warte_kehrt_mit_neuem_bild_zurueck(self):
h = Bildhalter()
threading.Timer(0.05, lambda: h.zeige(BILD, None)).start()
self.assertEqual(h.warte(0, timeout=5.0), 1)
def test_warte_kehrt_sofort_zurueck_wenn_schon_neu(self):
h = Bildhalter()
h.zeige(BILD, None)
self.assertEqual(h.warte(0, timeout=5.0), 1)
class VerteilerTests(unittest.TestCase):
def setUp(self):
Handler._zuschauer = 0
self.v = Verteiler(port=0, bind="127.0.0.1")
self.v.start()
self.addCleanup(self.v.stop)
def url(self, pfad: str) -> str:
return f"http://127.0.0.1:{self.v.port}{pfad}"
def hole(self, pfad: str, timeout: float = 5.0):
return urllib.request.urlopen(self.url(pfad), timeout=timeout)
def test_seite_kommt(self):
with self.hole("/") as r:
self.assertEqual(r.status, 200)
self.assertIn(b"/live", r.read())
def test_einzelbild_erst_mit_bild(self):
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/bild.jpg")
self.assertEqual(fehler.exception.code, 503)
self.v.zeige(BILD, jpeg_bytes())
with self.hole("/bild.jpg") as r:
self.assertEqual(r.status, 200)
self.assertEqual(r.headers["Content-Type"], "image/jpeg")
self.assertEqual(r.read()[:2], b"\xff\xd8")
def test_unbekannter_pfad_ist_404(self):
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/gibtsnicht")
self.assertEqual(fehler.exception.code, 404)
def test_stream_liefert_multipart(self):
stop = threading.Event()
def speisen():
while not stop.is_set():
self.v.zeige(BILD, jpeg_bytes())
stop.wait(0.01)
threading.Thread(target=speisen, daemon=True).start()
self.addCleanup(stop.set)
strom = self.hole("/live")
self.addCleanup(strom.close)
self.assertIn("multipart/x-mixed-replace", strom.headers["Content-Type"])
anfang = strom.read(400)
self.assertIn(b"--BILDGRENZE", anfang)
self.assertIn(b"\xff\xd8", anfang)
def test_zwei_zuschauer_sind_erlaubt(self):
"""Der ganze Zweck: an der Kamera geht nur einer, hier beliebig viele."""
stop = threading.Event()
def speisen():
while not stop.is_set():
self.v.zeige(BILD, jpeg_bytes())
stop.wait(0.01)
threading.Thread(target=speisen, daemon=True).start()
self.addCleanup(stop.set)
erster = self.hole("/live")
self.addCleanup(erster.close)
erster.read(200)
zweiter = self.hole("/live")
self.addCleanup(zweiter.close)
self.assertEqual(zweiter.status, 200)
zweiter.read(200)
self.assertEqual(self.v.zuschauer, 2)
def test_obergrenze_weist_ab(self):
stop = threading.Event()
def speisen():
while not stop.is_set():
self.v.zeige(BILD, jpeg_bytes())
stop.wait(0.01)
threading.Thread(target=speisen, daemon=True).start()
self.addCleanup(stop.set)
Handler.max_zuschauer = 1
self.addCleanup(setattr, Handler, "max_zuschauer", 8)
erster = self.hole("/live")
self.addCleanup(erster.close)
erster.read(200)
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/live")
self.assertEqual(fehler.exception.code, 503)
if __name__ == "__main__":
unittest.main()
tests/test_vogelkamera.py
+352
View File
@@ -0,0 +1,352 @@
"""Tests fuer den MJPEG-Server der Kameraseite (``pi/vogelkamera.py``).
./venv/bin/python -m unittest discover tests
Laeuft ohne Raspberry Pi und ohne Kamera: ``rpicam-vid`` wird nie gestartet,
der Bildpuffer von Hand befuellt. Geprueft wird die HTTP-Schicht -- und die
ist der Teil, der schon Schaden angerichtet hat. Dass ein zweiter Zuschauer
abgewiesen wird, ist keine Kosmetik: ein stiller zweiter Client halbiert die
Bildrate und der Detector faellt in Timeouts, ohne dass irgendwo ein Fehler
steht.
⚠️ **Diese Datei war vom 25.09. bis zum 09.10.2026 rot** -- 19 Fehler. Commit
b57c1b3 ersetzte die picamera2-Fassung durch die rpicam-vid-Fassung, die
wirklich auf dem Geraet laeuft; die Tests blieben auf den alten Namen stehen
(``Bildpuffer`` statt ``FrameBuffer``, ``Handler`` statt ``MJPEGHandler``) und
pruefen Code, den es nicht mehr gab.
🔑 Beim Nachziehen kam heraus, dass ``pi/README.md`` die 503-Absage fuer den
zweiten Zuschauer die ganze Zeit **versprochen** hat, waehrend die Fassung vom
Geraet ihn nur zaehlte. Die Tests haben also nicht nur den Namen verfehlt,
sondern eine fehlende Schutzfunktion verdeckt.
"""
from __future__ import annotations
import importlib.util
import json
import threading
import time
import unittest
import urllib.error
import urllib.request
from pathlib import Path
_pfad = Path(__file__).resolve().parents[1] / "pi" / "vogelkamera.py"
_spec = importlib.util.spec_from_file_location("vogelkamera", _pfad)
vogelkamera = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(vogelkamera)
# Ein winziges, aber gueltiges JPEG waere hier Overkill -- der Server reicht
# Bytes durch, ohne sie zu deuten.
BILD = b"\xff\xd8" + b"x" * 64 + b"\xff\xd9"
class ServerTests(unittest.TestCase):
def setUp(self):
# Der Server haelt seinen Zustand in Modulvariablen, nicht in einer
# Instanz. Jeder Test setzt sie zurueck, sonst schleppt er den Stand
# des vorigen mit -- der Zuschauerplatz ganz besonders.
vogelkamera.frame_buffer = vogelkamera.FrameBuffer()
vogelkamera.zuschauer_platz = threading.Semaphore(1)
vogelkamera.bilder_gesamt = 0
vogelkamera.zuschauer = 0
vogelkamera.START_ZEIT = time.time()
self.server = vogelkamera.ThreadedHTTPServer(
("127.0.0.1", 0), vogelkamera.MJPEGHandler
)
self.port = self.server.server_address[1]
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.addCleanup(self._stop)
def _stop(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
def url(self, pfad: str) -> str:
return f"http://127.0.0.1:{self.port}{pfad}"
def hole(self, pfad: str, timeout: float = 5.0):
return urllib.request.urlopen(self.url(pfad), timeout=timeout)
def schiebe(self, bild: bytes = BILD) -> None:
"""Ein Bild einspeisen, so wie es ``capture_loop`` tut -- inklusive
Zaehlwerk, denn /status liest beides."""
vogelkamera.frame_buffer.update(bild)
with vogelkamera._zaehler_sperre:
vogelkamera.bilder_gesamt += 1
def speisen(self) -> threading.Event:
"""Bilder nachschieben, bis der Test fertig ist."""
stop = threading.Event()
def lauf():
while not stop.is_set():
self.schiebe()
stop.wait(0.01)
threading.Thread(target=lauf, daemon=True).start()
self.addCleanup(stop.set)
return stop
# -- die einfachen Endpunkte ------------------------------------------
def test_startseite_kommt(self):
with self.hole("/") as r:
self.assertEqual(r.status, 200)
self.assertIn(b"/stream", r.read())
def test_temperatur_endpunkt_antwortet_immer(self):
"""Auf einem Nicht-Pi gibt es die Datei nicht -- dann 'N/A', aber kein
Fehler. Der Endpunkt ist die einzige Gesundheitspruefung, die den
Stream nicht anfasst."""
with self.hole("/temp") as r:
self.assertEqual(r.status, 200)
self.assertTrue(r.read().strip())
def test_gesund_meldet_erst_mit_bild(self):
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/gesund")
self.assertEqual(fehler.exception.code, 503)
self.schiebe()
with self.hole("/gesund") as r:
self.assertEqual(r.status, 200)
def test_gesund_ist_auch_unter_health_erreichbar(self):
"""Der englische Name steht in fremden Werkzeugen (Uptime-Kuma und
Konsorten) -- beide Pfade muessen dasselbe tun."""
self.schiebe()
with self.hole("/health") as r:
self.assertEqual(r.status, 200)
def test_gesund_merkt_ein_altes_bild(self):
"""Der Server antwortet noch, die Aufnahme haengt. Genau das ist der
Fall, den ein Ping nie sieht."""
vogelkamera.frame_buffer.update(BILD)
with vogelkamera.frame_buffer._lock:
vogelkamera.frame_buffer._timestamp = time.time() - 30
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/gesund")
self.assertEqual(fehler.exception.code, 503)
def test_schnappschuss_gibt_das_bild_her(self):
"""Ein einzelnes Bild statt eines Streams -- der schonende Weg, einmal
nachzusehen, was die Kamera sieht."""
self.schiebe()
with self.hole("/snapshot") as r:
self.assertEqual(r.status, 200)
self.assertEqual(r.headers["Content-Type"], "image/jpeg")
self.assertEqual(r.read(), BILD)
def test_schnappschuss_ohne_bild_ist_503(self):
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/snapshot")
self.assertEqual(fehler.exception.code, 503)
def test_unbekannter_pfad_ist_404(self):
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/gibtsnicht")
self.assertEqual(fehler.exception.code, 404)
# -- der Stream --------------------------------------------------------
def test_stream_liefert_multipart_mit_bildern(self):
self.speisen()
with self.hole("/stream") as r:
self.assertEqual(r.status, 200)
self.assertIn("multipart/x-mixed-replace", r.headers["Content-Type"])
anfang = r.read(len(BILD) + 200)
self.assertIn(b"--frame", anfang)
self.assertIn(b"\xff\xd8", anfang)
def test_zweiter_zuschauer_wird_abgewiesen(self):
"""Der eigentliche Punkt: lieber eine klare Absage als zwei ruckelnde
Bilder und ein Detector in Timeouts.
⚠️ Dieser Test war der Grund, die Datei wieder anzufassen -- er stand
auf der alten Fassung und hat deshalb nicht gemeldet, dass die Absage
im Code fehlte, obwohl das README sie zusagte."""
self.speisen()
erster = self.hole("/stream")
self.addCleanup(erster.close)
erster.read(len(BILD)) # sicherstellen, dass er wirklich haengt
with self.assertRaises(urllib.error.HTTPError) as fehler:
self.hole("/stream")
self.assertEqual(fehler.exception.code, 503)
self.assertIn("Zuschauer", fehler.exception.read().decode("utf-8"))
def test_nach_dem_trennen_darf_der_naechste_wieder(self):
"""Ohne das waere die Kamera nach einem Browserfenster dauerhaft
dicht."""
self.speisen()
erster = self.hole("/stream")
erster.read(len(BILD))
erster.close()
# Der Server merkt das Trennen erst beim naechsten Schreibversuch.
for _ in range(250):
if vogelkamera.zuschauer_platz.acquire(blocking=False):
vogelkamera.zuschauer_platz.release()
break
time.sleep(0.02)
else:
self.fail("Der Zuschauerplatz wurde nach dem Trennen nie frei")
zweiter = self.hole("/stream")
self.addCleanup(zweiter.close)
self.assertEqual(zweiter.status, 200)
def test_stehende_kamera_gibt_den_platz_wieder_frei(self):
"""🔑 Die Falle, die der Zuschauerplatz selbst aufstellt.
Ohne neues Bild schreibt der Server nichts, merkt also auch nicht, dass
der Zuschauer weg ist -- und haelt den einen Platz fuer immer. Der
Detector bekaeme bei jedem Versuch 503, die Kamera waere dauerhaft
stumm. Deshalb endet ein Stream ohne Bild nach ``STREAM_LEERLAUF``."""
vogelkamera.STREAM_LEERLAUF = 0.3
self.addCleanup(setattr, vogelkamera, "STREAM_LEERLAUF", 15.0)
self.schiebe() # ein Bild, danach bleibt die Kamera stehen
erster = self.hole("/stream")
self.addCleanup(erster.close)
erster.read(len(BILD))
for _ in range(100):
if vogelkamera.zuschauer_platz.acquire(blocking=False):
vogelkamera.zuschauer_platz.release()
break
time.sleep(0.02)
else:
self.fail("Der Platz blieb an einer stehenden Kamera haengen")
zweiter = self.hole("/stream")
self.addCleanup(zweiter.close)
self.assertEqual(zweiter.status, 200)
# -- /status ----------------------------------------------------------
def test_status_antwortet_ohne_bild(self):
"""Genau dann braucht man ihn: die Kamera liefert nichts und man will
wissen, warum -- ohne den Stream anzufassen."""
with self.hole("/status") as r:
self.assertEqual(r.status, 200)
self.assertEqual(r.headers["Content-Type"], "application/json")
d = json.loads(r.read())
self.assertEqual(d["kamera"], "fehlt")
self.assertEqual(d["fehler"], "0x1")
self.assertEqual(d["bilder"], 0)
self.assertEqual(d["letztes_bild_s"], -1.0)
self.assertFalse(d["zuschauer"])
def test_status_zaehlt_bilder_auch_ohne_zuschauer(self):
"""Die Kamera liefert ins Leere, wenn niemand zuschaut -- /status muss
trotzdem sagen koennen, dass sie arbeitet."""
for _ in range(3):
self.schiebe()
with self.hole("/status") as r:
d = json.loads(r.read())
self.assertEqual(d["kamera"], "bereit")
self.assertEqual(d["fehler"], "0x0")
self.assertEqual(d["bilder"], 3)
self.assertGreaterEqual(d["letztes_bild_s"], 0.0)
self.assertFalse(d["zuschauer"])
def test_status_hat_die_felder_der_esp32_seite(self):
"""Gleiches Schema auf beiden Kameraseiten, sonst braucht jedes
Werkzeug zwei Faelle. ``psram_frei`` fehlt absichtlich -- das gibt es
nur auf dem ESP32."""
self.schiebe()
with self.hole("/status") as r:
d = json.loads(r.read())
for feld in ("kamera", "fehler", "laufzeit_s", "bilder", "fps",
"letztes_bild_s", "zuschauer", "rssi", "temperatur_c",
"speicher_frei"):
self.assertIn(feld, d)
def test_status_sieht_den_zuschauer(self):
self.speisen()
strom = self.hole("/stream")
self.addCleanup(strom.close)
strom.read(len(BILD))
with self.hole("/status") as r:
self.assertTrue(json.loads(r.read())["zuschauer"])
def test_status_geht_waehrend_der_stream_laeuft(self):
"""Der Stream haelt seinen Socket dauerhaft. Waere der Server dann
blockiert, waere /status nutzlos -- gefragt ist er ja im Betrieb."""
self.speisen()
strom = self.hole("/stream")
self.addCleanup(strom.close)
strom.read(len(BILD))
with self.hole("/status", timeout=5.0) as r:
self.assertEqual(r.status, 200)
class MesswerteTests(unittest.TestCase):
"""Die drei Ablesungen aus /proc und /sys. Auf einem Nicht-Pi fehlen sie
teils -- dann ist None richtig und ein Absturz falsch."""
def test_temperatur_ist_zahl_oder_none(self):
c = vogelkamera.read_cpu_temp()
self.assertTrue(c is None or isinstance(c, float))
def test_funkstaerke_ist_zahl_oder_none(self):
r = vogelkamera.funkstaerke()
self.assertTrue(r is None or isinstance(r, int))
def test_speicher_ist_zahl_oder_none(self):
m = vogelkamera.speicher_frei()
self.assertTrue(m is None or m > 0)
def test_status_json_ueberlebt_fehlende_messwerte(self):
"""Auf einem Nicht-Pi fehlt alles, was aus /sys kommt. Trotzdem muss
das JSON vollstaendig und serialisierbar sein -- die Bruecke nach Home
Assistant liest es blind."""
vogelkamera.frame_buffer = vogelkamera.FrameBuffer()
d = vogelkamera.status_json()
json.dumps(d) # wirft, wenn ein Wert nicht serialisierbar ist
self.assertEqual(d["kamera"], "fehlt")
class FrameBufferTests(unittest.TestCase):
def test_haelt_nur_das_neueste_bild(self):
"""Kein Rueckstau: wer sich spaet einklinkt, will das aktuelle Bild
und nicht die Warteschlange von vor zehn Sekunden."""
p = vogelkamera.FrameBuffer()
p.update(b"alt")
p.update(b"neu")
self.assertEqual(p.get()[0], b"neu")
def test_leerer_puffer_hat_keinen_zeitstempel(self):
"""Die Null ist der Wert, an dem /status 'fehlt' erkennt -- ein
geratener Zeitstempel wuerde eine stehende Kamera gesund aussehen
lassen."""
p = vogelkamera.FrameBuffer()
frame, ts = p.get()
self.assertIsNone(frame)
self.assertEqual(ts, 0)
def test_datiert_jedes_bild(self):
p = vogelkamera.FrameBuffer()
p.update(b"eins")
erster = p.get()[1]
self.assertGreater(erster, 0)
time.sleep(0.02)
p.update(b"zwei")
self.assertGreater(p.get()[1], erster)
if __name__ == "__main__":
unittest.main()