we_monitor/we_connect.py

import logging
import sys
from datetime import datetime, timezone

from data_model import _phys, _str

try:
    from weconnect import weconnect
except ImportError:
    print("weconnect is not installed.  Run:  pip install weconnect", file=sys.stderr)
    sys.exit(1)

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,
        password=password,
        updateAfterLogin=False,
        loginOnInit=True,
    )
    wc.login()
    wc.update()
    return wc


def select_vehicle(wc, vin: str | None):
    """Return (vehicle, error_message). error_message is None on success."""
    vehicles = list(wc.vehicles.values())
    if not vehicles:
        return None, "No vehicles found in this WeConnect account"
    if vin:
        vehicle = next((v for v in vehicles if v.vin.value == vin), None)
        if not vehicle:
            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