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.
353 lines
14 KiB
Python
353 lines
14 KiB
Python
"""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()
|