feat: add listen_ActiveRoute* listeners that clear when navigation ends

Port of the Home Assistant Teslemetry listen_active_route helper.

chore: bump version to 1.2.0

Author: Bre77

README.md

@@ -342,6 +342,11 @@ Listen for vehicle error events. The callback receives a list of dictionaries co
 ### `listen_ChargerPower(callback: Callable[[float | None], None]) -> Callable[[],None]`
 Listen for charger power whether the vehicle is AC or DC charging. Combines `ACChargingPower` and `DCChargingPower` (DC preferred), gated by `DetailedChargeState`: the value drops to `0` when charging ends, since neither source is reliably reset then, and a lone zero while charging is held until both sources have reported. Enables all three fields; the returned function removes all three listeners.
 
+### `listen_ActiveRouteDestinationLocation(callback: Callable[[TeslaLocation | None], None]) -> Callable[[],None]`
+### `listen_ActiveRouteTrafficMinutesDelay(callback: Callable[[int | None], None]) -> Callable[[],None]`
+### `listen_ActiveRouteExpectedEnergyPercentAtTripArrival(callback: Callable[[int | None], None]) -> Callable[[],None]`
+Listen for a route field only while navigation is active. The car keeps reporting the last trip's `DestinationLocation`, `RouteTrafficMinutesDelay` and `ExpectedEnergyPercentAtTripArrival` after arriving, but `MinutesToArrival` goes null. Each listener pairs its field with `MinutesToArrival`, applying both from one event together: it reports `None` once `MinutesToArrival` is null, and otherwise the field's value once both have been seen. Enables both fields; the returned function removes all listeners.
+
 ### `listen_*` Methods
 The `TeslemetryStreamVehicle` class contains a `listen_*` methods for each telemetry signal.
 These methods allow you to listen to specific signals and handle their data in a type-safe manner.

pyproject.toml

@@ -4,7 +4,7 @@ requires = ["setuptools>=77.0"]
 
 [project]
 name = "teslemetry_stream"
-version = "1.1.0"
+version = "1.2.0"
 license = "Apache-2.0"
 description = "Teslemetry Streaming API library for Python"
 readme = "README.md"

teslemetry_stream/vehicle.py

@@ -6,7 +6,7 @@
 import logging
 from collections.abc import Callable
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 
 import aiohttp
 
@@ -66,6 +66,9 @@
 LOGGER = logging.getLogger(__package__)
 
 
+_T = TypeVar("_T")
+
+
 class TeslemetryStreamVehicle:
     """Handle streaming field updates."""
 
@@ -3135,6 +3138,79 @@ def _unsubscribe() -> None:
 
         return _unsubscribe
 
