mirror of
https://github.com/Dielee/volvo2mqtt.git
synced 2026-08-12 10:49:52 +02:00
Initial commit
This commit is contained in:
Linus Dietz
@@ -0,0 +1 @@
|
||||
.idea
|
||||
+17
@@ -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" ]
|
||||
@@ -0,0 +1,6 @@
|
||||
from dynaconf import Dynaconf
|
||||
|
||||
settings = Dynaconf(
|
||||
envvar_prefix="CONF",
|
||||
settings_files=["settings.json"],
|
||||
)
|
||||
@@ -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": ""}
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"name": "volvoaaos2mqtt",
|
||||
"ports": {},
|
||||
"env": {"TZ": "Europe/Amsterdam"},
|
||||
"volumes": [
|
||||
"/home/dockerhost/volvoAAOS2mqtt/settings.json:/volvoAAOS2mqtt/settings.json"
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
from volvo import authorize
|
||||
from mqtt import update_loop, connect
|
||||
|
||||
if __name__ == '__main__':
|
||||
authorize()
|
||||
connect()
|
||||
update_loop()
|
||||
@@ -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)
|
||||
@@ -0,0 +1,5 @@
|
||||
requests~=2.29.0
|
||||
dynaconf~=3.1.12
|
||||
paho-mqtt~=1.6.1
|
||||
pytz~=2023.3
|
||||
Babel~=2.12.1
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"updateInterval": 300,
|
||||
"babelLocale": "de",
|
||||
"mqtt": {
|
||||
"broker": "",
|
||||
"username": "",
|
||||
"password": ""
|
||||
},
|
||||
"volvoData": {
|
||||
"username": "",
|
||||
"password": "",
|
||||
"vin": ""
|
||||
}
|
||||
}
|
||||
@@ -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 ""
|
||||
Reference in New Issue
Block a user