Fahrterkennung: FMM003-Funklöcher reißen keine Fahrt mehr auseinander

Die Zündungs-Entität stammt vom FMM003-Tracker; verliert er unterwegs
kurz die Verbindung, meldet Home Assistant "unavailable" statt eines
echten Zündungszustands. Die Live-Erkennung wertete das bisher wie
"aus" - eine Funklücke über der Pausenzeit teilte eine durchgehende
Fahrt in zwei. zuendung_geaendert() ignoriert "unavailable"/"unknown"
jetzt vollständig und leitet "fährt gerade" aus dem eigenen
Zwischenstand (fahrt_start_ts) statt aus dem letzten Rohwert ab.

historienimport.py bekommt zusätzlich eine Plausibilitätsprüfung: eine
errechnete Durchschnittsgeschwindigkeit über 300 km/h deutet auf einen
Kilometerstand-Ausreißer an der Fahrtgrenze hin, nicht auf eine echte
Fahrt - die Strecke wird dann verworfen statt eine unmögliche Fahrt
anzuzeigen.

Tankerkennung: optionaler zweiter Sensor für das Tankvolumen in Litern
(TANK_LITER_SENSOR, direkt vom CAN) neben dem bisherigen
Prozent-Füllstand. Ist er zugeordnet, übernimmt er die automatische
Tankerkennung vollständig - genauer als der Umweg über das im
Fahrzeugprofil hinterlegte Tankvolumen, und der erkannte Anstieg
liefert gleich eine grobe Vorbelegung für die getankte Menge statt
eines leeren Feldes. Ohne den Sensor bleibt alles beim Alten. Der
historische Import zieht mit derselben Präferenz nach.

