jens/we_monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

server: replace carconnectivity with volkswagencarnet

Browse Source 
Switches the VW API backend from carconnectivity + connector-volkswagen
to volkswagencarnet (local fork with the cariad-hybrid-auth-fix patch
applied), fixing authentication against the current CARIAD BFF endpoints.

- requirements.txt: point at local volkswagencarnet git repo via
  git+file:// URL (editable path no longer needed now the patch is
  committed there)
- we_connect.py: rewrite around the async volkswagencarnet Connection
  class; a persistent asyncio event loop held in _Session lets the rest
  of the code stay synchronous
- data_model.py: remap all extractors to volkswagencarnet Vehicle
  properties and Paths constants; output shape preserved so
  apply_procedural and the GUI client continue to work unchanged
- collect.py: swap carconnectivity error import, drop
  TemporaryAuthenticationError (no equivalent), fix vehicle.vin.value →
  vehicle.vin

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jens
2026-05-28 22:50:57 +02:00
co-authored by Claude Sonnet 4.6
parent 63e807bdb0
commit bc424310b5
4 changed files with 223 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/data_model.py
+169 -105
View File
@@ -2,68 +2,83 @@ import logging
from collections.abc import Callable
from datetime import datetime, timezone
from volkswagencarnet.vw_const import Paths
from volkswagencarnet.vw_utilities import find_path
log = logging.getLogger(__name__)
# ── value helpers ─────────────────────────────────────────────────────────────
def _str(attr) -> str:
"""Return the string value of a CarConnectivity attribute."""
return str(attr)
def _phys(value, unit: str) -> dict:
return {"value": value, "unit": unit}
def _remaining_min(estimated_date_reached_attr) -> float | None:
"""Compute remaining minutes from an estimated_date_reached DateAttribute."""
edr = estimated_date_reached_attr.value
if edr is None:
return None
return max(0.0, (edr - datetime.now(timezone.utc)).total_seconds() / 60)
# ── domain extractors ─────────────────────────────────────────────────────────
def extract_charging(vehicle) -> dict | None:
result: dict = {}
try:
ch = vehicle.charging
result["state"] = _str(ch.state)
result["type"] = _str(ch.type)
try:
result["power"] = _phys(ch.power.value, "kW")
except Exception:
pass
try:
rem = _remaining_min(ch.estimated_date_reached)
if rem is not None:
result["remain"] = _phys(round(rem), "min")
except Exception:
pass
result["state"] = vehicle.charging_state
except Exception:
log.debug("vehicle.charging unavailable", exc_info=True)
log.debug("charging.state unavailable", exc_info=True)
try:
cfg = vehicle.charging.settings
result["settings"] = {
"target_level": _phys(cfg.target_level.value, "%"),
"maximum_current": _phys(cfg.maximum_current.value, "A"),
"auto_unlock": _str(cfg.auto_unlock),
}
if vehicle.is_charger_type_supported:
result["type"] = vehicle.charger_type
except Exception:
log.debug("vehicle.charging.settings unavailable", exc_info=True)
pass
try:
conn = vehicle.charging.connector
result["plugStatus"] = {
"connection_state": _str(conn.connection_state),
"lock_state": _str(conn.lock_state),
"external_power": _str(conn.external_power),
}
power = vehicle.charging_power
if power is not None:
result["power"] = _phys(power, "kW")
except Exception:
log.debug("charging.power unavailable", exc_info=True)
try:
remain = vehicle.charging_time_left
if remain is not None:
result["remain"] = _phys(remain, "min")
except Exception:
log.debug("charging.remain unavailable", exc_info=True)
settings: dict = {}
try:
target = vehicle.battery_target_charge_level
if target is not None:
settings["target_level"] = _phys(target, "%")
except Exception:
pass
try:
max_a = vehicle.charge_max_ac_ampere
if max_a is not None:
settings["maximum_current"] = _phys(max_a, "A")
except Exception:
pass
try:
auto_unlock = vehicle.auto_release_ac_connector_state
if auto_unlock is not None:
settings["auto_unlock"] = auto_unlock
except Exception:
pass
if settings:
result["settings"] = settings
try:
plug: dict = {}
conn_state = find_path(vehicle.attrs, Paths.PLUG_CONN)
if conn_state is not None:
plug["connection_state"] = conn_state
lock_state = find_path(vehicle.attrs, Paths.PLUG_LOCK)
if lock_state is not None:
plug["lock_state"] = lock_state
ext_pwr = find_path(vehicle.attrs, Paths.PLUG_EXT_PWR)
if ext_pwr is not None:
plug["external_power"] = ext_pwr
if plug:
result["plugStatus"] = plug
except Exception:
log.debug("plugStatus unavailable", exc_info=True)
@@ -74,40 +89,39 @@ def extract_climatisation(vehicle) -> dict | None:
result: dict = {}
try:
cl = vehicle.climatization
result["state"] = _str(cl.state)
try:
rem = _remaining_min(cl.estimated_date_reached)
if rem is not None:
result["remain"] = _phys(round(rem), "min")
except Exception:
pass
state = vehicle.climatisation_state
if state is not None:
result["state"] = state
except Exception:
log.debug("vehicle.climatization unavailable", exc_info=True)
log.debug("climatisation.state unavailable", exc_info=True)
try:
cs = vehicle.climatization.settings
result["settings"] = {
"target_temperature": _phys(cs.target_temperature.value, "degC"),
}
remain = vehicle.electric_remaining_climatisation_time
if remain is not None:
result["remain"] = _phys(remain, "min")
except Exception:
log.debug("climatization.settings unavailable", exc_info=True)
pass
try:
wh = vehicle.window_heatings
result["window_heatings"] = {
name: _str(win.heating_state)
for name, win in wh.heatings.items()
}
target = vehicle.climatisation_target_temperature
if target is not None:
result["settings"] = {"target_temperature": _phys(target, "degC")}
except Exception:
log.debug("vehicle.window_heatings unavailable", exc_info=True)
log.debug("climatisation.settings unavailable", exc_info=True)
try:
result["environment"] = {
"temperature_outside": _phys(vehicle.outside_temperature.value, "degC"),
}
window_status = find_path(vehicle.attrs, Paths.CLIMATISATION_WINDOW_HEATING_STATUS)
if window_status:
heatings: dict = {}
for entry in window_status:
loc = entry.get("windowLocation")
state = entry.get("windowHeatingState")
if loc and state is not None:
heatings[loc] = state
if heatings:
result["window_heatings"] = heatings
except Exception:
log.debug("vehicle.outside_temperature unavailable", exc_info=True)
log.debug("window_heatings unavailable", exc_info=True)
return result or None
@@ -116,23 +130,45 @@ def extract_electric_drive(vehicle) -> dict | None:
result: dict = {}
try:
ed = vehicle.get_electric_drive()
result["range"] = _phys(ed.range.value, "km")
result["range_full"] = _phys(round(float(ed.range_estimated_full.value), 1), "km")
try:
bat = ed.battery
result["battery"] = {
"soc": _phys(ed.level.value, "%"),
"temperature_max": _phys(round(float(bat.temperature_max.value) - 273.15, 1), "degC"),
"temperature_min": _phys(round(float(bat.temperature_min.value) - 273.15, 1), "degC"),
}
except Exception:
log.debug("electric_drive/battery unavailable", exc_info=True)
rng = vehicle.electric_range
if rng is not None:
result["range"] = _phys(rng, "km")
except Exception:
log.debug("electric_drive unavailable", exc_info=True)
log.debug("electric_range unavailable", exc_info=True)
try:
result["odometer"] = {"odometer": _phys(vehicle.odometer.value, "km")}
rng_full = vehicle.battery_cruising_range
if rng_full is not None:
result["range_full"] = _phys(round(float(rng_full), 1), "km")
except Exception:
log.debug("battery_cruising_range unavailable", exc_info=True)
battery: dict = {}
try:
soc = vehicle.battery_level
if soc is not None:
battery["soc"] = _phys(soc, "%")
except Exception:
log.debug("battery.soc unavailable", exc_info=True)
try:
tmax = vehicle.hv_battery_max_temperature
if tmax is not None:
battery["temperature_max"] = _phys(round(tmax, 1), "degC")
except Exception:
pass
try:
tmin = vehicle.hv_battery_min_temperature
if tmin is not None:
battery["temperature_min"] = _phys(round(tmin, 1), "degC")
except Exception:
pass
if battery:
result["battery"] = battery
try:
odo = vehicle.distance
if odo is not None:
result["odometer"] = {"odometer": _phys(odo, "km")}
except Exception:
log.debug("odometer unavailable", exc_info=True)
@@ -143,32 +179,43 @@ def extract_connectivity(vehicle) -> dict | None:
result: dict = {}
try:
state = str(vehicle.connection_state)
result["state"] = state
is_online = vehicle.connection_state_is_online
is_active = vehicle.connection_state_is_active
if is_online:
result["state"] = "online"
elif is_active:
result["state"] = "active"
else:
result["state"] = "offline"
except Exception:
log.debug("connectivity.state unavailable", exc_info=True)
return result or None
def extract_vehicle(vehicle) -> dict | None:
result: dict = {}
try:
state = str(vehicle.state)
result["state"] = state
overall = find_path(vehicle.attrs, Paths.ACCESS_OVERALL_STATUS)
if overall is not None:
result["state"] = overall
except Exception:
log.debug("vehicle.state unavailable", exc_info=True)
return result or None
def extract_position(vehicle) -> dict | None:
result: dict = {}
try:
pos = vehicle.position
result["lat"] = pos.latitude.value
result["lon"] = pos.longitude.value
lat = pos.get("lat")
lng = pos.get("lng")
if lat is not None and lat != "?" and lng is not None and lng != "?":
result["lat"] = lat
result["lon"] = lng # volkswagencarnet uses "lng" internally; we keep "lon" for compat
except Exception:
log.debug("position unavailable", exc_info=True)
@@ -179,20 +226,29 @@ def extract_doors(vehicle) -> dict | None:
result: dict = {}
try:
doors = vehicle.doors
if doors.lock_state.value is not None:
result["lock_state"] = _str(doors.lock_state)
if doors.open_state.value is not None:
result["open_state"] = _str(doors.open_state)
door_entries = {}
for name, door in doors.doors.items():
entry = {}
if door.lock_state.value is not None:
entry["lock_state"] = _str(door.lock_state)
if door.open_state.value is not None:
entry["open_state"] = _str(door.open_state)
lock_state = find_path(vehicle.attrs, Paths.ACCESS_DOOR_LOCK)
if lock_state is not None:
result["lock_state"] = lock_state
doors_list = find_path(vehicle.attrs, Paths.ACCESS_DOORS) or []
door_entries: dict = {}
all_closed: bool | None = None
for door in doors_list:
name = door.get("name")
status = door.get("status") or []
if not name:
continue
entry: dict = {}
if "closed" in status or "open" in status:
open_state = "closed" if "closed" in status else "open"
entry["open_state"] = open_state
all_closed = open_state == "closed" if all_closed is None else (all_closed and open_state == "closed")
if "locked" in status or "unlocked" in status:
entry["lock_state"] = "locked" if "locked" in status else "unlocked"
if entry:
door_entries[name] = entry
if all_closed is not None:
result["open_state"] = "closed" if all_closed else "open"
if door_entries:
result["doors"] = door_entries
except Exception:
@@ -205,12 +261,20 @@ def extract_windows(vehicle) -> dict | None:
result: dict = {}
try:
if vehicle.windows.open_state.value is not None:
result["open_state"] = _str(vehicle.windows.open_state)
window_entries = {}
for name, win in vehicle.windows.windows.items():
if win.open_state.value is not None:
window_entries[name] = {"open_state": _str(win.open_state)}
windows_list = find_path(vehicle.attrs, Paths.ACCESS_WINDOWS) or []
window_entries: dict = {}
all_closed: bool | None = None
for win in windows_list:
name = win.get("name")
status = win.get("status") or []
if not name or "unsupported" in status:
continue
if "closed" in status or "open" in status:
open_state = "closed" if "closed" in status else "open"
window_entries[name] = {"open_state": open_state}
all_closed = open_state == "closed" if all_closed is None else (all_closed and open_state == "closed")
if all_closed is not None:
result["open_state"] = "closed" if all_closed else "open"
if window_entries:
result["windows"] = window_entries
except Exception:
@@ -278,4 +342,4 @@ def apply_procedural(record: dict) -> dict:
if proc:
record["procedural"] = proc
return record
return record