Hi everyone, I made some scripts in python and openhab rules to get my data of the car. it is likely the same as BredmichMichael but for me it is easier adjustable in python so her is how I did it. Just so you know it was with help of a good friend claude 
,
First get your client_id like described above or in the Connect to new BMW CarData service through MQTT
Then I made a uv environment that is in my directory where the dir rules, scripts, items… are, for me /etc/openhab/. I have 3 files in the scripts directory, 2 python and one txt that is made once you run the tokens script.
BMW_Tokens.py
import os
import base64
import hashlib
import requests
import time
# =========================================================
# CONFIGURATION
# =========================================================
BMW_DEVICE_CODE_URL = "https://customer.bmwgroup.com/gcdm/oauth/device/code"
BMW_TOKEN_URL = "https://customer.bmwgroup.com/gcdm/oauth/token"
TOKENS_FILE = "bmw_tokens.txt"
# =========================================================
# HELPER FUNCTIONS
# =========================================================
def generate_code_verifier(length: int = 64) -> str:
"""Generate PKCE code_verifier."""
return base64.urlsafe_b64encode(os.urandom(length)).rstrip(b'=').decode('utf-8')
def generate_code_challenge(verifier: str) -> str:
"""Generate S256 code_challenge from code_verifier."""
digest = hashlib.sha256(verifier.encode('ascii')).digest()
return base64.urlsafe_b64encode(digest).rstrip(b'=').decode('utf-8')
def save_tokens(tokens):
"""Save tokens to file for later use."""
with open(TOKENS_FILE, "w") as f:
f.write("BMW ConnectedDrive API Tokens\n")
f.write("=" * 70 + "\n\n")
f.write(f"Access Token: {tokens.get('access_token')}\n")
f.write(f"Refresh Token: {tokens.get('refresh_token')}\n")
f.write(f"ID Token: {tokens.get('id_token')}\n")
f.write(f"Token Type: {tokens.get('token_type')}\n")
f.write(f"Expires In: {tokens.get('expires_in')} seconds\n")
f.write(f"Scope: {tokens.get('scope')}\n")
f.write(f"GCID: {tokens.get('gcid')}\n")
f.write(f"Client ID: {CLIENT_ID}\n")
f.write(f"\nGenerated at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
def load_tokens():
"""Load tokens from file if present."""
if not os.path.exists(TOKENS_FILE):
return None
tokens = {}
with open(TOKENS_FILE, "r") as f:
for line in f:
if ":" in line:
key, value = line.split(":", 1)
tokens[key.strip()] = value.strip()
return tokens
def print_curl(url, headers, data):
print("\n📋 Equivalent cURL (for debugging):")
print("curl --request POST \\")
print(f" '{url}' \\")
for k, v in headers.items():
print(f" -H '{k}: {v}' \\")
for k, v in data.items():
print(f" -d '{k}={v}' \\")
def refresh_access_token(client_id, refresh_token):
"""Refresh access and ID tokens using refresh_token."""
print("\n🔄 Refreshing token using refresh_token...")
payload = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id
}
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
resp = requests.post(BMW_TOKEN_URL, data=payload, headers=headers)
if resp.ok:
tokens = resp.json()
save_tokens(tokens)
print("✅ Tokens refreshed and saved.")
return tokens
else:
print("❌ Refresh failed:", resp.text)
return None
# =========================================================
# STEP 1 — GET CLIENT ID & PKCE
# =========================================================
print("🚗 BMW CarData Device Code Flow")
print("=" * 70)
# Try to load CLIENT_ID from file first
CLIENT_ID = None
if os.path.exists(TOKENS_FILE):
try:
with open(TOKENS_FILE, "r") as f:
for line in f:
if line.strip().startswith("Client ID:"):
CLIENT_ID = line.split(":", 1)[1].strip().lower()
break
except Exception as e:
print(f"⚠️ Could not read Client ID from file: {e}")
# If not found in file, ask user
if not CLIENT_ID:
CLIENT_ID = input("Enter your BMW Client ID: ").strip().lower()
if not CLIENT_ID:
print("❌ Client ID is required.")
exit(1)
else:
print(f"✅ Loaded Client ID from {TOKENS_FILE}")
# Check if refresh token exists
existing_tokens = load_tokens()
if existing_tokens and "Refresh Token" in existing_tokens:
print("✅ Found existing refresh token. Refreshing tokens...")
new_tokens = refresh_access_token(CLIENT_ID, existing_tokens["Refresh Token"])
if new_tokens:
exit(0)
# Generate PKCE
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
print("\n✅ Generated code_verifier and code_challenge")
print("STEP 1 Verifier:", code_verifier)
print("STEP 1 Challenge:", code_challenge)
# =========================================================
# STEP 2 — REQUEST DEVICE CODE
# =========================================================
print("\n" + "=" * 70)
print("🔐 STEP 2: Requesting device & user code")
print("=" * 70)
device_code_payload = {
"client_id": CLIENT_ID,
"response_type": "device_code",
"scope": "authenticate_user openid cardata:api:read cardata:streaming:read",
"code_challenge": code_challenge,
"code_challenge_method": "S256"
}
device_code_headers = {
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded"
}
resp = requests.post(BMW_DEVICE_CODE_URL, data=device_code_payload, headers=device_code_headers)
if resp.status_code != 200:
print("❌ Device Code request failed")
print("Status Code:", resp.status_code)
print("Response Body:", resp.text)
print_curl(BMW_DEVICE_CODE_URL, device_code_headers, device_code_payload)
exit(1)
device_response = resp.json()
user_code = device_response["user_code"]
device_code = device_response["device_code"]
verification_uri = device_response.get("verification_uri_complete") or device_response.get("verification_uri")
interval = device_response.get("interval", 5)
expires_in = device_response.get("expires_in", 300)
print("\n✅ Device Code successfully retrieved!")
print("=" * 70)
print(f"📱 User Code: {user_code}")
print(f"🔐 Device Code: {device_code}")
print(f"🌐 Verification URI: {verification_uri}")
print(f"⏳ Expires In: {expires_in} seconds")
print(f"🔁 Poll Interval: {interval} seconds")
print("=" * 70)
print("\n👉 Please open this URL in your browser to authenticate your BMW account:")
print(f"{verification_uri}")
input("⏳ Press Enter after you finished authorizing the device in the browser...")
print("STEP 2 Verifier (reused):", code_verifier)
# =========================================================
# STEP 3 — EXCHANGE DEVICE CODE FOR TOKENS
# =========================================================
print("\n⏳ Waiting 15 seconds for BMW servers to process authorization...")
time.sleep(15)
print("\n" + "=" * 70)
print("🔑 STEP 3: Exchanging device code for tokens")
print("=" * 70)
token_headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json"
}
token_payload = {
"client_id": CLIENT_ID,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"code_verifier": code_verifier
}
# Better formatted debug output
print("\n🔍 Request Details:")
print(f" URL: {BMW_TOKEN_URL}")
print(f" Method: POST")
print(f" Headers: {token_headers}")
print(f" Payload: {token_payload}")
# Or as a formatted string
from urllib.parse import urlencode
print(f"\n🔗 As URL query string (for reference): {BMW_TOKEN_URL}?{urlencode(token_payload)}")
token_resp = requests.post(BMW_TOKEN_URL, data=token_payload, headers=token_headers)
if token_resp.status_code != 200:
print("❌ Token request failed")
print("Status Code:", token_resp.status_code)
print("Response Body:", token_resp.text)
print_curl(BMW_TOKEN_URL, token_headers, token_payload)
exit(1)
tokens = token_resp.json()
print("\n✅ Tokens successfully received!")
print("=" * 70)
print(f"Access Token: {tokens.get('access_token', '')[:50]}...")
print(f"Refresh Token: {tokens.get('refresh_token', '')[:50]}...")
print(f"ID Token: {tokens.get('id_token', '')[:50]}...")
print(f"Token Type: {tokens.get('token_type')}")
print(f"Expires In: {tokens.get('expires_in')} seconds")
print(f"Scope: {tokens.get('scope')}")
print(f"GCID: {tokens.get('gcid')}")
print("=" * 70)
# Save tokens
save_tokens(tokens)
print(f"💾 Tokens saved to {TOKENS_FILE}")
# =========================================================
# STEP 4 — TOKEN USAGE INSTRUCTIONS
# =========================================================
print("\n📚 Token Usage Guide")
print("=" * 70)
print("🔹 Access Token:")
print(" - Use for BMW CarData REST API calls")
print(" - Header: Authorization: Bearer <access_token>")
print(" - Valid for 1 hour")
print()
print("🔹 ID Token:")
print(" - Use for streaming vehicle data")
print(" - Valid for 1 hour")
print()
print("🔹 Refresh Token:")
print(" - Valid for 2 weeks")
print(" - Use to get new tokens without re-authenticating")
print("=" * 70)
# =========================================================
# STEP 5 — TEST API CALL (EXAMPLE)
# =========================================================
print("\n🧪 Testing API Access")
print("=" * 70)
# Extract the access token
access_token = tokens.get('access_token')
# Set up headers with Bearer token
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# Example: Get vehicles list (adjust the endpoint based on BMW API documentation)
api_url = "https://cardata.bmwgroup.com/api/v1/vehicles" # Replace with actual endpoint
try:
api_response = requests.get(api_url, headers=headers)
if api_response.status_code == 200:
print("✅ API call successful!")
print("Response:", api_response.json())
else:
print(f"❌ API call failed with status {api_response.status_code}")
print("Response:", api_response.text)
except Exception as e:
print(f"❌ Error making API call: {e}")
print("=" * 70)
then run it the first time out of openhab. sudo -u openhab /etc/openhab/bmw_venv/bin/python3 /etc/openhab/scripts/BMW_Token.py that will make the tokens file
It asks to go to the bmw website to confirm the code if it is done then press enter in terminal it will wait for 15s and then get the next set of tokens
second file : BMW_refresh_mqtt.py
#!/usr/bin/env python3
"""
Update MQTT broker password in OpenHAB things file with new BMW ID token
"""
import re
import os
# File paths
TOKEN_FILE = "/etc/openhab/scripts/bmw_tokens.txt"
MQTT_THINGS_FILE = "/etc/openhab/things/BMW_mqtt.things"
def read_id_token():
"""Read the ID token from the tokens file"""
try:
with open(TOKEN_FILE, 'r') as f:
content = f.read()
# Find the ID Token line
match = re.search(r'ID Token:\s*(.+)', content)
if match:
return match.group(1).strip()
else:
print("Error: ID Token not found in token file")
return None
except FileNotFoundError:
print(f"Error: Token file not found: {TOKEN_FILE}")
return None
except Exception as e:
print(f"Error reading token file: {e}")
return None
def update_mqtt_password(new_token):
"""Update the password in the MQTT things file"""
try:
# Read the current file
with open(MQTT_THINGS_FILE, 'r') as f:
content = f.read()
# Replace the password using regex
# This matches the password parameter and replaces its value
pattern = r'(password=")[^"]*(")'
replacement = r'\1' + new_token + r'\2'
new_content = re.sub(pattern, replacement, content)
# Check if any replacement was made
if new_content == content:
print("Warning: No password field found to update")
return False
# Write the updated content back
with open(MQTT_THINGS_FILE, 'w') as f:
f.write(new_content)
print("MQTT password updated successfully in things file")
return True
except FileNotFoundError:
print(f"Error: MQTT things file not found: {MQTT_THINGS_FILE}")
return False
except Exception as e:
print(f"Error updating MQTT things file: {e}")
return False
def main():
"""Main function"""
print("=" * 70)
print("BMW MQTT Password Updater")
print("=" * 70)
# Step 1: Read the new ID token
print("\n1. Reading ID token from file...")
id_token = read_id_token()
if not id_token:
print("Failed to read ID token. Exiting.")
return 1
print(f" ID Token found (length: {len(id_token)} chars)")
# Step 2: Update the password
print("\n2. Updating MQTT password in things file...")
if update_mqtt_password(id_token):
print("\n✓ MQTT password updated successfully!")
print(" Note: OpenHAB will reload the configuration automatically.")
return 0
else:
print("\n✗ Failed to update MQTT password")
return 1
if __name__ == "__main__":
exit(main())
then you need a rule (first part) for refreshing the tokens and updating the pasword on the mqtt file later
The second rule changes the raw data from mqtt to an item you select in here I made an example for the batterysoc of my car
rule "Refresh BMW Tokens Hourly"
when
Time cron "48 0 * * * ?" or // Every hour at minute 45
System started
then
logInfo("BMW Token Refresh", "Starting token refresh by running BMW_Tokens.py")
// Step 1: Get new tokens
val result = executeCommandLine(Duration.ofSeconds(30), "/bin/bash", "-c", "cd /etc/openhab/scripts && source /etc/openhab/bmw_venv/bin/activate && python BMW_Tokens.py")
if (result !== null) {
logInfo("BMW Token Refresh", "Tokens refreshed successfully: " + result)
// Wait a moment for the token file to be updated
Thread::sleep(2000)
// Step 2: Update MQTT password in things file
logInfo("BMW Token Refresh", "Updating MQTT broker password...")
val updateResult = executeCommandLine(Duration.ofSeconds(30), "/bin/bash", "-c", "cd /etc/openhab/scripts && source /etc/openhab/bmw_venv/bin/activate && python BMW_refresh_mqtt.py")
if (updateResult !== null) {
logInfo("BMW Token Refresh", "MQTT password updated: " + updateResult)
} else {
logError("BMW Token Refresh", "Failed to update MQTT password")
}
} else {
logError("BMW Token Refresh", "Token refresh failed or timed out")
}
end
rule "Parse BMW iX2 MQTT Data"
when
Item BMW_iX2_RawData changed
then
val String jsonData = BMW_iX2_RawData.state.toString()
// Parse the JSON
val data = transform("JSONPATH", "$.data", jsonData)
val timestamp = transform("JSONPATH", "$.timestamp", jsonData)
val vin = transform("JSONPATH", "$.vin", jsonData)
// Update VIN if received
if (vin !== null && !vin.equals("null")) {
BMW_iX2_VIN.postUpdate(vin)
}
// Update timestamp
if (timestamp !== null && !timestamp.equals("null")) {
BMW_iX2_LastUpdate.postUpdate(timestamp)
}
// Check which field is in this message and extract it
if (jsonData.contains("vehicle.drivetrain.batteryManagement.header")) {
val value = transform("JSONPATH", "$.data['vehicle.drivetrain.batteryManagement.header'].value", jsonData)
if (value !== null && !value.equals("null")) {
iX2BatterySoc.postUpdate(value as Number)
logInfo("BMW iX2", "Battery Level: " + value + "%")
}
}
end
then the BMW_mqtt.things
Bridge mqtt:broker:iX2_bmw_streaming "BMW ConnectedDrive Stream" [ host="customer.streaming-cardata.bmwgroup.com", port=9000, secure=true, certificationpin=false, mqttVersion="V5", publickeypin=false, enableDiscovery = false, clientid="openhab_client_stream_1985", username="your_gcid", password="you_id_token", qos=1,
keepAlive=60]{ Thing topic bmw_vehicle_iX2 "iX2 BMW Vehicle Data" { Channels: Type string : raw_data "Raw Vehicle Data" [ stateTopic="your_gcid/your_VIN/#" ] }
also in the a items file
String BMW_iX2_RawData "BMW iX2 Raw Data \[%s\]" { channel="mqtt:topic:iX2_bmw_streaming:bmw_vehicle_iX2:raw_data" }
I think you have it all. If I forgot something or something isn’t working let me know. 