diff --git a/docs/teslemetry.md b/docs/teslemetry.md index fb8edef..e676511 100644 --- a/docs/teslemetry.md +++ b/docs/teslemetry.md @@ -159,3 +159,186 @@ async def main(): asyncio.run(main()) ``` + +## Get Streaming Fields + +The `fields` method retrieves streaming field parameters and metadata. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + response = await teslemetry.fields() + print(response) + +asyncio.run(main()) +``` + +## Get Vehicle Configuration + +The `vehicle_config` method retrieves the saved vehicle configuration for a specific vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vin = "" + + response = await teslemetry.vehicle_config(vin) + print(response) + +asyncio.run(main()) +``` + +## Get Streaming Configuration + +The `streaming_config` method retrieves the streaming configuration for a specific vehicle, including certificate, hostname, port, and configurable telemetry fields. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vin = "" + + response = await teslemetry.streaming_config(vin) + print(response) + +asyncio.run(main()) +``` + +## Stop Streaming + +The `stop_streaming` method stops streaming data from a specific vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vin = "" + + response = await teslemetry.stop_streaming(vin) + print(response) + +asyncio.run(main()) +``` + +## Modify Streaming Configuration + +The `modify_streaming_config` method modifies the streaming configuration for a specific vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vin = "" + + # Configure specific fields to stream + fields = { + "BatteryLevel": { + "interval_seconds": 60, + "minimum_delta": 0.1 + }, + "VehicleSpeed": { + "interval_seconds": 30, + "minimum_delta": 0.5 + } + } + + response = await teslemetry.modify_streaming_config(vin, fields) + print(response) + +asyncio.run(main()) +``` + +## Create Streaming Configuration + +The `create_streaming_config` method creates or updates the streaming configuration for a specific vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vin = "" + + # Configure fields to stream + fields = { + "BatteryLevel": { + "interval_seconds": 60 + }, + "Location": { + "interval_seconds": 120 + } + } + + response = await teslemetry.create_streaming_config(vin, fields) + print(response) + +asyncio.run(main()) +``` + +## Vehicle Custom Commands + +### Clear PIN to Drive + +The `clear_pin_to_drive` method deactivates PIN to Drive on the vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vehicle = teslemetry.vehicles.create("") + + # Clear PIN to Drive with your 4-digit PIN + response = await vehicle.clear_pin_to_drive("1234") + print(response) + +asyncio.run(main()) +``` + +### Remove All Impermanent Keys + +The `remove_key` method removes all impermanent keys from the vehicle. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + teslemetry = Teslemetry( + session=session, + access_token="", + ) + + vehicle = teslemetry.vehicles.create("") + + response = await vehicle.remove_key() + print(response) + +asyncio.run(main()) +``` diff --git a/tesla_fleet_api/teslemetry/teslemetry.py b/tesla_fleet_api/teslemetry/teslemetry.py index 128541b..ddc087e 100644 --- a/tesla_fleet_api/teslemetry/teslemetry.py +++ b/tesla_fleet_api/teslemetry/teslemetry.py @@ -122,3 +122,75 @@ async def migrate_to_oauth( new_token["expires_in"] = int(new_token["expires_in"]) new_token["expires_at"] = time() + new_token["expires_in"] return new_token + + async def fields(self) -> dict[str, Any]: + """Get streaming field parameters and metadata.""" + return await self._request( + Method.GET, + "fields.json", + ) + + async def vehicle_config(self, vin: str) -> dict[str, Any]: + """Get the saved vehicle configuration. + + Args: + vin: Vehicle identification number + """ + return await self._request( + Method.GET, + f"api/vehicle_config/{vin}", + ) + + async def streaming_config(self, vin: str) -> dict[str, Any]: + """Get the streaming configuration for a specific vehicle. + + Returns certificate, hostname, port, and configurable telemetry fields. + + Args: + vin: Vehicle identification number + """ + return await self._request( + Method.GET, + f"api/config/{vin}", + ) + + async def stop_streaming(self, vin: str) -> dict[str, Any]: + """Stop streaming data from a specific vehicle. + + Args: + vin: Vehicle identification number + """ + return await self._request( + Method.DELETE, + f"api/config/{vin}", + ) + + async def modify_streaming_config( + self, vin: str, fields: dict[str, Any] + ) -> dict[str, Any]: + """Modify the streaming configuration for a specific vehicle. + + Args: + vin: Vehicle identification number + fields: Fields to stream with their configuration + """ + return await self._request( + Method.PATCH, + f"api/config/{vin}", + json={"fields": fields}, + ) + + async def create_streaming_config( + self, vin: str, fields: dict[str, Any] + ) -> dict[str, Any]: + """Create/update the streaming configuration for a specific vehicle. + + Args: + vin: Vehicle identification number + fields: Fields to stream with their configuration + """ + return await self._request( + Method.POST, + f"api/config/{vin}", + json={"fields": fields}, + ) diff --git a/tesla_fleet_api/teslemetry/vehicles.py b/tesla_fleet_api/teslemetry/vehicles.py index 61860d2..e00da22 100644 --- a/tesla_fleet_api/teslemetry/vehicles.py +++ b/tesla_fleet_api/teslemetry/vehicles.py @@ -260,6 +260,25 @@ async def start_light_show( json=data, ) + async def clear_pin_to_drive(self, pin: str) -> dict[str, Any]: + """Deactivates PIN to Drive. + + Args: + pin: 4-digit PIN to clear + """ + return await self._request( + Method.POST, + f"api/1/vehicles/{self.vin}/custom_command/clear_pin_to_drive", + json={"pin": pin}, + ) + + async def remove_key(self) -> dict[str, Any]: + """Remove all impermanent keys from the vehicle.""" + return await self._request( + Method.POST, + f"api/1/vehicles/{self.vin}/custom_command/remove_key", + ) + class TeslemetryVehicles(Vehicles): """Class containing and creating vehicles."""