data_model: redesign domain extractors for CarConnectivity object model
Flatten the snapshot structure so each domain maps directly to its fields rather than mirroring the old WeConnect status-object nesting. New domains: electric_drive (range, battery/SOC/temps, odometer), connectivity, vehicle, position, doors — replacing the old readiness, parking, access, and measurements domains. Update apply_procedural paths and collect.py defaults accordingly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
+76
-85
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime, timezone
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -30,49 +31,38 @@ def extract_charging(vehicle) -> dict | None:
|
||||
|
||||
try:
|
||||
ch = vehicle.charging
|
||||
status: dict = {
|
||||
"chargingState": _str(ch.state),
|
||||
"chargeType": _str(ch.type),
|
||||
}
|
||||
result["state"] = _str(ch.state)
|
||||
result["type"] = _str(ch.type)
|
||||
try:
|
||||
status["chargePower"] = _phys(ch.power.value, "kW")
|
||||
result["power"] = _phys(ch.power.value, "kW")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
rem = _remaining_min(ch.estimated_date_reached)
|
||||
if rem is not None:
|
||||
status["remainingChargingTimeToComplete"] = _phys(round(rem), "min")
|
||||
result["remain"] = _phys(round(rem), "min")
|
||||
except Exception:
|
||||
pass
|
||||
result["chargingStatus"] = status
|
||||
except Exception:
|
||||
log.debug("chargingStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
ed = vehicle.get_electric_drive()
|
||||
result["batteryStatus"] = {
|
||||
"currentSOC": _phys(ed.level.value, "%"),
|
||||
"cruisingRangeElectric": _phys(ed.range.value, "km"),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("charging/batteryStatus unavailable", exc_info=True)
|
||||
log.debug("vehicle.charging unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
cfg = vehicle.charging.settings
|
||||
result["chargingSettings"] = {
|
||||
"targetSOC": _phys(cfg.target_level.value, "%"),
|
||||
"maxChargeCurrentAC": _phys(cfg.maximum_current.value, "A"),
|
||||
"autoUnlockPlugWhenCharged": _str(cfg.auto_unlock),
|
||||
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("chargingSettings unavailable", exc_info=True)
|
||||
log.debug("vehicle.charging.settings unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
conn = vehicle.charging.connector
|
||||
result["plugStatus"] = {
|
||||
"plugConnectionState": _str(conn.connection_state),
|
||||
"plugLockState": _str(conn.lock_state),
|
||||
"externalPower": _str(conn.external_power),
|
||||
"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)
|
||||
@@ -85,113 +75,113 @@ def extract_climatisation(vehicle) -> dict | None:
|
||||
|
||||
try:
|
||||
cl = vehicle.climatization
|
||||
status: dict = {"climatisationState": _str(cl.state)}
|
||||
result["state"] = _str(cl.state)
|
||||
try:
|
||||
rem = _remaining_min(cl.estimated_date_reached)
|
||||
if rem is not None:
|
||||
status["remainingClimatisationTime"] = _phys(round(rem), "min")
|
||||
result["remain"] = _phys(round(rem), "min")
|
||||
except Exception:
|
||||
pass
|
||||
result["climatisationStatus"] = status
|
||||
except Exception:
|
||||
log.debug("climatisationStatus unavailable", exc_info=True)
|
||||
log.debug("vehicle.climatization unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
cs = vehicle.climatization.settings
|
||||
result["climatisationSettings"] = {
|
||||
"targetTemperature": _phys(cs.target_temperature.value, "degC"),
|
||||
result["settings"] = {
|
||||
"target_temperature": _phys(cs.target_temperature.value, "degC"),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("climatisationSettings unavailable", exc_info=True)
|
||||
log.debug("climatization.settings unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
wh = vehicle.window_heatings
|
||||
result["windowHeatingStatus"] = {
|
||||
result["window_heatings"] = {
|
||||
name: _str(win.heating_state)
|
||||
for name, win in wh.heatings.items()
|
||||
}
|
||||
except Exception:
|
||||
log.debug("windowHeatingStatus unavailable", exc_info=True)
|
||||
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_measurements(vehicle) -> dict | None:
|
||||
def extract_electric_drive(vehicle) -> dict | None:
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
result["odometerStatus"] = {"odometer": _phys(vehicle.odometer.value, "km")}
|
||||
except Exception:
|
||||
log.debug("odometerStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
ed = vehicle.get_electric_drive()
|
||||
electric_range = ed.range.value
|
||||
result["rangeStatus"] = {
|
||||
"electricRange": _phys(electric_range, "km"),
|
||||
"totalRange": _phys(electric_range, "km"),
|
||||
}
|
||||
result["range"] = _phys(ed.range.value, "km")
|
||||
result["range_full"] = _phys(ed.range_estimated_full.value, "km")
|
||||
try:
|
||||
bat = ed.battery
|
||||
result["battery"] = {
|
||||
"soc": _phys(ed.level.value, "%"),
|
||||
"temperature_max": _phys(bat.temperature_max.value, "degC"),
|
||||
"temperature_min": _phys(bat.temperature_min.value, "degC"),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("electric_drive/battery unavailable", exc_info=True)
|
||||
except Exception:
|
||||
log.debug("rangeStatus unavailable", exc_info=True)
|
||||
log.debug("electric_drive unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
bat = vehicle.get_electric_drive().battery
|
||||
result["temperatureBatteryStatus"] = {
|
||||
"temperatureHvBatteryMax": _phys(bat.temperature_max.value, "degC"),
|
||||
"temperatureHvBatteryMin": _phys(bat.temperature_min.value, "degC"),
|
||||
}
|
||||
result["odometer"] = {"odometer": _phys(vehicle.odometer.value, "km")}
|
||||
except Exception:
|
||||
log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
result["temperatureOutsideStatus"] = {
|
||||
"temperatureOutside": _phys(vehicle.outside_temperature.value, "degC"),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("temperatureOutsideStatus unavailable", exc_info=True)
|
||||
log.debug("odometer unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_readiness(vehicle) -> dict | None:
|
||||
def extract_connectivity(vehicle) -> dict | None:
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
conn_state = str(vehicle.connection_state)
|
||||
result["readinessStatus"] = {
|
||||
"connectionState": {
|
||||
"isOnline": conn_state == "online",
|
||||
"isActive": conn_state in ("online", "reachable"),
|
||||
},
|
||||
}
|
||||
state = str(vehicle.connection_state)
|
||||
result["state"] = state
|
||||
except Exception:
|
||||
log.debug("readinessStatus unavailable", exc_info=True)
|
||||
log.debug("connectivity.state unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
def extract_vehicle(vehicle) -> dict | None:
|
||||
result: dict = {}
|
||||
|
||||
def extract_parking(vehicle) -> dict | None:
|
||||
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["parkingPosition"] = {
|
||||
"lat": pos.latitude.value,
|
||||
"lon": pos.longitude.value,
|
||||
}
|
||||
result["lat"] = pos.latitude.value
|
||||
result["lon"] = pos.longitude.value
|
||||
|
||||
except Exception:
|
||||
log.debug("parkingPosition unavailable", exc_info=True)
|
||||
log.debug("position unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_access(vehicle) -> dict | None:
|
||||
def extract_doors(vehicle) -> dict | None:
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
doors = vehicle.doors
|
||||
result["accessStatus"] = {
|
||||
"overallStatus": _str(doors.lock_state),
|
||||
result["doors"] = {
|
||||
"overallState": _str(doors.lock_state),
|
||||
"doors": {
|
||||
name: {
|
||||
"lockState": _str(door.lock_state),
|
||||
@@ -205,18 +195,19 @@ def extract_access(vehicle) -> dict | None:
|
||||
},
|
||||
}
|
||||
except Exception:
|
||||
log.debug("accessStatus unavailable", exc_info=True)
|
||||
log.debug("vehicle.doors unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
ALL_DOMAINS: dict[str, callable] = {
|
||||
ALL_DOMAINS: dict[str, Callable] = {
|
||||
"charging": extract_charging,
|
||||
"climatisation": extract_climatisation,
|
||||
"measurements": extract_measurements,
|
||||
"readiness": extract_readiness,
|
||||
"parking": extract_parking,
|
||||
"access": extract_access,
|
||||
"electric_drive": extract_electric_drive,
|
||||
"connectivity": extract_connectivity,
|
||||
"vehicle": extract_vehicle,
|
||||
"position": extract_position,
|
||||
"doors": extract_doors,
|
||||
}
|
||||
|
||||
|
||||
@@ -249,8 +240,8 @@ def apply_procedural(record: dict) -> dict:
|
||||
"""
|
||||
proc: dict = {}
|
||||
|
||||
range_at_soc = _get(record, "measurements", "rangeStatus", "totalRange", "value")
|
||||
soc = _get(record, "charging", "batteryStatus", "currentSOC", "value")
|
||||
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")
|
||||
|
||||
@@ -260,7 +251,7 @@ def apply_procedural(record: dict) -> dict:
|
||||
# format so the GUI picks up the field automatically.
|
||||
#
|
||||
# Example — usable energy estimated from SOC and battery capacity:
|
||||
# soc = _get(record, "charging", "batteryStatus", "currentSOC", "value")
|
||||
# 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"}
|
||||
# ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user