diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..c439123 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,29 @@ +# Project agent memory + +This file is the project's committed home for project-intrinsic agent knowledge: build, test, release, architecture, and sharp-edge notes that should travel with the code. + +- Add durable project-specific notes here as they are discovered through real work. + +## Maintainer scripts (`scripts/`) + +`scripts/` holds prod-touching maintainer tooling that is deliberately kept +out of the `teslemetry_stream` package and out of CI - it needs an +authenticated NATS context with prod access, which CI does not have and +should never be given. + +- `scripts/tesla_cache_validate.py` validates Tesla field semantics (tri-state + vs true-or-absent, observed value domains, presence rates) against the live + `na_cache`/`eu_cache` NATS KV buckets - vehicle REST snapshots, streamed + Fleet Telemetry signals, and energy `live_status`/`site_info`. Read-only, + run by hand by a maintainer. See `scripts/README.md` for usage and the + reliability guards (most importantly: reads MUST go through + `nats stream get --last-for ... -j`, never `nats kv get`, which silently + reads empty on these buckets). +- Its pure logic (field extraction, presence classification, sanitization) + is unit-tested offline in `tests/test_cache_validate_logic.py` - safe to + run in CI. The script itself is not exercised by any automated test suite + because doing so would require prod NATS access. +- This is phase 1 of a longer roadmap (declarative field registry + golden + fixtures, described in the originating design report) - phases 2+ + (fixtures module, `Field`/`Presence` registry) are intentionally not + implemented yet. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..47dc3e3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..b5aa098 --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,152 @@ +# Maintainer scripts + +## `tesla_cache_validate.py` - validate Tesla field semantics against prod + +Answers, with evidence instead of guesswork, the questions that keep coming +up when a client library or an HA integration has to decide how to treat a +Tesla field: + +- Is this field a genuine **tri-state** (explicit `true`/`false`/`null`), or + does it follow a **true-or-absent** shape (present when true, simply + omitted otherwise)? +- What is the **observed value domain** - every distinct value Tesla actually + sends for this field? +- What is the **presence rate** across the fleet, and does absence correlate + with some other field (asleep state, a sensor's own "invalid" reading, + location scope, etc.)? + +It queries the production `na_cache`/`eu_cache` NATS KV buckets (read-only) +that Teslemetry's backend already populates from real vehicle/energy +traffic. This promotes the one-off recipe from +`data/chargeport-null-validate-v4/report.md` (the report that settled +`charge_state.charge_port_door_open` as a genuine tri-state, n=723) into a +reusable, committed tool. + +### Why this matters + +Every one of these was, at some point, a guess that turned out wrong: + +- `charge_state.charge_port_door_open`: assumed "null means closed" - + refuted by 431/723 vehicles reporting explicit `false`. +- Commit `b72b9a9` in this repo: six streamed fields were silently mistyped + until someone hand-checked na_cache against real telemetry. +- Several Home Assistant integration bugs (`TypeError`/`IndexError` on + present-null or out-of-range values) trace back to a field whose + nullability or value domain was never actually confirmed. + +This tool exists so that confirmation is a five-minute command instead of a +one-off archaeology project. + +### Prerequisites + +- The `nats` CLI, with an authenticated context already configured + (`nats context select` / `nats context info` to confirm). This script + never hardcodes a context, server URL, or credential - pass `--context` + if you don't want to rely on whatever context is currently selected. +- Read access to the `na_cache`/`eu_cache` KV buckets. Nothing here needs + write access; the script never calls `nats kv put`, `nats pub`, or any + other write path. + +This is an **offline, maintainer-run** tool. CI does not have prod NATS +access and should never be given it - CI's job (a later phase) is to check +committed fixtures against a schema, not to touch prod. Run this by hand, +commit what it finds. + +### The three record shapes it understands + +The cache holds three distinct shapes; pick the matching `--kind`: + +| `--kind` | KV key shape | Record body | +|---|---|---| +| `vehicle_data` | `.vehicle_data` | Full REST `vehicle_data` snapshot (nested dict: `charge_state`, `climate_state`, `drive_state`, `vehicle_state`, ...) | +| `signal` | `.data.` | One streamed Fleet Telemetry signal - the whole record body IS the value (a bare scalar) | +| `energy_live_status` | `energy_sites..live_status` | REST energy `live_status`, wrapped as `{"statusCode":200,"json":{"response":{...}}}` (auto-unwrapped) | +| `energy_site_info` | `energy_sites..site_info` | REST energy `site_info`, same envelope (auto-unwrapped) | + +`--field` is a dotted path within the (unwrapped) record for the three REST +kinds, e.g. `charge_state.charge_port_door_open` or `grid_status`. For +`signal` it is the bare `Signal` name as it appears in `const.py`, e.g. +`ChargePortDoorOpen` - no dots, since the KV key already names the field. + +### Examples + +```bash +# Settle a tri-state-vs-true-or-absent question on a vehicle REST field, +# cross-tabbed against a related field, both regions, every cached record: +python3 scripts/tesla_cache_validate.py \ + --kind vehicle_data \ + --field charge_state.charge_port_door_open \ + --cross-tab charge_state.charge_port_latch \ + --out-md /tmp/charge_port_door_open.md + +# The same question for a streamed Fleet Telemetry signal (na region only, +# capped so it doesn't have to walk the entire fleet): +python3 scripts/tesla_cache_validate.py \ + --kind signal --field ChargePortDoorOpen --buckets na --sample 500 + +# An energy field, both regions: +python3 scripts/tesla_cache_validate.py \ + --kind energy_live_status --field grid_status + +# Save sanitized copies of every record touched, to seed phase-2 fixtures: +python3 scripts/tesla_cache_validate.py \ + --kind energy_site_info --field components.battery \ + --save-raw /tmp/tesla-fixture-seed +``` + +### Reliability guards (do not remove these) + +These are the difference between evidence and a fabricated conclusion - see +the original recipe report for how each one was discovered the hard way: + +1. **Never uses `nats kv get`.** It reliably times out or silently reads + empty against these buckets. The only read path used is + `nats stream get KV_ --last-for '$KV..' -j`, decoding + the JSON envelope's base64 `data` field. +2. **Has-key semantics, never truthiness.** Presence is classified as + `explicit_value` / `present_null` / `key_absent` / `parent_absent` by + walking the dotted path and checking key membership at each level - an + explicit `false`/`0`/`null` is never mistaken for "absent". +3. **The non-empty-read guard.** Before any "this field is absent" or + "this field is always X" conclusion is trustworthy, the tool must have + actually read at least one non-empty record. If every fetch in a run + failed or decoded empty, it prints `GUARD TRIPPED` and exits non-zero + instead of emitting a report - this is exactly the silent-empty failure + mode that made the original recipe insist on proving reads first. +4. **Read-only, always.** No write path exists in this script. It is safe + to run repeatedly against prod at any time. +5. **Fetch errors are never counted as absent.** A `nats stream get` that + times out or returns "no message found" is a `fetch_error`, tallied and + reported separately from `key_absent`. This matters most for + `energy_live_status`/`energy_site_info`: those cache entries carry a + short NATS message TTL (refreshed roughly every 30s), so a nontrivial + fetch-error rate on energy kinds is expected background noise, not a + tool bug - it means the entry expired between listing and reading, not + that the field was absent. + +### Output + +Markdown (human-readable) and/or JSON (machine-readable) reports include: +presence classification counts and percentages, the observed value domain, +an optional cross-tab against a second field, and an inferred presence shape +(`ALWAYS` / `NULLABLE` / `TRUE_OR_ABSENT` / `CONDITIONALLY_PRESENT` / +`ALWAYS_ABSENT` / `MIXED` / `INSUFFICIENT_DATA`) with a one-line rationale. +Each report ends with a `provenance` line +(`validated na_cache+eu_cache kind=... field=... n=... shape=...`) meant to +be pasted directly into a field's `provenance` string once the phase-3 +`Field`/`Presence` registry exists (see +`api-tesla-schema-fixtures-n6/report.md` for that roadmap). This tool +deliberately stops at producing evidence - it does not implement the +registry or the fixtures module itself. + +### Sanitization + +`--save-raw DIR` writes one sanitized JSON file per record touched. Real +customer values may appear in the statistical report (value domains, +presence counts) since those aren't personally identifying, but full raw +records are redacted before being saved or displayed unless they belong to +the configured reference vehicle/site (`--preferred-vin`/`--preferred-site`, +defaulting to the maintainer's own vehicle and energy site). Redaction +replaces sensitive fields (VIN, id, tokens, precise geo-coordinates, wall +connector VIN/din, email, site name) with a same-typed placeholder rather +than deleting the key, so "is this field present" stays checkable. diff --git a/scripts/tesla_cache_validate.py b/scripts/tesla_cache_validate.py new file mode 100644 index 0000000..7266c1f --- /dev/null +++ b/scripts/tesla_cache_validate.py @@ -0,0 +1,716 @@ +#!/usr/bin/env python3 +"""Validate Tesla field semantics against the live prod NATS KV caches. + +Read-only maintainer tool. It answers per-field questions that are otherwise +guessed at or re-derived by every consumer: is this field a genuine tri-state +(explicit true/false/null), does it follow a true-or-absent shape (present +when true, omitted rather than false), what value domain is actually +observed, and how often is it present across the fleet? + +It queries the `na_cache`/`eu_cache` NATS KV buckets, which hold three record +shapes: + + vehicle_data .vehicle_data REST vehicle_data snapshot (nested dict) + signal .data. one streamed Fleet Telemetry signal (scalar) + energy_live_status energy_sites..live_status REST energy live_status (wrapped) + energy_site_info energy_sites..site_info REST energy site_info (wrapped) + +Reliability guards baked in (see data/chargeport-null-validate-v4/report.md, +the recipe this script promotes to a committed tool): + + * `nats kv get` reliably times out / silently reads empty against these + buckets. This tool ONLY reads via + `nats stream get KV_ --last-for '$KV..' -j` and + base64-decodes the envelope `data` field. Never uses `nats kv get`. + * Presence is classified with has-key semantics, never truthiness, so an + explicit `false`/`0`/`null` is never confused with "absent". + * Before trusting ANY "field is absent" conclusion, the tool proves the + scan actually read at least one non-empty record. If every fetch in a + run failed or decoded empty, it refuses to emit a report and exits + non-zero instead of fabricating an "absent" finding. + * Read-only: this tool never writes to NATS (no `kv put`, no `pub`). It is + meant to run on a maintainer box against an authenticated `nats` CLI + context; no endpoint or credentials are hardcoded here. CI never runs + this script (it has no prod access and none should be granted). + +Usage examples: + + # Vehicle REST snapshot field, both regions, all cached records + python3 scripts/tesla_cache_validate.py \\ + --kind vehicle_data \\ + --field charge_state.charge_port_door_open \\ + --cross-tab charge_state.charge_port_latch \\ + --out-md /tmp/charge_port_door_open.md + + # One streamed Fleet Telemetry signal, na region only, capped sample + python3 scripts/tesla_cache_validate.py \\ + --kind signal --field ChargePortDoorOpen --buckets na --sample 200 + + # Energy live_status field across both regions + python3 scripts/tesla_cache_validate.py \\ + --kind energy_live_status --field grid_status + + # Use a non-default nats CLI context (never hardcode one here) + python3 scripts/tesla_cache_validate.py --context my-prod-context \\ + --kind energy_site_info --field components.battery + +See scripts/README.md for the full recipe and rationale. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import re +import subprocess +import sys +from collections import Counter +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field as dc_field +from pathlib import Path +from typing import Any + +# The maintainer's own reference vehicle/site (captain's ruling, 2026-07-07): +# real customer values may appear in outputs, but should default to this +# vehicle/site where possible; other customers' records are sanitized. +DEFAULT_PREFERRED_VIN = "LRW3F7EK4NC716336" +DEFAULT_PREFERRED_SITE = "2533979794926773" + +REGION_BUCKETS = {"na": "na_cache", "eu": "eu_cache"} + +# Keys redacted (case-insensitive, anywhere in the record) before a record is +# saved to --save-raw or printed as a --show-samples example, unless the +# record belongs to the configured preferred VIN/site. +SENSITIVE_KEYS = { + "vin", "id", "id_s", "vehicle_id", "user_id", "tokens", "token", + "backseat_token", "backseat_token_updated_at", "access_token", + "refresh_token", "email", "site_name", "din", "latitude", "longitude", + "native_latitude", "native_longitude", "address", +} + + +class RecordKind: + VEHICLE_DATA = "vehicle_data" + SIGNAL = "signal" + ENERGY_LIVE_STATUS = "energy_live_status" + ENERGY_SITE_INFO = "energy_site_info" + + ALL = (VEHICLE_DATA, SIGNAL, ENERGY_LIVE_STATUS, ENERGY_SITE_INFO) + + +# --------------------------------------------------------------------------- +# nats CLI plumbing (read-only) +# --------------------------------------------------------------------------- + + +class FetchError(Exception): + """A read against NATS failed (timeout, no message, bad decode).""" + + +def _nats_base_args(context: str | None, timeout: str) -> list[str]: + args = ["nats", f"--timeout={timeout}"] + if context: + args.append(f"--context={context}") + return args + + +def nats_kv_ls(bucket: str, context: str | None, timeout: str) -> list[str]: + """List every key in a KV bucket. Read-only (`nats kv ls`).""" + cmd = _nats_base_args(context, timeout) + ["kv", "ls", bucket] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + if proc.returncode != 0: + raise FetchError(f"kv ls {bucket} failed: {proc.stderr.strip()}") + return [line.strip() for line in proc.stdout.splitlines() if line.strip()] + + +def nats_stream_get_raw( + bucket: str, key: str, context: str | None, timeout: str +) -> bytes: + """Read the current value for one KV key via the reliable path. + + Uses `nats stream get KV_ --last-for '$KV..' -j` + and base64-decodes the envelope's `data` field. Deliberately never uses + `nats kv get`, which reliably times out / silently reads empty on these + buckets (see module docstring). + """ + subject = f"$KV.{bucket}.{key}" + cmd = _nats_base_args(context, timeout) + [ + "stream", "get", f"KV_{bucket}", f"--last-for={subject}", "-j", + ] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + if proc.returncode != 0: + raise FetchError(proc.stderr.strip() or "nats stream get failed") + try: + envelope = json.loads(proc.stdout) + return base64.b64decode(envelope["data"]) + except Exception as exc: # noqa: BLE001 - surfaced as FetchError below + raise FetchError(f"could not decode envelope for {key}: {exc}") from exc + + +# --------------------------------------------------------------------------- +# Record-kind specific key discovery + field extraction +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class RecordRef: + """One KV key that holds a record we want to inspect for a given kind.""" + + bucket: str + key: str + identifier: str # VIN or energy site id, used for sanitization + cross-tab labels + + +def discover_records( + kind: str, bucket: str, field: str, keys: list[str] +) -> list[RecordRef]: + if kind == RecordKind.VEHICLE_DATA: + pattern = re.compile(r"^(?P[^.]+)\.vehicle_data$") + elif kind == RecordKind.ENERGY_LIVE_STATUS: + pattern = re.compile(r"^energy_sites\.(?P[^.]+)\.live_status$") + elif kind == RecordKind.ENERGY_SITE_INFO: + pattern = re.compile(r"^energy_sites\.(?P[^.]+)\.site_info$") + elif kind == RecordKind.SIGNAL: + pattern = re.compile(r"^(?P[^.]+)\.data\." + re.escape(field) + r"$") + else: + raise ValueError(f"unknown kind: {kind}") + + refs = [] + for key in keys: + match = pattern.match(key) + if match: + refs.append(RecordRef(bucket=bucket, key=key, identifier=match.group("id"))) + return refs + + +def discover_signal_universe(bucket: str, keys: list[str]) -> list[str]: + """VINs known to this bucket, used as the presence-rate denominator for + the `signal` kind (a `.data.` key simply won't exist for a VIN + that never streamed that signal, so absence is a listing fact, not a + fetch).""" + pattern = re.compile(r"^(?P[^.]+)\.state$") + vins = [] + for key in keys: + match = pattern.match(key) + if match: + vins.append(match.group("vin")) + return vins + + +ENERGY_ENVELOPE_PREFIX = ("json", "response") + + +def unwrap_energy_envelope(record: Any) -> Any: + """Energy live_status/site_info records are cached as + `{"statusCode": 200, "json": {"response": {...actual fields...}}}`. + Unwrap to the actual field dict; fall back to the raw record if the + envelope shape doesn't match (so a shape change surfaces as ABSENT + fields rather than a crash).""" + node = record + for key in ENERGY_ENVELOPE_PREFIX: + if isinstance(node, dict) and key in node: + node = node[key] + else: + return record + return node + + +class Presence: + EXPLICIT_VALUE = "explicit_value" + PRESENT_NULL = "present_null" + KEY_ABSENT = "key_absent" + PARENT_ABSENT = "parent_absent" # absent because a containing object is missing entirely + FETCH_ERROR = "fetch_error" + + +@dataclass +class ClassifiedField: + identifier: str + bucket: str + presence: str + value: Any = None + cross_tab_value: Any = None + error: str | None = None + + +def extract_dotted(record: Any, path: str) -> tuple[str, Any]: + """Walk a dotted path with has-key semantics (never truthiness). + + Returns (presence, value). `value` is only meaningful for EXPLICIT_VALUE. + """ + node = record + parts = path.split(".") + for i, part in enumerate(parts): + is_last = i == len(parts) - 1 + if not isinstance(node, dict) or part not in node: + # A missing leaf key means its container exists but doesn't + # carry this field; a missing intermediate segment means some + # ancestor object (e.g. drive_state on an asleep-vehicle + # payload) is absent entirely - worth telling apart when + # diagnosing conditionally-present fields. + return (Presence.KEY_ABSENT if is_last else Presence.PARENT_ABSENT), None + node = node[part] + if node is None: + return Presence.PRESENT_NULL, None + return Presence.EXPLICIT_VALUE, node + + +# --------------------------------------------------------------------------- +# Sanitization (captain's ruling, 2026-07-07): real values may appear in +# outputs, prefer the maintainer's own vehicle/site; other customers' +# records must be sanitized but stay shape-valid. +# --------------------------------------------------------------------------- + + +def sanitize_record(record: Any, preferred: bool) -> Any: + if preferred: + return record + if isinstance(record, dict): + out = {} + for k, v in record.items(): + if k.lower() in SENSITIVE_KEYS: + out[k] = _redact_shape_valid(v) + else: + out[k] = sanitize_record(v, preferred) + return out + if isinstance(record, list): + return [sanitize_record(v, preferred) for v in record] + return record + + +def _redact_shape_valid(value: Any) -> Any: + """Replace a sensitive leaf with a shape-valid placeholder (same type, + fixed value) rather than deleting the key, so consumers validating + "field is present" still see it.""" + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return type(value)(0) + if isinstance(value, str): + return "REDACTED" + if value is None: + return None + return "REDACTED" + + +# --------------------------------------------------------------------------- +# Presence-shape inference (labels a finding with the report's taxonomy; +# does NOT build the Field/Presence registry itself - that is phase 3) +# --------------------------------------------------------------------------- + + +def infer_presence_shape(counts: Counter[str], value_domain: Counter[str]) -> tuple[str, str]: + explicit = counts[Presence.EXPLICIT_VALUE] + null = counts[Presence.PRESENT_NULL] + absent = counts[Presence.KEY_ABSENT] + counts[Presence.PARENT_ABSENT] + total_conclusive = explicit + null + absent + + if total_conclusive == 0: + return "INSUFFICIENT_DATA", "no records were successfully classified" + + if absent == 0 and null == 0: + return "ALWAYS", "never null or absent in this sample" + + if null > 0 and explicit > 0: + return ( + "NULLABLE", + f"present-null in {null}/{total_conclusive} records alongside " + f"{explicit} explicit non-null value(s) - genuine tri-state candidate", + ) + + if absent > 0 and null == 0 and explicit > 0: + distinct = len(value_domain) + if distinct == 1: + return ( + "TRUE_OR_ABSENT", + f"only one distinct explicit value observed ({next(iter(value_domain))!r}), " + f"omitted in {absent}/{total_conclusive} - matches the true-or-absent hypothesis", + ) + return ( + "CONDITIONALLY_PRESENT", + f"absent in {absent}/{total_conclusive} with {distinct} distinct explicit " + "values observed when present - confirm the gating condition (scope/sleep/etc.) " + "before assuming true-or-absent", + ) + + if absent > 0 and explicit == 0 and null == 0: + return "ALWAYS_ABSENT", "field never observed present in this sample - check the field path" + + return "MIXED", "does not cleanly match one shape - inspect the raw counts" + + +# --------------------------------------------------------------------------- +# Scan orchestration +# --------------------------------------------------------------------------- + + +@dataclass +class ScanResult: + kind: str + field: str + cross_tab: str | None + buckets: list[str] + sampled: int + counts: Counter[str] = dc_field(default_factory=Counter) + value_domain: Counter[str] = dc_field(default_factory=Counter) + cross_tab_table: dict[tuple[str, str], int] = dc_field(default_factory=dict) + fetch_errors: list[str] = dc_field(default_factory=list) + non_empty_reads: int = 0 + saved_raw: int = 0 + + +def _fetch_and_classify( + ref: RecordRef, + kind: str, + field: str, + cross_tab: str | None, + context: str | None, + timeout: str, +) -> tuple[ClassifiedField, Any]: + try: + raw = nats_stream_get_raw(ref.bucket, ref.key, context, timeout) + except FetchError as exc: + return ClassifiedField(ref.identifier, ref.bucket, Presence.FETCH_ERROR, error=str(exc)), None + + if kind == RecordKind.SIGNAL: + # The whole decoded body IS the field's value (a bare scalar). + text = raw.decode("utf-8", errors="replace").strip() + try: + value = json.loads(text) + except json.JSONDecodeError: + value = text + presence = Presence.PRESENT_NULL if value is None else Presence.EXPLICIT_VALUE + return ClassifiedField(ref.identifier, ref.bucket, presence, value=value), {field: value} + + try: + record = json.loads(raw) + except json.JSONDecodeError as exc: + return ClassifiedField( + ref.identifier, ref.bucket, Presence.FETCH_ERROR, error=f"non-JSON body: {exc}" + ), None + + if kind in (RecordKind.ENERGY_LIVE_STATUS, RecordKind.ENERGY_SITE_INFO): + record = unwrap_energy_envelope(record) + + presence, value = extract_dotted(record, field) + cross_val = None + if cross_tab: + cross_presence, cross_val = extract_dotted(record, cross_tab) + if cross_presence != Presence.EXPLICIT_VALUE: + cross_val = f"<{cross_presence}>" + + classified = ClassifiedField(ref.identifier, ref.bucket, presence, value=value, cross_tab_value=cross_val) + return classified, record + + +def run_scan( + kind: str, + field: str, + regions: list[str], + context: str | None, + timeout: str, + sample: int | None, + concurrency: int, + cross_tab: str | None, + preferred_vin: str, + preferred_site: str, + save_raw_dir: Path | None, +) -> ScanResult: + result = ScanResult(kind=kind, field=field, cross_tab=cross_tab, buckets=[], sampled=0) + + all_refs: list[RecordRef] = [] + for region in regions: + bucket = REGION_BUCKETS[region] + result.buckets.append(bucket) + keys = nats_kv_ls(bucket, context, timeout) + + if kind == RecordKind.SIGNAL: + # Cap the fleet universe itself (not just the present-record + # fetch), so KEY_ABSENT is always counted against the same + # denominator as EXPLICIT_VALUE/PRESENT_NULL. Capping only the + # present-record fetch while leaving KEY_ABSENT computed against + # the full uncapped universe would silently skew every + # percentage once --sample is smaller than the true universe. + universe = discover_signal_universe(bucket, keys) + if sample is not None and sample > 0: + universe = universe[:sample] + universe_set = set(universe) + present_refs = [ + ref + for ref in discover_records(kind, bucket, field, keys) + if ref.identifier in universe_set + ] + present_vins = {ref.identifier for ref in present_refs} + result.counts[Presence.KEY_ABSENT] += len(universe_set) - len(present_vins) + all_refs.extend(present_refs) + else: + refs = discover_records(kind, bucket, field, keys) + if sample is not None and sample > 0: + refs = refs[:sample] + all_refs.extend(refs) + + result.sampled = len(all_refs) + + if save_raw_dir: + save_raw_dir.mkdir(parents=True, exist_ok=True) + + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = [ + pool.submit(_fetch_and_classify, ref, kind, field, cross_tab, context, timeout) + for ref in all_refs + ] + for ref, future in zip(all_refs, futures): + classified, raw_record = future.result() + result.counts[classified.presence] += 1 + if classified.presence == Presence.FETCH_ERROR: + result.fetch_errors.append(f"{ref.bucket}:{ref.key}: {classified.error}") + continue + + result.non_empty_reads += 1 + + if classified.presence == Presence.EXPLICIT_VALUE: + result.value_domain[repr(classified.value)] += 1 + + if cross_tab and classified.presence == Presence.EXPLICIT_VALUE: + key = (repr(classified.value), repr(classified.cross_tab_value)) + result.cross_tab_table[key] = result.cross_tab_table.get(key, 0) + 1 + + if save_raw_dir and raw_record is not None: + is_preferred = ref.identifier in (preferred_vin, preferred_site) + sanitized = sanitize_record(raw_record, is_preferred) + out_path = save_raw_dir / f"{ref.bucket}.{ref.identifier}.{kind}.json" + out_path.write_text(json.dumps(sanitized, indent=2, sort_keys=True)) + result.saved_raw += 1 + + return result + + +# --------------------------------------------------------------------------- +# Reporting +# --------------------------------------------------------------------------- + + +def render_markdown(result: ScanResult) -> str: + shape, shape_note = infer_presence_shape(result.counts, result.value_domain) + total = sum(result.counts.values()) + lines = [ + f"# Field validation: `{result.field}` ({result.kind})", + "", + f"**Buckets scanned:** {', '.join(result.buckets)} ", + f"**Records fetched over the network:** {result.sampled} ", + f"**Records classified (fetched + known-absent-from-listing):** {total} ", + f"**Non-empty reads:** {result.non_empty_reads} ", + f"**Fetch errors:** {len(result.fetch_errors)}", + "", + ] + if result.kind == RecordKind.SIGNAL: + lines.insert( + 3, + "_For `signal`, absence is known for free from the key listing " + "(no fetch needed), so `classified` can exceed `fetched`._ ", + ) + + if result.non_empty_reads == 0: + lines += [ + "## GUARD TRIPPED - no absence conclusion is trustworthy", + "", + "Every read in this scan failed or decoded empty. This matches the " + "known silent-empty failure mode (see module docstring) - do NOT " + "treat any KEY_ABSENT count below as evidence the field is absent. " + "Fix connectivity/context/timeout and re-run.", + "", + ] + + lines += [ + f"**Inferred presence shape:** `{shape}` ", + f"_{shape_note}_", + "", + "## Presence classification", + "", + "| Classification | Count | % of total |", + "|---|---|---|", + ] + for presence in ( + Presence.EXPLICIT_VALUE, + Presence.PRESENT_NULL, + Presence.KEY_ABSENT, + Presence.PARENT_ABSENT, + Presence.FETCH_ERROR, + ): + count = result.counts.get(presence, 0) + pct = (count / total * 100) if total else 0.0 + lines.append(f"| {presence} | {count} | {pct:.1f}% |") + + lines += ["", "## Observed value domain", "", "| Value | Count |", "|---|---|"] + for value, count in result.value_domain.most_common(30): + lines.append(f"| `{value}` | {count} |") + if len(result.value_domain) > 30: + lines.append(f"| ... {len(result.value_domain) - 30} more distinct values | |") + + if result.cross_tab: + lines += [ + "", + f"## Cross-tab: `{result.field}` vs `{result.cross_tab}`", + "", + f"| {result.field} | {result.cross_tab} | Count |", + "|---|---|---|", + ] + for (value, cross_val), count in sorted( + result.cross_tab_table.items(), key=lambda kv: -kv[1] + ): + lines.append(f"| `{value}` | `{cross_val}` | {count} |") + + if result.fetch_errors: + lines += ["", "## Fetch errors (excluded from presence counts)", ""] + for err in result.fetch_errors[:20]: + lines.append(f"- `{err}`") + if len(result.fetch_errors) > 20: + lines.append(f"- ... {len(result.fetch_errors) - 20} more") + + lines += [ + "", + "## Provenance", + "", + "```", + f"validated {'+'.join(result.buckets)} " + f"kind={result.kind} field={result.field} " + f"n={result.sampled} non_empty={result.non_empty_reads} " + f"shape={shape}", + "```", + "", + "Paste the line above into a field's `provenance` string once the " + "phase-3 registry exists.", + ] + return "\n".join(lines) + + +def render_json(result: ScanResult) -> dict[str, Any]: + shape, shape_note = infer_presence_shape(result.counts, result.value_domain) + return { + "kind": result.kind, + "field": result.field, + "cross_tab": result.cross_tab, + "buckets": result.buckets, + "sampled": result.sampled, + "non_empty_reads": result.non_empty_reads, + "counts": dict(result.counts), + "value_domain": dict(result.value_domain), + "cross_tab_table": {f"{k[0]}|{k[1]}": v for k, v in result.cross_tab_table.items()}, + "fetch_errors": result.fetch_errors, + "saved_raw": result.saved_raw, + "inferred_shape": shape, + "inferred_shape_note": shape_note, + } + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--kind", required=True, choices=RecordKind.ALL) + parser.add_argument( + "--field", + required=True, + help="Dotted path within the record (vehicle_data/energy kinds), or the " + "bare Signal name (signal kind), e.g. charge_state.charge_port_door_open " + "or ChargePortDoorOpen.", + ) + parser.add_argument( + "--cross-tab", + default=None, + help="Optional second dotted path to cross-tabulate against (same record). " + "Not supported for --kind signal.", + ) + parser.add_argument( + "--buckets", + default="na,eu", + help="Comma-separated regions to scan: na, eu, or na,eu (default).", + ) + parser.add_argument( + "--context", + default=None, + help="nats CLI context name to use. Defaults to whatever context is " + "currently selected (`nats context select`) - never hardcode a " + "context, endpoint, or credential here.", + ) + parser.add_argument("--timeout", default="20s", help="Per-read nats timeout (default 20s).") + parser.add_argument( + "--sample", + type=int, + default=None, + help="Cap the number of records fetched (default: fetch every discovered key).", + ) + parser.add_argument("--concurrency", type=int, default=10, help="Parallel fetches (default 10).") + parser.add_argument("--out-json", type=Path, default=None, help="Write the JSON report here.") + parser.add_argument("--out-md", type=Path, default=None, help="Write the Markdown report here.") + parser.add_argument( + "--save-raw", + type=Path, + default=None, + help="Directory to save sanitized per-record JSON, seeding phase-2 fixtures. " + "Off by default.", + ) + parser.add_argument("--preferred-vin", default=DEFAULT_PREFERRED_VIN) + parser.add_argument("--preferred-site", default=DEFAULT_PREFERRED_SITE) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + regions = [r.strip() for r in args.buckets.split(",") if r.strip()] + for region in regions: + if region not in REGION_BUCKETS: + print(f"error: unknown region {region!r}, expected one of {list(REGION_BUCKETS)}", file=sys.stderr) + return 2 + + if args.kind == RecordKind.SIGNAL and args.cross_tab: + print("error: --cross-tab is not supported for --kind signal", file=sys.stderr) + return 2 + + try: + result = run_scan( + kind=args.kind, + field=args.field, + regions=regions, + context=args.context, + timeout=args.timeout, + sample=args.sample, + concurrency=args.concurrency, + cross_tab=args.cross_tab, + preferred_vin=args.preferred_vin, + preferred_site=args.preferred_site, + save_raw_dir=args.save_raw, + ) + except FetchError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + if result.non_empty_reads == 0: + print( + "GUARD TRIPPED: every read failed or decoded empty - refusing to " + "report an absence conclusion. See --out-md/--out-json for details " + "if you still want the raw counts, or fix connectivity and re-run.", + file=sys.stderr, + ) + + markdown = render_markdown(result) + if args.out_md: + args.out_md.write_text(markdown) + else: + print(markdown) + + if args.out_json: + args.out_json.write_text(json.dumps(render_json(result), indent=2, sort_keys=True)) + + return 0 if result.non_empty_reads > 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_cache_validate_logic.py b/tests/test_cache_validate_logic.py new file mode 100644 index 0000000..fd3a0fd --- /dev/null +++ b/tests/test_cache_validate_logic.py @@ -0,0 +1,172 @@ +"""Offline unit tests for the pure logic in scripts/tesla_cache_validate.py. + +These exercise field extraction, envelope unwrapping, sanitization, presence- +shape inference, and key discovery against synthetic records shaped like the +real na_cache/eu_cache corpus (see +data/chargeport-null-validate-v4/raw/ for the originals this is modeled on). +Nothing here touches NATS - safe to run anywhere, including CI. +""" +from __future__ import annotations + +import sys +from collections import Counter +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +import tesla_cache_validate as tcv # noqa: E402 + +results: list[tuple[str, bool]] = [] + + +def check(name: str, got: object, expected: object) -> None: + ok = got == expected + results.append((name, ok)) + if not ok: + print(f"FAIL {name}: expected {expected!r}, got {got!r}") + + +def test_extract_dotted() -> None: + record = { + "charge_state": { + "charge_port_door_open": False, + "charge_port_latch": "Engaged", + "user_charge_enable_request": None, + } + } + check( + "extract_dotted explicit false", + tcv.extract_dotted(record, "charge_state.charge_port_door_open"), + (tcv.Presence.EXPLICIT_VALUE, False), + ) + check( + "extract_dotted present null", + tcv.extract_dotted(record, "charge_state.user_charge_enable_request"), + (tcv.Presence.PRESENT_NULL, None), + ) + check( + "extract_dotted key absent (leaf missing)", + tcv.extract_dotted(record, "charge_state.nonexistent_field"), + (tcv.Presence.KEY_ABSENT, None), + ) + check( + "extract_dotted parent absent (no drive_state at all)", + tcv.extract_dotted(record, "drive_state.latitude"), + (tcv.Presence.PARENT_ABSENT, None), + ) + + +def test_unwrap_energy_envelope() -> None: + wrapped = {"statusCode": 200, "json": {"response": {"grid_status": "Active"}}} + check( + "unwrap_energy_envelope unwraps", + tcv.unwrap_energy_envelope(wrapped), + {"grid_status": "Active"}, + ) + unwrapped_shape = {"grid_status": "Active"} + check( + "unwrap_energy_envelope falls back on unrecognized shape", + tcv.unwrap_energy_envelope(unwrapped_shape), + unwrapped_shape, + ) + + +def test_sanitize_record() -> None: + record = { + "vin": "5YJ3E1EA0KF495312", + "battery_level": 40, + "wall_connectors": [{"vin": "5YJ3E1EA0KF495312", "din": "abc", "wall_connector_power": 0}], + } + check( + "sanitize_record passthrough for preferred identifier", + tcv.sanitize_record(record, preferred=True), + record, + ) + sanitized = tcv.sanitize_record(record, preferred=False) + check("sanitize_record redacts top-level vin", sanitized["vin"], "REDACTED") + check("sanitize_record keeps non-sensitive numeric field", sanitized["battery_level"], 40) + check( + "sanitize_record redacts nested vin/din but keeps numeric shape", + (sanitized["wall_connectors"][0]["vin"], sanitized["wall_connectors"][0]["din"], + sanitized["wall_connectors"][0]["wall_connector_power"]), + ("REDACTED", "REDACTED", 0), + ) + + +def test_infer_presence_shape() -> None: + always = Counter({tcv.Presence.EXPLICIT_VALUE: 720, tcv.Presence.KEY_ABSENT: 0}) + check("infer_presence_shape ALWAYS", tcv.infer_presence_shape(always, Counter({"0": 720}))[0], "ALWAYS") + + nullable = Counter({tcv.Presence.EXPLICIT_VALUE: 710, tcv.Presence.PRESENT_NULL: 13}) + check("infer_presence_shape NULLABLE", tcv.infer_presence_shape(nullable, Counter({"False": 431, "True": 279}))[0], "NULLABLE") + + true_or_absent = Counter({tcv.Presence.EXPLICIT_VALUE: 40, tcv.Presence.KEY_ABSENT: 3448}) + check( + "infer_presence_shape TRUE_OR_ABSENT", + tcv.infer_presence_shape(true_or_absent, Counter({"True": 40}))[0], + "TRUE_OR_ABSENT", + ) + + conditional = Counter({tcv.Presence.EXPLICIT_VALUE: 23, tcv.Presence.KEY_ABSENT: 5}) + check( + "infer_presence_shape CONDITIONALLY_PRESENT", + tcv.infer_presence_shape(conditional, Counter({"Active": 22, "Unknown": 1}))[0], + "CONDITIONALLY_PRESENT", + ) + + empty: Counter[str] = Counter() + check("infer_presence_shape INSUFFICIENT_DATA", tcv.infer_presence_shape(empty, Counter())[0], "INSUFFICIENT_DATA") + + always_absent = Counter({tcv.Presence.KEY_ABSENT: 10}) + check("infer_presence_shape ALWAYS_ABSENT", tcv.infer_presence_shape(always_absent, Counter())[0], "ALWAYS_ABSENT") + + +def test_discover_records() -> None: + keys = [ + "5YJ3E1EA0KF495312.vehicle_data", + "5YJ3E1EA0KF495312.data.ChargePortDoorOpen", + "5YJ3E1EA0KF495312.state", + "5YJ3E1EA0KF495312.alerts", + "5YJ3E1EA0KF495312.connectivity.wifi", + "energy_sites.2533979794926773.live_status", + "energy_sites.2533979794926773.site_info", + "energy_sites.2533979794926773.calendar_history.energy.day", + ] + + vd_refs = tcv.discover_records(tcv.RecordKind.VEHICLE_DATA, "na_cache", "unused", keys) + check("discover_records vehicle_data finds exactly one", [r.key for r in vd_refs], ["5YJ3E1EA0KF495312.vehicle_data"]) + + sig_refs = tcv.discover_records(tcv.RecordKind.SIGNAL, "na_cache", "ChargePortDoorOpen", keys) + check("discover_records signal matches exact signal name", [r.key for r in sig_refs], ["5YJ3E1EA0KF495312.data.ChargePortDoorOpen"]) + + live_refs = tcv.discover_records(tcv.RecordKind.ENERGY_LIVE_STATUS, "na_cache", "unused", keys) + check("discover_records energy_live_status", [r.identifier for r in live_refs], ["2533979794926773"]) + + site_refs = tcv.discover_records(tcv.RecordKind.ENERGY_SITE_INFO, "na_cache", "unused", keys) + check("discover_records energy_site_info", [r.identifier for r in site_refs], ["2533979794926773"]) + + universe = tcv.discover_signal_universe("na_cache", keys) + check("discover_signal_universe only counts .state keys", universe, ["5YJ3E1EA0KF495312"]) + + +def main() -> None: + test_extract_dotted() + test_unwrap_energy_envelope() + test_sanitize_record() + test_infer_presence_shape() + test_discover_records() + + print(f"{'test':<70} result") + print("-" * 80) + all_ok = True + for name, ok in results: + all_ok = all_ok and ok + print(f"{name:<70} {'PASS' if ok else 'FAIL'}") + print("-" * 80) + print("ALL PASS" if all_ok else "FAILURES PRESENT") + if not all_ok: + raise SystemExit(1) + + +if __name__ == "__main__": + main()