cleanup repo

src/config.py

@@ -0,0 +1,6 @@
+from dynaconf import Dynaconf
+
+settings = Dynaconf(
+    envvar_prefix="CONF",
+    settings_files=["settings.json"],
+)

src/const.py

@@ -0,0 +1,36 @@
+VERSION = "v1.0.7"
+
+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": ""}
+]

src/main.py

@@ -0,0 +1,9 @@
+from volvo import authorize
+from mqtt import update_loop, connect
+from const import VERSION
+
+if __name__ == '__main__':
+    print("Starting volvo2mqtt version " + VERSION)
+    authorize()
+    connect()
+    update_loop()

src/mqtt.py

@@ -0,0 +1,215 @@
+import time
+import paho.mqtt.client as mqtt
+import json
+import volvo
+import threading
+from datetime import datetime
+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:
+        vin = msg.topic.split('/')[2].split('_')[0]
+        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", vin))
+                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", vin))
+                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", vin)
+                update_car_data()
+            elif payload == "UNLOCK":
+                volvo.api_call(CAR_UNLOCK_URL, "POST", vin)
+                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()
+        print("Mqtt update done. Next run in " + str(settings["updateInterval"]) + " seconds.")
+        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 vin in volvo.vins:
+        for lock in supported_locks:
+            state = volvo.api_call(lock["url"], "GET", vin, lock["id"])
+            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
+            else:
+                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
+            else:
+                state = volvo.api_call(sensor["url"], "GET", vin, sensor["id"])
+            mqtt_client.publish(
+                f"homeassistant/sensor/{vin}_{sensor['id']}/state",
+                state
+            )
+
+
+def create_ha_devices():
+    for vin in volvo.vins:
+        car_details = volvo.api_call(VEHICLE_DETAILS_URL, "GET", vin)
+
+        for button in supported_buttons:
+            command_topic = f"homeassistant/button/{vin}_{button['id']}/command"
+            config = {
+                        "name": button['name'],
+                        "object_id": button['id'],
+                        "schema": "state",
+                        "icon": f"mdi:{button['icon']}",
+                        "state_topic": f"homeassistant/button/{vin}_{button['id']}/state",
+                        "command_topic": command_topic,
+                        "device": {
+                            "identifiers": [f"volvoAAOS2mqtt_{vin}"],
+                            "manufacturer": "Volvo",
+                            "model": car_details['descriptions']['model'],
+                            "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {vin}",
+                        },
+                        "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:
+            command_topic = f"homeassistant/lock/{vin}_{lock['id']}/command"
+            config = {
+                        "name": lock['name'],
+                        "object_id": lock['id'],
+                        "schema": "state",
+                        "icon": f"mdi:{lock['icon']}",
+                        "state_topic": f"homeassistant/lock/{vin}_{lock['id']}/state",
+                        "command_topic": command_topic,
+                        "optimistic": False,
+                        "device": {
+                            "identifiers": [f"volvoAAOS2mqtt_{vin}"],
+                            "manufacturer": "Volvo",
+                            "model": car_details['descriptions']['model'],
+                            "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {vin}",
+                        },
+                        "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": 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": {
+                            "identifiers": [f"volvoAAOS2mqtt_{vin}"],
+                            "manufacturer": "Volvo",
+                            "model": car_details['descriptions']['model'],
+                            "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {vin}",
+                        },
+                        "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": sensor['id'],
+                        "schema": "state",
+                        "icon": f"mdi:{sensor['icon']}",
+                        "state_topic": f"homeassistant/sensor/{vin}_{sensor['id']}/state",
+                        "device": {
+                            "identifiers": [f"volvoAAOS2mqtt_{vin}"],
+                            "manufacturer": "Volvo",
+                            "model": car_details['descriptions']['model'],
+                            "name": f"{car_details['descriptions']['model']} ({car_details['modelYear']}) - {vin}",
+                        },
+                        "unique_id": f"volvoAAOS2mqtt_{vin}_{sensor['id']}",
+                    }
+            if "unit" in sensor:
+                config["unit_of_measurement"] = sensor["unit"]
+
+            mqtt_client.publish(
+                f"homeassistant/sensor/volvoAAOS2mqtt/{vin}_{sensor['id']}/config",
+                json.dumps(config),
+            )
+    time.sleep(2)

