Ortsnamen aus dem Backend, Verbrauchsfaktor aus dem Tankbeleg, Radzaehler ueber Fahrten

Ortsauflösung serverseitig (geokodierung.py): der Ort steht jetzt in der Fahrt
statt im Zwischenspeicher jedes Geräts. Beide Oberflächen fragen nichts mehr ab.

Verbrauchskorrektur (verbrauchskorrektur.py): Faktor = Beleg-Liter / Summe der
Einzelfahrten, voll zu voll. Statt einer Schranke am Faktor wird der Nenner
geprüft - decken die erkannten Fahrten 90-110 % der Tacho-Spanne ab? Ausgelöst
beim Schreiben eines Tankvorgangs, nicht erst bei der nächsten Fahrt.

Radzähler: summiert die Fahrten des montierten Satzes statt Tacho-Deltas. Der
alte Weg hatte Fahrzeugwechsel als gefahrene Strecke verbucht (375.962 km bei
einem Tacho von 61.823).

Null heißt unbekannt: ein Kilometerstand von 0 oder ein leeres Feld werden zu
None normalisiert, statt an sechs Lesestellen als "Tachostand null" zu gelten.
Momentanwerte der Sensoren gelten nur für einen Tankvorgang von jetzt.

Oberfläche, beide Codebasen: Wertespalte der Fahrtenliste ausgerichtet, Ort ->
Ort in "Zuletzt", "Räder" statt "Reifen", "Termin vereinbart" entfernt,
Markenlogo 20 % größer, Übersicht-Symbol auf volle Rasterbreite.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-02 11:28:37 +02:00
parent 78a8017e95
commit 2af5c469d6
27 changed files with 1575 additions and 352 deletions
+76 -11
View File
@@ -36,7 +36,7 @@ from . import shell_beleg_parser
from . import tankerkennung
from .tankerkennung import leerer_tankvorgang
from .veroeffentlichung import zustand_oder_none
from .verlauf import zeit_normalisiert
from .verlauf import als_kilometerstand, zeit_normalisiert
if TYPE_CHECKING:
from .koordinator import Koordinator
@@ -169,7 +169,7 @@ async def _nachtragen(
aenderungen = dict(beleg_felder)
aenderungen["status"] = "vollständig"
if await k.ablage.tankvorgang_aktualisieren(tank_id, aenderungen):
await k.tankvorgaenge_veroeffentlichen()
await k.tankvorgang_nachbereiten()
await k.beleg_ergebnis_veroeffentlichen({"erfolg": True, "tank_id": tank_id})
_LOGGER.info("Beleg %s in Tankvorgang %s nachgetragen", receipt_key, tank_id)
else:
@@ -192,17 +192,23 @@ async def tankvorgang_manuell(k: Koordinator, daten: dict) -> None:
_LOGGER.warning("liter und kosten sind Pflichtfelder")
return
km = daten.get("km")
odometer_km = km if km is not None else _km_aus_sensor(k)
ts_roh = daten.get("ts")
ts = zeit_normalisiert(ts_roh) if ts_roh else datetime.datetime.now(datetime.UTC)
# Eine 0 oder ein geleertes Feld heissen "kenne ich nicht" - siehe
# als_kilometerstand().
odometer_km = als_kilometerstand(daten.get("km"))
if odometer_km is None:
odometer_km = _km_aus_sensor_wenn_jetzt(k, ts)
distanz = daten.get("distanz")
if distanz is None:
# Vorschlag aus TANK_DISTANZ_SENSOR oder, ohne ihn, der Differenz zum
# letzten Tankvorgang - aber wie odometer_km selbst frei
# überschreibbar, ein übergebener Wert hat deshalb immer Vorrang.
distanz = await tankerkennung.distanz_seit_tankung(k, odometer_km)
distanz = await tankerkennung.distanz_seit_tankung(
k, odometer_km, jetzt=_ist_jetzt(ts)
)
ts_roh = daten.get("ts")
ts = zeit_normalisiert(ts_roh) if ts_roh else datetime.datetime.now(datetime.UTC)
receipt_key = daten.get("receipt_key")
tankvorgang = leerer_tankvorgang(
ts.isoformat(),
@@ -223,10 +229,15 @@ async def tankvorgang_manuell(k: Koordinator, daten: dict) -> None:
})
await k.ablage.tankvorgang_anhaengen(tankvorgang)
await k.tankvorgaenge_veroeffentlichen()
await k.tankvorgang_nachbereiten()
_LOGGER.info("Tankvorgang %s angelegt", tankvorgang["tank_id"])
# Wie nah der eingetragene Zeitpunkt an "jetzt" liegen muss, damit der
# aktuelle Tachostand als Ersatz taugt.
SENSOR_ERSATZ_FENSTER = datetime.timedelta(hours=2)
def _km_aus_sensor(k: Koordinator) -> float | None:
wert = zustand_oder_none(k.hass, k.zuordnung.werte.KM_SENSOR)
try:
@@ -235,6 +246,39 @@ def _km_aus_sensor(k: Koordinator) -> float | None:
return None
def _ist_jetzt(ts: datetime.datetime | None) -> bool:
"""Liegt der eingetragene Zeitpunkt nah genug an jetzt, dass die
Momentanwerte der Sensoren zu ihm passen?"""
if ts is None:
return False
jetzt = datetime.datetime.now(datetime.UTC)
return abs((jetzt - ts).total_seconds()) <= SENSOR_ERSATZ_FENSTER.total_seconds()
def _km_aus_sensor_wenn_jetzt(
k: Koordinator, ts: datetime.datetime | None
) -> float | None:
"""Der aktuelle Tachostand - aber nur fuer einen Tankvorgang von jetzt.
Der Sensor kennt genau einen Wert: den von diesem Augenblick. Fuer eine
Tankung, die gerade passiert ist, ist das der richtige Ersatz und der
Normalfall - der Nutzer muss nichts eintippen.
Fuer eine nachgetragene Tankung von vorgestern ist es der falsche Wert,
und er richtet zweifachen Schaden an: die gefahrene Distanz wird gegen
einen zu hohen Stand gerechnet, und die Datumspruefung der
Verbrauchskorrektur verwirft das Intervall, weil der gespeicherte Stand
nicht zum Tacho des eingetragenen Zeitpunkts passt. Dann lieber gar kein
Kilometerstand - genau das, was eine 0 im Formular ausdruecken soll."""
if not _ist_jetzt(ts):
_LOGGER.info(
"Tankvorgang auf %s datiert - der aktuelle Tachostand passt dazu "
"nicht und wird nicht eingesetzt", ts.isoformat(),
)
return None
return _km_aus_sensor(k)
async def tankvorgang_aktualisieren(k: Koordinator, daten: dict) -> None:
"""Bearbeitet einen bestehenden Tankvorgang (Einzelbeleg-Seite, egal ob
ursprünglich per Beleg oder manuell angelegt).
@@ -252,6 +296,12 @@ async def tankvorgang_aktualisieren(k: Koordinator, daten: dict) -> None:
_LOGGER.warning("tankvorgang_aktualisieren ohne tank_id aufgerufen")
return
# edited_fields fortschreiben statt ersetzen: was frueher von Hand gesetzt
# wurde, bleibt geschuetzt.
vorhanden = await k.ablage.tankvorgaenge_lesen()
alt = next((t for t in vorhanden if t.get("tank_id") == tank_id), None)
hand = set((alt or {}).get("edited_fields") or [])
aenderungen: dict = {}
if "ts" in daten:
ts_normalisiert = zeit_normalisiert(daten.get("ts"))
@@ -262,8 +312,20 @@ async def tankvorgang_aktualisieren(k: Koordinator, daten: dict) -> None:
aenderungen["liters"] = liter
aenderungen["fuel_total_eur"] = kosten
aenderungen["price_per_l"] = preis_je_liter(kosten, liter)
if liter is not None:
# Von Hand eingetippte Liter sind der Beleg, nur ohne PDF - und
# damit dieselbe Wahrheit wie ein geparster Beleg. Ohne diesen
# Vermerk haelt verbrauchskorrektur.beleg_liter() sie fuer den
# Anstieg des Gebers und verwirft sie als Zirkelschluss; der
# Verbrauchsfaktor entstuende nie.
#
# Nebenwirkung, bewusst in Kauf genommen: edited_fields schuetzt
# das Feld vor der Automatik. Wird spaeter doch noch ein PDF zu
# demselben Tankvorgang hochgeladen, bleibt die getippte Zahl
# stehen - der Mensch hatte den Beleg in der Hand.
hand.add("liters")
if "km" in daten:
aenderungen["odometer_km"] = daten.get("km")
aenderungen["odometer_km"] = als_kilometerstand(daten.get("km"))
if "ersparnis" in daten:
aenderungen["discount"] = daten.get("ersparnis")
if "station" in daten:
@@ -277,8 +339,11 @@ async def tankvorgang_aktualisieren(k: Koordinator, daten: dict) -> None:
aenderungen["receipt_file"] = daten.get("receipt_file")
aenderungen["source"] = "beleg"
if hand:
aenderungen["edited_fields"] = sorted(hand)
if await k.ablage.tankvorgang_aktualisieren(tank_id, aenderungen):
await k.tankvorgaenge_veroeffentlichen()
await k.tankvorgang_nachbereiten()
_LOGGER.info("Tankvorgang %s aktualisiert", tank_id)
else:
_LOGGER.warning("Tankvorgang %s nicht gefunden", tank_id)
@@ -289,7 +354,7 @@ async def tankvorgang_loeschen(k: Koordinator, tank_id: str | None) -> None:
_LOGGER.warning("tankvorgang_loeschen ohne tank_id aufgerufen")
return
if await k.ablage.tankvorgang_loeschen(tank_id):
await k.tankvorgaenge_veroeffentlichen()
await k.tankvorgang_nachbereiten()
_LOGGER.info("Tankvorgang %s gelöscht", tank_id)
else:
_LOGGER.warning("Tankvorgang %s nicht gefunden", tank_id)