From a05abe5ef97f5a75e34ef88e70b70a61e977b289 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Mar 2025 20:51:59 +1000 Subject: [PATCH 1/2] Update README.md with examples and documentation --- For more details, open the [Copilot Workspace session](https://copilot-workspace.githubnext.com/Teslemetry/python-teslemetry-stream?shareId=XXXX-XXXX-XXXX-XXXX). --- README.md | 85 ++++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 75 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index c647de9..7c21559 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,11 @@ # Teslemetry Stream Library -This is an asynchronous Python 3 library that connects to the Teslemetry Stream server and provides Tesla Fleet Telemetry using server side events. +This is an asynchronous Python 3 library that connects to the Teslemetry Stream service and provides Tesla Fleet Telemetry using server sent events. The library allows you to listen to various telemetry signals from Tesla vehicles, and provides a convenient way to handle these signals using typed listen methods. + +## Capabilities +- Connect to the Teslemetry Stream service +- Listen to various telemetry signals from Tesla vehicles +- Handle signals using typed listen methods +- Write custom listeners for multiple signals ## Installation @@ -27,20 +33,79 @@ The following example puts the listening loop in the background, then stopping a ``` async def main(): async with aiohttp.ClientSession() as session: - stream = TeslemetryStream( + async with TeslemetryStream( + access_token="", + vin="", # for single vehicles + server="na.teslemetry.com" # or "eu.teslemetry.com" + session=session, + ) as stream: + + def callback(event): + print(event["data"]) + + remove = stream.async_add_listener(callback) + + print("Running") + await asyncio.sleep(60) + remove() +``` + +## Using Typed Listen Methods + +The library provides typed listen methods for various telemetry signals. These methods allow you to listen to specific signals and handle their data in a type-safe manner. Here is an example of using the typed listen methods: + +```python +async def main(): + async with aiohttp.ClientSession() as session: + async with TeslemetryStream( + access_token="", + vin="", # for single vehicles + server="na.teslemetry.com" # or "eu.teslemetry.com" + session=session, + ) as stream: + + vehicle = stream.get_vehicle("") + + def battery_level_callback(battery_level): + print(f"Battery Level: {battery_level}") + + def vehicle_speed_callback(vehicle_speed): + print(f"Vehicle Speed: {vehicle_speed}") + + remove_battery_level_listener = vehicle.listen_BatteryLevel(battery_level_callback) + remove_vehicle_speed_listener = vehicle.listen_VehicleSpeed(vehicle_speed_callback) + + print("Running") + await asyncio.sleep(60) + remove_battery_level_listener() + remove_vehicle_speed_listener() +``` + +## Writing Your Own Listener with Multiple Signals + +You can also write your own listener that listens to multiple signals. Here is an example of writing a custom listener: + +```python +async def main(): + async with aiohttp.ClientSession() as session: + async with TeslemetryStream( access_token="", vin="", # for single vehicles server="na.teslemetry.com" # or "eu.teslemetry.com" session=session, - ) - await stream.connect() + ) as stream: + + vehicle = stream.get_vehicle("") - def callback(event): - print(event["data"]) + def custom_listener(event): + if "BatteryLevel" in event["data"]: + print(f"Battery Level: {event['data']['BatteryLevel']}") + if "VehicleSpeed" in event["data"]: + print(f"Vehicle Speed: {event['data']['VehicleSpeed']}") - remove = stream.async_add_listener(callback) + remove_custom_listener = stream.async_add_listener(custom_listener, {"vin": "", "data": {"BatteryLevel": None, "VehicleSpeed": None}}) - print("Running") - await asyncio.sleep(60) - remove() + print("Running") + await asyncio.sleep(60) + remove_custom_listener() ``` From 3fa77201326f6afe4e5185c4c81f2d66ccc5cda3 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Mar 2025 20:55:14 +1000 Subject: [PATCH 2/2] * Update `README.md` example to use explicit connect and disconnect methods * **Example Update** - Replace `async with` context syntax with explicit `connect` and `disconnect` methods - Add `stream.connect()` and `stream.disconnect()` calls - Adjust indentation and remove `async with` block - Add `stream` variable assignment outside the context block --- README.md | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 7c21559..0167198 100644 --- a/README.md +++ b/README.md @@ -88,24 +88,28 @@ You can also write your own listener that listens to multiple signals. Here is a ```python async def main(): async with aiohttp.ClientSession() as session: - async with TeslemetryStream( + stream = TeslemetryStream( access_token="", vin="", # for single vehicles server="na.teslemetry.com" # or "eu.teslemetry.com" session=session, - ) as stream: + ) - vehicle = stream.get_vehicle("") + await stream.connect() - def custom_listener(event): - if "BatteryLevel" in event["data"]: - print(f"Battery Level: {event['data']['BatteryLevel']}") - if "VehicleSpeed" in event["data"]: - print(f"Vehicle Speed: {event['data']['VehicleSpeed']}") + vehicle = stream.get_vehicle("") - remove_custom_listener = stream.async_add_listener(custom_listener, {"vin": "", "data": {"BatteryLevel": None, "VehicleSpeed": None}}) + def custom_listener(event): + if "BatteryLevel" in event["data"]: + print(f"Battery Level: {event['data']['BatteryLevel']}") + if "VehicleSpeed" in event["data"]: + print(f"Vehicle Speed: {event['data']['VehicleSpeed']}") - print("Running") - await asyncio.sleep(60) - remove_custom_listener() + remove_custom_listener = stream.async_add_listener(custom_listener, {"vin": "", "data": {"BatteryLevel": None, "VehicleSpeed": None}}) + + print("Running") + await asyncio.sleep(60) + remove_custom_listener() + + await stream.disconnect() ```