Add procedural field post-processing to snapshots
server/data_model.py: - apply_procedural(record) — stub function; add computed fields to the proc dict inside and they appear under record['procedural'] in every record stored and broadcast. - _get(data, *keys) — safe nested dict accessor for use inside apply_procedural. collect.py: - New snapshots: apply_procedural() called after collect_snapshot(), before appending to the store and broadcasting. - Preloaded 24-h history: guarded apply (skip if 'procedural' already present) applied after load_last_24h_records(). - Current-day store loaded from disk: same guarded apply after load_store(), so legacy records sent to clients are enriched on the fly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -243,3 +243,39 @@ def collect_snapshot(vehicle, domains: list[str]) -> dict:
|
||||
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 = {}
|
||||
|
||||
# ── 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, "charging", "batteryStatus", "currentSOC", "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
|
||||
|
||||
Reference in New Issue
Block a user