Add metadata listeners

Author: Bre77

README.md

@@ -18,7 +18,8 @@ The TeslemetryStream class requires:
 - session: an aiohttp.ClientSession
 - access_token: an access token from the [Teslemetry console](https://teslemetry.com/console)
 - vin: If you only want to use a single vehicle, otherwise use `create_vehicle`
-- server: The Teslemetry server to connect to, otherwise use `find_server`
+- server: Override the Teslemetry server to connect to:
+  - api.teslemetry.com (recommended and default)
   - na.teslemetry.com
   - eu.teslemetry.com
 
@@ -60,7 +61,6 @@ async def main():
         async with TeslemetryStream(
             access_token="<token>",
             vin="<vin>", # for single vehicles
-            server="na.teslemetry.com" # or "eu.teslemetry.com"
             session=session,
         ) as stream:
 
@@ -175,6 +175,25 @@ Replace Fleet Telemetry configuration for the vehicle.
 ### `config(self) -> dict`
 Return current configuration for the vehicle.
 
+### `listen_State(callback: Callable[[bool], None]) -> Callable[[],None]`
+Listen for vehicle online state polling. The callback receives a boolean value representing whether the vehicle is online.
+
+### `listen_VehicleData(callback: Callable[[dict], None]) -> Callable[[],None]`
+Listen for vehicle data polling events. The callback receives a dictionary containing the complete vehicle data.
+
+### `listen_Cellular(callback: Callable[[bool], None]) -> Callable[[],None]`
+Listen for cellular connectivity events. The callback receives a boolean value indicating whether the cellular connection is established.
+
+### `listen_Wifi(callback: Callable[[bool], None]) -> Callable[[],None]`
+Listen for WiFi connectivity events. The callback receives a boolean value indicating whether the WiFi connection is established.
+
+### `listen_Alerts(callback: Callable[[list[dict]], None]) -> Callable[[],None]`
+Listen for vehicle alert events. The callback receives a list of dictionaries containing alert information.
+
+### `listen_Errors(callback: Callable[[list[dict]], None]) -> Callable[[],None]`
+Listen for vehicle error events. The callback receives a list of dictionaries containing error information.
+
 ### `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.
+A full list of fields and metadata can be found at [api.teslemetry.com/fields.json](https://api.teslemetry.com/fields.json)

teslemetry_stream/const.py

@@ -10,6 +10,17 @@ class StrEnum(str, Enum):
     """String Enum"""
 
 
+class Key(StrEnum):
+    """Topics available in Fleet Telemetry streams"""
+    VIN = "vin"
+    DATA = "data"
+    ALERTS = "alerts"
+    ERRORS = "errors"
+    VEHICLE_DATA = "vehicle_data"
+    STATE = "state"
+    STATUS = "status"
+    NETWORK_INTERFACE = "networkInterface"
+
 class Signal(StrEnum):
     """Signals available in Fleet Telemetry streams"""
 
@@ -248,6 +259,24 @@ class Alert(StrEnum):
     SERVICE = "Service"
     SERVICE_FIX = "ServiceFix"
 
+class State(StrEnum):
+    """States available in Fleet Telemetry streams"""
+
+    ONLINE = "online"
+    OFFLINE = "offline"
+    ASLEEP = "asleep"
+
+class NetworkInterface(StrEnum):
+    """Network interfaces available in Fleet Telemetry streams"""
+
+    WIFI = "wifi"
+    CELLULAR = "cellular"
+
+class Status(StrEnum):
+    """Statuses available in Fleet Telemetry streams"""
+
+    CONNECTED = "CONNECTED"
+    DISCONNECTED = "DISCONNECTED"
 
 @dataclass
 class TeslaLocation:

teslemetry_stream/vehicle.py

@@ -28,6 +28,7 @@
     HvacAutoModeState,
     HvacPowerState,
     HvilStatus,
+    Key,
     LaneAssistLevel,
     MediaStatus,
     PowershareState,
@@ -38,6 +39,9 @@
     SentryModeState,
     ShiftState,
     Signal,
+    State,
+    Status,
+    NetworkInterface,
     SunroofInstalledState,
     TemperatureUnit,
     TeslaLocation,
@@ -162,6 +166,49 @@ def _enable_field(self, field: Signal) -> None:
         """Enable a field for streaming from a listener."""
         asyncio.create_task(self.add_field(field))
 
+    # Add listeners for other streaming fields
+    def listen_State(self, callback: Callable[[bool], None]) -> Callable[[],None]:
+        """Listen for State polling."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.STATE] == State.ONLINE),
+            {Key.VIN: self.vin, Key.STATE: None}
+        )
+
+    def listen_VehicleData(self, callback: Callable[[dict], None]) -> Callable[[],None]:
+        """Listen for Vehicle Data polling."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.VEHICLE_DATA]),
+            {Key.VIN: self.vin, Key.VEHICLE_DATA: None}
+        )
+
+    def listen_Cellular(self, callback: Callable[[bool], None]) -> Callable[[],None]:
+        """Listen for Cellular connectivity."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.STATUS] == Status.CONNECTED),
+            {Key.VIN: self.vin, Key.NETWORK_INTERFACE: NetworkInterface.CELLULAR}
+        )
+
+    def listen_Wifi(self, callback: Callable[[bool], None]) -> Callable[[],None]:
+        """Listen for WiFi connectivity."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.STATUS] == Status.CONNECTED),
+            {Key.VIN: self.vin, Key.NETWORK_INTERFACE: NetworkInterface.WIFI}
+        )
+
+    def listen_Alerts(self, callback: Callable[[list[dict]], None]) -> Callable[[],None]:
+        """Listen for Alerts."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.ALERTS]),
+            {Key.VIN: self.vin, Key.ALERTS: None}
+        )
+
+    def listen_Errors(self, callback: Callable[[list[dict]], None]) -> Callable[[],None]:
+        """Listen for Errors."""
+        return self.stream.async_add_listener(
+            lambda x: callback(x[Key.ERRORS]),
+            {Key.VIN: self.vin, Key.ERRORS: None}
+        )
+
     # Add listeners for each signal
     def listen_ACChargingEnergyIn(self, callback: Callable[[float | None], None]) -> Callable[[],None]:
         """Listen for AC Charging Energy In."""