bmw-cardata 0.1.0a0

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

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)

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

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