diff --git a/collect.py b/collect.py index d5577d1..594ae6b 100644 --- a/collect.py +++ b/collect.py @@ -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) diff --git a/data_model.py b/data_model.py index c557783..98b9124 100644 --- a/data_model.py +++ b/data_model.py @@ -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" diff --git a/we_connect.py b/we_connect.py index 29e30f8..61fa8f6 100644 --- a/we_connect.py +++ b/we_connect.py @@ -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 \ No newline at end of file + return vehicles[0], None \ No newline at end of file