Move domain extractors and collect_snapshot into data_model.py

we_connect.py now only handles WeConnect login and vehicle selection. All extract_*() functions, ALL_DOMAINS, and collect_snapshot() live in data_model.py alongside the store and value helpers. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-25 19:47:25 +02:00
parent bc016cd7cb
commit 097bc5ad51
3 changed files with 238 additions and 240 deletions
@@ -58,8 +58,7 @@ from pathlib import Path
 import log_config
 import network
 import we_connect
-from data_model import auto_out_path, load_store, save_store
-from we_connect import ALL_DOMAINS
+from data_model import ALL_DOMAINS, auto_out_path, collect_snapshot, load_store, save_store

 log = logging.getLogger(__name__)

@@ -118,7 +117,7 @@ def run(args: argparse.Namespace) -> None:

            try:
                wc.update()
-                snapshot = we_connect.collect_snapshot(vehicle, domains)
+                snapshot = collect_snapshot(vehicle, domains)
                store["records"].append(snapshot)
                save_store(out, store, args.max_records)
                network.broadcast(snapshot, push_clients, push_lock)
@@ -6,6 +6,8 @@ from pathlib import Path
 log = logging.getLogger(__name__)


+# ── value helpers ─────────────────────────────────────────────────────────────
+
 def _str(v) -> str:
    return str(v.value) if hasattr(v, "value") else str(v)

@@ -14,6 +16,239 @@ def _phys(value, unit: str) -> dict:
    return {"value": value, "unit": unit}


+# ── 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),
+        }
+    except Exception:
+        log.debug("chargingStatus unavailable", exc_info=True)
+
+    try:
+        bs = d["batteryStatus"]
+        result["batteryStatus"] = {
+            "currentSOC": _phys(bs.currentSOC_pct.value, "%"),
+            "cruisingRangeElectric": _phys(bs.cruisingRangeElectric_km.value, "km"),
+        }
+    except Exception:
+        log.debug("charging/batteryStatus unavailable", exc_info=True)
+
+    try:
+        cfg = d["chargingSettings"]
+        result["chargingSettings"] = {
+            "targetSOC": _phys(cfg.targetSOC_pct.value, "%"),
+            "maxChargeCurrentAC": _str(cfg.maxChargeCurrentAC),
+            "autoUnlockPlugWhenCharged": _str(cfg.autoUnlockPlugWhenCharged),
+        }
+    except Exception:
+        log.debug("chargingSettings unavailable", exc_info=True)
+
+    try:
+        ps = d["plugStatus"]
+        result["plugStatus"] = {
+            "plugConnectionState": _str(ps.plugConnectionState),
+            "plugLockState": _str(ps.plugLockState),
+            "externalPower": _str(ps.externalPower),
+        }
+    except Exception:
+        log.debug("plugStatus unavailable", exc_info=True)
+
+    return result or 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"),
+        }
+    except Exception:
+        log.debug("climatisationStatus unavailable", exc_info=True)
+
+    try:
+        cs = d["climatisationSettings"]
+        result["climatisationSettings"] = {
+            "targetTemperature": _phys(cs.targetTemperature_C.value, "degC"),
+            "unitInCar": _str(cs.unitInCar),
+        }
+    except Exception:
+        log.debug("climatisationSettings unavailable", exc_info=True)
+
+    try:
+        wh = d["windowHeatingStatus"]
+        result["windowHeatingStatus"] = {
+            name: _str(win.windowHeatingState)
+            for name, win in wh.windows.items()
+        }
+    except Exception:
+        log.debug("windowHeatingStatus unavailable", exc_info=True)
+
+    return result or 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")}
+    except Exception:
+        log.debug("odometerStatus unavailable", exc_info=True)
+
+    try:
+        rs = d["rangeStatus"]
+        result["rangeStatus"] = {
+            "electricRange": _phys(rs.electricRange.value, "km"),
+            "totalRange": _phys(rs.totalRange_km.value, "km"),
+        }
+    except Exception:
+        log.debug("rangeStatus unavailable", exc_info=True)
+
+    try:
+        bs = d["temperatureBatteryStatus"]
+        result["temperatureBatteryStatus"] = {
+            "temperatureHvBatteryMax": _phys(float(bs.temperatureHvBatteryMax_K.value - 273.15), "degC"),
+            "temperatureHvBatteryMin": _phys(float(bs.temperatureHvBatteryMin_K.value - 273.15), "degC"),
+        }
+    except Exception:
+        log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True)
+
+    try:
+        temp = d["temperatureOutsideStatus"]
+        result["temperatureOutsideStatus"] = {
+            "temperatureOutside": _phys(temp.temperatureOutside_C.value, "degC"),
+        }
+    except Exception:
+        log.debug("temperatureOutsideStatus unavailable", exc_info=True)
+
+    return result or 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
+        result["readinessStatus"] = {
+            "connectionState": {
+                "isOnline": conn.isOnline.value,
+                "isActive": conn.isActive.value,
+            },
+            "connectionWarning": {
+                "insufficientBatteryLevelWarning": warn.insufficientBatteryLevelWarning.value,
+                "insufficientBatteryLevelError": warn.insufficientBatteryLevelError.value,
+            },
+        }
+    except Exception:
+        log.debug("readinessStatus unavailable", exc_info=True)
+
+    return result or None
+
+
+def extract_parking(vehicle) -> dict | None:
+    try:
+        d = vehicle.domains["parking"]
+    except KeyError:
+        return None
+
+    result: dict = {}
+
+    try:
+        pos = d["parkingPosition"]
+        result["parkingPosition"] = {
+            "lat": pos.lat.value,
+            "lon": pos.lon.value,
+        }
+    except Exception:
+        log.debug("parkingPosition unavailable", exc_info=True)
+
+    return result or None
+
+
+def extract_access(vehicle) -> dict | None:
+    try:
+        d = vehicle.domains["access"]
+    except KeyError:
+        return None
+
+    result: dict = {}
+
+    try:
+        acc = d["accessStatus"]
+        result["accessStatus"] = {
+            "overallStatus": _str(acc.overallStatus),
+            "doors": {
+                name: {
+                    "lockState": _str(door.lockState),
+                    "openState": _str(door.openState),
+                }
+                for name, door in acc.doors.items()
+            },
+            "windows": {
+                name: {"openState": _str(win.openState)}
+                for name, win in acc.windows.items()
+            },
+        }
+    except Exception:
+        log.debug("accessStatus unavailable", exc_info=True)
+
+    return result or None
+
+
+ALL_DOMAINS: dict[str, callable] = {
+    "charging": extract_charging,
+    "climatisation": extract_climatisation,
+    "measurements": extract_measurements,
+    "readiness": extract_readiness,
+    "parking": extract_parking,
+    "access": extract_access,
+}
+
+
+def collect_snapshot(vehicle, domains: list[str]) -> dict:
+    snapshot: dict = {"ts": datetime.now(timezone.utc).isoformat()}
+    for domain in domains:
+        data = ALL_DOMAINS[domain](vehicle)
+        if data is not None:
+            snapshot[domain] = data
+    return snapshot
+
+
+# ── store helpers ─────────────────────────────────────────────────────────────
+
 def auto_out_path(logs_dir: Path, vin: str) -> Path:
    stamp = datetime.now(timezone.utc).strftime("%Y_%m_%d_%H_%M_%S")
    return logs_dir / f"{stamp}_{vin}.json"