Manifest auf 2026.8.25.2, beide Änderungsrunden live im Testcontainer
verifiziert (py_compile, Neustart, sauberes Setup-Log).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-25 22:04:26 +02:00
parent f97ceab3ad
commit 70253e83a5
7 changed files with 295 additions and 37 deletions
@@ -60,9 +60,18 @@ class Sensorzuordnung:
# Kilometerstand der EU-Data-Act-Integration ist der richtige.
KM_SENSOR: str = ""
# Tankfüllstand in Prozent - für die automatische Tankerkennung.
# Tankfüllstand in Prozent - für die automatische Tankerkennung und die
# Anzeige in der Übersicht.
TANK_SENSOR: str = ""
# Tankvolumen in Litern, direkt vom CAN (nicht aus dem Prozentwert
# umgerechnet). Optional - ist er gesetzt, übernimmt er die automatische
# Tankerkennung von TANK_SENSOR: ein Anstieg misst sich dann direkt in
# Litern statt über den Umweg Prozentpunkte -> Tankvolumen aus dem
# Fahrzeugprofil, und der erkannte Anstieg selbst dient als grobe
# Vorbelegung für die getankte Menge (siehe tankerkennung.py).
TANK_LITER_SENSOR: str = ""
# Reichweite (Übersicht).
RANGE_SENSOR: str = ""
@@ -144,6 +153,11 @@ FELDER: list[dict] = [
"hinweis": "Füllstand in Prozent - für die automatische Tankerkennung.",
"domains": ["sensor"], "device_classes": [], "units": ["%"], "liste": False, "pflicht": False,
"stichworte": ["tank", "fuel", "kraftstoff", "füllstand", "level"]},
{"key": "TANK_LITER_SENSOR", "label": "Tankvolumen (Liter)", "gruppe": "fahrterkennung",
"hinweis": "Optional, vom CAN statt aus dem Prozentwert umgerechnet - macht die automatische "
"Tankerkennung genauer und liefert eine Vorbelegung für die getankte Menge.",
"domains": ["sensor"], "device_classes": ["volume"], "units": ["L", "l"], "liste": False, "pflicht": False,
"stichworte": ["tankvolumen", "liter", "litres", "fuel volume", "kraftstoffvolumen"]},
{"key": "RANGE_SENSOR", "label": "Reichweite", "gruppe": "uebersicht",
"hinweis": "Für die Übersicht - bleibt sie leer, zeigt die Oberfläche \"unbekannt\".",
"domains": ["sensor"], "device_classes": ["distance"], "units": ["km", "mi"], "liste": False, "pflicht": False,
@@ -78,24 +78,39 @@ async def pausenzeit_sekunden(k: Koordinator) -> int:
async def zuendung_geaendert(k: Koordinator, neu: str | None, alt: str | None) -> None:
"""Reagiert auf jede Zustandsänderung der Zündungs-Entität."""
"""Reagiert auf jede Zustandsänderung der Zündungs-Entität.
"unavailable"/"unknown" ist kein Zündungszustand, sondern Funkstille des
Trackers - beim FMM003 z. B. in Tiefgaragen oder Funklöchern, auch mitten
in einer laufenden Fahrt. Würde das wie "aus" gewertet, risse jede solche
Lücke eine einzelne Fahrt in zwei auseinander, sobald sie länger als die
Pausenzeit dauert (der Tracker meldet sich zurück, die Zündung steht laut
letztem echten Wert immer noch auf "on", also sieht es wie eine neue
Fahrt aus). Ohne Signal ist deshalb "keine Aussage", nicht "aus": der
Zustand bleibt unangetastet, bis wieder ein echter Wert kommt."""
if neu not in ("on", "off"):
return
# Eine noch wartende Ende-Bestätigung aus einer vorherigen Änderung
# abbrechen - das ist der Mechanismus hinter der Pausenregel.
k.warte_ende_ab_abbrechen()
an_jetzt = neu == "on"
an_vorher = alt == "on"
# Ob eine Fahrt läuft, entscheidet der eigene Zwischenstand
# (fahrt_start_ts) - nicht der vorherige Rohwert der Entität. Der wäre bei
# einer Funklücke "unavailable" statt "on" gewesen, obwohl die Fahrt die
# ganze Zeit lief.
fahrt_laeuft = k.fahrt_start_ts is not None
if an_jetzt and not an_vorher:
if k.fahrt_start_ts is None:
await k.fahrt_start_setzen(datetime.datetime.now(datetime.UTC))
_LOGGER.info("Fahrt gestartet um %s", k.fahrt_start_ts)
if an_jetzt and not fahrt_laeuft:
await k.fahrt_start_setzen(datetime.datetime.now(datetime.UTC))
_LOGGER.info("Fahrt gestartet um %s", k.fahrt_start_ts)
# Die Anzeige "fährt/steht" hängt an derselben Entität - sofort neu
# veröffentlichen, statt bis zum nächsten 20-Sekunden-Takt zu warten.
await k.fahrzeugstatus_veroeffentlichen()
return
if an_vorher and not an_jetzt and k.fahrt_start_ts is not None:
if not an_jetzt and fahrt_laeuft:
start_ts = k.fahrt_start_ts
abbruch_ts = datetime.datetime.now(datetime.UTC)
wartezeit = await pausenzeit_sekunden(k)
@@ -42,7 +42,7 @@ import logging
from typing import TYPE_CHECKING
from .fahrterkennung import leere_fahrt, pausenzeit_sekunden
from .tankerkennung import leerer_tankvorgang, schwelle_prozent
from .tankerkennung import LITER_SCHWELLE, leerer_tankvorgang, schwelle_prozent
from .verlauf import Verlaufspunkt, verlauf_lesen, wert_bei, zahl
if TYPE_CHECKING:
@@ -60,6 +60,15 @@ TANK_DUBLETTE_MIN = 90
# fluten sie den Bestand aber mit Nulleinträgen. Bewusst konservativ.
MINDESTDAUER_S = 60
# Kein Auto dieser Art erreicht diesen Schnitt. Eine errechnete
# Durchschnittsgeschwindigkeit darüber ist kein Beleg für eine schnelle
# Fahrt, sondern für einen fehlerhaften Kilometerstand-Sprung am Fahrtrand
# (z. B. ein Ausreißer im Verlauf oder ein knapp daneben liegender
# wert_bei()-Treffer über eine Lücke hinweg). Die Strecke wird dann verworfen
# statt eine physikalisch unmögliche Fahrt anzuzeigen - die Fahrt bleibt
# "offen", genau wie bei fehlendem Kilometerstand.
UNPLAUSIBLE_KMH = 300
def als_zeit(wert: object) -> datetime.datetime | None:
"""Akzeptiert, was die Oberfläche schickt: ISO mit oder ohne Zeitzone.
@@ -162,6 +171,16 @@ async def _fahrten_importieren(k: Koordinator, verlaeufe: dict) -> dict:
durchschnitt = None
if distanz is not None and dauer_s > 0:
durchschnitt = round(distanz / (dauer_s / 3600.0), 1)
if durchschnitt > UNPLAUSIBLE_KMH:
_LOGGER.warning(
"Kilometerstand für Fahrt %s bis %s verworfen: unmögliche %s km/h "
"im Schnitt (%s km in %s s) - vermutlich ein Ausreißer im Verlauf. "
"Die Fahrt wird trotzdem angelegt, bleibt aber ohne Strecke.",
f_start.isoformat(), f_ende.isoformat(), durchschnitt, distanz, dauer_s,
)
distanz = None
odo_start = odo_end = None
durchschnitt = None
fahrt = leere_fahrt(f_start, f_ende, "import")
fahrt.update({
@@ -185,15 +204,19 @@ async def _fahrten_importieren(k: Koordinator, verlaeufe: dict) -> dict:
async def _tankvorgaenge_importieren(k: Koordinator, verlaeufe: dict) -> dict:
"""Tankvorgänge aus dem Füllstandsverlauf - dieselbe Tiefststand-Logik wie
in der Live-Erkennung: jeder Anstieg über die Schwelle gegen den zuletzt
gesehenen Tiefststand ist ein Tankvorgang, nicht jeder Anstieg gegen den
unmittelbar vorherigen Wert."""
verlauf = verlaeufe["tank"]
"""Tankvorgänge aus dem Füllstands- oder Tankvolumenverlauf - dieselbe
Tiefststand-Logik wie in der Live-Erkennung (tankerkennung.py): jeder
Anstieg über die Schwelle gegen den zuletzt gesehenen Tiefststand ist ein
Tankvorgang, nicht jeder Anstieg gegen den unmittelbar vorherigen Wert.
Dieselbe Präferenz wie live auch: ist ein Litersensor zugeordnet,
übernimmt der - genauer, und der Anstieg dient gleich als grobe
Vorbelegung für die getankte Menge."""
mit_liter = bool(k.zuordnung.werte.TANK_LITER_SENSOR)
verlauf = verlaeufe["tank_liter"] if mit_liter else verlaeufe["tank"]
if not verlauf:
return {"angelegt": 0, "uebersprungen": 0}
schwelle = schwelle_prozent(await k.ablage.profil_lesen())
schwelle = LITER_SCHWELLE if mit_liter else schwelle_prozent(await k.ablage.profil_lesen())
km_verlauf = verlaeufe["km"]
bestehende = await k.ablage.tankvorgaenge_lesen()
fenster_s = TANK_DUBLETTE_MIN * 60
@@ -218,7 +241,8 @@ async def _tankvorgaenge_importieren(k: Koordinator, verlaeufe: dict) -> dict:
if tiefststand is None or aktuell <= tiefststand:
tiefststand = aktuell
continue
if aktuell - tiefststand < schwelle:
anstieg = aktuell - tiefststand
if anstieg < schwelle:
continue
if any(abs((bekannt - ts).total_seconds()) < fenster_s for bekannt in bekannte_zeiten):
@@ -229,6 +253,8 @@ async def _tankvorgaenge_importieren(k: Koordinator, verlaeufe: dict) -> dict:
odometer_km = wert_bei(km_verlauf, ts)
tankvorgang = leerer_tankvorgang(ts.isoformat(), "import")
tankvorgang["odometer_km"] = odometer_km
if mit_liter:
tankvorgang["liters"] = round(anstieg, 1)
tankvorgang["distance_km"] = _distanz_zum_vorherigen(
ts.isoformat(), odometer_km, neue + bestehende
)
@@ -339,6 +365,7 @@ async def importieren(k: Koordinator, start: object, ende: object) -> None:
"zuendung": await verlauf_lesen(k.hass, werte.ZUENDUNG_SENSOR, von, bis),
"km": await verlauf_lesen(k.hass, werte.KM_SENSOR, von, bis),
"tank": await verlauf_lesen(k.hass, werte.TANK_SENSOR, von, bis),
"tank_liter": await verlauf_lesen(k.hass, werte.TANK_LITER_SENSOR, von, bis),
"batterie": await verlauf_lesen(k.hass, werte.BATTERIE_SENSOR, von, bis),
"lat": await verlauf_lesen(k.hass, werte.STANDORT_LAT_SENSOR, von, bis),
"lon": await verlauf_lesen(k.hass, werte.STANDORT_LON_SENSOR, von, bis),
@@ -99,6 +99,7 @@ class Koordinator:
self._store = Store[dict](hass, STORE_VERSION, f"{entry.entry_id}_laufzeit")
self.fahrt_start_ts: datetime.datetime | None = None
self.tiefststand_pct: float | None = None
self.tiefststand_liter: float | None = None
# Ergebnis des letzten "Auf Update prüfen"/"Update installieren" -
# nur auf Tastendruck gesetzt (siehe dienste.py), nie im normalen
# Veröffentlichungs-Takt neu berechnet, damit keine automatischen
@@ -198,6 +199,7 @@ class Koordinator:
self._beobachten(werte.ZUENDUNG_SENSOR, self._zuendung_geaendert)
self._beobachten(werte.KM_SENSOR, self._kilometerstand_geaendert)
self._beobachten(werte.TANK_SENSOR, self._tankfuellstand_geaendert)
self._beobachten(werte.TANK_LITER_SENSOR, self._tankvolumen_geaendert)
def _beobachten(
self, entity_id: str, rueckruf: Callable[[Event[EventStateChangedData]], Coroutine[Any, Any, None]]
@@ -229,6 +231,10 @@ class Koordinator:
neu, _alt = self._zustaende(ereignis)
await tankerkennung.fuellstand_geaendert(self, neu)
async def _tankvolumen_geaendert(self, ereignis: Event[EventStateChangedData]) -> None:
neu, _alt = self._zustaende(ereignis)
await tankerkennung.tankvolumen_geaendert(self, neu)
# ------------------------------------------------- Laufende Fahrt merken
async def _laufzeit_laden(self) -> None:
@@ -243,11 +249,13 @@ class Koordinator:
except ValueError:
self.fahrt_start_ts = None
self.tiefststand_pct = gespeichert.get("tiefststand_pct")
self.tiefststand_liter = gespeichert.get("tiefststand_liter")
async def _laufzeit_sichern(self) -> None:
await self._store.async_save({
"fahrt_start_ts": self.fahrt_start_ts.isoformat() if self.fahrt_start_ts else None,
"tiefststand_pct": self.tiefststand_pct,
"tiefststand_liter": self.tiefststand_liter,
})
async def fahrt_start_setzen(self, ts: datetime.datetime | None) -> None:
@@ -258,6 +266,10 @@ class Koordinator:
self.tiefststand_pct = wert
await self._laufzeit_sichern()
async def tiefststand_liter_setzen(self, wert: float | None) -> None:
self.tiefststand_liter = wert
await self._laufzeit_sichern()
def warte_ende_ab(self, koroutine: Coroutine[Any, Any, None]) -> None:
self.warte_ende_ab_abbrechen()
self._ende_aufgabe = self.entry.async_create_background_task(
@@ -1,7 +1,7 @@
{
"domain": "audi_dashboard",
"name": "Audi Dashboard",
"version": "2026.8.24.16",
"version": "2026.8.25.2",
"documentation": "https://gitea.nothaft.cloud/paul/audi-app/src/branch/main/README.md",
"issue_tracker": "https://gitea.nothaft.cloud/paul/audi-app/issues",
"codeowners": ["@paul"],
@@ -1,23 +1,35 @@
"""Automatische Tankerkennung über den Füllstandssensor.
"""Automatische Tankerkennung über den Füllstandssensor - oder, wenn
zugeordnet, direkt über einen Tankvolumen-Sensor in Litern.
Beobachtung: der Füllstand (in Prozent) springt beim Fahren nie nach oben, er
sinkt nur - jeder Anstieg ist also ein Tankvorgang. Schwelle nach Vorgabe: ab
+5 Liter ODER +9 Prozentpunkten Anstieg gilt als nachgetankt. Die
5-Liter-Vorgabe wird über fahrzeug.tankvolumen_liter aus dem Fahrzeugprofil in
Prozentpunkte umgerechnet, damit beide Angaben auf derselben Einheit
verglichen werden können; es gilt jeweils die empfindlichere (kleinere) der
beiden Schwellen.
Beobachtung: der Füllstand springt beim Fahren nie nach oben, er sinkt nur -
jeder Anstieg ist also ein Tankvorgang. Zwei Datenquellen dafür, in
absteigender Genauigkeit:
Tiefststand-Tracking statt einfachem Vorher/Nachher-Vergleich: die Datenquelle
liefert einen Tankvorgang oft in mehreren kleinen Schritten (40% -> 45% ->
60%), von denen keiner allein die Schwelle überschreiten muss. Deshalb wird
der zuletzt bekannte Tiefststand gespeichert und der Anstieg dagegen gemessen.
Nach Anlage eines Tankvorgangs wird der Tiefststand auf den aktuellen Wert
zurückgesetzt, damit derselbe Vorgang nicht mehrfach Datensätze erzeugt.
- TANK_LITER_SENSOR (optional, z. B. vom CAN): Anstieg direkt in Litern,
Schwelle LITER_SCHWELLE. Braucht keine Umrechnung und liefert den Anstieg
selbst gleich als Vorbelegung für die getankte Menge mit.
- TANK_SENSOR (Prozent, siehe fuellstand_geaendert()): Schwelle nach Vorgabe
ab +5 Liter ODER +9 Prozentpunkten Anstieg, wobei die 5-Liter-Vorgabe über
fahrzeug.tankvolumen_liter aus dem Fahrzeugprofil in Prozentpunkte
umgerechnet wird (beide Angaben so auf derselben Einheit); es gilt jeweils
die empfindlichere (kleinere) der beiden Schwellen.
Ist der Litersensor zugeordnet, übernimmt er die Erkennung vollständig -
beide gleichzeitig auszuwerten würde denselben Tankvorgang zweimal anlegen.
Ohne ihn bleibt es beim Prozentsensor wie bisher.
Tiefststand-Tracking statt einfachem Vorher/Nachher-Vergleich, für beide
Quellen gleichermaßen: die Datenquelle liefert einen Tankvorgang oft in
mehreren kleinen Schritten (40% -> 45% -> 60%), von denen keiner allein die
Schwelle überschreiten muss. Deshalb wird der zuletzt bekannte Tiefststand
gespeichert und der Anstieg dagegen gemessen. Nach Anlage eines Tankvorgangs
wird der Tiefststand auf den aktuellen Wert zurückgesetzt, damit derselbe
Vorgang nicht mehrfach Datensätze erzeugt.
Der so angelegte Tankvorgang ist bewusst ein unvollständiger Platzhalter
(status "unvollständig", ohne Liter/Kosten/Station): erwartet wird ein
automatisch erkannter Tankvorgang mit Zeitstempel, Kilometerstand und
(status "unvollständig", ohne Kosten/Station, Liter nur als grobe Schätzung
vom Litersensor oder ganz ohne bei reiner Prozent-Erkennung): erwartet wird
ein automatisch erkannter Tankvorgang mit Zeitstempel, Kilometerstand und
gefahrener Distanz, den der Nutzer später per Beleg-Nachtrag vervollständigt.
Auch hier gilt die Härtung aus fahrterkennung.py: der Tiefststand lebt nicht
@@ -97,21 +109,58 @@ async def fuellstand_geaendert(k: Koordinator, neu: str | None) -> None:
if anstieg < schwelle_prozent(await k.ablage.profil_lesen()):
return
await _automatisch_anlegen(k, anstieg)
# Ist ein direkter Litersensor zugeordnet, übernimmt der die Erkennung
# (siehe tankvolumen_geaendert) - genauer, weil ohne den Umweg über das im
# Fahrzeugprofil hinterlegte Tankvolumen. Beide gleichzeitig auswerten
# würde denselben Tankvorgang zweimal anlegen.
if not k.zuordnung.werte.TANK_LITER_SENSOR:
await _automatisch_anlegen(k, f"Füllstandsanstieg {round(anstieg, 1)} Prozentpunkte")
await k.tiefststand_setzen(aktuell)
async def _automatisch_anlegen(k: Koordinator, anstieg_pct: float) -> None:
async def tankvolumen_geaendert(k: Koordinator, neu: str | None) -> None:
"""Wie fuellstand_geaendert(), aber direkt in Litern statt in Prozent -
braucht deshalb keine Umrechnung über das Fahrzeugprofil und liefert mit
dem gemessenen Anstieg selbst eine sinnvolle Vorbelegung für die getankte
Menge (siehe leerer_tankvorgang() - "liters" bleibt sonst None, bis der
Nutzer sie von Hand einträgt oder ein Beleg sie liefert)."""
aktuell = _als_zahl(neu)
if aktuell is None:
return
tiefststand = k.tiefststand_liter
if tiefststand is None or aktuell <= tiefststand:
await k.tiefststand_liter_setzen(aktuell)
return
anstieg = aktuell - tiefststand
if anstieg < LITER_SCHWELLE:
return
await _automatisch_anlegen(
k, f"Tankvolumenanstieg {round(anstieg, 1)} l", liter_schaetzung=anstieg
)
await k.tiefststand_liter_setzen(aktuell)
async def _automatisch_anlegen(
k: Koordinator, anlass: str, liter_schaetzung: float | None = None
) -> None:
odometer_km = _als_zahl(zustand_oder_none(k.hass, k.zuordnung.werte.KM_SENSOR))
tankvorgang = leerer_tankvorgang(
datetime.datetime.now(datetime.UTC).isoformat(), "auto"
)
tankvorgang["odometer_km"] = odometer_km
tankvorgang["distance_km"] = await k.ablage.distanz_seit_letzter_tankung(odometer_km)
if liter_schaetzung is not None:
# Grobe Vorbelegung, keine feststehende Menge - der Status bleibt
# "unvollständig" (siehe leerer_tankvorgang), bis Preis/Tankstelle
# dazukommen. Der Nutzer korrigiert sie bei Bedarf selbst, ein
# nachgetragener Beleg überschreibt sie ohnehin.
tankvorgang["liters"] = round(liter_schaetzung, 1)
await k.ablage.tankvorgang_anhaengen(tankvorgang)
await k.tankvorgaenge_veroeffentlichen()
_LOGGER.info(
"Tankvorgang %s automatisch erkannt (Füllstandsanstieg %s Prozentpunkte)",
tankvorgang["tank_id"], round(anstieg_pct, 1),
"Tankvorgang %s automatisch erkannt (%s)", tankvorgang["tank_id"], anlass
)