diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..723ef36 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.idea \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9e58b07 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,17 @@ +# set base image (host OS) +FROM python:3.9-slim + +# set the working directory in the container +WORKDIR /volvoAAOS2mqtt + +# copy the dependencies file to the working directory +COPY requirements.txt . + +# install dependencies +RUN pip install -r requirements.txt + +# copy the content of the local src directory to the working directory +COPY / . + +# command to run on container start +CMD [ "python", "-u", "./main.py" ] \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..a038814 --- /dev/null +++ b/config.py @@ -0,0 +1,6 @@ +from dynaconf import Dynaconf + +settings = Dynaconf( + envvar_prefix="CONF", + settings_files=["settings.json"], +) \ No newline at end of file diff --git a/const.py b/const.py new file mode 100644 index 0000000..4f730cf --- /dev/null +++ b/const.py @@ -0,0 +1,34 @@ +OAUTH_URL = "https://volvoid.eu.volvocars.com/as/token.oauth2" +VEHICLES_URL = "https://api.volvocars.com/connected-vehicle/v1/vehicles" +VEHICLE_DETAILS_URL = "https://api.volvocars.com/connected-vehicle/v1/vehicles/{0}" +WINDOW_STATUS_URL = "https://api.volvocars.com/connected-vehicle/v1/vehicles/{0}/windows" +CLIMATE_START_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/commands/climatization-start" +CLIMATE_STOP_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/commands/climatization-stop" +CAR_LOCK_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/doors" +CAR_LOCK_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/commands/lock" +CAR_UNLOCK_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/commands/unlock" +RECHARGE_STATUS_URL = "https://api.volvocars.com/energy/v1/vehicles/{0}/recharge-status" + +charging_system_states = {"CHARGING_SYSTEM_CHARGING": "Charging", "CHARGING_SYSTEM_IDLE": "Idle", + "CHARGING_SYSTEM_FAULT": "Fault", "CHARGING_SYSTEM_UNSPECIFIED": "UNSPECIFIED"} + +supported_sensors = [ + {"name": "Battery Charge Level", "id": "battery_charge_level", "unit": "%", "icon": "car-battery", "url": RECHARGE_STATUS_URL}, + {"name": "Electric Range", "id": "electric_range", "unit": "km", "icon": "map-marker-distance", "url": RECHARGE_STATUS_URL}, + {"name": "Estimated Charging Time", "id": "estimated_charging_time", "unit": "minutes", "icon": "timer-sync-outline", "url": RECHARGE_STATUS_URL}, + {"name": "Charging System Status", "id": "charging_system_status", "icon": "ev-plug-ccs2", "url": RECHARGE_STATUS_URL}, + {"name": "Estimated Charging Finish Time", "id": "estimated_charging_finish_time", "icon": "timer-sync-outline", "url": RECHARGE_STATUS_URL}, + {"name": "Last Data Update", "id": "last_data_update", "icon": "timer", "url": ""} +] + +supported_switches = [ + {"name": "Air Conditioning", "id": "climate_status", "icon": "air-conditioner"}, +] + +supported_locks = [ + {"name": "Lock state", "id": "lock_status", "icon": "lock", "url": CAR_LOCK_STATE_URL} +] + +supported_buttons = [ + {"name": "Update Data", "id": "update_data", "icon": "update", "url": ""} +] \ No newline at end of file diff --git a/container_config.json b/container_config.json new file mode 100644 index 0000000..1319fa2 --- /dev/null +++ b/container_config.json @@ -0,0 +1,8 @@ +{ + "name": "volvoaaos2mqtt", + "ports": {}, + "env": {"TZ": "Europe/Amsterdam"}, + "volumes": [ + "/home/dockerhost/volvoAAOS2mqtt/settings.json:/volvoAAOS2mqtt/settings.json" + ] +} \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..1937dd8 --- /dev/null +++ b/main.py @@ -0,0 +1,7 @@ +from volvo import authorize +from mqtt import update_loop, connect + +if __name__ == '__main__': + authorize() + connect() + update_loop() diff --git a/mqtt.py b/mqtt.py new file mode 100644 index 0000000..ce7a6e4 --- /dev/null +++ b/mqtt.py @@ -0,0 +1,211 @@ +import time +import paho.mqtt.client as mqtt +import json +import volvo +import threading +from datetime import datetime, timedelta +from babel.dates import format_datetime +from config import settings +from const import VEHICLE_DETAILS_URL,CLIMATE_START_URL, CLIMATE_STOP_URL, CAR_LOCK_URL, \ + CAR_UNLOCK_URL, supported_sensors, supported_buttons, supported_switches, supported_locks + + +mqtt_client: mqtt.Client +subscribed_topics = [] +assumed_climate_state = "OFF" +last_data_update = None + + +def connect(): + client = mqtt.Client("volvoAAOS2mqtt") + if settings["mqtt"]["username"] and settings["mqtt"]["password"]: + client.username_pw_set(settings["mqtt"]["username"], settings["mqtt"]["password"]) + client.connect(settings["mqtt"]["broker"]) + client.loop_start() + client.on_message = on_message + client.on_disconnect = on_disconnect + client.on_connect = on_connect + + global mqtt_client + mqtt_client = client + + +def on_connect(client, userdata, flags, rc): + if len(subscribed_topics) > 0: + for topic in subscribed_topics: + mqtt_client.subscribe(topic) + + +def on_disconnect(client, userdata, rc): + print("MQTT disconnected, reconnecting automatically") + + +def on_message(client, userdata, msg): + if msg.topic in subscribed_topics: + payload = msg.payload.decode("UTF-8") + if "climate_status" in msg.topic: + global assumed_climate_state + if payload == "ON": + api_thread = threading.Thread(target=volvo.api_call, args=(CLIMATE_START_URL, "POST", )) + api_thread.start() + assumed_climate_state = "ON" + # Starting timer to disable climate after 30 mins + threading.Timer(30 * 60, volvo.disable_climate).start() + update_car_data() + elif payload == "OFF": + api_thread = threading.Thread(target=volvo.api_call, args=(CLIMATE_STOP_URL, "POST", )) + api_thread.start() + assumed_climate_state = "OFF" + update_car_data() + elif "lock_status" in msg.topic: + if payload == "LOCK": + volvo.api_call(CAR_LOCK_URL, "POST") + update_car_data() + elif payload == "UNLOCK": + volvo.api_call(CAR_UNLOCK_URL, "POST") + update_car_data() + elif "update_data" in msg.topic: + if payload == "PRESS": + update_car_data() + + +def update_loop(): + create_ha_devices() + while True: + print("Sending mqtt update...") + update_car_data() + time.sleep(settings["updateInterval"]) + + +def update_car_data(): + global last_data_update + last_data_update = format_datetime(datetime.now(), format="medium", locale=settings["babelLocale"]) + for lock in supported_locks: + state = volvo.api_call(lock["url"], "GET", lock["id"]) + mqtt_client.publish( + f"homeassistant/lock/{volvo.vin}_{lock['id']}/state", + state + ) + + for switch in supported_switches: + if switch["id"] == "climate_status": + state = assumed_climate_state + else: + state = "OFF" + + mqtt_client.publish( + f"homeassistant/switch/{volvo.vin}_{switch['id']}/state", + state + ) + + for sensor in supported_sensors: + if sensor["id"] == "last_data_update": + state = last_data_update + else: + state = volvo.api_call(sensor["url"], "GET", sensor["id"]) + mqtt_client.publish( + f"homeassistant/sensor/{volvo.vin}_{sensor['id']}/state", + state + ) + + +def create_ha_devices(): + car_details = volvo.api_call(VEHICLE_DETAILS_URL, "GET") + + for button in supported_buttons: + command_topic = f"homeassistant/button/{volvo.vin}_{button['id']}/command" + config = { + "name": button['name'], + "object_id": button['id'], + "schema": "state", + "icon": f"mdi:{button['icon']}", + "state_topic": f"homeassistant/button/{volvo.vin}_{button['id']}/state", + "command_topic": command_topic, + "device": { + "identifiers": ["volvoAAOS2mqtt"], + "manufacturer": "Volvo", + "model": car_details['descriptions']['model'], + "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {volvo.vin}", + }, + "unique_id": f"volvoAAOS2mqtt_{volvo.vin}_{button['id']}", + } + mqtt_client.publish( + f"homeassistant/button/volvoAAOS2mqtt/{volvo.vin}_{button['id']}/config", + json.dumps(config), + ) + subscribed_topics.append(command_topic) + mqtt_client.subscribe(command_topic) + + for lock in supported_locks: + command_topic = f"homeassistant/lock/{volvo.vin}_{lock['id']}/command" + config = { + "name": lock['name'], + "object_id": lock['id'], + "schema": "state", + "icon": f"mdi:{lock['icon']}", + "state_topic": f"homeassistant/lock/{volvo.vin}_{lock['id']}/state", + "command_topic": command_topic, + "optimistic": False, + "device": { + "identifiers": ["volvoAAOS2mqtt"], + "manufacturer": "Volvo", + "model": car_details['descriptions']['model'], + "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {volvo.vin}", + }, + "unique_id": f"volvoAAOS2mqtt_{volvo.vin}_{lock['id']}", + } + mqtt_client.publish( + f"homeassistant/lock/volvoAAOS2mqtt/{volvo.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/{volvo.vin}_{switch['id']}/command" + config = { + "name": switch['name'], + "object_id": switch['id'], + "schema": "state", + "icon": f"mdi:{switch['icon']}", + "state_topic": f"homeassistant/switch/{volvo.vin}_{switch['id']}/state", + "command_topic": command_topic, + "optimistic": False, + "device": { + "identifiers": ["volvoAAOS2mqtt"], + "manufacturer": "Volvo", + "model": car_details['descriptions']['model'], + "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {volvo.vin}", + }, + "unique_id": f"volvoAAOS2mqtt_{volvo.vin}_{switch['id']}", + } + mqtt_client.publish( + f"homeassistant/switch/volvoAAOS2mqtt/{volvo.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": sensor['id'], + "schema": "state", + "icon": f"mdi:{sensor['icon']}", + "state_topic": f"homeassistant/sensor/{volvo.vin}_{sensor['id']}/state", + "device": { + "identifiers": ["volvoAAOS2mqtt"], + "manufacturer": "Volvo", + "model": car_details['descriptions']['model'], + "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {volvo.vin}", + }, + "unique_id": f"volvoAAOS2mqtt_{volvo.vin}_{sensor['id']}", + } + if "unit" in sensor: + config["unit_of_measurement"] = sensor["unit"] + + mqtt_client.publish( + f"homeassistant/sensor/volvoAAOS2mqtt/{volvo.vin}_{sensor['id']}/config", + json.dumps(config), + ) + time.sleep(2) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..dc0ef4c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +requests~=2.29.0 +dynaconf~=3.1.12 +paho-mqtt~=1.6.1 +pytz~=2023.3 +Babel~=2.12.1 \ No newline at end of file diff --git a/settings.json b/settings.json new file mode 100644 index 0000000..72ad4d0 --- /dev/null +++ b/settings.json @@ -0,0 +1,14 @@ +{ + "updateInterval": 300, + "babelLocale": "de", + "mqtt": { + "broker": "", + "username": "", + "password": "" + }, + "volvoData": { + "username": "", + "password": "", + "vin": "" + } +} \ No newline at end of file diff --git a/volvo.py b/volvo.py new file mode 100644 index 0000000..a424a4f --- /dev/null +++ b/volvo.py @@ -0,0 +1,170 @@ +import requests +from datetime import datetime, timedelta +import mqtt +from config import settings +from babel.dates import format_datetime +from const import charging_system_states, CLIMATE_START_URL, \ + OAUTH_URL, VEHICLES_URL, VEHICLE_DETAILS_URL, RECHARGE_STATUS_URL + +session = requests.Session() +session.headers = { + "vcc-api-key": "f0d0419bf51d420c8efb21cf9a127227", + "content-type": "application/json", + "accept": "*/*" +} + +token_expires_at = None +refresh_token = None +vin = "" +recharge_response = {} +recharge_last_update = None + + +def authorize(): + headers = { + "authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=", + "content-type": "application/x-www-form-urlencoded", + "accept": "application/json" + } + + body = { + "username": settings.volvoData["username"], + "password": settings.volvoData["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" + } + auth = requests.post(OAUTH_URL, data=body, headers=headers) + if auth.status_code == 200: + data = auth.json() + session.headers.update({'authorization': "Bearer " + data["access_token"]}) + + global token_expires_at, refresh_token + token_expires_at = datetime.now() + timedelta(seconds=(data["expires_in"] - 30)) + refresh_token = data["refresh_token"] + + get_vehicle() + else: + message = auth.json() + raise Exception(message["error_description"]) + + +def refresh_auth(): + print("Refreshing credentials") + global refresh_token + headers = { + "authorization": "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc=", + "content-type": "application/x-www-form-urlencoded", + "accept": "application/json" + } + + body = { + "grant_type": "refresh_token", + "refresh_token": refresh_token + } + auth = requests.post(OAUTH_URL, data=body, headers=headers) + if auth.status_code == 200: + data = auth.json() + session.headers.update({'authorization': "Bearer " + data["access_token"]}) + + global token_expires_at + token_expires_at = datetime.now() + timedelta(seconds=(data["expires_in"] - 30)) + refresh_token = data["refresh_token"] + + +def get_vehicle(): + global vin + if not settings.volvoData["vin"]: + vehicles = session.get(VEHICLES_URL) + if vehicles.status_code == 200: + data = vehicles.json() + if len(data["data"]) > 0: + vin = data["data"][0]["vin"] + else: + print("No vehicle in account " + settings.volvoData["username"] + " found.") + else: + print("Error getting Vehicles " + str(vehicles.status_code)) + else: + vin = settings.volvoData["vin"] + + if not vin: + raise Exception("No vehicle found, exiting application!") + else: + print("Vin: " + vin + " found!") + + +def disable_climate(): + mqtt.assumed_climate_state = "OFF" + mqtt.update_car_data() + + +def api_call(url, method, sensor_id=None): + global token_expires_at + if datetime.now() >= token_expires_at: + refresh_auth() + + global vin, recharge_response, recharge_last_update + if url == RECHARGE_STATUS_URL: + # Minimize API calls for recharge API + if not bool(recharge_response): + # No API Data cached, get fresh data from API + print("Starting " + method + " call against " + url) + response = session.get(url.format(vin), timeout=15) + recharge_response = response + recharge_last_update = datetime.now() + else: + if (datetime.now() - recharge_last_update).total_seconds() >= settings["updateInterval"]: + # Old Data in Cache, updateing + print("Starting " + method + " call against " + url) + response = session.get(url.format(vin), timeout=15) + recharge_response = response + recharge_last_update = datetime.now() + else: + # Data is up do date, returning cached data + response = recharge_response + elif method == "GET": + print("Starting " + method + " call against " + url) + response = session.get(url.format(vin), timeout=15) + elif method == "POST": + print("Starting " + method + " call against " + url) + response = session.post(url.format(vin), timeout=20) + else: + print("Unkown method posted: " + method + ". Returning nothing") + return "" + + if response.status_code == 200: + data = response.json() + else: + if url == CLIMATE_START_URL and response.status_code == 503: + print("Car in use, cannot start pre climatization") + mqtt.assumed_climate_state = "OFF" + mqtt.update_car_data() + else: + print("API Call failed. Status Code: " + str(response.status_code) + ". Error: " + response.text) + return "" + + if url == VEHICLE_DETAILS_URL: + return data["data"] + elif sensor_id == "battery_charge_level": + return data["data"]["batteryChargeLevel"]["value"] + elif sensor_id == "electric_range": + return data["data"]["electricRange"]["value"] + elif sensor_id == "charging_system_status": + return charging_system_states[data["data"]["chargingSystemStatus"]["value"]] + elif sensor_id == "estimated_charging_time": + charging_system_state = charging_system_states[data["data"]["chargingSystemStatus"]["value"]] + if charging_system_state == "Charging": + return data["data"]["estimatedChargingTime"]["value"] + else: + return 0 + elif sensor_id == "estimated_charging_finish_time": + charging_system_state = charging_system_states[data["data"]["chargingSystemStatus"]["value"]] + if charging_system_state == "Charging": + charging_time = int(data["data"]["estimatedChargingTime"]["value"]) + charging_finished = datetime.now() + timedelta(minutes=charging_time) + return format_datetime(charging_finished, format="medium", locale=settings["babelLocale"]) + else: + return None + elif sensor_id == "lock_status": + return data["data"]["carLocked"]["value"] + else: + return ""