christian
a97f066e92
Die Sichtung merkt sich jetzt, WO der Vogel im Foto sitzt (vier Zahlen); die Anzeige schneidet daraus bei jedem Aufruf. Eine geaenderte Regel wirkt damit sofort auf den ganzen Bestand. Altbestand faellt auf das alte Bild zurueck.
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""Die Zustandsmaschine, die aus Einzelbildern eine "Sichtung" macht.
|
|
|
|
Eine Sichtung ist ein Besuch: Vogel landet, bleibt ein paar Sekunden, fliegt
|
|
weg. Über die Dauer sammeln wir Stimmen und behalten das schärfste Foto.
|
|
|
|
Warum abstimmen statt einfach das erste Ergebnis nehmen? Weil ein Vogel sich
|
|
dreht. Von hinten ist eine Kohlmeise nur eine olivgrüne Kugel; von vorn ist
|
|
sie eindeutig. Über 30 Frames gewinnt die richtige Art zuverlässig — eine
|
|
Einzelmessung kippt bei jedem Flügelschlag.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
|
|
from PIL import Image
|
|
|
|
from .classify import Box, Pipeline, Result
|
|
from .motion import MotionDetector, sharpness
|
|
from .species_de import Species
|
|
|
|
|
|
@dataclass
|
|
class Sighting:
|
|
started_at: datetime
|
|
ended_at: datetime
|
|
species: Species | None
|
|
confidence: float
|
|
votes: dict[str, float]
|
|
frames_total: int
|
|
frames_with_bird: int
|
|
best_image: Image.Image | None
|
|
best_crop: Image.Image | None
|
|
best_sharpness: float
|
|
mean_bird_score: float
|
|
# Wo der Vogel im besten Bild sitzt, relativ. Die Sammlung schneidet
|
|
# daraus selbst zu -- siehe web/dex/zuschnitt.py.
|
|
best_box: tuple[float, float, float, float] | None = None
|
|
|
|
@property
|
|
def duration_s(self) -> float:
|
|
return (self.ended_at - self.started_at).total_seconds()
|
|
|
|
def summary(self) -> str:
|
|
who = self.species.german if self.species else "unbestimmt"
|
|
return (
|
|
f"{self.started_at:%H:%M:%S} {who} ({self.confidence:.0%}) "
|
|
f"{self.duration_s:.1f}s, {self.frames_with_bird}/{self.frames_total} Frames"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class WatcherConfig:
|
|
classify_every: int = 3
|
|
"""Nur jedes n-te bewegte Bild klassifizieren. Der Vogel ändert sich
|
|
zwischen zwei Frames kaum, das Modell aber kostet 25 ms."""
|
|
|
|
cooldown_s: float = 4.0
|
|
"""So lange ohne Bewegung, bis der Besuch als beendet gilt."""
|
|
|
|
max_visit_s: float = 120.0
|
|
"""Notbremse: dauerhafte Bewegung (Regen, schwankender Ast) soll nicht zu
|
|
einer stundenlangen Sichtung führen."""
|
|
|
|
min_bird_frames: int = 2
|
|
"""Weniger Treffer als das -> verwerfen. Ein einzelnes Frame ist Rauschen."""
|
|
|
|
min_confidence: float = 0.45
|
|
"""Darunter wird die Sichtung als 'unbestimmt' gespeichert statt geraten."""
|
|
|
|
min_mean_bird_score: float = 0.0
|
|
"""Mittlerer Vogelwert ueber den ganzen Besuch; darunter kein Besuch.
|
|
|
|
⚠ Nicht zu verwechseln mit ``Pipeline.min_bird_score``: der gilt je
|
|
EINZELBILD, dieser fuer den DURCHSCHNITT. Genau diese Luecke liess
|
|
Zufallstreffer durch -- ein einziges Bild von 73 mit einem Hauch
|
|
Vogelwert reichte fuer einen Besuch.
|
|
|
|
📌 Vorgabe 0: Der Wert will gemessen und nicht geraten sein. Aus den
|
|
Daten vom 01.10.2026 lagen echte Sichtungen bei 0,269 bis 0,600,
|
|
eine Befestigungsschraube bei 0,098 und Mondspiegelungen bei 0,000.
|
|
Die Web-App gibt den Wert je Kamera vor (``/api/config/``).
|
|
"""
|
|
|
|
gate: str = "motion"
|
|
"""Was einen Besuch beginnt und beendet — ``"motion"`` oder ``"detector"``.
|
|
|
|
``motion`` ist der sparsame Weg: Frame-Differencing kostet unter 1 ms und
|
|
hält das Modell aus dem Spiel, solange sich nichts rührt. Das setzt aber
|
|
einen ruhigen Hintergrund voraus.
|
|
|
|
``detector`` lässt den Vogeldetektor selbst entscheiden und ignoriert
|
|
Bewegung ganz. Nötig, sobald die Kamera auf etwas dauerhaft Bewegtes
|
|
schaut — eine Hecke im Wind löst sonst rund um die Uhr aus. Kostet bei
|
|
4 fps nur wenige Prozent CPU und ist damit fast immer bezahlbar.
|
|
"""
|
|
|
|
save_debug: bool = False
|
|
|
|
|
|
class Watcher:
|
|
def __init__(
|
|
self,
|
|
pipeline: Pipeline | None = None,
|
|
motion: MotionDetector | None = None,
|
|
config: WatcherConfig | None = None,
|
|
beobachter=None,
|
|
):
|
|
"""``beobachter`` wird nach jeder Analyse mit ``(box, result)``
|
|
aufgerufen — auch wenn nichts gefunden wurde. Damit kann ein
|
|
Protokoll mitschreiben, ohne dass die Zustandsmaschine davon weiß."""
|
|
self.pipeline = pipeline or Pipeline()
|
|
self.motion = motion or MotionDetector()
|
|
self.cfg = config or WatcherConfig()
|
|
self.beobachter = beobachter
|
|
|
|
self._reset_visit()
|
|
|
|
def _reset_visit(self) -> None:
|
|
self._active = False
|
|
self._start: datetime | None = None
|
|
self._last_motion = 0.0
|
|
self._votes: dict[str, float] = defaultdict(float)
|
|
self._species_by_slug: dict[str, Species] = {}
|
|
self._frames = 0
|
|
self._bird_frames = 0
|
|
self._best_sharp = -1.0
|
|
self._best_image: Image.Image | None = None
|
|
self._best_crop: Image.Image | None = None
|
|
self._best_box: tuple[float, float, float, float] | None = None
|
|
self._bird_scores: list[float] = []
|
|
self._frame_i = 0
|
|
|
|
def process(self, frame: Image.Image) -> Sighting | None:
|
|
"""Ein Bild einspeisen. Gibt eine Sichtung zurück, wenn ein Besuch
|
|
gerade zu Ende gegangen ist, sonst None."""
|
|
if self.cfg.gate == "detector":
|
|
return self._process_detector_gated(frame)
|
|
return self._process_motion_gated(frame)
|
|
|
|
def _process_motion_gated(self, frame: Image.Image) -> Sighting | None:
|
|
now = time.monotonic()
|
|
ratio = self.motion.update(frame)
|
|
moving = self.motion.is_motion(ratio)
|
|
|
|
if not self._active:
|
|
if not moving:
|
|
return None
|
|
self._active = True
|
|
self._start = datetime.now(timezone.utc)
|
|
self._last_motion = now
|
|
|
|
self._frames += 1
|
|
if moving:
|
|
self._last_motion = now
|
|
|
|
self._frame_i += 1
|
|
if self._frame_i % self.cfg.classify_every == 0:
|
|
self._analyse(frame)
|
|
|
|
return self._maybe_close(now)
|
|
|
|
def _process_detector_gated(self, frame: Image.Image) -> Sighting | None:
|
|
"""Der Vogeldetektor entscheidet selbst, wann ein Besuch läuft.
|
|
|
|
Für Kameras, die auf etwas dauerhaft Bewegtes schauen. Jedes n-te Bild
|
|
geht durchs Modell, unabhängig davon ob sich etwas bewegt hat — Wind in
|
|
der Hecke ist damit schlicht kein Thema mehr.
|
|
"""
|
|
now = time.monotonic()
|
|
self._frame_i += 1
|
|
if self._frame_i % self.cfg.classify_every != 0:
|
|
if self._active:
|
|
self._frames += 1
|
|
return None
|
|
|
|
before = self._bird_frames
|
|
# Beim ersten Treffer eines Besuchs steht der Startzeitpunkt noch nicht;
|
|
# _analyse zählt _bird_frames hoch, daran erkennen wir den Vogel.
|
|
pending_start = datetime.now(timezone.utc)
|
|
self._analyse(frame)
|
|
saw_bird = self._bird_frames > before
|
|
|
|
if not self._active:
|
|
if not saw_bird:
|
|
# Kein Vogel und kein laufender Besuch: die Zwischenergebnisse
|
|
# aus _analyse gehören zu nichts und müssen weg.
|
|
self._reset_visit()
|
|
return None
|
|
self._active = True
|
|
self._start = pending_start
|
|
self._last_motion = now
|
|
|
|
self._frames += 1
|
|
if saw_bird:
|
|
self._last_motion = now
|
|
|
|
return self._maybe_close(now)
|
|
|
|
def _maybe_close(self, now: float) -> Sighting | None:
|
|
too_long = self._start is not None and (
|
|
(datetime.now(timezone.utc) - self._start).total_seconds()
|
|
> self.cfg.max_visit_s
|
|
)
|
|
if (now - self._last_motion) > self.cfg.cooldown_s or too_long:
|
|
return self._close()
|
|
return None
|
|
|
|
def _analyse(self, frame: Image.Image) -> None:
|
|
box, result = self.pipeline.run(frame)
|
|
if self.beobachter is not None:
|
|
# Auch Beinahe-Treffer melden — das Protokoll lebt genau davon.
|
|
self.beobachter(box, result)
|
|
if not self.pipeline.zaehlt_als_vogel(box):
|
|
return
|
|
self._bird_frames += 1
|
|
|
|
if result is not None:
|
|
self._bird_scores.append(result.bird_score)
|
|
self._vote(box, result)
|
|
|
|
# Bestes Foto: Schärfe, gewichtet mit der Detektor-Sicherheit. Ein
|
|
# gestochen scharfes Bild von einem halb verdeckten Vogel nützt weniger
|
|
# als ein leicht weicheres, auf dem er ganz zu sehen ist.
|
|
# Schaerfe am ENGEN Ausschnitt messen: Umgebung soll die Bewertung
|
|
# nicht verwaessern. Gespeichert wird dann der weitere -- siehe
|
|
# Box.anzeige_ausschnitt.
|
|
crop = box.crop(frame)
|
|
score = sharpness(crop) * (0.5 + 0.5 * box.score)
|
|
if score > self._best_sharp:
|
|
self._best_sharp = score
|
|
self._best_image = frame.copy()
|
|
self._best_crop = box.anzeige_ausschnitt(frame)
|
|
self._best_box = (box.x0, box.y0, box.x1, box.y1)
|
|
|
|
def _vote(self, box: Box, result: Result) -> None:
|
|
# Gewicht: wie sicher ist der Detektor, wie groß ist der Vogel im Bild,
|
|
# und wie viel Wahrscheinlichkeitsmasse liegt überhaupt auf heimischen
|
|
# Arten. Ein 20-Pixel-Fleck am Bildrand soll die Abstimmung nicht
|
|
# gewinnen.
|
|
size_w = min(1.0, box.area / 0.05)
|
|
weight = box.score * (0.4 + 0.6 * size_w) * result.bird_score
|
|
for guess in result.guesses[:3]:
|
|
self._votes[guess.species.slug] += weight * guess.confidence
|
|
self._species_by_slug[guess.species.slug] = guess.species
|
|
|
|
def _close(self) -> Sighting | None:
|
|
if self._bird_frames < self.cfg.min_bird_frames:
|
|
self._reset_visit()
|
|
return None
|
|
|
|
# ⚠ Der Durchschnitt ueber den ganzen Besuch, nicht das beste Bild.
|
|
# Hier faellt der Zufallstreffer durch: ein Bild von 73 mit einem
|
|
# Hauch Vogelwert zieht den Mittelwert kaum nach oben. Lieber hier
|
|
# verwerfen als senden -- ein Bild hochzuladen, das die Web-App
|
|
# ohnehin abweist, kostet eine Kamera am Gartenzaun echte Bandbreite.
|
|
mittel = (
|
|
sum(self._bird_scores) / len(self._bird_scores)
|
|
if self._bird_scores else 0.0
|
|
)
|
|
if mittel < self.cfg.min_mean_bird_score:
|
|
self._reset_visit()
|
|
return None
|
|
|
|
species: Species | None = None
|
|
confidence = 0.0
|
|
total = sum(self._votes.values())
|
|
if total > 0:
|
|
slug, score = max(self._votes.items(), key=lambda kv: kv[1])
|
|
confidence = score / total
|
|
if confidence >= self.cfg.min_confidence:
|
|
species = self._species_by_slug[slug]
|
|
|
|
sighting = Sighting(
|
|
started_at=self._start or datetime.now(timezone.utc),
|
|
ended_at=datetime.now(timezone.utc),
|
|
species=species,
|
|
confidence=confidence,
|
|
votes={k: v / total for k, v in self._votes.items()} if total else {},
|
|
frames_total=self._frames,
|
|
frames_with_bird=self._bird_frames,
|
|
best_image=self._best_image,
|
|
best_crop=self._best_crop,
|
|
best_box=self._best_box,
|
|
best_sharpness=max(0.0, self._best_sharp),
|
|
mean_bird_score=mittel,
|
|
)
|
|
self._reset_visit()
|
|
return sighting
|
|
|
|
def flush(self) -> Sighting | None:
|
|
"""Laufenden Besuch abschließen — beim Beenden des Dienstes."""
|
|
return self._close() if self._active else None
|