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.
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""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()
|