jarvis/volvo2mqtt
1
0
Fork 0
forked from peter/volvo2mqtt
Code Pull Requests Activity

Create only sensors that are supported by the car // Rework sensor creation

Browse Source
This commit is contained in:
Linus Dietz
2023-06-23 08:58:24 +02:00
parent b3ab5acf0f
commit 17853210c3
3 changed files with 125 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/const.py
+23 -35
View File
@@ -18,39 +18,27 @@ LOCATION_STATE_URL = "https://api.volvocars.com/location/v1/vehicles/{0}/locatio
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 = [ supported_entities = [
{"name": "Battery Charge Level", "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": RECHARGE_STATE_URL},
{"name": "Electric Range", "id": "electric_range", "unit": "km" if settings["babelLocale"] != "en_US" else "mi", "icon": "map-marker-distance", "url": RECHARGE_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", "id": "estimated_charging_time", "unit": "minutes", "icon": "timer-sync-outline", "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", "id": "charging_system_status", "icon": "ev-plug-ccs2", "url": RECHARGE_STATE_URL}, {"name": "Charging System Status", "domain": "sensor", "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": "Estimated Charging Finish Time", "domain": "sensor", "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": "Odometer", "domain": "sensor", "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": "Last Data Update", "domain": "sensor", "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 Left", "domain": "sensor", "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 Front Right", "domain": "sensor", "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 Left", "domain": "sensor", "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": "Window Rear Right", "domain": "sensor", "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 Left", "domain": "sensor", "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 Front Right", "domain": "sensor", "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 Left", "domain": "sensor", "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": "Door Rear Right", "domain": "sensor", "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": "Tailgate", "domain": "sensor", "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": "Engine Hood", "domain": "sensor", "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} {"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},
supported_switches = [ {"name": "Force Update Data", "domain": "button", "id": "update_data", "icon": "update", "url": ""},
{"name": "Air Conditioning", "id": "climate_status", "icon": "air-conditioner"}, {"name": "Location", "domain": "device_tracker", "id": "location", "icon": "map-marker-radius", "url": LOCATION_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}
] ]
src/mqtt.py
+36 -114
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,8 +42,12 @@ 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]
except IndexError:
print("Error - Cannot get vin from MQTT topic!")
return None
payload = msg.payload.decode("UTF-8") payload = msg.payload.decode("UTF-8")
if "climate_status" in msg.topic: if "climate_status" in msg.topic:
global assumed_climate_state, climate_timer global assumed_climate_state, climate_timer
@@ -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'], if entity.get('domain') in ["switch", "lock", "button"]:
"object_id": f"volvo_{vin}_{button['id']}", command_topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/command"
"schema": "state", config["command_topic"] = command_topic
"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) subscribed_topics.append(command_topic)
mqtt_client.subscribe(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": f"volvo_{vin}_{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": device,
"unique_id": f"volvoAAOS2mqtt_{vin}_{lock['id']}",
}
mqtt_client.publish( mqtt_client.publish(
f"homeassistant/lock/volvoAAOS2mqtt/{vin}_{lock['id']}/config", f"homeassistant/{entity['domain']}/volvoAAOS2mqtt/{vin}_{entity['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(
f"homeassistant/sensor/volvoAAOS2mqtt/{vin}_{sensor['id']}/config",
json.dumps(config), json.dumps(config),
) )
time.sleep(2) time.sleep(2)
src/volvo.py
+36 -20
View File
@@ -6,7 +6,7 @@ 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, 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, supported_entities
session = requests.Session() session = requests.Session()
session.headers = { session.headers = {
@@ -24,6 +24,7 @@ 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 = {}
supported_endpoints = {}
def authorize(): def authorize():
@@ -49,6 +50,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"])
@@ -133,6 +135,20 @@ 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.get('url'):
state = api_call(entity["url"], "GET", vin, entity["id"])
else:
state = ""
if state is not None:
supported_endpoints[vin].append(entity)
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"
@@ -250,11 +266,11 @@ 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 == "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 +279,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 +291,33 @@ 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 int(data["odometer"]["value"]) * 10 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 == "location": elif sensor_id == "location":
coordinates = {} coordinates = {}
if keys_exists(data, "geometry"): if keys_exists(data, "geometry"):
@@ -312,4 +328,4 @@ def parse_api_data(data, sensor_id=None):
"gps_accuracy": 1} "gps_accuracy": 1}
return coordinates return coordinates
else: else:
return "" return None