Metadata-Version: 2.4
Name: bmw_cardata
Version: 0.1.0a0
Summary: Async Python client for the BMW CarData customer API (REST + MQTT streaming).
Project-URL: Homepage, https://github.com/zweckj/bmw_cardata
Author: Josef Zweck
License: MIT
License-File: LICENSE
Keywords: aiohttp,async,bmw,cardata,mqtt
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: aiohttp>=3.9
Requires-Dist: aiomqtt>=2.3
Requires-Dist: mashumaro>=3.13
Provides-Extra: dev
Requires-Dist: ruff>=0.6; extra == 'dev'
Requires-Dist: ty>=0.0.1a5; extra == 'dev'
Description-Content-Type: text/markdown

# bmw_cardata - BMW CarData Python Library

Async Python client for the **BMW CarData** customer API (REST) and the
customer **MQTT streaming** feed.

## Authentication (OAuth 2.0 Device Code Flow with PKCE)

```python
import asyncio

from bmw_cardata import DeviceCodeFlowClient


async def main() -> None:
    async with DeviceCodeFlowClient(client_id="<your-client-id>") as auth:
        device = await auth.request_device_code()
        print(f"Open {device.verification_uri} and enter code {device.user_code}")
        token = await auth.poll_for_token(device.device_code, interval=device.interval)
        print("access_token:", token.access_token)
        print("id_token:", token.id_token)
        print("gcid:", token.gcid)


asyncio.run(main())
```

## REST client

```python
import asyncio

from bmw_cardata import CarDataClient
from bmw_cardata.models import CreateContainerRequest


async def main() -> None:
    async with CarDataClient(access_token="<bearer-access-token>") as client:
        containers = await client.list_containers()
        for c in containers.containers:
            print(c.container_id, c.name, c.state)

        new = await client.create_container(
            CreateContainerRequest(
                name="my-container",
                purpose="example",
                technical_descriptors=["vehicle.powertrain.electric.battery.stateOfCharge"],
            )
        )
        data = await client.get_telematic_data("WBA...", container_id=new.container_id or "")
        for key, entry in data.telematic_data.items():
            print(key, entry.value, entry.unit, entry.timestamp)


asyncio.run(main())
```

You can also pass an async callable that returns a fresh token (handy for
refresh flows):

```python
async def token_provider() -> str:
    return await refresh_if_needed()

client = CarDataClient(access_token=token_provider)
```

## Streaming (MQTT)

The customer MQTT broker authenticates with the user's `gcid` as username and
the OAuth `id_token` (issued with the `openid` scope) as password.

```python
import asyncio

from bmw_cardata import StreamingClient


async def main() -> None:
    async with StreamingClient(gcid="<gcid>", id_token="<id-token>") as stream:
        async for message in stream.stream():  # all VINs of this gcid
            print(message.vin, message.data)


asyncio.run(main())
```

To subscribe to a single VIN:

```python
async for message in stream.stream(vin="WBA..."):
    ...
```

## Endpoints covered

REST (`CarDataClient`):

- `list_containers`, `create_container`, `get_container`, `delete_container`
- `get_mappings`
- `get_basic_data(vin)`
- `get_telematic_data(vin, container_id)`
- `get_vehicle_image(vin)` → raw bytes
- `get_smart_maintenance_tyre_diagnosis(vin)`
- `get_charging_history(vin, from_, to, next_token=...)`
- `get_location_based_charging_settings(vin, next_token=...)`

Streaming (`StreamingClient`):

- `stream(vin=None)` — async iterator of `StreamingMessage`

## Development

```bash
pip install -e .[dev]
ruff format src
ruff check src
ty check src
```

## License

MIT
