From 17853210c3b82bfc27c08f66fc50f6f252f7e4fe Mon Sep 17 00:00:00 2001 From: Linus Dietz <45101649+Dielee@users.noreply.github.com> Date: Fri, 23 Jun 2023 08:58:24 +0200 Subject: [PATCH] Create only sensors that are supported by the car // Rework sensor creation --- src/const.py | 58 ++++++-------- src/mqtt.py | 210 ++++++++++++++++----------------------------------- src/volvo.py | 56 +++++++++----- 3 files changed, 125 insertions(+), 199 deletions(-) diff --git a/src/const.py b/src/const.py index 99faaa6..f75c06a 100644 --- a/src/const.py +++ b/src/const.py @@ -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_FAULT": "Fault", "CHARGING_SYSTEM_UNSPECIFIED": "Unspecified"} -supported_sensors = [ - {"name": "Battery Charge Level", "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": "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_entities = [ + {"name": "Battery Charge Level", "domain": "sensor", "id": "battery_charge_level", "unit": "%", "icon": "car-battery", "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", "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-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} ] - -supported_switches = [ - {"name": "Air Conditioning", "id": "climate_status", "icon": "air-conditioner"}, -] - -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} -] \ No newline at end of file diff --git a/src/mqtt.py b/src/mqtt.py index e6915bd..acd5edd 100644 --- a/src/mqtt.py +++ b/src/mqtt.py @@ -7,9 +7,7 @@ from datetime import datetime from babel.dates import format_datetime from config import settings from const import CLIMATE_START_URL, CLIMATE_STOP_URL, CAR_LOCK_URL, \ - CAR_UNLOCK_URL, supported_sensors, supported_buttons, supported_switches, \ - supported_locks, supported_device_trackers - + CAR_UNLOCK_URL mqtt_client: mqtt.Client @@ -44,37 +42,41 @@ def on_disconnect(client, userdata, rc): def on_message(client, userdata, msg): - if msg.topic in subscribed_topics: + try: vin = msg.topic.split('/')[2].split('_')[0] - payload = msg.payload.decode("UTF-8") - if "climate_status" in msg.topic: - global assumed_climate_state, climate_timer - if payload == "ON": - api_thread = Thread(target=volvo.api_call, args=(CLIMATE_START_URL, "POST", vin)) - api_thread.start() - assumed_climate_state[vin] = "ON" - # Starting timer to disable climate after 30 mins - climate_timer = Timer(1 * 60, volvo.disable_climate, (vin, )) - climate_timer.start() - update_car_data() - elif payload == "OFF": - api_thread = Thread(target=volvo.api_call, args=(CLIMATE_STOP_URL, "POST", vin)) - api_thread.start() - assumed_climate_state[vin] = "OFF" - # Stop timer if active - if climate_timer.is_alive(): - climate_timer.cancel() - 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(True) + except IndexError: + print("Error - Cannot get vin from MQTT topic!") + return None + + payload = msg.payload.decode("UTF-8") + if "climate_status" in msg.topic: + global assumed_climate_state, climate_timer + if payload == "ON": + api_thread = Thread(target=volvo.api_call, args=(CLIMATE_START_URL, "POST", vin)) + api_thread.start() + assumed_climate_state[vin] = "ON" + # Starting timer to disable climate after 30 mins + climate_timer = Timer(1 * 60, volvo.disable_climate, (vin, )) + climate_timer.start() + update_car_data() + elif payload == "OFF": + api_thread = Thread(target=volvo.api_call, args=(CLIMATE_STOP_URL, "POST", vin)) + api_thread.start() + assumed_climate_state[vin] = "OFF" + # Stop timer if active + if climate_timer.is_alive(): + climate_timer.cancel() + 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(True) def update_loop(): @@ -90,136 +92,56 @@ def update_car_data(force_update=False): global last_data_update last_data_update = format_datetime(datetime.now(), format="medium", locale=settings["babelLocale"]) for vin in volvo.vins: - for device_tracker in supported_device_trackers: - state = volvo.api_call(device_tracker["url"], "GET", vin, device_tracker["id"], force_update) - mqtt_client.publish( - f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/attributes", - json.dumps(state) - ) + for entity in volvo.supported_endpoints[vin]: + if entity["domain"] == "button": + continue - for lock in supported_locks: - 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": + if entity["id"] == "climate_status": state = assumed_climate_state[vin] - 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": + elif entity["id"] == "last_data_update": state = last_data_update 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( - f"homeassistant/sensor/{vin}_{sensor['id']}/state", - state + topic, + json.dumps(state) if isinstance(state, dict) else state ) def create_ha_devices(): + global subscribed_topics for vin in volvo.vins: device = volvo.get_vehicle_details(vin) - - for device_tracker in supported_device_trackers: + for entity in volvo.supported_endpoints[vin]: config = { - "name": device_tracker['name'], - "object_id": f"volvo_{vin}_{device_tracker['id']}", + "name": entity['name'], + "object_id": f"volvo_{vin}_{entity['id']}", "schema": "state", - "icon": f"mdi:{device_tracker['icon']}", - "state_topic": f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/state", + "icon": f"mdi:{entity['icon']}", + "state_topic": f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/state", "device": device, - "unique_id": f"volvoAAOS2mqtt_{vin}_{device_tracker['id']}", - "json_attributes_topic": f"homeassistant/device_tracker/{vin}_{device_tracker['id']}/attributes" + "unique_id": f"volvoAAOS2mqtt_{vin}_{entity['id']}" } - mqtt_client.publish( - f"homeassistant/device_tracker/volvoAAOS2mqtt/{vin}_{device_tracker['id']}/config", - json.dumps(config), - ) + if entity.get('unit'): + config["unit_of_measurement"] = entity["unit"] - for button in supported_buttons: - command_topic = f"homeassistant/button/{vin}_{button['id']}/command" - 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) + if entity.get('domain') == "device_tracker": + config["json_attributes_topic"] = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes" - 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( - 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"] + if entity.get('domain') in ["switch", "lock", "button"]: + command_topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/command" + config["command_topic"] = command_topic + subscribed_topics.append(command_topic) + mqtt_client.subscribe(command_topic) mqtt_client.publish( - f"homeassistant/sensor/volvoAAOS2mqtt/{vin}_{sensor['id']}/config", + f"homeassistant/{entity['domain']}/volvoAAOS2mqtt/{vin}_{entity['id']}/config", json.dumps(config), ) time.sleep(2) diff --git a/src/volvo.py b/src/volvo.py index 8f6932b..fb5f7b3 100644 --- a/src/volvo.py +++ b/src/volvo.py @@ -6,7 +6,7 @@ from babel.dates import format_datetime from util import keys_exists from const import charging_system_states, CLIMATE_START_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.headers = { @@ -24,6 +24,7 @@ window_cached_api_response = {} window_api_last_update = {} door_cached_api_response = {} door_api_last_update = {} +supported_endpoints = {} def authorize(): @@ -49,6 +50,7 @@ def authorize(): refresh_token = data["refresh_token"] get_vehicles() + check_supported_endpoints() else: message = auth.json() raise Exception(message["error_description"]) @@ -133,6 +135,20 @@ def get_vehicle_details(vin): 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): for vin in vins: 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): data = data["data"] 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": - 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": - 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": if keys_exists(data, "chargingSystemStatus"): charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]] @@ -263,7 +279,7 @@ def parse_api_data(data, sensor_id=None): else: return 0 else: - return "" + return None elif sensor_id == "estimated_charging_finish_time": if keys_exists(data, "chargingSystemStatus"): charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]] @@ -275,33 +291,33 @@ def parse_api_data(data, sensor_id=None): else: return "" else: - return "" + return None 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": - 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": coordinates = {} if keys_exists(data, "geometry"): @@ -312,4 +328,4 @@ def parse_api_data(data, sensor_id=None): "gps_accuracy": 1} return coordinates else: - return "" + return None