414 lines
14 KiB
Python
414 lines
14 KiB
Python
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, "sensor", "electric_range", "value")
|
|
soc = _get(record, "sensor", "battery_level", "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, "sensor"|"binary_sensor"|..., "attr_name", "value")
|
|
# Always use {"value": ..., "unit": ...} format so the GUI picks it up.
|
|
#
|
|
# Example — usable energy estimated from SOC and battery capacity:
|
|
# soc = _get(record, "sensor", "battery_level", "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
|