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

Move energy api to v2

Browse Source
This commit is contained in:
linus
2025-11-16 08:36:30 +01:00
parent f95348769b
commit 7acfdea1d8
5 changed files with 220 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/CHANGELOG.md
+6
View File
@@ -1,3 +1,9 @@
## v1.13.0
### 🚀 Features:
- Move energy api to v2 (https://developer.volvocars.com/apis/energy/v2/release-notes/)
## v1.12.2
### 🚀 Features:
src/config.yaml
+1 -1
View File
@@ -1,6 +1,6 @@
name: "Volvo2Mqtt"
description: "Volvo AAOS MQTT bridge"
version: "1.12.2"
version: "1.13.0"
slug: "volvo2mqtt"
init: false
url: "https://github.com/Dielee/volvo2mqtt"
src/const.py
+7 -8
View File
@@ -1,6 +1,6 @@
from config import settings
VERSION = "v1.12.2"
VERSION = "v1.13.0"
OAUTH_TOKEN_URL = "https://volvoid.eu.volvocars.com/as/token.oauth2"
OAUTH_AUTH_URL = "https://volvoid.eu.volvocars.com/as/authorization.oauth2"
@@ -12,7 +12,7 @@ CLIMATE_STOP_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/
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_STATE_URL = "https://api.volvocars.com/energy/v2/vehicles/{0}/recharge-status"
RECHARGE_STATE_URL = "https://api.volvocars.com/energy/v2/vehicles/{0}/state"
ODOMETER_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/odometer"
LOCATION_STATE_URL = "https://api.volvocars.com/location/v1/vehicles/{0}/location"
TYRE_STATE_URL = "https://api.volvocars.com/connected-vehicle/v2/vehicles/{0}/tyres"
@@ -59,13 +59,12 @@ units = {
availability_topic = "volvoAAOS2mqtt/availability"
charging_system_states = {"CHARGING_SYSTEM_CHARGING": "Charging", "CHARGING_SYSTEM_IDLE": "Idle",
"CHARGING_SYSTEM_FAULT": "Fault", "CHARGING_SYSTEM_UNSPECIFIED": "Unspecified",
"CHARGING_SYSTEM_DONE": "Done", "CHARGING_SYSTEM_SCHEDULED": "Scheduled"}
charging_system_states = {"CHARGING": "Charging", "IDLE": "Idle",
"FAULT": "Fault", "UNSPECIFIED": "Unspecified",
"DONE": "Done", "SCHEDULED": "Scheduled"}
charging_connection_states = {"CONNECTION_STATUS_DISCONNECTED": "Disconnected", "CONNECTION_STATUS_UNSPECIFIED": "Unspecified",
"CONNECTION_STATUS_CONNECTED_DC": "Connected DC", "CONNECTION_STATUS_CONNECTED_AC": "Connected AC",
"CONNECTION_STATUS_FAULT": "Fault"}
charging_connection_states = {"DISCONNECTED": "Disconnected", "UNSPECIFIED": "Unspecified",
"FAULT": "Fault", "CONNECTED": "Connected"}
window_states = {"CLOSED": "OFF", "OPEN": "ON", "UNSPECIFIED": "UNKNOWN", "AJAR": "ON"}
door_states = {"CLOSED": "OFF", "OPEN": "ON", "UNSPECIFIED": "UNKNOWN", "AJAR": "ON"}
src/mqtt.py
+13
View File
@@ -176,16 +176,24 @@ def on_message(client, userdata, msg):
if "climate_status" in msg.topic:
if payload == "ON":
start_climate(vin)
return None
elif payload == "OFF":
stop_climate(vin)
return None
return None
elif "lock_status" in msg.topic:
if payload == "LOCK":
lock_car(vin)
return None
elif payload == "UNLOCK":
unlock_car(vin)
return None
return None
elif "update_data" in msg.topic:
if payload == "PRESS":
update_car_data(True)
return None
return None
elif "update_interval" in msg.topic:
try:
update_interval = int(payload)
@@ -194,8 +202,10 @@ def on_message(client, userdata, msg):
else:
logging.warning(f"Interval {update_interval} seconds is to low. Doing nothing!")
update_car_data()
return None
except ValueError as error:
logging.error(f"Unable to change update_interval {error} payload={payload}")
return None
elif "schedule" in msg.topic:
try:
d = json.loads(payload)
@@ -205,8 +215,11 @@ def on_message(client, userdata, msg):
if d["mode"] == "timer":
start_climate_timer(d, vin)
return None
else:
logging.warning("No schedule mode found, doing nothing")
return None
return None
def start_climate_timer(d, vin):
src/volvo.py
+22 -16
View File
@@ -600,43 +600,50 @@ def cached_request(url, method, vin, force_update=False, key_change=False):
def parse_api_data(data, sensor_id=None):
if sensor_id != "api_backend_status":
data = data["data"]
if sensor_id in ["battery_charge_level", "electric_range", "charging_system_status",
"charging_connection_status", "estimated_charging_time", "estimated_charging_finish_time"]:
if sensor_id == "battery_charge_level":
return data["batteryChargeLevel"]["value"] if util.keys_exists(data, "batteryChargeLevel") else None
elif sensor_id == "battery_capacity":
return data["batteryCapacityKWH"] if util.keys_exists(data, "batteryCapacityKWH") else None
elif sensor_id == "electric_range":
return util.convert_metric_values(data["electricRange"]["value"]) \
if util.keys_exists(data, "electricRange") else None
elif sensor_id == "charging_system_status":
return charging_system_states[data["chargingSystemStatus"]["value"]] \
if util.keys_exists(data, "chargingSystemStatus") else None
return charging_system_states[data["chargingStatus"]["value"]] \
if util.keys_exists(data, "chargingStatus") else None
elif sensor_id == "charging_connection_status":
return charging_connection_states[data["chargingConnectionStatus"]["value"]] \
if util.keys_exists(data, "chargingConnectionStatus") else None
return charging_connection_states[data["chargerConnectionStatus"]["value"]] \
if util.keys_exists(data, "chargerConnectionStatus") else None
elif sensor_id == "estimated_charging_time":
if util.keys_exists(data, "chargingSystemStatus"):
charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]]
if util.keys_exists(data, "chargingStatus"):
charging_system_state = charging_system_states[data["chargingStatus"]["value"]]
if charging_system_state == "Charging":
return data["estimatedChargingTime"]["value"] if util.keys_exists(data, "estimatedChargingTime") else ""
return data["estimatedChargingTimeToTargetBatteryChargeLevel"]["value"] if util.keys_exists(data,
"estimatedChargingTimeToTargetBatteryChargeLevel") else ""
else:
return 0
return None
elif sensor_id == "estimated_charging_finish_time":
if util.keys_exists(data, "chargingSystemStatus"):
charging_system_state = charging_system_states[data["chargingSystemStatus"]["value"]]
if util.keys_exists(data, "estimatedChargingTimeToTargetBatteryChargeLevel"):
charging_system_state = charging_system_states[data["chargingStatus"]["value"]]
if charging_system_state == "Charging":
charging_time = int(
data["estimatedChargingTime"]["value"] if util.keys_exists(data, "estimatedChargingTime")
data["estimatedChargingTimeToTargetBatteryChargeLevel"]["value"] if util.keys_exists(data, "estimatedChargingTimeToTargetBatteryChargeLevel")
else 0)
charging_finished = datetime.now(util.TZ) + timedelta(minutes=charging_time)
return format_datetime(charging_finished, format="medium", locale=settings["babelLocale"])
else:
return ""
return None
elif sensor_id == "lock_status":
return None
else:
if sensor_id != "api_backend_status":
data = data["data"]
if sensor_id == "lock_status":
return data["centralLock"]["value"] if util.keys_exists(data, "centralLock") else None
elif sensor_id == "battery_capacity":
return data["batteryCapacityKWH"] if util.keys_exists(data, "batteryCapacityKWH") else None
elif sensor_id == "odometer":
return util.convert_metric_values(int(data["odometer"]["value"])) \
if util.keys_exists(data, "odometer") else None
@@ -788,7 +795,6 @@ def parse_api_data(data, sensor_id=None):
if trip_speed != 0:
return util.convert_metric_values(trip_speed)
return None
else:
return None