From bf33737654e99f3456b0eb00f82665273f344a5c Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Wed, 27 May 2026 09:01:15 +0200 Subject: [PATCH] data_model: redesign domain extractors for CarConnectivity object model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flatten the snapshot structure so each domain maps directly to its fields rather than mirroring the old WeConnect status-object nesting. New domains: electric_drive (range, battery/SOC/temps, odometer), connectivity, vehicle, position, doors — replacing the old readiness, parking, access, and measurements domains. Update apply_procedural paths and collect.py defaults accordingly. Co-Authored-By: Claude Sonnet 4.6 --- collect.py | 6 +- server/data_model.py | 161 ++++++++++++++++++++----------------------- 2 files changed, 79 insertions(+), 88 deletions(-) diff --git a/collect.py b/collect.py index a633c15..35f5276 100644 --- a/collect.py +++ b/collect.py @@ -18,7 +18,7 @@ Usage examples # Override domains on the CLI: python collect.py -c credentials/alex.json -d charging,measurements -Available domains: charging, climatisation, measurements, readiness, parking, access +Available domains: charging, climatisation, electric_drive, connectivity, vehicle, position, doors Credentials file format (JSON) ------------------------------- @@ -191,7 +191,7 @@ def main() -> None: help=( "Comma-separated domains to collect, or 'all' (overrides credentials file). " f"Available: {', '.join(ALL_DOMAINS)}. " - "Default: charging,measurements,readiness" + "Default: charging,electric_drive,connectivity" ), ) p.add_argument( @@ -273,7 +273,7 @@ def main() -> None: if isinstance(raw_domains, list): args.domains = ",".join(raw_domains) else: - args.domains = raw_domains or "charging,measurements,readiness" + args.domains = raw_domains or "charging,electric_drive,connectivity" if args.interval is None: args.interval = int(creds_data.get("interval_s", 300)) diff --git a/server/data_model.py b/server/data_model.py index 6472792..6214557 100644 --- a/server/data_model.py +++ b/server/data_model.py @@ -1,4 +1,5 @@ import logging +from collections.abc import Callable from datetime import datetime, timezone log = logging.getLogger(__name__) @@ -30,49 +31,38 @@ def extract_charging(vehicle) -> dict | None: try: ch = vehicle.charging - status: dict = { - "chargingState": _str(ch.state), - "chargeType": _str(ch.type), - } + result["state"] = _str(ch.state) + result["type"] = _str(ch.type) try: - status["chargePower"] = _phys(ch.power.value, "kW") + result["power"] = _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") + result["remain"] = _phys(round(rem), "min") except Exception: pass - result["chargingStatus"] = status - except Exception: - log.debug("chargingStatus unavailable", exc_info=True) - try: - ed = vehicle.get_electric_drive() - result["batteryStatus"] = { - "currentSOC": _phys(ed.level.value, "%"), - "cruisingRangeElectric": _phys(ed.range.value, "km"), - } except Exception: - log.debug("charging/batteryStatus unavailable", exc_info=True) + log.debug("vehicle.charging unavailable", exc_info=True) try: cfg = vehicle.charging.settings - result["chargingSettings"] = { - "targetSOC": _phys(cfg.target_level.value, "%"), - "maxChargeCurrentAC": _phys(cfg.maximum_current.value, "A"), - "autoUnlockPlugWhenCharged": _str(cfg.auto_unlock), + result["settings"] = { + "target_level": _phys(cfg.target_level.value, "%"), + "maximum_current": _phys(cfg.maximum_current.value, "A"), + "auto_unlock": _str(cfg.auto_unlock), } except Exception: - log.debug("chargingSettings unavailable", exc_info=True) + log.debug("vehicle.charging.settings unavailable", exc_info=True) try: conn = vehicle.charging.connector result["plugStatus"] = { - "plugConnectionState": _str(conn.connection_state), - "plugLockState": _str(conn.lock_state), - "externalPower": _str(conn.external_power), + "connection_state": _str(conn.connection_state), + "lock_state": _str(conn.lock_state), + "external_power": _str(conn.external_power), } except Exception: log.debug("plugStatus unavailable", exc_info=True) @@ -85,113 +75,113 @@ def extract_climatisation(vehicle) -> dict | None: try: cl = vehicle.climatization - status: dict = {"climatisationState": _str(cl.state)} + result["state"] = _str(cl.state) try: rem = _remaining_min(cl.estimated_date_reached) if rem is not None: - status["remainingClimatisationTime"] = _phys(round(rem), "min") + result["remain"] = _phys(round(rem), "min") except Exception: pass - result["climatisationStatus"] = status except Exception: - log.debug("climatisationStatus unavailable", exc_info=True) + log.debug("vehicle.climatization unavailable", exc_info=True) try: cs = vehicle.climatization.settings - result["climatisationSettings"] = { - "targetTemperature": _phys(cs.target_temperature.value, "degC"), + result["settings"] = { + "target_temperature": _phys(cs.target_temperature.value, "degC"), } except Exception: - log.debug("climatisationSettings unavailable", exc_info=True) + log.debug("climatization.settings unavailable", exc_info=True) try: wh = vehicle.window_heatings - result["windowHeatingStatus"] = { + result["window_heatings"] = { name: _str(win.heating_state) for name, win in wh.heatings.items() } except Exception: - log.debug("windowHeatingStatus unavailable", exc_info=True) + log.debug("vehicle.window_heatings unavailable", exc_info=True) + + try: + result["environment"] = { + "temperature_outside": _phys(vehicle.outside_temperature.value, "degC"), + } + except Exception: + log.debug("vehicle.outside_temperature unavailable", exc_info=True) return result or None -def extract_measurements(vehicle) -> dict | None: +def extract_electric_drive(vehicle) -> dict | None: result: dict = {} - try: - result["odometerStatus"] = {"odometer": _phys(vehicle.odometer.value, "km")} - except Exception: - log.debug("odometerStatus unavailable", exc_info=True) - try: ed = vehicle.get_electric_drive() - electric_range = ed.range.value - result["rangeStatus"] = { - "electricRange": _phys(electric_range, "km"), - "totalRange": _phys(electric_range, "km"), - } + result["range"] = _phys(ed.range.value, "km") + result["range_full"] = _phys(ed.range_estimated_full.value, "km") + try: + bat = ed.battery + result["battery"] = { + "soc": _phys(ed.level.value, "%"), + "temperature_max": _phys(bat.temperature_max.value, "degC"), + "temperature_min": _phys(bat.temperature_min.value, "degC"), + } + except Exception: + log.debug("electric_drive/battery unavailable", exc_info=True) except Exception: - log.debug("rangeStatus unavailable", exc_info=True) + log.debug("electric_drive unavailable", exc_info=True) try: - bat = vehicle.get_electric_drive().battery - result["temperatureBatteryStatus"] = { - "temperatureHvBatteryMax": _phys(bat.temperature_max.value, "degC"), - "temperatureHvBatteryMin": _phys(bat.temperature_min.value, "degC"), - } + result["odometer"] = {"odometer": _phys(vehicle.odometer.value, "km")} except Exception: - log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True) - - try: - result["temperatureOutsideStatus"] = { - "temperatureOutside": _phys(vehicle.outside_temperature.value, "degC"), - } - except Exception: - log.debug("temperatureOutsideStatus unavailable", exc_info=True) + log.debug("odometer unavailable", exc_info=True) return result or None -def extract_readiness(vehicle) -> dict | None: +def extract_connectivity(vehicle) -> dict | None: result: dict = {} try: - conn_state = str(vehicle.connection_state) - result["readinessStatus"] = { - "connectionState": { - "isOnline": conn_state == "online", - "isActive": conn_state in ("online", "reachable"), - }, - } + state = str(vehicle.connection_state) + result["state"] = state except Exception: - log.debug("readinessStatus unavailable", exc_info=True) + log.debug("connectivity.state unavailable", exc_info=True) return result or None +def extract_vehicle(vehicle) -> dict | None: + result: dict = {} -def extract_parking(vehicle) -> dict | None: + try: + state = str(vehicle.state) + result["state"] = state + 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["parkingPosition"] = { - "lat": pos.latitude.value, - "lon": pos.longitude.value, - } + result["lat"] = pos.latitude.value + result["lon"] = pos.longitude.value + except Exception: - log.debug("parkingPosition unavailable", exc_info=True) + log.debug("position unavailable", exc_info=True) return result or None -def extract_access(vehicle) -> dict | None: +def extract_doors(vehicle) -> dict | None: result: dict = {} try: doors = vehicle.doors - result["accessStatus"] = { - "overallStatus": _str(doors.lock_state), + result["doors"] = { + "overallState": _str(doors.lock_state), "doors": { name: { "lockState": _str(door.lock_state), @@ -205,18 +195,19 @@ def extract_access(vehicle) -> dict | None: }, } except Exception: - log.debug("accessStatus unavailable", exc_info=True) + log.debug("vehicle.doors unavailable", exc_info=True) return result or None -ALL_DOMAINS: dict[str, callable] = { +ALL_DOMAINS: dict[str, Callable] = { "charging": extract_charging, "climatisation": extract_climatisation, - "measurements": extract_measurements, - "readiness": extract_readiness, - "parking": extract_parking, - "access": extract_access, + "electric_drive": extract_electric_drive, + "connectivity": extract_connectivity, + "vehicle": extract_vehicle, + "position": extract_position, + "doors": extract_doors, } @@ -249,8 +240,8 @@ def apply_procedural(record: dict) -> dict: """ proc: dict = {} - range_at_soc = _get(record, "measurements", "rangeStatus", "totalRange", "value") - soc = _get(record, "charging", "batteryStatus", "currentSOC", "value") + range_at_soc = _get(record, "electric_drive", "range", "value") + soc = _get(record, "electric_drive", "battery", "soc", "value") if range_at_soc is not None and soc: proc["range_at_100"] = _phys(round(100 * float(range_at_soc) / float(soc), 1), "km") @@ -260,7 +251,7 @@ def apply_procedural(record: dict) -> dict: # format so the GUI picks up the field automatically. # # Example — usable energy estimated from SOC and battery capacity: - # soc = _get(record, "charging", "batteryStatus", "currentSOC", "value") + # soc = _get(record, "electric_drive", "battery", "soc", "value") # if soc is not None: # proc["energy_stored"] = {"value": round(soc / 100 * 77.0, 1), "unit": "kWh"} # ─────────────────────────────────────────────────────────────────────