we_monitor/server/data_model.py

import logging
import re
from collections.abc import Callable
from datetime import datetime, timezone

from volkswagencarnet.vw_const import Paths
from volkswagencarnet.vw_dashboard import Dashboard
from volkswagencarnet.vw_utilities import find_path

log = logging.getLogger(__name__)


# ── value helpers ─────────────────────────────────────────────────────────────

def _phys(value, unit: str) -> dict:
    return {"value": value, "unit": unit}


# ── domain extractors ─────────────────────────────────────────────────────────

def extract_charging(vehicle) -> dict | None:
    result: dict = {}

    try:
        result["state"] = vehicle.charging_state
    except Exception:
        log.debug("charging.state unavailable", exc_info=True)

    try:
        if vehicle.is_charger_type_supported:
            result["type"] = vehicle.charger_type
    except Exception:
        pass

    try:
        power = vehicle.charging_power
        if power is not None:
            result["power"] = _phys(power, "kW")
    except Exception:
        log.debug("charging.power unavailable", exc_info=True)

    try:
        remain = vehicle.charging_time_left
        if remain is not None:
            result["remain"] = _phys(remain, "min")
    except Exception:
        log.debug("charging.remain unavailable", exc_info=True)

    settings: dict = {}
    try:
        target = vehicle.battery_target_charge_level
        if target is not None:
            settings["target_level"] = _phys(target, "%")
    except Exception:
        pass
    try:
        max_a = vehicle.charge_max_ac_ampere
        if max_a is not None:
            settings["maximum_current"] = _phys(max_a, "A")
    except Exception:
        pass
    try:
        auto_unlock = vehicle.auto_release_ac_connector_state
        if auto_unlock is not None:
            settings["auto_unlock"] = auto_unlock
    except Exception:
        pass
    if settings:
        result["settings"] = settings

    try:
        plug: dict = {}
        conn_state = find_path(vehicle.attrs, Paths.PLUG_CONN)
        if conn_state is not None:
            plug["connection_state"] = conn_state
        lock_state = find_path(vehicle.attrs, Paths.PLUG_LOCK)
        if lock_state is not None:
            plug["lock_state"] = lock_state
        ext_pwr = find_path(vehicle.attrs, Paths.PLUG_EXT_PWR)
        if ext_pwr is not None:
            plug["external_power"] = ext_pwr
        if plug:
            result["plugStatus"] = plug
    except Exception:
        log.debug("plugStatus unavailable", exc_info=True)

    return result or None


def extract_climatisation(vehicle) -> dict | None:
    result: dict = {}

    try:
        state = vehicle.climatisation_state
        if state is not None:
            result["state"] = state
    except Exception:
        log.debug("climatisation.state unavailable", exc_info=True)

    try:
        remain = vehicle.electric_remaining_climatisation_time
        if remain is not None:
            result["remain"] = _phys(remain, "min")
    except Exception:
        pass

    try:
        target = vehicle.climatisation_target_temperature
        if target is not None:
            result["settings"] = {"target_temperature": _phys(target, "degC")}
    except Exception:
        log.debug("climatisation.settings unavailable", exc_info=True)

    try:
        window_status = find_path(vehicle.attrs, Paths.CLIMATISATION_WINDOW_HEATING_STATUS)
        if window_status:
            heatings: dict = {}
            for entry in window_status:
                loc = entry.get("windowLocation")
                state = entry.get("windowHeatingState")
                if loc and state is not None:
                    heatings[loc] = state
            if heatings:
                result["window_heatings"] = heatings
    except Exception:
        log.debug("window_heatings unavailable", exc_info=True)

    return result or None


def extract_electric_drive(vehicle) -> dict | None:
    result: dict = {}

    try:
        rng = vehicle.electric_range
        if rng is not None:
            result["range"] = _phys(rng, "km")
    except Exception:
        log.debug("electric_range unavailable", exc_info=True)

    try:
        rng_full = vehicle.battery_cruising_range
        if rng_full is not None:
            result["range_full"] = _phys(round(float(rng_full), 1), "km")
    except Exception:
        log.debug("battery_cruising_range unavailable", exc_info=True)

    battery: dict = {}
    try:
        soc = vehicle.battery_level
        if soc is not None:
            battery["soc"] = _phys(soc, "%")
    except Exception:
        log.debug("battery.soc unavailable", exc_info=True)
    try:
        tmax = vehicle.hv_battery_max_temperature
        if tmax is not None:
            battery["temperature_max"] = _phys(round(tmax, 1), "degC")
    except Exception:
        pass
    try:
        tmin = vehicle.hv_battery_min_temperature
        if tmin is not None:
            battery["temperature_min"] = _phys(round(tmin, 1), "degC")
    except Exception:
        pass
    if battery:
        result["battery"] = battery

    try:
        odo = vehicle.distance
        if odo is not None:
            result["odometer"] = {"odometer": _phys(odo, "km")}
    except Exception:
        log.debug("odometer unavailable", exc_info=True)

    return result or None


def extract_connectivity(vehicle) -> dict | None:
    result: dict = {}

    try:
        is_online = vehicle.connection_state_is_online
        is_active = vehicle.connection_state_is_active
        if is_online:
            result["state"] = "online"
        elif is_active:
            result["state"] = "active"
        else:
            result["state"] = "offline"
    except Exception:
        log.debug("connectivity.state unavailable", exc_info=True)

    return result or None


def extract_vehicle(vehicle) -> dict | None:
    result: dict = {}

    try:
        overall = find_path(vehicle.attrs, Paths.ACCESS_OVERALL_STATUS)
        if overall is not None:
            result["state"] = overall
    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
        lat = pos.get("lat")
        lng = pos.get("lng")
        if lat is not None and lat != "?" and lng is not None and lng != "?":
            result["lat"] = lat
            result["lon"] = lng  # volkswagencarnet uses "lng" internally; we keep "lon" for compat
    except Exception:
        log.debug("position unavailable", exc_info=True)

    return result or None