@@ -1,8 +1,5 @@
 import logging
 import sys
-from datetime import datetime, timezone
-
-from data_model import _phys, _str

 try:
    from weconnect import weconnect
@@ -13,230 +10,6 @@ except ImportError:
 log = logging.getLogger(__name__)


-# ── 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),
-        }
-    except Exception:
-        log.debug("chargingStatus unavailable", exc_info=True)
-
-    try:
-        bs = d["batteryStatus"]
-        result["batteryStatus"] = {
-            "currentSOC": _phys(bs.currentSOC_pct.value, "%"),
-            "cruisingRangeElectric": _phys(bs.cruisingRangeElectric_km.value, "km"),
-        }
-    except Exception:
-        log.debug("charging/batteryStatus unavailable", exc_info=True)
-
-    try:
-        cfg = d["chargingSettings"]
-        result["chargingSettings"] = {
-            "targetSOC": _phys(cfg.targetSOC_pct.value, "%"),
-            "maxChargeCurrentAC": _str(cfg.maxChargeCurrentAC),
-            "autoUnlockPlugWhenCharged": _str(cfg.autoUnlockPlugWhenCharged),
-        }
-    except Exception:
-        log.debug("chargingSettings unavailable", exc_info=True)
-
-    try:
-        ps = d["plugStatus"]
-        result["plugStatus"] = {
-            "plugConnectionState": _str(ps.plugConnectionState),
-            "plugLockState": _str(ps.plugLockState),
-            "externalPower": _str(ps.externalPower),
-        }
-    except Exception:
-        log.debug("plugStatus unavailable", exc_info=True)
-
-    return result or 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"),
-        }
-    except Exception:
-        log.debug("climatisationStatus unavailable", exc_info=True)
-
-    try:
-        cs = d["climatisationSettings"]
-        result["climatisationSettings"] = {
-            "targetTemperature": _phys(cs.targetTemperature_C.value, "degC"),
-            "unitInCar": _str(cs.unitInCar),
-        }
-    except Exception:
-        log.debug("climatisationSettings unavailable", exc_info=True)
-
-    try:
-        wh = d["windowHeatingStatus"]
-        result["windowHeatingStatus"] = {
-            name: _str(win.windowHeatingState)
-            for name, win in wh.windows.items()
-        }
-    except Exception:
-        log.debug("windowHeatingStatus unavailable", exc_info=True)
-
-    return result or 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")}
-    except Exception:
-        log.debug("odometerStatus unavailable", exc_info=True)
-
-    try:
-        rs = d["rangeStatus"]
-        result["rangeStatus"] = {
-            "electricRange": _phys(rs.electricRange.value, "km"),
-            "totalRange": _phys(rs.totalRange_km.value, "km"),
-        }
-    except Exception:
-        log.debug("rangeStatus unavailable", exc_info=True)
-
-    try:
-        bs = d["temperatureBatteryStatus"]
-        result["temperatureBatteryStatus"] = {
-            "temperatureHvBatteryMax": _phys(float(bs.temperatureHvBatteryMax_K.value - 273.15), "degC"),
-            "temperatureHvBatteryMin": _phys(float(bs.temperatureHvBatteryMin_K.value - 273.15), "degC"),
-        }
-    except Exception:
-        log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True)
-
-    try:
-        temp = d["temperatureOutsideStatus"]
-        result["temperatureOutsideStatus"] = {
-            "temperatureOutside": _phys(temp.temperatureOutside_C.value, "degC"),
-        }
-    except Exception:
-        log.debug("temperatureOutsideStatus unavailable", exc_info=True)
-
-    return result or 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
-        result["readinessStatus"] = {
-            "connectionState": {
-                "isOnline": conn.isOnline.value,
-                "isActive": conn.isActive.value,
-            },
-            "connectionWarning": {
-                "insufficientBatteryLevelWarning": warn.insufficientBatteryLevelWarning.value,
-                "insufficientBatteryLevelError": warn.insufficientBatteryLevelError.value,
-            },
-        }
-    except Exception:
-        log.debug("readinessStatus unavailable", exc_info=True)
-
-    return result or None
-
-
-def extract_parking(vehicle) -> dict | None:
-    try:
-        d = vehicle.domains["parking"]
-    except KeyError:
-        return None
-
-    result: dict = {}
-
-    try:
-        pos = d["parkingPosition"]
-        result["parkingPosition"] = {
-            "lat": pos.lat.value,
-            "lon": pos.lon.value,
-        }
-    except Exception:
-        log.debug("parkingPosition unavailable", exc_info=True)
-
-    return result or None
-
-
-def extract_access(vehicle) -> dict | None:
-    try:
-        d = vehicle.domains["access"]
-    except KeyError:
-        return None
-
-    result: dict = {}
-
-    try:
-        acc = d["accessStatus"]
-        result["accessStatus"] = {
-            "overallStatus": _str(acc.overallStatus),
-            "doors": {
-                name: {
-                    "lockState": _str(door.lockState),
-                    "openState": _str(door.openState),
-                }
-                for name, door in acc.doors.items()
-            },
-            "windows": {
-                name: {"openState": _str(win.openState)}
-                for name, win in acc.windows.items()
-            },
-        }
-    except Exception:
-        log.debug("accessStatus unavailable", exc_info=True)
-
-    return result or None
-
-
-ALL_DOMAINS: dict[str, callable] = {
-    "charging": extract_charging,
-    "climatisation": extract_climatisation,
-    "measurements": extract_measurements,
-    "readiness": extract_readiness,
-    "parking": extract_parking,
-    "access": extract_access,
-}
-
-
-# ── connection helpers ───────────────────────────────────────────────────────
-
 def connect(username: str, password: str):
    wc = weconnect.WeConnect(
        username=username,
@@ -260,13 +33,4 @@ def select_vehicle(wc, vin: str | None):
            available = [v.vin.value for v in vehicles]
            return None, f"VIN {vin} not found.  Available: {available}"
        return vehicle, None
-    return vehicles[0], None
-
-
-def collect_snapshot(vehicle, domains: list[str]) -> dict:
-    snapshot: dict = {"ts": datetime.now(timezone.utc).isoformat()}
-    for domain in domains:
-        data = ALL_DOMAINS[domain](vehicle)
-        if data is not None:
-            snapshot[domain] = data
-    return snapshot
+    return vehicles[0], None