bmw-cardata 0.1.0a3

Async Python client for the BMW CarData customer API.

bmw-cardata - BMW CarData Python Library

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

[!IMPORTANT] The CarData REST API is rate-limited to 50 requests per day per user (BMW-side limit, not enforced by this library). Once exceeded, calls return HTTP 403 CU-429 API rate limit reached until the next day. For frequent updates use the MQTT streaming feed instead.

Prerequisites (one-time setup in the BMW customer portal)

Before this library can talk to the API you must complete a few steps in the BMW CarData customer portal:

1. Generate a CarData client ID — Portal → CarData → Create CarData Client. Note the client ID; you will pass it to DeviceCodeFlowClient.
2. Subscribe to services — subscribe the client to both CarData API and CarData Stream. Without this the OAuth scopes (cardata:api:read, cardata:streaming:read) will not be granted and the API/stream calls will be rejected.
3. For REST telematic data: create a container selecting the telematic keys you want. See spec/telematic_catalogue.json for valid technical_descriptor values. You can also create containers from code via CarDataClient.create_container(...).
4. For MQTT streaming: open Configure data stream and pick the streamable keys (those with "streamable": true in spec/telematic_catalogue.json). Without this step the MQTT broker will reject the subscribe with code:128.

Notes:

• The REST API has a daily rate limit of 50 requests per user. Prefer streaming for frequent updates.
• Only one streaming connection per gcid is allowed at a time.
• Refresh tokens are valid for 14 days; access/id tokens for 1 hour.

Authentication (OAuth 2.0 Device Code Flow with PKCE)

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)
        print("refresh_token:", token.refresh_token)


asyncio.run(main())

Persist the refresh_token and use it to resume the session without prompting the user again:

auth = DeviceCodeFlowClient(client_id="...", refresh_token="<stored-refresh-token>")
token = await auth.refresh()  # new access/id/refresh tokens

REST client

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):

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.

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:

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

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

License

MIT