Merge pull request #26 from Teslemetry/claude/teslemetry-api-endpoints-5bi2h

Add missing Teslemetry API endpoints

Author: Bre77

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="<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="<access_token>",
+        )
+
+        vin = "<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="<access_token>",
+        )
+
+        vin = "<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="<access_token>",
+        )
+
+        vin = "<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="<access_token>",
+        )
+
+        vin = "<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="<access_token>",
+        )
+
+        vin = "<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="<access_token>",
+        )
+
+        vehicle = teslemetry.vehicles.create("<vin>")
+
+        # 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="<access_token>",
+        )
+
+        vehicle = teslemetry.vehicles.create("<vin>")
+
+        response = await vehicle.remove_key()
+        print(response)
+
+asyncio.run(main())
+```

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},
+        )

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."""