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

collect: migrate from weconnect to CarConnectivity + VW connector

Browse Source 
Replace the weconnect library with carconnectivity and
carconnectivity-connector-volkswagen.  The connector is configured via
a dict passed to CarConnectivity(); startup()/fetch_all() replace
login()/update().  All domain extractors in data_model.py are rewritten
for the new object model (vehicle.charging.*, vehicle.get_electric_drive(),
vehicle.climatization.*, etc.).  Remaining time fields are now derived
from estimated_date_reached DateAttributes.  Battery temperatures are
read directly in °C (no Kelvin conversion).  Auth error handling drops
the manual re-login since the connector manages token refresh internally.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jens
2026-05-27 08:02:34 +02:00
co-authored by Claude Sonnet 4.6
parent ef7b99e7e8
commit f050de7001
4 changed files with 122 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/data_model.py
+74 -90
View File
@@ -6,60 +6,73 @@ log = logging.getLogger(__name__)
# ── value helpers ─────────────────────────────────────────────────────────────
def _str(v) -> str:
return str(v.value) if hasattr(v, "value") else str(v)
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:
try:
d = vehicle.domains["charging"]
except KeyError:
return None
result: dict = {}
try:
cs = d["chargingStatus"]
result["chargingStatus"] = {
"chargingState": _str(cs.chargingState),
"chargePower": _phys(cs.chargePower_kW.value, "kW"),
"remainingChargingTimeToComplete": _phys(cs.remainingChargingTimeToComplete_min.value, "min"),
"chargeType": _str(cs.chargeType),
ch = vehicle.charging
status: dict = {
"chargingState": _str(ch.state),
"chargeType": _str(ch.type),
}
try:
status["chargePower"] = _phys(ch.power.value, "kW")
except Exception:
pass
try:
rem = _remaining_min(ch.estimated_date_reached)
if rem is not None:
status["remainingChargingTimeToComplete"] = _phys(round(rem), "min")
except Exception:
pass
result["chargingStatus"] = status
except Exception:
log.debug("chargingStatus unavailable", exc_info=True)
try:
bs = d["batteryStatus"]
ed = vehicle.get_electric_drive()
result["batteryStatus"] = {
"currentSOC": _phys(bs.currentSOC_pct.value, "%"),
"cruisingRangeElectric": _phys(bs.cruisingRangeElectric_km.value, "km"),
"currentSOC": _phys(ed.level.value, "%"),
"cruisingRangeElectric": _phys(ed.range.value, "km"),
}
except Exception:
log.debug("charging/batteryStatus unavailable", exc_info=True)
try:
cfg = d["chargingSettings"]
cfg = vehicle.charging.settings
result["chargingSettings"] = {
"targetSOC": _phys(cfg.targetSOC_pct.value, "%"),
"maxChargeCurrentAC": _str(cfg.maxChargeCurrentAC),
"autoUnlockPlugWhenCharged": _str(cfg.autoUnlockPlugWhenCharged),
"targetSOC": _phys(cfg.target_level.value, "%"),
"maxChargeCurrentAC": _phys(cfg.maximum_current.value, "A"),
"autoUnlockPlugWhenCharged": _str(cfg.auto_unlock),
}
except Exception:
log.debug("chargingSettings unavailable", exc_info=True)
try:
ps = d["plugStatus"]
conn = vehicle.charging.connector
result["plugStatus"] = {
"plugConnectionState": _str(ps.plugConnectionState),
"plugLockState": _str(ps.plugLockState),
"externalPower": _str(ps.externalPower),
"plugConnectionState": _str(conn.connection_state),
"plugLockState": _str(conn.lock_state),
"externalPower": _str(conn.external_power),
}
except Exception:
log.debug("plugStatus unavailable", exc_info=True)
@@ -68,36 +81,34 @@ def extract_charging(vehicle) -> dict | None:
def extract_climatisation(vehicle) -> dict | None:
try:
d = vehicle.domains["climatisation"]
except KeyError:
return None
result: dict = {}
try:
cl = d["climatisationStatus"]
result["climatisationStatus"] = {
"climatisationState": _str(cl.climatisationState),
"remainingClimatisationTime": _phys(cl.remainingClimatisationTime_min.value, "min"),
}
cl = vehicle.climatization
status: dict = {"climatisationState": _str(cl.state)}
try:
rem = _remaining_min(cl.estimated_date_reached)
if rem is not None:
status["remainingClimatisationTime"] = _phys(round(rem), "min")
except Exception:
pass
result["climatisationStatus"] = status
except Exception:
log.debug("climatisationStatus unavailable", exc_info=True)
try:
cs = d["climatisationSettings"]
cs = vehicle.climatization.settings
result["climatisationSettings"] = {
"targetTemperature": _phys(cs.targetTemperature_C.value, "degC"),
"unitInCar": _str(cs.unitInCar),
"targetTemperature": _phys(cs.target_temperature.value, "degC"),
}
except Exception:
log.debug("climatisationSettings unavailable", exc_info=True)
try:
wh = d["windowHeatingStatus"]
wh = vehicle.window_heatings
result["windowHeatingStatus"] = {
name: _str(win.windowHeatingState)
for name, win in wh.windows.items()
name: _str(win.heating_state)
for name, win in wh.heatings.items()
}
except Exception:
log.debug("windowHeatingStatus unavailable", exc_info=True)
@@ -106,41 +117,35 @@ def extract_climatisation(vehicle) -> dict | None:
def extract_measurements(vehicle) -> dict | None:
try:
d = vehicle.domains["measurements"]
except KeyError:
return None
result: dict = {}
try:
od = d["odometerStatus"]
result["odometerStatus"] = {"odometer": _phys(od.odometer.value, "km")}
result["odometerStatus"] = {"odometer": _phys(vehicle.odometer.value, "km")}
except Exception:
log.debug("odometerStatus unavailable", exc_info=True)
try:
rs = d["rangeStatus"]
ed = vehicle.get_electric_drive()
electric_range = ed.range.value
result["rangeStatus"] = {
"electricRange": _phys(rs.electricRange.value, "km"),
"totalRange": _phys(rs.totalRange_km.value, "km"),
"electricRange": _phys(electric_range, "km"),
"totalRange": _phys(electric_range, "km"),
}
except Exception:
log.debug("rangeStatus unavailable", exc_info=True)
try:
bs = d["temperatureBatteryStatus"]
bat = vehicle.get_electric_drive().battery
result["temperatureBatteryStatus"] = {
"temperatureHvBatteryMax": _phys(float(bs.temperatureHvBatteryMax_K.value - 273.15), "degC"),
"temperatureHvBatteryMin": _phys(float(bs.temperatureHvBatteryMin_K.value - 273.15), "degC"),
"temperatureHvBatteryMax": _phys(bat.temperature_max.value, "degC"),
"temperatureHvBatteryMin": _phys(bat.temperature_min.value, "degC"),
}
except Exception:
log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True)
try:
temp = d["temperatureOutsideStatus"]
result["temperatureOutsideStatus"] = {
"temperatureOutside": _phys(temp.temperatureOutside_C.value, "degC"),
"temperatureOutside": _phys(vehicle.outside_temperature.value, "degC"),
}
except Exception:
log.debug("temperatureOutsideStatus unavailable", exc_info=True)
@@ -149,25 +154,14 @@ def extract_measurements(vehicle) -> dict | None:
def extract_readiness(vehicle) -> dict | None:
try:
d = vehicle.domains["readiness"]
except KeyError:
return None
result: dict = {}
try:
rs = d["readinessStatus"]
conn = rs.connectionState
warn = rs.connectionWarning
conn_state = str(vehicle.connection_state)
result["readinessStatus"] = {
"connectionState": {
"isOnline": conn.isOnline.value,
"isActive": conn.isActive.value,
},
"connectionWarning": {
"insufficientBatteryLevelWarning": warn.insufficientBatteryLevelWarning.value,
"insufficientBatteryLevelError": warn.insufficientBatteryLevelError.value,
"isOnline": conn_state == "online",
"isActive": conn_state in ("online", "reachable"),
},
}
except Exception:
@@ -177,18 +171,13 @@ def extract_readiness(vehicle) -> dict | None:
def extract_parking(vehicle) -> dict | None:
try:
d = vehicle.domains["parking"]
except KeyError:
return None
result: dict = {}
try:
pos = d["parkingPosition"]
pos = vehicle.position
result["parkingPosition"] = {
"lat": pos.lat.value,
"lon": pos.lon.value,
"lat": pos.latitude.value,
"lon": pos.longitude.value,
}
except Exception:
log.debug("parkingPosition unavailable", exc_info=True)
@@ -197,27 +186,22 @@ def extract_parking(vehicle) -> dict | None:
def extract_access(vehicle) -> dict | None:
try:
d = vehicle.domains["access"]
except KeyError:
return None
result: dict = {}
try:
acc = d["accessStatus"]
doors = vehicle.doors
result["accessStatus"] = {
"overallStatus": _str(acc.overallStatus),
"overallStatus": _str(doors.lock_state),
"doors": {
name: {
"lockState": _str(door.lockState),
"openState": _str(door.openState),
"lockState": _str(door.lock_state),
"openState": _str(door.open_state),
}
for name, door in acc.doors.items()
for name, door in doors.doors.items()
},
"windows": {
name: {"openState": _str(win.openState)}
for name, win in acc.windows.items()
name: {"openState": _str(win.open_state)}
for name, win in vehicle.windows.windows.items()
},
}
except Exception:
@@ -283,4 +267,4 @@ def apply_procedural(record: dict) -> dict:
if proc:
record["procedural"] = proc
return record
return record