def extract_doors(vehicle) -> dict | None:
    result: dict = {}

    try:
        lock_state = find_path(vehicle.attrs, Paths.ACCESS_DOOR_LOCK)
        if lock_state is not None:
            result["lock_state"] = lock_state

        doors_list = find_path(vehicle.attrs, Paths.ACCESS_DOORS) or []
        door_entries: dict = {}
        all_closed: bool | None = None
        for door in doors_list:
            name = door.get("name")
            status = door.get("status") or []
            if not name:
                continue
            entry: dict = {}
            if "closed" in status or "open" in status:
                open_state = "closed" if "closed" in status else "open"
                entry["open_state"] = open_state
                all_closed = open_state == "closed" if all_closed is None else (all_closed and open_state == "closed")
            if "locked" in status or "unlocked" in status:
                entry["lock_state"] = "locked" if "locked" in status else "unlocked"
            if entry:
                door_entries[name] = entry
        if all_closed is not None:
            result["open_state"] = "closed" if all_closed else "open"
        if door_entries:
            result["doors"] = door_entries
    except Exception:
        log.debug("vehicle.doors unavailable", exc_info=True)

    return result or None


def extract_windows(vehicle) -> dict | None:
    result: dict = {}

    try:
        windows_list = find_path(vehicle.attrs, Paths.ACCESS_WINDOWS) or []
        window_entries: dict = {}
        all_closed: bool | None = None
        for win in windows_list:
            name = win.get("name")
            status = win.get("status") or []
            if not name or "unsupported" in status:
                continue
            if "closed" in status or "open" in status:
                open_state = "closed" if "closed" in status else "open"
                window_entries[name] = {"open_state": open_state}
                all_closed = open_state == "closed" if all_closed is None else (all_closed and open_state == "closed")
        if all_closed is not None:
            result["open_state"] = "closed" if all_closed else "open"
        if window_entries:
            result["windows"] = window_entries
    except Exception:
        log.debug("vehicle.windows unavailable", exc_info=True)

    return result or None


_UNIT_WHITELIST: frozenset[str] = frozenset({"min", "%", "d", "kW", "°C", "km", "V", "W", "A"})
_UNIT_CONVERSIONS: dict[str, str] = {"d": "days", "°C": "degC"}


def _normalise_unit(unit: str) -> str:
    if unit not in _UNIT_WHITELIST:
        return ""
    return _UNIT_CONVERSIONS.get(unit, unit)


def _unit_from_str_state(str_state: str, state) -> str:
    """Parse the unit suffix from a str_state like '42 %' or '21.5 degC'."""
    state_repr = str(state)
    if str_state.startswith(state_repr):
        suffix = str_state[len(state_repr):].strip()
        if suffix:
            return suffix
    # fallback: last whitespace-separated token if it looks like a unit
    parts = str_state.rsplit(None, 1)
    if len(parts) == 2 and re.fullmatch(r"[^\d\s]\S*", parts[1]):
        return parts[1]
    return ""


def extract_all(vehicle) -> dict | None:
    """Read every supported instrument via the Dashboard and return a flat snapshot.

    Structure: snapshot[component][attr] = {"value": state, "unit": unit, "str_state": str_state}
    Unit is taken from instrument.unit when available, otherwise parsed from str_state.
    """
    snapshot: dict = {"ts": datetime.now(timezone.utc).isoformat()}
    try:
        dashboard = Dashboard(vehicle)
    except Exception:
        log.debug("extract_all: Dashboard setup failed", exc_info=True)
        return None

    for instrument in dashboard.instruments:
        try:
            state = instrument.state
            if state is None:
                continue

            try:
                str_state = str(instrument.str_state)
            except Exception:
                str_state = str(state)

            if hasattr(instrument, "unit") and instrument.unit:
                raw_unit = instrument.unit
            else:
                raw_unit = _unit_from_str_state(str_state, state)
            unit = _normalise_unit(raw_unit)

            component = instrument.component
            attr = instrument.attr
            snapshot.setdefault(component, {})[attr] = {
                "value": state,
                "unit": unit,
                "str_state": str_state,
            }
        except Exception:
            log.debug("extract_all: %s unavailable", instrument.attr, exc_info=True)

    return snapshot or None


ALL_DOMAINS: dict[str, Callable] = {
    "charging": extract_charging,
    "climatisation": extract_climatisation,
    "electric_drive": extract_electric_drive,
    "connectivity": extract_connectivity,
    "vehicle": extract_vehicle,
    "position": extract_position,
    "doors": extract_doors,
    "windows": extract_windows,
}


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


# ── procedural fields ─────────────────────────────────────────────────────────

def _get(data: dict, *keys):
    """Navigate nested dicts safely; returns None if any key is missing."""
    for k in keys:
        if not isinstance(data, dict):
            return None
        data = data.get(k)
    return data


def apply_procedural(record: dict) -> dict:
    """Compute derived fields and store them under record['procedural'].

    Called for every new snapshot before storage and broadcast, and for
    historical records that lack the 'procedural' key before they are sent
    to clients.  Add computed fields to the proc dict below.
    """
    proc: dict = {}

    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")

    # ── add computed fields here ──────────────────────────────────────────
    # Use _get(record, "domain", "statusObject", "field", "value") to
    # safely read any nested value.  Always use {"value": ..., "unit": ...}
    # format so the GUI picks up the field automatically.
    #
    # Example — usable energy estimated from SOC and battery capacity:
    # 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"}
    # ─────────────────────────────────────────────────────────────────────

    if proc:
        record["procedural"] = proc
    return record