Add warnings endpoint #151

src/CHANGELOG.md

@@ -2,6 +2,7 @@
 ### 🚀 Features:
 
 - Add option for dynamic interval settings #160
+- Add warnings endpoint #151
 
 ## v1.8.23
 ### 🚀 Features:

src/const.py

@@ -133,6 +133,7 @@ supported_entities = [
                         {"name": "Washer Fluid Level warning", "domain": "sensor", "id": "washer_fluid_warning", "icon": "alert-outline", "url": VEHICLE_DIAGNOSTICS_URL},
                         {"name": "API Backend status", "domain": "sensor", "id": "api_backend_status", "icon": "alert"},
                         {"name": "Update Interval", "domain": "number", "id": "update_interval", "unit": "seconds", "icon": "timer", "min": 60, "max": 600, "mode": "box"},
+                        {"name": "Warnings", "domain": "sensor", "id": "warnings", "icon": "alert", "url": WARNINGS_URL}
 ]
 
 old_entity_ids = ["months_to_service", "service_warning_trigger", "distance_to_empty"]

src/mqtt.py

@@ -260,6 +260,13 @@ def update_car_data(force_update=False, overwrite={}):
 
             if entity["domain"] == "device_tracker" or entity["id"] == "active_schedules":
                 topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes"
+            elif entity["id"] == "warnings":
+                mqtt_client.publish(
+                    f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes",
+                    json.dumps(state)
+                )
+                state = sum(value == "FAILURE" for value in state.values())
+                topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/state"
             else:
                 topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/state"
 
@@ -340,6 +347,7 @@ def create_ha_devices():
                 "unique_id": f"volvoAAOS2mqtt_{vin}_{entity['id']}",
                 "availability_topic": availability_topic
             }
+
             if entity.get("device_class"):
                 config["device_class"] = entity["device_class"]
 
@@ -349,9 +357,11 @@ def create_ha_devices():
             if entity.get("state_class"):
                 config["state_class"] = entity["state_class"]
 
-            if entity.get("domain") == "device_tracker" or entity.get("id") == "active_schedules":
+            if (entity.get("domain") == "device_tracker" or entity.get("id") == "active_schedules"
+                    or entity.get("id") == "warnings"):
                 config["json_attributes_topic"] = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/attributes"
-            elif entity.get("domain") in ["switch", "lock", "button", "number"]:
+
+            if entity.get("domain") in ["switch", "lock", "button", "number"]:
                 command_topic = f"homeassistant/{entity['domain']}/{vin}_{entity['id']}/command"
                 config["command_topic"] = command_topic
                 subscribed_topics.append(command_topic)

src/volvo.py

@@ -12,7 +12,7 @@ from json import JSONDecodeError
 from const import charging_system_states, charging_connection_states, door_states, window_states, \
     OAUTH_URL, VEHICLES_URL, VEHICLE_DETAILS_URL, RECHARGE_STATE_URL, CLIMATE_START_URL, \
     WINDOWS_STATE_URL, LOCK_STATE_URL, TYRE_STATE_URL, supported_entities, FUEL_BATTERY_STATE_URL, \
-    STATISTICS_URL, ENGINE_DIAGNOSTICS_URL, API_BACKEND_STATUS, engine_states
+    STATISTICS_URL, ENGINE_DIAGNOSTICS_URL, API_BACKEND_STATUS, WARNINGS_URL, engine_states
 
 session = requests.Session()
 session.headers = {
@@ -346,7 +346,7 @@ def api_call(url, method, vin, sensor_id=None, force_update=False, key_change=Fa
         refresh_auth()
 
     if url in [RECHARGE_STATE_URL, WINDOWS_STATE_URL, LOCK_STATE_URL, TYRE_STATE_URL,
-               STATISTICS_URL, ENGINE_DIAGNOSTICS_URL, FUEL_BATTERY_STATE_URL]:
+               STATISTICS_URL, ENGINE_DIAGNOSTICS_URL, FUEL_BATTERY_STATE_URL, WARNINGS_URL]:
         # Minimize API calls for endpoints with multiple values
         response = cached_request(url, method, vin, force_update, key_change)
         if response is None:
@@ -595,5 +595,17 @@ def parse_api_data(data, sensor_id=None):
         return data["averageEnergyConsumption"]["value"] if util.keys_exists(data, "averageEnergyConsumption") else None
     elif sensor_id == "washer_fluid_warning":
         return data["washerFluidLevelWarning"]["value"] if util.keys_exists(data, "washerFluidLevelWarning") else None
+    elif sensor_id == "warnings":
+        warnings = 0
+        cleaned_data = {}
+        for key, dicts in data.items():
+            contains_data = sum(value == "NO_WARNING" or value == "FAILURE" for value in dicts.values())
+            if contains_data:
+                warnings = warnings + 1
+                cleaned_data[key] = dicts["value"]
+
+        return cleaned_data if warnings > 0 else None
     else:
         return None
+
+