src/settings.json

@@ -0,0 +1,15 @@
+{
+  "updateInterval": 300,
+  "babelLocale": "de",
+  "mqtt": {
+    "broker": "",
+    "username": "",
+    "password": ""
+  },
+  "volvoData": {
+    "username": "",
+    "password": "",
+    "vin": "",
+	"vccapikey": ""
+  }
+}

src/volvo.py

@@ -0,0 +1,194 @@
+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": settings["volvoData"]["vccapikey"],
+                "content-type": "application/json",
+                "accept": "*/*"
+}
+
+token_expires_at: datetime
+refresh_token = None
+vins = []
+recharge_response = {}
+recharge_last_update = {}
+
+
+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_vehicles()
+    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_vehicles():
+    global vins
+    if not settings.volvoData["vin"]:
+        vehicles = session.get(VEHICLES_URL)
+        if vehicles.status_code == 200:
+            data = vehicles.json()
+            if len(data["data"]) > 0:
+                for vehicle in data["data"]:
+                    vins.append(vehicle["vin"])
+            else:
+                raise Exception("No vehicle in account " + settings.volvoData["username"] + " found.")
+        else:
+            raise Exception("Error getting vehicles: " + str(vehicles.status_code))
+    else:
+        if isinstance(settings.volvoData["vin"], list):
+            # If setting is a list, copy
+            vins = settings.volvoData["vin"]
+        else:
+            # If setting is a string, append to list
+            vins.append(settings.volvoData["vin"])
+
+    if len(vins) == 0:
+        raise Exception("No vehicle found, exiting application!")
+    else:
+        print("Vin: " + str(vins) + " found!")
+
+
+def disable_climate():
+    mqtt.assumed_climate_state = "OFF"
+    mqtt.update_car_data()
+
+
+def api_call(url, method, vin, sensor_id=None):
+    global token_expires_at
+    if datetime.now() >= token_expires_at:
+        refresh_auth()
+
+    global recharge_response, recharge_last_update
+    if url == RECHARGE_STATUS_URL:
+        # Minimize API calls for recharge API
+        if not vin in recharge_response:
+            # 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)
+            recharge_response[vin] = response
+            recharge_last_update[vin] = datetime.now()
+        else:
+            if (datetime.now() - recharge_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)
+                recharge_response[vin] = response
+                recharge_last_update[vin] = datetime.now()
+            else:
+                # Data is up do date, returning cached data
+                response = recharge_response[vin]
+    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 ""
+    return parse_api_data(url, data, sensor_id)
+
+
+def parse_api_data(url, data, sensor_id=None):
+    if url == VEHICLE_DETAILS_URL:
+        return data["data"]
+    elif sensor_id == "battery_charge_level":
+        if "batteryChargeLevel" in data["data"]:
+            return data["data"]["batteryChargeLevel"]["value"]
+        else:
+            return ""
+    elif sensor_id == "electric_range":
+        if "electricRange" in data["data"]:
+            return data["data"]["electricRange"]["value"]
+        else:
+            return ""
+    elif sensor_id == "charging_system_status":
+        if "charging_system_status" in data["data"]:
+            return charging_system_states[data["data"]["chargingSystemStatus"]["value"]]
+        else:
+            return ""
+    elif sensor_id == "estimated_charging_time":
+        if "chargingSystemStatus" in data["data"]:
+            charging_system_state = charging_system_states[data["data"]["chargingSystemStatus"]["value"]]
+            if charging_system_state == "Charging":
+                return data["data"]["estimatedChargingTime"]["value"]
+            else:
+                return 0
+        else:
+            return ""
+    elif sensor_id == "estimated_charging_finish_time":
+        if "chargingSystemStatus" in data["data"]:
+            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
+        else:
+            return None
+    elif sensor_id == "lock_status":
+        return data["data"]["carLocked"]["value"]
+    else:
+        return ""