Compare commits

...
9 Commits
5 changed files with 215 additions and 228 deletions
+3 -1
View File
@@ -10,6 +10,8 @@ Maybe this component works also with other Volvo cars. Please try out the native
- C40 PHEV (2022) - C40 PHEV (2022)
- XC90 T8 PHEV (2023) - XC90 T8 PHEV (2023)
- XC60 PHEV (2023) - XC60 PHEV (2023)
- XC60 PHEV (2022)
- V90 PHEV T8 (2019)
Please let me know if your car works with this addon so I can expand the list! Please let me know if your car works with this addon so I can expand the list!
@@ -30,7 +32,7 @@ NOTE: Energy status currently available only for cars in the Europe / Middle Eas
Just install this addon with the following command. Just install this addon with the following command.
Please note to fill in your settings inside the environment variables. Please note to fill in your settings inside the environment variables.
`docker run --pull=always -e CONF_updateInterval=300 -e CONF_babelLocale='de' -e CONF_mqtt='@json {"broker": "", "username": "", "password": ""}' -e CONF_volvoData='@json {"username": "", "password": "", "vin": "", "vccapikey": ""}' -e TZ='Europe/Berlin' --name volvo2mqtt ghcr.io/dielee/volvo2mqtt:latest` `docker run -d --pull=always -e CONF_updateInterval=300 -e CONF_babelLocale='de' -e CONF_mqtt='@json {"broker": "", "username": "", "password": ""}' -e CONF_volvoData='@json {"username": "", "password": "", "vin": "", "vccapikey": ""}' -e TZ='Europe/Berlin' --name volvo2mqtt ghcr.io/dielee/volvo2mqtt:latest`
<b>HA Add-On:</b><br> <b>HA Add-On:</b><br>
+1 -1
View File
@@ -1,6 +1,6 @@
name: "Volvo2Mqtt" name: "Volvo2Mqtt"
description: "Volvo AAOS MQTT bridge" description: "Volvo AAOS MQTT bridge"
version: "1.2.2" version: "1.3.0"
slug: "volvo2mqtt" slug: "volvo2mqtt"
init: false init: false
url: "https://github.com/Dielee/volvo2mqtt" url: "https://github.com/Dielee/volvo2mqtt"
+36 -35
View File
@@ -1,6 +1,6 @@
from config import settings from config import settings
VERSION = "v1.2.2" VERSION = "v1.3.0"
OAUTH_URL = "https://volvoid.eu.volvocars.com/as/token.oauth2" OAUTH_URL = "https://volvoid.eu.volvocars.com/as/token.oauth2"
VEHICLES_URL = "https://api.volvocars.com/connected-vehicle/v1/vehicles" VEHICLES_URL = "https://api.volvocars.com/connected-vehicle/v1/vehicles"
@@ -14,43 +14,44 @@ CAR_UNLOCK_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/co
RECHARGE_STATE_URL = "https://api.volvocars.com/energy/v1/vehicles/{0}/recharge-status" RECHARGE_STATE_URL = "https://api.volvocars.com/energy/v1/vehicles/{0}/recharge-status"
ODOMETER_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/odometer" ODOMETER_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/odometer"
LOCATION_STATE_URL = "https://api.volvocars.com/location/v1/vehicles/{0}/location" LOCATION_STATE_URL = "https://api.volvocars.com/location/v1/vehicles/{0}/location"
TYRE_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/tyres"
ENGINE_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/engine-status"
BATTERY_CHARGE_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/battery-charge-level"
charging_system_states = {"CHARGING_SYSTEM_CHARGING": "Charging", "CHARGING_SYSTEM_IDLE": "Idle", charging_system_states = {"CHARGING_SYSTEM_CHARGING": "Charging", "CHARGING_SYSTEM_IDLE": "Idle",
"CHARGING_SYSTEM_FAULT": "Fault", "CHARGING_SYSTEM_UNSPECIFIED": "Unspecified"} "CHARGING_SYSTEM_FAULT": "Fault", "CHARGING_SYSTEM_UNSPECIFIED": "Unspecified"}
supported_sensors = [ charging_connection_states = {"CONNECTION_STATUS_DISCONNECTED": "Disconnected", "CONNECTION_STATUS_UNSPECIFIED": "Unspecified",
{"name": "Battery Charge Level", "id": "battery_charge_level", "unit": "%", "icon": "car-battery", "url": RECHARGE_STATE_URL}, "CONNECTION_STATUS_CONNECTED_DC": "Connected DC", "CONNECTION_STATUS_CONNECTED_AC": "Connected AC"}
{"name": "Electric Range", "id": "electric_range", "unit": "km" if settings["babelLocale"] != "en_US" else "mi", "icon": "map-marker-distance", "url": RECHARGE_STATE_URL},
{"name": "Estimated Charging Time", "id": "estimated_charging_time", "unit": "minutes", "icon": "timer-sync-outline", "url": RECHARGE_STATE_URL},
{"name": "Charging System Status", "id": "charging_system_status", "icon": "ev-plug-ccs2", "url": RECHARGE_STATE_URL},
{"name": "Estimated Charging Finish Time", "id": "estimated_charging_finish_time", "icon": "timer-sync-outline", "url": RECHARGE_STATE_URL},
{"name": "Odometer", "id": "odometer", "unit": "km" if settings["babelLocale"] != "en_US" else "mi", "icon": "counter", "url": ODOMETER_STATE_URL},
{"name": "Last Data Update", "id": "last_data_update", "icon": "timer", "url": ""},
{"name": "Window Front Left", "id": "window_front_left", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Front Right", "id": "window_front_right", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Rear Left", "id": "window_rear_left", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Rear Right", "id": "window_rear_right", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Door Front Left", "id": "door_front_left", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Front Right", "id": "door_front_right", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Rear Left", "id": "door_rear_left", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Rear Right", "id": "door_rear_right", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Tailgate", "id": "tailgate", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Engine Hood", "id": "engine_hood", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Tank Lid", "id": "tank_lid", "icon": "car-door-lock", "url": LOCK_STATE_URL}
]
supported_switches = [ supported_entities = [
{"name": "Air Conditioning", "id": "climate_status", "icon": "air-conditioner"}, {"name": "Battery Charge Level", "domain": "sensor", "id": "battery_charge_level", "unit": "%", "icon": "car-battery", "url": RECHARGE_STATE_URL},
{"name": "Battery Charge Level", "domain": "sensor", "id": "battery_charge_level", "unit": "%", "icon": "car-battery", "url": BATTERY_CHARGE_STATE_URL},
{"name": "Electric Range", "domain": "sensor", "id": "electric_range", "unit": "km" if settings["babelLocale"] != "en_US" else "mi", "icon": "map-marker-distance", "url": RECHARGE_STATE_URL},
{"name": "Estimated Charging Time", "domain": "sensor", "id": "estimated_charging_time", "unit": "minutes", "icon": "timer-sync-outline", "url": RECHARGE_STATE_URL},
{"name": "Charging System Status", "domain": "sensor", "id": "charging_system_status", "icon": "ev-station", "url": RECHARGE_STATE_URL},
{"name": "Charging Connection Status", "domain": "sensor", "id": "charging_connection_status", "icon": "ev-plug-ccs2", "url": RECHARGE_STATE_URL},
{"name": "Estimated Charging Finish Time", "domain": "sensor", "id": "estimated_charging_finish_time", "icon": "timer-sync-outline", "url": RECHARGE_STATE_URL},
{"name": "Odometer", "domain": "sensor", "id": "odometer", "unit": "km" if settings["babelLocale"] != "en_US" else "mi", "icon": "counter", "url": ODOMETER_STATE_URL},
{"name": "Last Data Update", "domain": "sensor", "id": "last_data_update", "icon": "timer", "url": ""},
{"name": "Window Front Left", "domain": "sensor", "id": "window_front_left", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Front Right", "domain": "sensor", "id": "window_front_right", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Rear Left", "domain": "sensor", "id": "window_rear_left", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Window Rear Right", "domain": "sensor", "id": "window_rear_right", "icon": "car-door-lock", "url": WINDOWS_STATE_URL},
{"name": "Door Front Left", "domain": "sensor", "id": "door_front_left", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Front Right", "domain": "sensor", "id": "door_front_right", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Rear Left", "domain": "sensor", "id": "door_rear_left", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Door Rear Right", "domain": "sensor", "id": "door_rear_right", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Tailgate", "domain": "sensor", "id": "tailgate", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Engine Hood", "domain": "sensor", "id": "engine_hood", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Tank Lid", "domain": "sensor", "id": "tank_lid", "icon": "car-door-lock", "url": LOCK_STATE_URL},
{"name": "Air Conditioning", "domain": "switch", "id": "climate_status", "icon": "air-conditioner"},
{"name": "Lock state", "domain": "lock", "id": "lock_status", "icon": "lock", "url": LOCK_STATE_URL},
{"name": "Force Update Data", "domain": "button", "id": "update_data", "icon": "update", "url": ""},
{"name": "Location", "domain": "device_tracker", "id": "location", "icon": "map-marker-radius", "url": LOCATION_STATE_URL},
{"name": "Tyre Front Left", "domain": "sensor", "id": "tyre_front_left", "icon": "car-tire-alert", "url": TYRE_STATE_URL},
{"name": "Tyre Front Right", "domain": "sensor", "id": "tyre_front_right", "icon": "car-tire-alert", "url": TYRE_STATE_URL},
{"name": "Tyre Rear Left", "domain": "sensor", "id": "tyre_rear_left", "icon": "car-tire-alert", "url": TYRE_STATE_URL},
{"name": "Tyre Rear Right", "domain": "sensor", "id": "tyre_rear_right", "icon": "car-tire-alert", "url": TYRE_STATE_URL},
{"name": "Engine State", "domain": "sensor", "id": "engine_state", "icon": "engine", "url": ENGINE_STATE_URL}
] ]
supported_locks = [
{"name": "Lock state", "id": "lock_status", "icon": "lock", "url": LOCK_STATE_URL}
]
supported_buttons = [
{"name": "Force Update Data", "id": "update_data", "icon": "update", "url": ""}
]
supported_device_trackers = [
{"name": "Location", "id": "location", "icon": "map-marker-radius", "url": LOCATION_STATE_URL}
]
+66 -144
View File
@@ -7,9 +7,7 @@ from datetime import datetime
from babel.dates import format_datetime from babel.dates import format_datetime
from config import settings from config import settings
from const import CLIMATE_START_URL, CLIMATE_STOP_URL, CAR_LOCK_URL, \ from const import CLIMATE_START_URL, CLIMATE_STOP_URL, CAR_LOCK_URL, \
CAR_UNLOCK_URL, supported_sensors, supported_buttons, supported_switches, \ CAR_UNLOCK_URL
supported_locks, supported_device_trackers
mqtt_client: mqtt.Client mqtt_client: mqtt.Client
@@ -44,37 +42,41 @@ def on_disconnect(client, userdata, rc):
def on_message(client, userdata, msg): def on_message(client, userdata, msg):
if msg.topic in subscribed_topics: try:
vin = msg.topic.split('/')[2].split('_')[0] vin = msg.topic.split('/')[2].split('_')[0]
payload = msg.payload.decode("UTF-8") except IndexError:
if "climate_status" in msg.topic: print("Error - Cannot get vin from MQTT topic!")
global assumed_climate_state, climate_timer return None
if payload == "ON":
api_thread = Thread(target=volvo.api_call, args=(CLIMATE_START_URL, "POST", vin)) payload = msg.payload.decode("UTF-8")
api_thread.start() if "climate_status" in msg.topic:
assumed_climate_state[vin] = "ON" global assumed_climate_state, climate_timer
# Starting timer to disable climate after 30 mins if payload == "ON":
climate_timer = Timer(1 * 60, volvo.disable_climate, (vin, )) api_thread = Thread(target=volvo.api_call, args=(CLIMATE_START_URL, "POST", vin))
climate_timer.start() api_thread.start()
update_car_data() assumed_climate_state[vin] = "ON"
elif payload == "OFF": # Starting timer to disable climate after 30 mins
api_thread = Thread(target=volvo.api_call, args=(CLIMATE_STOP_URL, "POST", vin)) climate_timer = Timer(1 * 60, volvo.disable_climate, (vin, ))
api_thread.start() climate_timer.start()
assumed_climate_state[vin] = "OFF" update_car_data()
# Stop timer if active elif payload == "OFF":
if climate_timer.is_alive(): api_thread = Thread(target=volvo.api_call, args=(CLIMATE_STOP_URL, "POST", vin))
climate_timer.cancel() api_thread.start()
update_car_data() assumed_climate_state[vin] = "OFF"
elif "lock_status" in msg.topic: # Stop timer if active
if payload == "LOCK": if climate_timer.is_alive():
volvo.api_call(CAR_LOCK_URL, "POST", vin) climate_timer.cancel()
update_car_data() update_car_data()
elif payload == "UNLOCK": elif "lock_status" in msg.topic:
volvo.api_call(CAR_UNLOCK_URL, "POST", vin) if payload == "LOCK":
update_car_data() volvo.api_call(CAR_LOCK_URL, "POST", vin)
elif "update_data" in msg.topic: update_car_data()
if payload == "PRESS": elif payload == "UNLOCK":
update_car_data(True) volvo.api_call(CAR_UNLOCK_URL, "POST", vin)
update_car_data()
elif "update_data" in msg.topic:
if payload == "PRESS":
update_car_data(True)
def update_loop(): def update_loop():
@@ -90,136 +92,56 @@ def update_car_data(force_update=False):
global last_data_update global last_data_update
last_data_update = format_datetime(datetime.now(), format="medium", locale=settings["babelLocale"]) last_data_update = format_datetime(datetime.now(), format="medium", locale=settings["babelLocale"])
for vin in volvo.vins: for vin in volvo.vins:
for device_tracker in supported_device_trackers: for entity in volvo.supported_endpoints[vin]:
state = volvo.api_call(device_tracker["url"], "GET", vin, device_tracker["id"], force_update) if entity["domain"] == "button":
mqtt_client.publish( continue
f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/attributes",
json.dumps(state)
)
for lock in supported_locks: if entity["id"] == "climate_status":
state = volvo.api_call(lock["url"], "GET", vin, lock["id"], force_update)
mqtt_client.publish(
f"homeassistant/lock/{vin}_{lock['id']}/state",
state
)
for switch in supported_switches:
if switch["id"] == "climate_status":
state = assumed_climate_state[vin] state = assumed_climate_state[vin]
else: elif entity["id"] == "last_data_update":
state = "OFF"
mqtt_client.publish(
f"homeassistant/switch/{vin}_{switch['id']}/state",
state
)
for sensor in supported_sensors:
if sensor["id"] == "last_data_update":
state = last_data_update state = last_data_update
else: else:
state = volvo.api_call(sensor["url"], "GET", vin, sensor["id"], force_update) state = volvo.api_call(entity["url"], "GET", vin, entity["id"], force_update)
if entity["domain"] == "device_tracker":
topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes"
else:
topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/state"
mqtt_client.publish( mqtt_client.publish(
f"homeassistant/sensor/{vin}_{sensor['id']}/state", topic,
state json.dumps(state) if isinstance(state, dict) else state
) )
def create_ha_devices(): def create_ha_devices():
global subscribed_topics
for vin in volvo.vins: for vin in volvo.vins:
device = volvo.get_vehicle_details(vin) device = volvo.get_vehicle_details(vin)
for entity in volvo.supported_endpoints[vin]:
for device_tracker in supported_device_trackers:
config = { config = {
"name": device_tracker['name'], "name": entity['name'],
"object_id": f"volvo_{vin}_{device_tracker['id']}", "object_id": f"volvo_{vin}_{entity['id']}",
"schema": "state", "schema": "state",
"icon": f"mdi:{device_tracker['icon']}", "icon": f"mdi:{entity['icon']}",
"state_topic": f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/state", "state_topic": f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/state",
"device": device, "device": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{device_tracker['id']}", "unique_id": f"volvoAAOS2mqtt_{vin}_{entity['id']}"
"json_attributes_topic": f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/attributes"
} }
mqtt_client.publish( if entity.get('unit'):
f"homeassistant/device_tracker/volvoAAOS2mqtt/{vin}_{device_tracker['id']}/config", config["unit_of_measurement"] = entity["unit"]
json.dumps(config),
)
for button in supported_buttons: if entity.get('domain') == "device_tracker":
command_topic = f"homeassistant/button/{vin}_{button['id']}/command" config["json_attributes_topic"] = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes"
config = {
"name": button['name'],
"object_id": f"volvo_{vin}_{button['id']}",
"schema": "state",
"icon": f"mdi:{button['icon']}",
"state_topic": f"homeassistant/button/{vin}_{button['id']}/state",
"command_topic": command_topic,
"device": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{button['id']}",
}
mqtt_client.publish(
f"homeassistant/button/volvoAAOS2mqtt/{vin}_{button['id']}/config",
json.dumps(config),
)
subscribed_topics.append(command_topic)
mqtt_client.subscribe(command_topic)
for lock in supported_locks: if entity.get('domain') in ["switch", "lock", "button"]:
command_topic = f"homeassistant/lock/{vin}_{lock['id']}/command" command_topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/command"
config = { config["command_topic"] = command_topic
"name": lock['name'], subscribed_topics.append(command_topic)
"object_id": f"volvo_{vin}_{lock['id']}", mqtt_client.subscribe(command_topic)
"schema": "state",
"icon": f"mdi:{lock['icon']}",
"state_topic": f"homeassistant/lock/{vin}_{lock['id']}/state",
"command_topic": command_topic,
"optimistic": False,
"device": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{lock['id']}",
}
mqtt_client.publish(
f"homeassistant/lock/volvoAAOS2mqtt/{vin}_{lock['id']}/config",
json.dumps(config),
)
subscribed_topics.append(command_topic)
mqtt_client.subscribe(command_topic)
for switch in supported_switches:
command_topic = f"homeassistant/switch/{vin}_{switch['id']}/command"
config = {
"name": switch['name'],
"object_id": f"volvo_{vin}_{switch['id']}",
"schema": "state",
"icon": f"mdi:{switch['icon']}",
"state_topic": f"homeassistant/switch/{vin}_{switch['id']}/state",
"command_topic": command_topic,
"optimistic": False,
"device": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{switch['id']}",
}
mqtt_client.publish(
f"homeassistant/switch/volvoAAOS2mqtt/{vin}_{switch['id']}/config",
json.dumps(config),
)
subscribed_topics.append(command_topic)
mqtt_client.subscribe(command_topic)
for sensor in supported_sensors:
config = {
"name": sensor['name'],
"object_id": f"volvo_{vin}_{sensor['id']}",
"schema": "state",
"icon": f"mdi:{sensor['icon']}",
"state_topic": f"homeassistant/sensor/{vin}_{sensor['id']}/state",
"device": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{sensor['id']}",
}
if "unit" in sensor:
config["unit_of_measurement"] = sensor["unit"]
mqtt_client.publish( mqtt_client.publish(
f"homeassistant/sensor/volvoAAOS2mqtt/{vin}_{sensor['id']}/config", f"homeassistant/{entity['domain']}/volvoAAOS2mqtt/{vin}_{entity['id']}/config",
json.dumps(config), json.dumps(config),
) )
time.sleep(2) time.sleep(2)
+109 -47
View File
@@ -4,15 +4,15 @@ import mqtt
from config import settings from config import settings
from babel.dates import format_datetime from babel.dates import format_datetime
from util import keys_exists from util import keys_exists
from const import charging_system_states, CLIMATE_START_URL, \ from const import charging_system_states, charging_connection_states, CLIMATE_START_URL, \
OAUTH_URL, VEHICLES_URL, VEHICLE_DETAILS_URL, RECHARGE_STATE_URL, \ OAUTH_URL, VEHICLES_URL, VEHICLE_DETAILS_URL, RECHARGE_STATE_URL, \
WINDOWS_STATE_URL, LOCK_STATE_URL WINDOWS_STATE_URL, LOCK_STATE_URL, TYRE_STATE_URL, supported_entities, BATTERY_CHARGE_STATE_URL
session = requests.Session() session = requests.Session()
session.headers = { session.headers = {
"vcc-api-key": settings["volvoData"]["vccapikey"], "vcc-api-key": settings["volvoData"]["vccapikey"],
"content-type": "application/json", "content-type": "application/json",
"accept": "*/*" "accept": "*/*"
} }
token_expires_at: datetime token_expires_at: datetime
@@ -24,20 +24,23 @@ window_cached_api_response = {}
window_api_last_update = {} window_api_last_update = {}
door_cached_api_response = {} door_cached_api_response = {}
door_api_last_update = {} door_api_last_update = {}
tyre_cached_api_response = {}
tyre_api_last_update = {}
supported_endpoints = {}
def authorize(): def authorize():
headers = { headers = {
"authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=", "authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=",
"content-type": "application/x-www-form-urlencoded", "content-type": "application/x-www-form-urlencoded",
"accept": "application/json" "accept": "application/json"
} }
body = { body = {
"username": settings.volvoData["username"], "username": settings.volvoData["username"],
"password": settings.volvoData["password"], "password": settings.volvoData["password"],
"grant_type": "password", "grant_type": "password",
"scope": "openid email profile care_by_volvo:financial_information:invoice:read care_by_volvo:financial_information:payment_method care_by_volvo:subscription:read customer:attributes customer:attributes:write order:attributes vehicle:attributes tsp_customer_api:all conve:brake_status conve:climatization_start_stop conve:command_accessibility conve:commands conve:diagnostics_engine_status conve:diagnostics_workshop conve:doors_status conve:engine_status conve:environment conve:fuel_status conve:honk_flash conve:lock conve:lock_status conve:navigation conve:odometer_status conve:trip_statistics conve:tyre_status conve:unlock conve:vehicle_relation conve:warnings conve:windows_status energy:battery_charge_level energy:charging_connection_status energy:charging_system_status energy:electric_range energy:estimated_charging_time energy:recharge_status vehicle:attributes" "scope": "openid email profile care_by_volvo:financial_information:invoice:read care_by_volvo:financial_information:payment_method care_by_volvo:subscription:read customer:attributes customer:attributes:write order:attributes vehicle:attributes tsp_customer_api:all conve:brake_status conve:climatization_start_stop conve:command_accessibility conve:commands conve:diagnostics_engine_status conve:diagnostics_workshop conve:doors_status conve:engine_status conve:environment conve:fuel_status conve:honk_flash conve:lock conve:lock_status conve:navigation conve:odometer_status conve:trip_statistics conve:tyre_status conve:unlock conve:vehicle_relation conve:warnings conve:windows_status energy:battery_charge_level energy:charging_connection_status energy:charging_system_status energy:electric_range energy:estimated_charging_time energy:recharge_status vehicle:attributes"
} }
auth = requests.post(OAUTH_URL, data=body, headers=headers) auth = requests.post(OAUTH_URL, data=body, headers=headers)
if auth.status_code == 200: if auth.status_code == 200:
@@ -49,6 +52,7 @@ def authorize():
refresh_token = data["refresh_token"] refresh_token = data["refresh_token"]
get_vehicles() get_vehicles()
check_supported_endpoints()
else: else:
message = auth.json() message = auth.json()
raise Exception(message["error_description"]) raise Exception(message["error_description"])
@@ -58,14 +62,14 @@ def refresh_auth():
print("Refreshing credentials") print("Refreshing credentials")
global refresh_token global refresh_token
headers = { headers = {
"authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=", "authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=",
"content-type": "application/x-www-form-urlencoded", "content-type": "application/x-www-form-urlencoded",
"accept": "application/json" "accept": "application/json"
} }
body = { body = {
"grant_type": "refresh_token", "grant_type": "refresh_token",
"refresh_token": refresh_token "refresh_token": refresh_token
} }
auth = requests.post(OAUTH_URL, data=body, headers=headers) auth = requests.post(OAUTH_URL, data=body, headers=headers)
if auth.status_code == 200: if auth.status_code == 200:
@@ -113,19 +117,19 @@ def get_vehicle_details(vin):
if settings["debug"]: if settings["debug"]:
print(response.text) print(response.text)
device = { device = {
"identifiers": [f"volvoAAOS2mqtt_{vin}"], "identifiers": [f"volvoAAOS2mqtt_{vin}"],
"manufacturer": "Volvo", "manufacturer": "Volvo",
"model": data['descriptions']['model'], "model": data['descriptions']['model'],
"name": f"{data['descriptions']['model']} ({data['modelYear']}) - {vin}", "name": f"{data['descriptions']['model']} ({data['modelYear']}) - {vin}",
} }
elif response.status_code == 500 and not settings.volvoData["vin"]: elif response.status_code == 500 and not settings.volvoData["vin"]:
# Workaround for some cars that are not returning vehicle details # Workaround for some cars that are not returning vehicle details
device = { device = {
"identifiers": [f"volvoAAOS2mqtt_{vin}"], "identifiers": [f"volvoAAOS2mqtt_{vin}"],
"manufacturer": "Volvo", "manufacturer": "Volvo",
"model": vin, "model": vin,
"name": f"Volvo - {vin}", "name": f"Volvo - {vin}",
} }
else: else:
raise Exception("Getting vehicle details failed. Status Code: " + str(response.status_code) + raise Exception("Getting vehicle details failed. Status Code: " + str(response.status_code) +
". Error: " + response.text) ". Error: " + response.text)
@@ -133,6 +137,28 @@ def get_vehicle_details(vin):
return device return device
def check_supported_endpoints():
global supported_endpoints
for vin in vins:
supported_endpoints[vin] = []
for entity in supported_entities:
if entity["id"] == "battery_charge_level" and entity["url"] == BATTERY_CHARGE_STATE_URL \
and any("battery_charge_level" in d["id"] for d in supported_endpoints[vin]):
# If battery charge level could be found in recharge-api, skip the second battery charge sensor
continue
if entity.get('url'):
state = api_call(entity["url"], "GET", vin, entity["id"])
else:
state = ""
if state is not None:
print("Success! " + entity["name"] + " is supported by your vehicle.")
supported_endpoints[vin].append(entity)
else:
print("Failed, " + entity["name"] + " is unfortunately not supported by your vehicle.")
def initialize_climate(vins): def initialize_climate(vins):
for vin in vins: for vin in vins:
mqtt.assumed_climate_state[vin] = "OFF" mqtt.assumed_climate_state[vin] = "OFF"
@@ -158,6 +184,9 @@ def api_call(url, method, vin, sensor_id=None, force_update=False):
elif url == LOCK_STATE_URL: elif url == LOCK_STATE_URL:
# Minimize API calls for door state # Minimize API calls for door state
response = pull_door_api(url, method, vin, force_update) response = pull_door_api(url, method, vin, force_update)
elif url == TYRE_STATE_URL:
# Minimize API calls for tyre state
response = pull_tyre_api(url, method, vin, force_update)
elif method == "GET": elif method == "GET":
print("Starting " + method + " call against " + url) print("Starting " + method + " call against " + url)
response = session.get(url.format(vin), timeout=15) response = session.get(url.format(vin), timeout=15)
@@ -176,7 +205,7 @@ def api_call(url, method, vin, sensor_id=None, force_update=False):
else: else:
if url == CLIMATE_START_URL and response.status_code == 503: if url == CLIMATE_START_URL and response.status_code == 503:
print("Car in use, cannot start pre climatization") print("Car in use, cannot start pre climatization")
mqtt.assumed_climate_state = "OFF" mqtt.assumed_climate_state[vin] = "OFF"
mqtt.update_car_data() mqtt.update_car_data()
else: else:
print("API Call failed. Status Code: " + str(response.status_code) + ". Error: " + response.text) print("API Call failed. Status Code: " + str(response.status_code) + ". Error: " + response.text)
@@ -184,6 +213,27 @@ def api_call(url, method, vin, sensor_id=None, force_update=False):
return parse_api_data(data, sensor_id) return parse_api_data(data, sensor_id)
def pull_tyre_api(url, method, vin, force_update=False):
global tyre_cached_api_response, tyre_api_last_update
if not vin in tyre_cached_api_response or force_update:
# No API Data for vin cached, get fresh data from API
print("Starting " + method + " call against " + url)
response = session.get(url.format(vin), timeout=15)
tyre_cached_api_response[vin] = response
tyre_api_last_update[vin] = datetime.now()
else:
if (datetime.now() - tyre_api_last_update[vin]).total_seconds() >= settings["updateInterval"]:
# Old Data in Cache, updating
print("Starting " + method + " call against " + url)
response = session.get(url.format(vin), timeout=15)
tyre_cached_api_response[vin] = response
tyre_api_last_update[vin] = datetime.now()
else:
# Data is up do date, returning cached data
response = tyre_cached_api_response[vin]
return response
def pull_door_api(url, method, vin, force_update=False): def pull_door_api(url, method, vin, force_update=False):
global door_cached_api_response, door_api_last_update global door_cached_api_response, door_api_last_update
if not vin in door_cached_api_response or force_update: if not vin in door_cached_api_response or force_update:
@@ -250,11 +300,13 @@ def pull_recharge_api(url, method, vin, force_update=False):
def parse_api_data(data, sensor_id=None): def parse_api_data(data, sensor_id=None):
data = data["data"] data = data["data"]
if sensor_id == "battery_charge_level": if sensor_id == "battery_charge_level":
return data["batteryChargeLevel"]["value"] if keys_exists(data, "batteryChargeLevel") else 0 return data["batteryChargeLevel"]["value"] if keys_exists(data, "batteryChargeLevel") else None
elif sensor_id == "electric_range": elif sensor_id == "electric_range":
return data["electricRange"]["value"] if keys_exists(data, "electricRange") else 0 return data["electricRange"]["value"] if keys_exists(data, "electricRange") else None
elif sensor_id == "charging_system_status": elif sensor_id == "charging_system_status":
return charging_system_states[data["chargingSystemStatus"]["value"]] if keys_exists(data, "chargingSystemStatus") else "" return charging_system_states[data["chargingSystemStatus"]["value"]] if keys_exists(data, "chargingSystemStatus") else None
elif sensor_id == "charging_connection_status":
return charging_connection_states[data["chargingConnectionStatus"]["value"]] if keys_exists(data, "chargingConnectionStatus") else None
elif sensor_id == "estimated_charging_time": elif sensor_id == "estimated_charging_time":
if keys_exists(data, "chargingSystemStatus"): if keys_exists(data, "chargingSystemStatus"):
charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]] charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]]
@@ -263,7 +315,7 @@ def parse_api_data(data, sensor_id=None):
else: else:
return 0 return 0
else: else:
return "" return None
elif sensor_id == "estimated_charging_finish_time": elif sensor_id == "estimated_charging_finish_time":
if keys_exists(data, "chargingSystemStatus"): if keys_exists(data, "chargingSystemStatus"):
charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]] charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]]
@@ -275,33 +327,43 @@ def parse_api_data(data, sensor_id=None):
else: else:
return "" return ""
else: else:
return "" return None
elif sensor_id == "lock_status": elif sensor_id == "lock_status":
return data["carLocked"]["value"] if keys_exists(data, "carLocked") else "" return data["carLocked"]["value"] if keys_exists(data, "carLocked") else None
elif sensor_id == "odometer": elif sensor_id == "odometer":
return data["odometer"]["value"] if keys_exists(data, "odometer") else 0 return int(data["odometer"]["value"]) * 10 if keys_exists(data, "odometer") else None
elif sensor_id == "window_front_left": elif sensor_id == "window_front_left":
return data["frontLeftWindowOpen"]["value"] if keys_exists(data, "frontLeftWindowOpen") else "" return data["frontLeftWindowOpen"]["value"] if keys_exists(data, "frontLeftWindowOpen") else None
elif sensor_id == "window_front_right": elif sensor_id == "window_front_right":
return data["frontRightWindowOpen"]["value"] if keys_exists(data, "frontRightWindowOpen") else "" return data["frontRightWindowOpen"]["value"] if keys_exists(data, "frontRightWindowOpen") else None
elif sensor_id == "window_rear_left": elif sensor_id == "window_rear_left":
return data["rearLeftWindowOpen"]["value"] if keys_exists(data, "rearLeftWindowOpen") else "" return data["rearLeftWindowOpen"]["value"] if keys_exists(data, "rearLeftWindowOpen") else None
elif sensor_id == "window_rear_right": elif sensor_id == "window_rear_right":
return data["rearRightWindowOpen"]["value"] if keys_exists(data, "rearRightWindowOpen") else "" return data["rearRightWindowOpen"]["value"] if keys_exists(data, "rearRightWindowOpen") else None
elif sensor_id == "door_front_left": elif sensor_id == "door_front_left":
return data["frontLeftDoorOpen"]["value"] if keys_exists(data, "frontLeftDoorOpen") else "" return data["frontLeftDoorOpen"]["value"] if keys_exists(data, "frontLeftDoorOpen") else None
elif sensor_id == "door_front_right": elif sensor_id == "door_front_right":
return data["frontRightDoorOpen"]["value"] if keys_exists(data, "frontRightDoorOpen") else "" return data["frontRightDoorOpen"]["value"] if keys_exists(data, "frontRightDoorOpen") else None
elif sensor_id == "door_rear_left": elif sensor_id == "door_rear_left":
return data["rearLeftDoorOpen"]["value"] if keys_exists(data, "rearLeftDoorOpen") else "" return data["rearLeftDoorOpen"]["value"] if keys_exists(data, "rearLeftDoorOpen") else None
elif sensor_id == "door_rear_right": elif sensor_id == "door_rear_right":
return data["rearRightDoorOpen"]["value"] if keys_exists(data, "rearRightDoorOpen") else "" return data["rearRightDoorOpen"]["value"] if keys_exists(data, "rearRightDoorOpen") else None
elif sensor_id == "tailgate": elif sensor_id == "tailgate":
return data["tailGateOpen"]["value"] if keys_exists(data, "tailGateOpen") else "" return data["tailGateOpen"]["value"] if keys_exists(data, "tailGateOpen") else None
elif sensor_id == "engine_hood": elif sensor_id == "engine_hood":
return data["hoodOpen"]["value"] if keys_exists(data, "hoodOpen") else "" return data["hoodOpen"]["value"] if keys_exists(data, "hoodOpen") else None
elif sensor_id == "tank_lid": elif sensor_id == "tank_lid":
return data["tankLidOpen"]["value"] if keys_exists(data, "tankLidOpen") else "" return data["tankLidOpen"]["value"] if keys_exists(data, "tankLidOpen") else None
elif sensor_id == "tyre_front_left":
return data["frontLeftTyrePressure"]["value"] if keys_exists(data, "frontLeftTyrePressure") else None
elif sensor_id == "tyre_front_right":
return data["frontRightTyrePressure"]["value"] if keys_exists(data, "frontRightTyrePressure") else None
elif sensor_id == "tyre_rear_left":
return data["rearLeftTyrePressure"]["value"] if keys_exists(data, "rearLeftTyrePressure") else None
elif sensor_id == "tyre_rear_right":
return data["rearRightTyrePressure"]["value"] if keys_exists(data, "rearRightTyrePressure") else None
elif sensor_id == "engine_state":
return data["engineRunning"]["value"] if keys_exists(data, "engineRunning") else None
elif sensor_id == "location": elif sensor_id == "location":
coordinates = {} coordinates = {}
if keys_exists(data, "geometry"): if keys_exists(data, "geometry"):
@@ -312,4 +374,4 @@ def parse_api_data(data, sensor_id=None):
"gps_accuracy": 1} "gps_accuracy": 1}
return coordinates return coordinates
else: else:
return "" return None