jensandClaude Sonnet 4.6
63e807bdb0
- Add extract_windows() and register 'windows' in ALL_DOMAINS - Remove windows extraction from extract_doors; no more double keys - extract_windows also captures the overall windows.open_state Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
281 lines
9.3 KiB
Python
281 lines
9.3 KiB
Python
import logging
|
|
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# ── value helpers ─────────────────────────────────────────────────────────────
|
|
|
|
def _str(attr) -> str:
|
|
"""Return the string value of a CarConnectivity attribute."""
|
|
return str(attr)
|
|
|
|
|
|
def _phys(value, unit: str) -> dict:
|
|
return {"value": value, "unit": unit}
|
|
|
|
|
|
def _remaining_min(estimated_date_reached_attr) -> float | None:
|
|
"""Compute remaining minutes from an estimated_date_reached DateAttribute."""
|
|
edr = estimated_date_reached_attr.value
|
|
if edr is None:
|
|
return None
|
|
return max(0.0, (edr - datetime.now(timezone.utc)).total_seconds() / 60)
|
|
|
|
|
|
# ── domain extractors ─────────────────────────────────────────────────────────
|
|
|
|
def extract_charging(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
ch = vehicle.charging
|
|
result["state"] = _str(ch.state)
|
|
result["type"] = _str(ch.type)
|
|
try:
|
|
result["power"] = _phys(ch.power.value, "kW")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
rem = _remaining_min(ch.estimated_date_reached)
|
|
if rem is not None:
|
|
result["remain"] = _phys(round(rem), "min")
|
|
except Exception:
|
|
pass
|
|
|
|
except Exception:
|
|
log.debug("vehicle.charging unavailable", exc_info=True)
|
|
|
|
try:
|
|
cfg = vehicle.charging.settings
|
|
result["settings"] = {
|
|
"target_level": _phys(cfg.target_level.value, "%"),
|
|
"maximum_current": _phys(cfg.maximum_current.value, "A"),
|
|
"auto_unlock": _str(cfg.auto_unlock),
|
|
}
|
|
except Exception:
|
|
log.debug("vehicle.charging.settings unavailable", exc_info=True)
|
|
|
|
try:
|
|
conn = vehicle.charging.connector
|
|
result["plugStatus"] = {
|
|
"connection_state": _str(conn.connection_state),
|
|
"lock_state": _str(conn.lock_state),
|
|
"external_power": _str(conn.external_power),
|
|
}
|
|
except Exception:
|
|
log.debug("plugStatus unavailable", exc_info=True)
|
|
|
|
return result or None
|
|
|
|
|
|
def extract_climatisation(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
cl = vehicle.climatization
|
|
result["state"] = _str(cl.state)
|
|
try:
|
|
rem = _remaining_min(cl.estimated_date_reached)
|
|
if rem is not None:
|
|
result["remain"] = _phys(round(rem), "min")
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
log.debug("vehicle.climatization unavailable", exc_info=True)
|
|
|
|
try:
|
|
cs = vehicle.climatization.settings
|
|
result["settings"] = {
|
|
"target_temperature": _phys(cs.target_temperature.value, "degC"),
|
|
}
|
|
except Exception:
|
|
log.debug("climatization.settings unavailable", exc_info=True)
|
|
|
|
try:
|
|
wh = vehicle.window_heatings
|
|
result["window_heatings"] = {
|
|
name: _str(win.heating_state)
|
|
for name, win in wh.heatings.items()
|
|
}
|
|
except Exception:
|
|
log.debug("vehicle.window_heatings unavailable", exc_info=True)
|
|
|
|
try:
|
|
result["environment"] = {
|
|
"temperature_outside": _phys(vehicle.outside_temperature.value, "degC"),
|
|
}
|
|
except Exception:
|
|
log.debug("vehicle.outside_temperature unavailable", exc_info=True)
|
|
|
|
return result or None
|
|
|
|
|
|
def extract_electric_drive(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
ed = vehicle.get_electric_drive()
|
|
result["range"] = _phys(ed.range.value, "km")
|
|
result["range_full"] = _phys(round(float(ed.range_estimated_full.value), 1), "km")
|
|
try:
|
|
bat = ed.battery
|
|
result["battery"] = {
|
|
"soc": _phys(ed.level.value, "%"),
|
|
"temperature_max": _phys(round(float(bat.temperature_max.value) - 273.15, 1), "degC"),
|
|
"temperature_min": _phys(round(float(bat.temperature_min.value) - 273.15, 1), "degC"),
|
|
}
|
|
except Exception:
|
|
log.debug("electric_drive/battery unavailable", exc_info=True)
|
|
except Exception:
|
|
log.debug("electric_drive unavailable", exc_info=True)
|
|
|
|
try:
|
|
result["odometer"] = {"odometer": _phys(vehicle.odometer.value, "km")}
|
|
except Exception:
|
|
log.debug("odometer unavailable", exc_info=True)
|
|
|
|
return result or None
|
|
|
|
|
|
def extract_connectivity(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
state = str(vehicle.connection_state)
|
|
result["state"] = state
|
|
except Exception:
|
|
log.debug("connectivity.state unavailable", exc_info=True)
|
|
|
|
return result or None
|
|
|
|
def extract_vehicle(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
state = str(vehicle.state)
|
|
result["state"] = state
|
|
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
|
|
result["lat"] = pos.latitude.value
|
|
result["lon"] = pos.longitude.value
|
|
|
|
except Exception:
|
|
log.debug("position unavailable", exc_info=True)
|
|
|
|
return result or None
|
|
|
|
|
|
def extract_doors(vehicle) -> dict | None:
|
|
result: dict = {}
|
|
|
|
try:
|
|
doors = vehicle.doors
|
|
if doors.lock_state.value is not None:
|
|
result["lock_state"] = _str(doors.lock_state)
|
|
if doors.open_state.value is not None:
|
|
result["open_state"] = _str(doors.open_state)
|
|
door_entries = {}
|
|
for name, door in doors.doors.items():
|
|
entry = {}
|
|
if door.lock_state.value is not None:
|
|
entry["lock_state"] = _str(door.lock_state)
|
|
if door.open_state.value is not None:
|
|
entry["open_state"] = _str(door.open_state)
|
|
if entry:
|
|
door_entries[name] = entry
|
|
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:
|
|
if vehicle.windows.open_state.value is not None:
|
|
result["open_state"] = _str(vehicle.windows.open_state)
|
|
window_entries = {}
|
|
for name, win in vehicle.windows.windows.items():
|
|
if win.open_state.value is not None:
|
|
window_entries[name] = {"open_state": _str(win.open_state)}
|
|
if window_entries:
|
|
result["windows"] = window_entries
|
|
except Exception:
|
|
log.debug("vehicle.windows unavailable", exc_info=True)
|
|
|
|
return result 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 |