+    def _listen_active_route(
+        self,
+        listen: Callable[[Callable[[_T | None], None]], Callable[[], None]],
+        callback: Callable[[_T | None], None],
+    ) -> Callable[[], None]:
+        """Listen for a route field, reporting None while no navigation is active.
+
+        After navigation ends Minutes to Arrival goes null, but the car keeps
+        the last trip's destination location, arrival energy and traffic delay.
+        Reports None once Minutes to Arrival is null, otherwise the value once
+        both have been seen.
+        """
+        value: _T | None = None
+        minutes: float | None = None
+        value_seen = minutes_seen = changed = False
+
+        def _value_callback(new_value: _T | None) -> None:
+            nonlocal value, value_seen, changed
+            value, value_seen, changed = new_value, True, True
+
+        def _minutes_callback(new_minutes: float | None) -> None:
+            nonlocal minutes, minutes_seen, changed
+            minutes, minutes_seen, changed = new_minutes, True, True
+
+        def _event_callback(event: dict[str, Any]) -> None:
+            # Registered last, so both fields from one event are applied
+            # together rather than briefly reporting a value from a route
+            # that just ended
+            nonlocal changed
+            if not changed:
+                return
+            changed = False
+            if minutes_seen and minutes is None:
+                callback(None)
+            elif minutes_seen and value_seen:
+                callback(value)
+
+        unsubs = (
+            listen(_value_callback),
+            self.listen_MinutesToArrival(_minutes_callback),
+            self.stream.async_add_listener(
+                _event_callback, {"vin": self.vin, "data": {}}
+            ),
+        )
+
+        def _unsubscribe() -> None:
+            for unsub in unsubs:
+                unsub()
+
+        return _unsubscribe
+
+    def listen_ActiveRouteDestinationLocation(
+        self, callback: Callable[[TeslaLocation | None], None]
+    ) -> Callable[[], None]:
+        """Listen for Destination Location during navigation."""
+        return self._listen_active_route(self.listen_DestinationLocation, callback)
+
+    def listen_ActiveRouteTrafficMinutesDelay(
+        self, callback: Callable[[int | None], None]
+    ) -> Callable[[], None]:
+        """Listen for Route Traffic Minutes Delay during navigation."""
+        return self._listen_active_route(
+            self.listen_RouteTrafficMinutesDelay, callback
+        )
+
+    def listen_ActiveRouteExpectedEnergyPercentAtTripArrival(
+        self, callback: Callable[[int | None], None]
+    ) -> Callable[[], None]:
+        """Listen for Expected Energy Percent at Trip Arrival during navigation."""
+        return self._listen_active_route(
+            self.listen_ExpectedEnergyPercentAtTripArrival, callback
+        )
+
 
 def make_int(
     signal: Signal, callback: Callable[[int | None], None]

tests/test_active_route.py

@@ -0,0 +1,141 @@
+"""Tests for the listen_ActiveRoute* composite listeners.
+
+After navigation ends Minutes to Arrival goes null, but the car keeps
+reporting the last trip's destination, arrival energy and traffic delay. The
+composites pair each route field with Minutes to Arrival. Scenarios mirror the
+Home Assistant Teslemetry tests the helper was ported from.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from teslemetry_stream.const import Signal, TeslaLocation
+from teslemetry_stream.stream import TeslemetryStream
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
+MINUTES = Signal.MINUTES_TO_ARRIVAL
+DELAY = Signal.ROUTE_TRAFFIC_MINUTES_DELAY
+ENERGY = Signal.EXPECTED_ENERGY_PERCENT_AT_TRIP_ARRIVAL
+DESTINATION = Signal.DESTINATION_LOCATION
+LOCATION = {"latitude": -27.824252, "longitude": 153.328079}
+
+
+def check(label: str, ok: bool, detail: str = "") -> bool:
+    print(f"{label:<72} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")
+    return ok
+
+
+def make_vehicle() -> tuple[TeslemetryStream, TeslemetryStreamVehicle]:
+    stream = TeslemetryStream(None, "token", manual=True)  # type: ignore[arg-type]
+    vehicle = TeslemetryStreamVehicle(stream, VIN)
+    vehicle.fields = {s.value: {} for s in (MINUTES, DELAY, ENERGY, DESTINATION)}
+    vehicle._populated = True
+    return stream, vehicle
+
+
+def run(listener: str, messages: list[dict[Signal, Any]]) -> list[Any]:
+    """Return every value the callback received, in order."""
+    stream, vehicle = make_vehicle()
+    values: list[Any] = []
+    unsub = getattr(vehicle, listener)(values.append)
+    for message in messages:
+        stream.ingest({k.value: v for k, v in message.items()}, vin=VIN)
+    unsub()
+    return values
+
+
+SEQUENCES: list[tuple[str, str, list[dict[Signal, Any]], list[Any]]] = [
+    (
+        "route_ends_delay",
+        "listen_ActiveRouteTrafficMinutesDelay",
+        [
+            {MINUTES: 12.5, DELAY: 3, ENERGY: 62},
+            {MINUTES: None, DELAY: 0, ENERGY: 62},
+        ],
+        [3, None],
+    ),
+    (
+        "route_ends_energy",
+        "listen_ActiveRouteExpectedEnergyPercentAtTripArrival",
+        [
+            {MINUTES: 12.5, DELAY: 3, ENERGY: 62},
+            {MINUTES: None, DELAY: 0, ENERGY: 62},
+        ],
+        [62, None],
+    ),
+    (
+        "route_restarts",
+        "listen_ActiveRouteTrafficMinutesDelay",
+        [
+            {MINUTES: 12.5, DELAY: 3},
+            {MINUTES: None, DELAY: 0},
+            {MINUTES: 30.0, DELAY: 5},
+        ],
+        [3, None, 5],
+    ),
+    (
+        "route_restarts_with_unchanged_value",
+        "listen_ActiveRouteTrafficMinutesDelay",
+        [{MINUTES: 12.5, DELAY: 3}, {MINUTES: None}, {MINUTES: 20.0}],
+        [3, None, 3],
+    ),
+    (
+        "waits_for_minutes_to_arrival",
+        "listen_ActiveRouteTrafficMinutesDelay",
+        [{DELAY: 3}, {MINUTES: 12.5}],
+        [3],
+    ),
+    (
+        "energy_never_reported",
+        "listen_ActiveRouteExpectedEnergyPercentAtTripArrival",
+        [{MINUTES: 12.5, DELAY: 3}, {MINUTES: None, DELAY: 0}, {MINUTES: 30.0}],
+        [None],
+    ),
+    (
+        "destination_route_ends",
+        "listen_ActiveRouteDestinationLocation",
+        [
+            {DESTINATION: LOCATION, MINUTES: 12.5},
+            {DESTINATION: LOCATION, MINUTES: None},
+        ],
+        [TeslaLocation(**LOCATION), None],
+    ),
+    (
+        "unrelated_event_ignored",
+        "listen_ActiveRouteTrafficMinutesDelay",
+        [{MINUTES: 12.5, DELAY: 3}, {Signal.VEHICLE_SPEED: 40}],
+        [3],
+    ),
+]
+
+
+async def main() -> None:
+    results: list[bool] = []
+    for name, listener, messages, expected in SEQUENCES:
+        got = run(listener, messages)
+        results.append(check(f"sequence {name}", got == expected, f"got {got!r}"))
+
+    stream, vehicle = make_vehicle()
+    before = len(stream._listeners)
+    unsub = vehicle.listen_ActiveRouteDestinationLocation(lambda v: None)
+    added = len(stream._listeners) - before
+    unsub()
+    results.append(
+        check(
+            "one unsubscribe removes all three listeners",
+            added == 3 and len(stream._listeners) == before,
+            f"added {added}, left {len(stream._listeners) - before}",
+        )
+    )
+    await asyncio.sleep(0)
+
+    print("-" * 72)
+    print("ALL PASS" if all(results) else "FAILURES PRESENT")
+    if not all(results):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())