Vogelalbum 1.0.0
Eine Kamera am Vogelhaus, ein Rechner der erkennt wer da war — und eine Sammlung, die sich von allein fuellt. Zwei neuronale Netze: eines findet den Vogel im Bild, eines bestimmt die Art. 148 heimische Arten, auf Deutschland eingeschraenkt und neu normiert — ohne diese Einschraenkung wird aus einer Kohlmeise eine nordamerikanische Schwesterart. Daraus entsteht ein Sammelalbum: nummerierte Seiten, gefundene Arten in Farbe, fehlende als Schatten, dazu Erstfang, Punkte, Stufen und 15 Abzeichen. Laeuft vollstaendig auf eigener Hardware. Kein Cloud-Dienst, kein Konto, keine Telemetrie; die Bilder verlassen das eigene Netz nicht. Zwei getrennt deploybare Teile: die Sammlung als Django-Anwendung, die Erkennung als eigener Dienst. Der Erkennungs-Dienst fragt selbst an und braucht deshalb weder eine erreichbare Adresse noch einen offenen Port — er laeuft hinter einem gewoehnlichen Router. Dabei: Firmware fuer Raspberry Pi und ESP32-S3, Anbindung an Home Assistant ueber MQTT, Oberflaeche auf Deutsch und Englisch, Einstellungen je Kamera aus dem Browser, und ein Weg fuer Fotos von unterwegs. AGPL-3.0.
This commit is contained in:
commit
cc9e0786c2
189 files changed
+29792
No files matched your search
@@ -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()
|
||||
Reference in new issue
Block a user