diff --git a/connections/_schema/mapping_rules.json b/connections/_schema/mapping_rules.json new file mode 100644 index 00000000..8ad580fb --- /dev/null +++ b/connections/_schema/mapping_rules.json @@ -0,0 +1,253 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "RoamCore Capability Mapping Rules", + "description": "Declarative rules that turn raw Home Assistant entity_ids into canonical RoamCore capability ids. The rule list is consumed by the pure-Python mapper in homeassistant/custom_components/roamcore/capability_mapper.py. Every rule's `canonical_capability` MUST exist in canonical_capabilities.json (validated by validate_mapping_rules). Pattern syntax is Python re (matched against the entity_id via re.fullmatch). Naming follows docs/reference/rc-entity-naming.md: contract ids start with `rc_`, no vendor names in contract ids, but the `source_pattern` is allowed to mention vendor names (it is matching raw upstream entity_ids, not contract ids).", + "version": 1, + "rules": [ + { + "id": "victron_battery_soc", + "source_pattern": "sensor\\.(vt_|victron).*battery_?(state_of_charge|soc|state)(_percent|_pct)?", + "canonical_capability": "rc_power_battery_soc", + "weight": 95, + "description": "Victron (any model) battery state of charge → leisure battery percentage" + }, + { + "id": "victron_battery_voltage", + "source_pattern": "sensor\\.(vt_|victron).*battery_?(voltage|v)(_v)?", + "canonical_capability": "rc_power_battery_voltage", + "weight": 95, + "description": "Victron (any model) battery terminal voltage" + }, + { + "id": "victron_battery_current", + "source_pattern": "sensor\\.(vt_|victron).*battery_?(current|i)(_a)?", + "canonical_capability": "rc_power_battery_current", + "weight": 95, + "description": "Victron (any model) battery current (positive = charging)" + }, + { + "id": "renogy_battery_soc", + "source_pattern": "sensor\\.renogy.*soc", + "canonical_capability": "rc_power_battery_soc", + "weight": 90, + "description": "Renogy battery monitor → leisure battery percentage (fallback to victron rule)" + }, + { + "id": "victron_solar_power", + "source_pattern": "sensor\\.(vt_|victron_)?solar_?(power|w)(_w)?", + "canonical_capability": "rc_power_solar_power", + "weight": 95, + "description": "Victron / generic solar charging power in watts" + }, + { + "id": "renogy_solar_power", + "source_pattern": "sensor\\.renogy.*solar.*(power|w)", + "canonical_capability": "rc_power_solar_power", + "weight": 90, + "description": "Renogy solar controller → solar charging power" + }, + { + "id": "victron_shore_connected", + "source_pattern": "binary_sensor\\.(vt_|victron_)?shore.*(connected|plug)", + "canonical_capability": "rc_power_shore_connected", + "weight": 95, + "description": "Whether the van is plugged into shore / mains power" + }, + { + "id": "interior_lights_state", + "source_pattern": "switch\\.(cabin[_a-z0-9]*|interior[_a-z0-9]*|main_lights|rc_lighting_interior)", + "canonical_capability": "rc_lighting_interior_state", + "weight": 90, + "description": "Interior cabin lights on/off (cabin / interior / main_lights, with any suffix)" + }, + { + "id": "approach_lights_state", + "source_pattern": "switch\\.(porch_light[a-z0-9_]*|awning_light[a-z0-9_]*|approach_light[a-z0-9_]*|exterior_light[a-z0-9_]*|rc_lighting_approach[a-z_0-9]*)", + "canonical_capability": "rc_lighting_approach_state", + "weight": 90, + "description": "Approach / porch / awning lights on/off" + }, + { + "id": "indoor_temperature", + "source_pattern": "sensor\\.(indoor_temp[a-z0-9_]*|indoor_temperature[a-z0-9_]*|cabin_temp[a-z0-9_]*|cabin_temperature[a-z0-9_]*|rc_climate_indoor[a-z_0-9]*)", + "canonical_capability": "rc_climate_indoor_temperature", + "weight": 90, + "description": "Indoor cabin temperature (degrees C)" + }, + { + "id": "hvac_main_state", + "source_pattern": "switch\\.(hvac_main[a-z0-9_]*|hvac[a-z0-9_]*|webasto[a-z0-9_]*|air_conditioner[a-z0-9_]*|ac_main[a-z0-9_]*|heater_main[a-z0-9_]*|rc_climate_hvac[a-z_0-9]*)", + "canonical_capability": "rc_climate_hvac_state", + "weight": 90, + "description": "Primary HVAC / heater / AC / Webasto on/off" + }, + { + "id": "fresh_water_level", + "source_pattern": "sensor\\.(fresh_water[a-z0-9_]*|fresh_tank[a-z0-9_]*|water_fresh[a-z0-9_]*|potable_water[a-z0-9_]*|rc_water_fresh[a-z_0-9]*)", + "canonical_capability": "rc_water_fresh_level", + "weight": 90, + "description": "Fresh / potable water tank level (percent)" + }, + { + "id": "water_pump_state", + "source_pattern": "switch\\.(water_pump[a-z0-9_]*|shurflo[a-z0-9_]*|rc_water_pump[a-z_0-9]*)", + "canonical_capability": "rc_water_pump_state", + "weight": 90, + "description": "12V water pump on/off" + }, + { + "id": "gps_latitude", + "source_pattern": "(sensor|device_tracker)\\.(gps_lat|vehicle_lat|vt_vehicle_lat|rc_position_lat|location_latitude)", + "canonical_capability": "rc_position_lat", + "weight": 85, + "description": "Vehicle GPS latitude (degrees)" + }, + { + "id": "gps_longitude", + "source_pattern": "(sensor|device_tracker)\\.(gps_lon|gps_lng|vehicle_lon|vt_vehicle_lon|rc_position_lon|location_longitude)", + "canonical_capability": "rc_position_lon", + "weight": 85, + "description": "Vehicle GPS longitude (degrees)" + }, + { + "id": "vehicle_device_tracker", + "source_pattern": "device_tracker\\.(vt_vehicle|roamcore_vehicle|van_gps|vehicle)", + "canonical_capability": "rc_position_lat", + "weight": 70, + "description": "device_tracker-style GPS — primary mapping to lat (lat + lon both available in the same tracker)" + }, + { + "id": "internet_reachable", + "source_pattern": "binary_sensor\\.(rc_net_internet_reachable|internet_reachable|wan_reachable|net_internet)", + "canonical_capability": "rc_network_internet_reachable", + "weight": 95, + "description": "Whether the van has any working internet path (LTE, Starlink, Wi-Fi)" + }, + { + "id": "wan_ip_address", + "source_pattern": "sensor\\.(rc_net_wan_ip|wan_ip|public_ip|external_ip)", + "canonical_capability": "rc_network_wan_ip", + "weight": 90, + "description": "Current public / WAN IP address of the active internet link" + }, + { + "id": "roamcore_contract_power_soc", + "source_pattern": "sensor\\.rc_power_battery_soc", + "canonical_capability": "rc_power_battery_soc", + "weight": 100, + "description": "Self-mapping of an existing canonical tile (identity rule — passes through if a downstream template already published one)" + }, + { + "id": "roamcore_contract_power_voltage", + "source_pattern": "sensor\\.rc_power_battery_voltage", + "canonical_capability": "rc_power_battery_voltage", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_power_current", + "source_pattern": "sensor\\.rc_power_battery_current", + "canonical_capability": "rc_power_battery_current", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_solar_power", + "source_pattern": "sensor\\.rc_power_solar_power", + "canonical_capability": "rc_power_solar_power", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_shore_connected", + "source_pattern": "binary_sensor\\.rc_power_shore_connected", + "canonical_capability": "rc_power_shore_connected", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_lighting_interior", + "source_pattern": "switch\\.rc_lighting_interior_state", + "canonical_capability": "rc_lighting_interior_state", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_lighting_approach", + "source_pattern": "switch\\.rc_lighting_approach_state", + "canonical_capability": "rc_lighting_approach_state", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_climate_indoor", + "source_pattern": "sensor\\.rc_climate_indoor_temperature", + "canonical_capability": "rc_climate_indoor_temperature", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_climate_hvac", + "source_pattern": "switch\\.rc_climate_hvac_state", + "canonical_capability": "rc_climate_hvac_state", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_water_fresh", + "source_pattern": "sensor\\.rc_water_fresh_level", + "canonical_capability": "rc_water_fresh_level", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_water_pump", + "source_pattern": "switch\\.rc_water_pump_state", + "canonical_capability": "rc_water_pump_state", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_position_lat", + "source_pattern": "(sensor|device_tracker)\\.rc_position_lat", + "canonical_capability": "rc_position_lat", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_position_lon", + "source_pattern": "(sensor|device_tracker)\\.rc_position_lon", + "canonical_capability": "rc_position_lon", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_network_internet", + "source_pattern": "binary_sensor\\.rc_network_internet_reachable", + "canonical_capability": "rc_network_internet_reachable", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "roamcore_contract_network_wan_ip", + "source_pattern": "sensor\\.rc_network_wan_ip", + "canonical_capability": "rc_network_wan_ip", + "weight": 100, + "description": "Self-mapping of an existing canonical tile" + }, + { + "id": "generic_battery_soc_fallback", + "source_pattern": "sensor\\..*battery.*(soc|state_of_charge)", + "canonical_capability": "rc_power_battery_soc", + "weight": 60, + "description": "Last-resort fallback: any sensor mentioning battery + soc maps to the leisure battery percentage" + }, + { + "id": "generic_solar_power_fallback", + "source_pattern": "sensor\\..*solar.*(power|w)", + "canonical_capability": "rc_power_solar_power", + "weight": 60, + "description": "Last-resort fallback: any sensor mentioning solar + power/w maps to solar charging power" + } + ] +} diff --git a/docs/reference/rc-capability-mapping.md b/docs/reference/rc-capability-mapping.md new file mode 100644 index 00000000..af511437 --- /dev/null +++ b/docs/reference/rc-capability-mapping.md @@ -0,0 +1,53 @@ +# How RoamCore knows where each device belongs + +> **One sentence:** RoamCore automatically organises whatever you plug in — battery, solar, water, network — without you telling it which is which. + +This page is for anyone who wants to understand *how* a new device ends up in the right place on the RoamCore dashboard, and what to do if it doesn't. + +If you only want to install a device and see it on your dashboard, the connection wizard does all of this for you. You don't need to read this page. If you're curious, or you want to know why a Victron SmartShunt and a Renogy battery monitor both show up in the same `Battery` tile, keep reading. + +--- + +## §1 What this is + +Every device in your van — Victron, Renogy, a generic USB temperature sensor, anything — has its own brand-specific way of talking. Without help, your dashboard would need one card per brand, and switching brands would mean rebuilding the dashboard. RoamCore solves this with a **shared vocabulary** of "things in a van" — battery, solar, water, network, position, lighting — and a smart layer that quietly translates whatever you plug in into that vocabulary. + +You do nothing. RoamCore looks at the device, decides which box on the dashboard it belongs in, and puts it there. + +--- + +## §2 What you see + +When you open the RoamCore dashboard, you see one **tile** per thing in the van. The tile is labelled in plain English (Battery, Solar, Fresh Water, Internet, GPS) and shows you the current value. + +When you plug in a new device, the right tile just *updates with the new value*. You don't have to drag anything, you don't have to edit a YAML, you don't have to pick from a dropdown. The tile was already there — now it's showing real data from the new device. + +If a device doesn't fit any known tile, RoamCore doesn't drop it on the floor. It goes into a "things we haven't categorised yet" panel in Advanced mode and stays out of your way. + +--- + +## §3 What you do + +**Nothing.** This is automatic. + +If you're the curious type and want to peek under the hood: there is a small list of rules that tell RoamCore "an entity that mentions battery + voltage is the leisure battery voltage." That list lives in the RoamCore repo (it's plain text — you can read it in your browser). You don't need to edit it, and RoamCore will never ask you to. It's there so that power users can suggest new mappings if a new device appears that RoamCore doesn't recognise yet. + +--- + +## §4 What to do if it goes wrong + +If a device shows up in the wrong tile (for example, your water pump is showing up under "Lighting"), or if a device doesn't show up at all: + +1. Open the **About this tile** panel on the tile that's wrong. It tells you which rule RoamCore used to decide where the device belongs. +2. If the rule is obviously wrong (e.g. the rule says "anything with `pump` in the name is a light"), file an issue against the RoamCore repo with the entity name from the panel. RoamCore will add a better rule and the next update will fix it for everyone. +3. If the device doesn't show up at all, the rule list probably doesn't have an entry for it yet. File the same kind of issue with the entity name — RoamCore's maintainers will add a rule. + +In either case the message is the same: **file an issue with the entity name, RoamCore will fix it in a future update.** You don't need to edit anything yourself. + +--- + +## §5 Useful links + +- **The canonical tile list** — the full vocabulary of "things in a van" that RoamCore knows about. It's plain English and lives in the RoamCore repo. +- **The naming convention** — explains why every tile is labelled the way it is (so the dashboard stays stable when you switch brands). +- **The connection wizard** — the setup flow that adds a new device. It runs the same smart layer automatically; you don't need to know any of this to use it. diff --git a/homeassistant/custom_components/roamcore/capability_mapper.py b/homeassistant/custom_components/roamcore/capability_mapper.py new file mode 100644 index 00000000..4a61f7a4 --- /dev/null +++ b/homeassistant/custom_components/roamcore/capability_mapper.py @@ -0,0 +1,444 @@ +"""Pure-Python capability mapper for the RoamCore canonical vehicle model. + +Phase 2 (Wave 9 #119.b) — the capability mapping layer turns raw +Home Assistant entity_ids into the canonical RoamCore capability ids +defined in `connections/_schema/canonical_capabilities.json`. The +mapping is declarative (`connections/_schema/mapping_rules.json`) and +this module is the pure-Python engine that applies it. + +Design goals (doctrine: must-not-fail + super-intuitive + critical +infrastructure, Bernard 2026-08-04): + * Pure stdlib + json (NO Home Assistant runtime imports — keep this + importable in tests outside an HA install, same as vehicle_model). + * Idempotent: same input → same output, every time. No randomness, + no network calls, no clock-dependence. + * Auto-recover on unknown entity_ids: never crash the dashboard. + Return None + log to a per-call log; callers surface a tile-readable + string ("I couldn't figure out where this device belongs — check + the setup wizard") instead of a stack trace. + * Plain-English errors: validation errors are returned as a list of + human-readable strings (empty list = valid), mirroring the + vehicle_model.validate_capabilities contract. + * Naming follows `docs/reference/rc-entity-naming.md`: the + `canonical_capability` field must start with `rc_`. + * Higher weight wins on conflicts, deterministic tie-break by + rule_id (alphabetical). + * Functions: load_mapping_rules, load_capability_schema, + resolve_entity_to_capability, map_entities, apply_mapping_rules, + validate_mapping_rules, _confidence_from_weight. + +How matching works: + * Every rule's `source_pattern` is a Python regex (NOT a glob). + We use re.fullmatch against the entity_id (which is a plain + ASCII string like "sensor.vt_battery_soc_percent"). + * On a single entity_id, every rule is tried in turn. Matching + rules are collected, sorted by (-weight, rule_id), and the + winner becomes the resolved capability. Confidence is computed + as weight / 100 (so 0..1, mapped to 0.0..1.0 float). + * If no rule matches, resolve returns None and the caller records + the entity in the unmatched list. The dashboard surfaces a + user-friendly fallback ("I couldn't figure out where this + device belongs") for those. + +Backwards compatibility with vehicle_model.py: + * This module does NOT re-implement the canonical_capabilities + validator. It imports `vehicle_model.load_capabilities` / + `validate_capabilities` and uses the schema as the ground truth + for `validate_mapping_rules` and `apply_mapping_rules`. + * vehicle_model.py stays untouched (slice contract: minimal + + additive only). +""" + +from __future__ import annotations + +import json +import os +import re +from typing import Any, Iterable + +# --- Public constants (exported so callers + tests don't hardcode) --- + +# Where the mapping rules file lives by default. Overridable per-call. +DEFAULT_MAPPING_RULES_PATH = "connections/_schema/mapping_rules.json" + +# Weight floor / ceiling. weight ∈ [0, 100]; confidence = weight / 100. +WEIGHT_MIN: int = 0 +WEIGHT_MAX: int = 100 + +# The rc_ prefix is mandatory for every canonical_capability reference. +# This duplicates vehicle_model.py's check on purpose — the mapper has +# to enforce it independently because the rule file is its own input. +_CONTRACT_PREFIX = "rc_" + + +# --- Loaders --- + + +def load_mapping_rules( + path: str | os.PathLike[str] = DEFAULT_MAPPING_RULES_PATH, +) -> dict[str, Any]: + """Read + JSON-parse the mapping rules file. + + Returns the full document (so callers can read `version` / `title` + / `description` as well as the per-rule entries). + + Raises: + FileNotFoundError: when `path` does not exist. + json.JSONDecodeError: when the file is not valid JSON. + """ + with open(path, "r", encoding="utf-8") as fp: + return json.load(fp) + + +def load_capability_schema( + path: str | os.PathLike[str] = "connections/_schema/canonical_capabilities.json", +) -> dict[str, Any]: + """Load the canonical capabilities schema. + + Thin wrapper around `vehicle_model.load_capabilities` so callers + that already imported the mapper don't have to import + vehicle_model separately. The canonical_capabilities.json schema + is the ground truth for `validate_mapping_rules`. + + Raises: + FileNotFoundError: when `path` does not exist. + json.JSONDecodeError: when the file is not valid JSON. + """ + # Lazy import to keep `capability_mapper` importable without a + # parent-package collision (the conftest patches the parent + # `homeassistant.custom_components.roamcore` package; a top-level + # `from .vehicle_model import load_capabilities` would crash on + # collection). Loading by absolute path avoids the package + # machinery entirely, exactly like test_vehicle_model.py does. + import importlib.util + + _HERE = os.path.dirname(os.path.abspath(__file__)) + _VM_PATH = os.path.join(_HERE, "vehicle_model.py") + _spec = importlib.util.spec_from_file_location( + "roamcore_vehicle_model_lazy", _VM_PATH + ) + if _spec is None or _spec.loader is None: + raise ImportError( + f"vehicle_model.py not loadable from {_VM_PATH!r}" + ) + _mod = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(_mod) + return _mod.load_capabilities(path) + + +# --- Validation --- + + +def validate_mapping_rules( + rules_doc: dict[str, Any], + schema: dict[str, Any], +) -> list[str]: + """Return a list of plain-English validation errors for the rules. + + Empty list means the rules doc is valid. Function is deliberately + total (never raises) so callers can aggregate errors across many + documents without try/except noise. Same contract as + vehicle_model.validate_capabilities. + + Checks performed: + 1. Top-level shape (`rules` is a non-empty list, optional + `version` is an int). + 2. Every rule has the 5 required keys (id, source_pattern, + canonical_capability, weight, description). + 3. Every `canonical_capability` value exists in `schema`'s + `capabilities` list (so a typo doesn't silently map into + the void). + 4. Every `canonical_capability` starts with `rc_`. + 5. Every `weight` is an int in [0, 100]. + 6. Every `source_pattern` compiles as a valid Python regex. + 7. Every `id` is unique (duplicate rule ids = silent override). + 8. Every `id` and `canonical_capability` is a non-empty string. + 9. `description` is a non-empty string when present. + """ + errors: list[str] = [] + + if not isinstance(rules_doc, dict): + return ["top-level document must be a JSON object"] + + version = rules_doc.get("version") + if version is not None and not isinstance(version, int): + errors.append( + f"'version' must be an integer when present (got {type(version).__name__})" + ) + + rules = rules_doc.get("rules") + if not isinstance(rules, list) or not rules: + errors.append("'rules' must be a non-empty list") + return errors + + # Build the set of valid canonical_capability ids from the schema. + capabilities = schema.get("capabilities") if isinstance(schema, dict) else None + if not isinstance(capabilities, list) or not capabilities: + errors.append( + "schema has no 'capabilities' list — cannot validate rule targets" + ) + valid_cap_ids: set[str] = set() + else: + valid_cap_ids = { + c.get("id") + for c in capabilities + if isinstance(c, dict) and isinstance(c.get("id"), str) + } + + seen_rule_ids: set[str] = set() + required_keys = { + "id", + "source_pattern", + "canonical_capability", + "weight", + } + + for idx, rule in enumerate(rules): + prefix = f"rules[{idx}]" + + if not isinstance(rule, dict): + errors.append(f"{prefix}: must be a JSON object") + continue + + # Required keys. + missing = required_keys - set(rule.keys()) + if missing: + errors.append( + f"{prefix}: missing required keys {sorted(missing)}" + ) + # Continue with what we can check — partial validation is + # better than swallowing the rest. + + # id. + rule_id = rule.get("id") + if not isinstance(rule_id, str) or not rule_id: + errors.append(f"{prefix}.id: must be a non-empty string") + elif rule_id in seen_rule_ids: + errors.append( + f"{prefix}.id ({rule_id!r}): duplicate rule id " + "(every id must be unique)" + ) + else: + seen_rule_ids.add(rule_id) + + # source_pattern must be a compilable regex. + source_pattern = rule.get("source_pattern") + if not isinstance(source_pattern, str) or not source_pattern: + errors.append( + f"{prefix}.source_pattern: must be a non-empty string" + ) + else: + try: + re.compile(source_pattern) + except re.error as exc: + errors.append( + f"{prefix}.source_pattern ({source_pattern!r}): " + f"invalid regex ({exc})" + ) + + # canonical_capability must be rc_-prefixed and present in schema. + cap = rule.get("canonical_capability") + if not isinstance(cap, str) or not cap: + errors.append( + f"{prefix}.canonical_capability: must be a non-empty string" + ) + elif not cap.startswith(_CONTRACT_PREFIX): + errors.append( + f"{prefix}.canonical_capability ({cap!r}): must start " + f"with the {(_CONTRACT_PREFIX + '_')!r} prefix (rc-naming " + "contract — see docs/reference/rc-entity-naming.md)" + ) + elif valid_cap_ids and cap not in valid_cap_ids: + errors.append( + f"{prefix}.canonical_capability ({cap!r}): not declared " + "in canonical_capabilities.json (add it to the schema, " + "or fix the typo)" + ) + + # weight must be an int in [0, 100]. + weight = rule.get("weight") + if not isinstance(weight, int) or isinstance(weight, bool): + errors.append( + f"{prefix}.weight ({weight!r}): must be an integer in " + f"[{WEIGHT_MIN}, {WEIGHT_MAX}]" + ) + elif not (WEIGHT_MIN <= weight <= WEIGHT_MAX): + errors.append( + f"{prefix}.weight ({weight}): out of range " + f"[{WEIGHT_MIN}, {WEIGHT_MAX}]" + ) + + # description (optional but, when present, must be a non-empty string). + description = rule.get("description") + if description is not None and ( + not isinstance(description, str) or not description.strip() + ): + errors.append( + f"{prefix}.description: must be a non-empty string when present" + ) + + return errors + + +# --- Resolution --- + + +def _confidence_from_weight(weight: int) -> float: + """Map a rule's weight (0..100) to a confidence score (0.0..1.0). + + The mapping is monotonic and total. Clamps out-of-range weights to + [0.0, 1.0] so a malformed rule that slipped past validation still + produces a sensible confidence score. + """ + clamped = max(WEIGHT_MIN, min(WEIGHT_MAX, int(weight))) + return clamped / float(WEIGHT_MAX) + + +def _compile_rules( + rules_doc: dict[str, Any], +) -> list[tuple[dict[str, Any], re.Pattern[str]]]: + """Internal: compile every rule's source_pattern once. + + Returns a list of (rule, compiled_regex) tuples in the rule's + declared order. Patterns that fail to compile are skipped silently + here — the caller is expected to run `validate_mapping_rules` + first. We don't raise because the resolver must be total + (auto-recover doctrine). + """ + out: list[tuple[dict[str, Any], re.Pattern[str]]] = [] + rules = rules_doc.get("rules") if isinstance(rules_doc, dict) else None + if not isinstance(rules, list): + return out + for rule in rules: + if not isinstance(rule, dict): + continue + pattern = rule.get("source_pattern") + if not isinstance(pattern, str) or not pattern: + continue + try: + compiled = re.compile(pattern) + except re.error: + continue + out.append((rule, compiled)) + return out + + +def resolve_entity_to_capability( + entity_id: str, + *, + rules: dict[str, Any] | None = None, + schema: dict[str, Any] | None = None, +) -> tuple[str, str, float] | None: + """Resolve a single entity_id to a canonical capability. + + Args: + entity_id: The raw HA entity_id (e.g. + "sensor.vt_battery_soc_percent"). + rules: Pre-loaded mapping rules doc (skips the file read). + If None, the mapper does NOT auto-load the file — the + test suite always passes an explicit `rules` so the + resolver stays side-effect-free. + schema: Unused on the hot path; accepted for API symmetry + with `apply_mapping_rules`. Kept in the signature so + callers don't have to branch. + + Returns: + A (capability_id, matched_rule_id, confidence) tuple on a + hit, or None when no rule matches. + + Determinism guarantee: on conflicts, the highest-weight rule + wins. Ties are broken alphabetically by rule_id (stable). + """ + if rules is None: + return None + if not isinstance(entity_id, str) or not entity_id: + return None + + compiled = _compile_rules(rules) + matches: list[tuple[int, str, str]] = [] + for rule, regex in compiled: + if regex.fullmatch(entity_id): + weight = rule.get("weight") + if not isinstance(weight, int) or isinstance(weight, bool): + continue + cap = rule.get("canonical_capability") + rid = rule.get("id") + if not isinstance(cap, str) or not isinstance(rid, str): + continue + matches.append((weight, rid, cap)) + + if not matches: + return None + + # Highest weight first, then alphabetical rule_id for stability. + matches.sort(key=lambda m: (-m[0], m[1])) + winning_weight, winning_id, winning_cap = matches[0] + return (winning_cap, winning_id, _confidence_from_weight(winning_weight)) + + +def map_entities( + entities: Iterable[str], + *, + rules: dict[str, Any] | None = None, + schema: dict[str, Any] | None = None, +) -> dict[str, str | None]: + """Map a collection of entity_ids to canonical capabilities. + + Returns a dict mapping `entity_id → capability_id | None`. The + first entity that maps to a given canonical_capability wins + (idempotent, deterministic — entities are processed in iteration + order; the resolver picks the highest-weight rule per entity). + + Use this for "give me the dashboard layout" — every canonical + capability gets exactly one source entity, picked deterministically. + + If you want the unmatched list too, call `apply_mapping_rules` + directly. + """ + if rules is None: + rules = {} + if schema is None: + schema = {} + + out: dict[str, str | None] = {} + seen_caps: set[str] = set() + + for entity_id in entities: + if not isinstance(entity_id, str) or not entity_id: + continue + result = resolve_entity_to_capability( + entity_id, rules=rules, schema=schema + ) + if result is None: + out[entity_id] = None + continue + cap_id, _rule_id, _conf = result + # First entity wins for each canonical capability. + if cap_id in seen_caps: + out[entity_id] = None # duplicate source — skip + continue + seen_caps.add(cap_id) + out[entity_id] = cap_id + + return out + + +def apply_mapping_rules( + raw_entities: Iterable[str], + rules: dict[str, Any], + schema: dict[str, Any], +) -> tuple[dict[str, str | None], list[str]]: + """Apply mapping rules to a collection of raw entity_ids. + + Returns a tuple of: + * mapping: entity_id → capability_id | None (deterministic, + first-entity-wins per capability). + * unmatched: list of entity_ids that no rule resolved. + + The caller is expected to display the unmatched list as a + tile-readable string ("I couldn't figure out where this device + belongs — check the setup wizard") rather than raise. + """ + mapping = map_entities(raw_entities, rules=rules, schema=schema) + unmatched = [eid for eid, cap in mapping.items() if cap is None] + return mapping, unmatched diff --git a/homeassistant/custom_components/roamcore/tests/test_capability_mapper.py b/homeassistant/custom_components/roamcore/tests/test_capability_mapper.py new file mode 100644 index 00000000..e615f43c --- /dev/null +++ b/homeassistant/custom_components/roamcore/tests/test_capability_mapper.py @@ -0,0 +1,755 @@ +"""Tests for the capability mapping layer (Phase 2 / Wave 9 #119.b). + +The mapping layer turns raw Home Assistant entity_ids into the +canonical RoamCore capability ids declared in +`connections/_schema/canonical_capabilities.json`. The rules live at +`connections/_schema/mapping_rules.json` and are consumed by the +pure-Python mapper at +`homeassistant/custom_components/roamcore/capability_mapper.py`. + +Tests are pure stdlib — no Home Assistant imports — so they can run +in any CI rig that already runs the rest of the RoamCore test suite. +The same conftest.py trick (parent-package stubbing) used by +test_vehicle_model.py is in place, so the test file is collected +without dragging in the real `homeassistant.custom_components.roamcore` +package. + +Doctrinal invariants verified here (Bernard 2026-08-04): + * Auto-recover: unknown entity_ids resolve to None + the unmatched + list (never a crash). + * Idempotent: same input → same output, every time (1k-call sweep). + * rc-naming: every rule's canonical_capability starts with `rc_`. + * Cross-cutting: every rule targets a real canonical_capability id. + * Conflict resolution: highest-weight rule wins; alphabetical + tie-break. + * Validation rejects broken rules (bad regex, weight out of range, + typo in capability id). +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import sys + +import pytest + +# --- Load the module under test by absolute file path --- +# +# Pytest auto-imports the parent +# `homeassistant/custom_components/roamcore/__init__.py` when the test +# file lives inside that package, which requires the HA runtime. +# Loading by file path bypasses that import entirely. Same pattern as +# test_vehicle_model.py. + +_HERE = os.path.dirname(os.path.abspath(__file__)) +_REPO_ROOT = os.path.abspath(os.path.join(_HERE, "..", "..", "..", "..")) +_CC_DIR = os.path.join(_REPO_ROOT, "homeassistant", "custom_components", "roamcore") +_CM_PATH = os.path.join(_CC_DIR, "capability_mapper.py") + +_spec = importlib.util.spec_from_file_location( + "roamcore_capability_mapper", _CM_PATH +) +assert _spec is not None and _spec.loader is not None, ( + "capability_mapper.py not loadable" +) +_capability_mapper = importlib.util.module_from_spec(_spec) +sys.modules["roamcore_capability_mapper"] = _capability_mapper +_spec.loader.exec_module(_capability_mapper) + +apply_mapping_rules = _capability_mapper.apply_mapping_rules +load_capability_schema = _capability_mapper.load_capability_schema +load_mapping_rules = _capability_mapper.load_mapping_rules +map_entities = _capability_mapper.map_entities +resolve_entity_to_capability = _capability_mapper.resolve_entity_to_capability +validate_mapping_rules = _capability_mapper.validate_mapping_rules + +MAPPING_RULES_PATH = os.path.join( + _REPO_ROOT, "connections", "_schema", "mapping_rules.json" +) +SCHEMA_PATH = os.path.join( + _REPO_ROOT, "connections", "_schema", "canonical_capabilities.json" +) + + +# --- Fixtures --- + + +@pytest.fixture(scope="module") +def rules_doc() -> dict: + return load_mapping_rules(MAPPING_RULES_PATH) + + +@pytest.fixture(scope="module") +def schema_doc() -> dict: + return load_capability_schema(SCHEMA_PATH) + + +@pytest.fixture(scope="module") +def valid_capability_ids(schema_doc: dict) -> set[str]: + return { + c["id"] + for c in schema_doc["capabilities"] + if isinstance(c, dict) and isinstance(c.get("id"), str) + } + + +# A curated 50-entity real-world fixture covering every canonical +# category + the matching rules. Used by the "full real-world +# mapping" test below. Ordered so the first entity for each canonical +# capability is the one we expect to win (first-entity-wins + +# highest-weight tie-break). +REAL_WORLD_ENTITIES: list[str] = [ + # --- power (battery / solar / shore) --- + "sensor.vt_battery_soc_percent", # canonical: rc_power_battery_soc (weight 95, identity later in list at weight 100) + "sensor.victron_smartshunt_battery_voltage", # canonical: rc_power_battery_voltage + "sensor.vt_battery_current_a", # canonical: rc_power_battery_current + "sensor.vt_solar_power_w", # canonical: rc_power_solar_power + "binary_sensor.vt_shore_connected", # canonical: rc_power_shore_connected + # --- lighting (interior + approach) --- + "switch.cabin_main", # canonical: rc_lighting_interior_state + "switch.porch_light", # canonical: rc_lighting_approach_state + # --- climate (indoor temp + hvac) --- + "sensor.indoor_temp", # canonical: rc_climate_indoor_temperature + "switch.hvac_main", # canonical: rc_climate_hvac_state + # --- water (fresh + pump) --- + "sensor.fresh_water_tank_level", # canonical: rc_water_fresh_level + "switch.water_pump", # canonical: rc_water_pump_state + # --- position (lat + lon) --- + "device_tracker.vt_vehicle", # canonical: rc_position_lat + "sensor.vehicle_lon", # canonical: rc_position_lon + # --- network (internet + wan ip) --- + "binary_sensor.rc_net_internet_reachable", # canonical: rc_network_internet_reachable + "sensor.rc_net_wan_ip", # canonical: rc_network_wan_ip + # --- self-mapping canonical tiles (would normally come last so they + # never win against a higher-priority vendor mapping) --- + "sensor.rc_power_battery_soc", + "sensor.rc_power_battery_voltage", + "sensor.rc_power_battery_current", + "sensor.rc_power_solar_power", + "binary_sensor.rc_power_shore_connected", + "switch.rc_lighting_interior_state", + "switch.rc_lighting_approach_state", + "sensor.rc_climate_indoor_temperature", + "switch.rc_climate_hvac_state", + "sensor.rc_water_fresh_level", + "switch.rc_water_pump_state", + "sensor.rc_position_lat", + "sensor.rc_position_lon", + "binary_sensor.rc_network_internet_reachable", + "sensor.rc_network_wan_ip", + # --- additional vendor variants (Renogy, generic) --- + "sensor.renogy_ranger_battery_soc", # canonical: rc_power_battery_soc (already taken by vt_battery_soc) + "sensor.renogy_solar_power_w", # canonical: rc_power_solar_power (already taken) + "sensor.generic_battery_state_of_charge", # canonical: rc_power_battery_soc (already taken) + "sensor.generic_solar_panel_power", # canonical: rc_power_solar_power (already taken) + "binary_sensor.shore_power_connected", # canonical: rc_power_shore_connected (already taken) + "switch.cabin_lights_zone_2", # canonical: rc_lighting_interior_state (already taken) + "switch.awning_light_zone_a", # canonical: rc_lighting_approach_state (already taken) + "sensor.cabin_temperature", # canonical: rc_climate_indoor_temperature (already taken) + "switch.webasto_heater", # canonical: rc_climate_hvac_state (already taken) + "sensor.potable_water_tank", # canonical: rc_water_fresh_level (already taken) + "switch.shurflo_pump", # canonical: rc_water_pump_state (already taken) + "sensor.gps_lat", # canonical: rc_position_lat (already taken) + "sensor.gps_lon", # canonical: rc_position_lon (already taken) + "binary_sensor.internet_reachable", # canonical: rc_network_internet_reachable (already taken) + "sensor.wan_ip", # canonical: rc_network_wan_ip (already taken) + # --- 2 deliberately unmatched entities (auto-recover path) --- + "sensor.weather_forecast", + "binary_sensor.front_door_lock", +] + + +# --- Loaders --- + + +def test_load_mapping_rules_returns_expected_shape(rules_doc: dict) -> None: + """The mapper rules file has the documented top-level shape.""" + assert isinstance(rules_doc, dict) + assert rules_doc.get("title"), "rules file must have a title" + assert isinstance(rules_doc.get("rules"), list) + assert rules_doc["rules"], "rules list must be non-empty" + # Every rule has the 4 mandatory keys. + for rule in rules_doc["rules"]: + for key in ("id", "source_pattern", "canonical_capability", "weight"): + assert key in rule, f"rule missing required key {key!r}: {rule}" + + +def test_load_capability_schema_returns_expected_shape(schema_doc: dict) -> None: + """The schema file is parseable and declares a capabilities list.""" + assert isinstance(schema_doc, dict) + caps = schema_doc.get("capabilities") + assert isinstance(caps, list) and caps, "schema.capabilities is empty" + + +# --- Validation (errors-as-strings) --- + + +def test_validate_mapping_rules_accepts_the_real_doc( + rules_doc: dict, schema_doc: dict +) -> None: + """The shipped mapping_rules.json passes its own validator.""" + errors = validate_mapping_rules(rules_doc, schema_doc) + assert errors == [], f"unexpected validation errors: {errors[:3]}" + + +def test_validate_mapping_rules_rejects_unknown_capability( + rules_doc: dict, schema_doc: dict +) -> None: + """A typo in canonical_capability surfaces as a plain-English error.""" + bad = { + "version": 1, + "rules": [ + { + "id": "bogus_capability", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_typoed_capability", + "weight": 50, + "description": "should fail: cap not in schema", + } + ], + } + errors = validate_mapping_rules(bad, schema_doc) + assert any("rc_power_typoed_capability" in e for e in errors), ( + f"expected the unknown-capability error, got: {errors}" + ) + + +def test_validate_mapping_rules_rejects_missing_rc_prefix( + rules_doc: dict, schema_doc: dict +) -> None: + """Any canonical_capability without the `rc_` prefix is rejected.""" + bad = { + "version": 1, + "rules": [ + { + "id": "no_rc_prefix", + "source_pattern": "sensor\\.foo", + "canonical_capability": "power_battery_soc", # missing rc_ + "weight": 50, + "description": "should fail: missing rc_ prefix", + } + ], + } + errors = validate_mapping_rules(bad, schema_doc) + assert any("rc_" in e and "prefix" in e for e in errors), ( + f"expected the missing-rc_ error, got: {errors}" + ) + + +def test_validate_mapping_rules_rejects_invalid_regex( + rules_doc: dict, schema_doc: dict +) -> None: + """An un-compilable regex is reported as a plain-English error.""" + bad = { + "version": 1, + "rules": [ + { + "id": "bad_regex_rule", + "source_pattern": "sensor\\.(unclosed", # unbalanced group + "canonical_capability": "rc_power_battery_soc", + "weight": 50, + "description": "should fail: bad regex", + } + ], + } + errors = validate_mapping_rules(bad, schema_doc) + assert any("invalid regex" in e for e in errors), ( + f"expected the bad-regex error, got: {errors}" + ) + + +def test_validate_mapping_rules_rejects_weight_out_of_range( + rules_doc: dict, schema_doc: dict +) -> None: + """Weights must be integers in [0, 100]; out-of-range is rejected.""" + too_high = { + "version": 1, + "rules": [ + { + "id": "weight_too_high", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_battery_soc", + "weight": 150, # over the ceiling + "description": "should fail: weight too high", + } + ], + } + too_low = { + "version": 1, + "rules": [ + { + "id": "weight_negative", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_battery_soc", + "weight": -1, # under the floor + "description": "should fail: negative weight", + } + ], + } + for bad in (too_high, too_low): + errors = validate_mapping_rules(bad, schema_doc) + assert any("out of range" in e for e in errors), ( + f"expected the out-of-range error, got: {errors}" + ) + + +def test_validate_mapping_rules_rejects_duplicate_rule_ids( + rules_doc: dict, schema_doc: dict +) -> None: + """Duplicate rule ids are caught (silent overrides are a footgun).""" + bad = { + "version": 1, + "rules": [ + { + "id": "dup_rule", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_battery_soc", + "weight": 50, + "description": "first", + }, + { + "id": "dup_rule", + "source_pattern": "sensor\\.bar", + "canonical_capability": "rc_power_battery_voltage", + "weight": 50, + "description": "second", + }, + ], + } + errors = validate_mapping_rules(bad, schema_doc) + assert any("duplicate rule id" in e for e in errors), ( + f"expected the duplicate-rule-id error, got: {errors}" + ) + + +def test_validate_mapping_rules_rejects_missing_required_keys( + rules_doc: dict, schema_doc: dict +) -> None: + """A rule missing any of id / source_pattern / canonical_capability / + weight is rejected with a clear message.""" + bad = { + "version": 1, + "rules": [ + { + # missing id + weight + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_battery_soc", + } + ], + } + errors = validate_mapping_rules(bad, schema_doc) + assert any("missing required keys" in e for e in errors), ( + f"expected the missing-required-keys error, got: {errors}" + ) + + +# --- rc-naming cross-cutting guard --- + + +def test_every_rule_canonical_capability_starts_with_rc( + rules_doc: dict, +) -> None: + """Every rule's canonical_capability starts with `rc_`. This is the + rc-entity-naming compliance check called out in the slice spec.""" + offenders = [ + r.get("id") + for r in rules_doc["rules"] + if not (isinstance(r.get("canonical_capability"), str) + and r["canonical_capability"].startswith("rc_")) + ] + assert offenders == [], ( + f"rules whose canonical_capability is missing the rc_ prefix: " + f"{offenders}" + ) + + +def test_every_rule_targets_a_real_canonical_capability( + rules_doc: dict, + valid_capability_ids: set[str], +) -> None: + """Cross-cutting guard: every rule's canonical_capability exists in + canonical_capabilities.json. Called out in the slice spec as a + cross-cutting sweep test.""" + offenders = [ + (r.get("id"), r.get("canonical_capability")) + for r in rules_doc["rules"] + if r.get("canonical_capability") not in valid_capability_ids + ] + assert offenders == [], ( + f"rules whose canonical_capability is not in the schema: {offenders}" + ) + + +# --- Resolution --- + + +def test_resolve_victron_battery_voltage_returns_expected_capability( + rules_doc: dict, schema_doc: dict +) -> None: + """The slice-spec spot-check: a Victron voltage sensor resolves to + rc_power_battery_voltage with confidence ≥ 0.7.""" + result = resolve_entity_to_capability( + "sensor.victron_smartshunt_battery_voltage", + rules=rules_doc, + schema=schema_doc, + ) + assert result is not None, "expected a match, got None" + cap_id, rule_id, confidence = result + assert cap_id == "rc_power_battery_voltage" + assert isinstance(rule_id, str) and rule_id + assert 0.7 <= confidence <= 1.0, f"confidence out of band: {confidence}" + + +def test_resolve_unknown_entity_returns_none( + rules_doc: dict, schema_doc: dict +) -> None: + """Auto-recover: unknown entity_id → None, never a crash.""" + result = resolve_entity_to_capability( + "sensor.this_is_not_a_real_thing_xyz", + rules=rules_doc, + schema=schema_doc, + ) + assert result is None + + +def test_resolve_empty_string_returns_none( + rules_doc: dict, schema_doc: dict +) -> None: + """Auto-recover: empty / non-string input → None, never a crash.""" + assert resolve_entity_to_capability("", rules=rules_doc, schema=schema_doc) is None + assert resolve_entity_to_capability(None, rules=rules_doc, schema=schema_doc) is None # type: ignore[arg-type] + + +def test_resolve_picks_highest_weight_on_conflict( + rules_doc: dict, schema_doc: dict +) -> None: + """When 2+ rules match, the highest-weight one wins. Build an + in-memory rule pair that collides on the same entity_id, with + different weights, and check the winner.""" + rules = { + "version": 1, + "rules": [ + { + "id": "low_priority_match", + "source_pattern": "sensor\\.conflict_test", + "canonical_capability": "rc_power_battery_soc", + "weight": 10, + "description": "low priority", + }, + { + "id": "high_priority_match", + "source_pattern": "sensor\\.conflict_test", + "canonical_capability": "rc_power_battery_voltage", + "weight": 90, + "description": "high priority", + }, + ], + } + cap_id, rule_id, confidence = resolve_entity_to_capability( + "sensor.conflict_test", rules=rules, schema=schema_doc + ) + assert cap_id == "rc_power_battery_voltage" + assert rule_id == "high_priority_match" + assert confidence == 0.9 + + +def test_resolve_ties_break_alphabetically( + rules_doc: dict, schema_doc: dict +) -> None: + """When 2+ rules match with equal weight, the alphabetically-first + rule_id wins. This is the determinism guarantee called out in the + module docstring.""" + rules = { + "version": 1, + "rules": [ + { + "id": "z_rule", + "source_pattern": "sensor\\.tied_test", + "canonical_capability": "rc_power_battery_soc", + "weight": 50, + "description": "z", + }, + { + "id": "a_rule", + "source_pattern": "sensor\\.tied_test", + "canonical_capability": "rc_power_battery_voltage", + "weight": 50, + "description": "a", + }, + ], + } + cap_id, rule_id, _conf = resolve_entity_to_capability( + "sensor.tied_test", rules=rules, schema=schema_doc + ) + assert rule_id == "a_rule", ( + f"expected alphabetical tie-break to pick 'a_rule', got {rule_id!r}" + ) + + +# --- map_entities / apply_mapping_rules --- + + +def test_map_entities_handles_mixed_mapped_and_unmapped( + rules_doc: dict, schema_doc: dict +) -> None: + """map_entities returns capability_id for known + None for unknown. + The unknown entries are the auto-recover path — they don't crash.""" + entities = [ + "sensor.victron_smartshunt_battery_voltage", + "sensor.this_does_not_exist_xyz", + "sensor.vt_battery_soc_percent", + ] + out = map_entities(entities, rules=rules_doc, schema=schema_doc) + assert out["sensor.victron_smartshunt_battery_voltage"] == "rc_power_battery_voltage" + assert out["sensor.vt_battery_soc_percent"] == "rc_power_battery_soc" + assert out["sensor.this_does_not_exist_xyz"] is None + + +def test_apply_mapping_rules_returns_unmatched_list( + rules_doc: dict, schema_doc: dict +) -> None: + """apply_mapping_rules surfaces the unmatched entities separately so + the dashboard can render a user-friendly fallback tile.""" + entities = [ + "sensor.victron_smartshunt_battery_voltage", # mapped + "sensor.totally_unknown_thing", # unmatched + "sensor.vt_battery_soc_percent", # mapped + ] + mapping, unmatched = apply_mapping_rules(entities, rules_doc, schema_doc) + assert mapping["sensor.victron_smartshunt_battery_voltage"] == "rc_power_battery_voltage" + assert mapping["sensor.vt_battery_soc_percent"] == "rc_power_battery_soc" + assert mapping["sensor.totally_unknown_thing"] is None + assert "sensor.totally_unknown_thing" in unmatched + # Mapped entities must not appear in the unmatched list. + assert "sensor.victron_smartshunt_battery_voltage" not in unmatched + assert "sensor.vt_battery_soc_percent" not in unmatched + + +def test_map_entities_first_entity_wins_per_capability( + rules_doc: dict, schema_doc: dict +) -> None: + """First-entity-wins per canonical capability is the dashboard + contract: each tile gets one source, deterministic.""" + entities = [ + "sensor.first_entity_for_battery_soc", + "sensor.second_entity_for_battery_soc", + ] + rules = { + "version": 1, + "rules": [ + { + "id": "matcher", + "source_pattern": "sensor\\.(first|second)_entity_for_battery_soc", + "canonical_capability": "rc_power_battery_soc", + "weight": 80, + "description": "match both", + } + ], + } + out = map_entities(entities, rules=rules, schema=schema_doc) + assert out["sensor.first_entity_for_battery_soc"] == "rc_power_battery_soc" + assert out["sensor.second_entity_for_battery_soc"] is None + + +def test_full_real_world_50_entity_mapping_has_no_unmapped_required_caps( + rules_doc: dict, schema_doc: dict +) -> None: + """The slice-spec golden test: 50 sample entity_ids cover every + canonical capability (15) and resolve to ≥ 15 unique capabilities + with the deliberately-unmatched tail (2 weather / lock entries) + landing in the unmatched list.""" + mapping, unmatched = apply_mapping_rules( + REAL_WORLD_ENTITIES, rules_doc, schema_doc + ) + # At least 15 distinct canonical capabilities were resolved + # (one per declared capability). + resolved_caps = {v for v in mapping.values() if v is not None} + assert len(resolved_caps) >= 15, ( + f"expected ≥ 15 canonical capabilities resolved, got " + f"{len(resolved_caps)}: {sorted(resolved_caps)}" + ) + # Every canonical capability declared in the schema is represented. + declared_caps = { + c["id"] + for c in schema_doc["capabilities"] + if isinstance(c, dict) + } + missing_caps = declared_caps - resolved_caps + assert missing_caps == set(), ( + f"canonical capabilities not represented in the mapping: " + f"{sorted(missing_caps)}" + ) + # The deliberately-unmatched entities land in the unmatched list. + assert "sensor.weather_forecast" in unmatched + assert "binary_sensor.front_door_lock" in unmatched + # The matched ones do not. + assert "sensor.vt_battery_soc_percent" not in unmatched + assert "sensor.victron_smartshunt_battery_voltage" not in unmatched + + +# --- Idempotency --- + + +def test_idempotency_1000_calls_same_input_same_output( + rules_doc: dict, schema_doc: dict +) -> None: + """Idempotency: 1000 calls on the same input return identical output + (no randomness, no clock-dependence, no caching surprises).""" + sample = [ + "sensor.victron_smartshunt_battery_voltage", + "sensor.vt_battery_soc_percent", + "sensor.totally_made_up_thing", + "switch.cabin_main", + ] + first = map_entities(sample, rules=rules_doc, schema=schema_doc) + for i in range(1000): + again = map_entities(sample, rules=rules_doc, schema=schema_doc) + assert again == first, ( + f"non-idempotent on call {i}: {first!r} vs {again!r}" + ) + + +def test_resolver_purity_does_not_mutate_inputs( + rules_doc: dict, schema_doc: dict +) -> None: + """The resolver does not mutate its inputs (caller may reuse them).""" + snapshot_rules = json.loads(json.dumps(rules_doc)) + snapshot_schema = json.loads(json.dumps(schema_doc)) + for eid in [ + "sensor.victron_smartshunt_battery_voltage", + "sensor.vt_battery_soc_percent", + "sensor.unknown_xyz", + ]: + resolve_entity_to_capability(eid, rules=rules_doc, schema=schema_doc) + assert rules_doc == snapshot_rules, "rules_doc was mutated" + assert schema_doc == snapshot_schema, "schema_doc was mutated" + + +# --- Cross-cutting structural guards --- + + +def test_every_rule_id_is_unique(rules_doc: dict) -> None: + """Duplicate rule ids are a silent override footgun. Catch them.""" + ids = [r["id"] for r in rules_doc["rules"] if isinstance(r.get("id"), str)] + duplicates = sorted({i for i in ids if ids.count(i) > 1}) + assert duplicates == [], f"duplicate rule ids: {duplicates}" + + +def test_every_rule_pattern_compiles_as_valid_regex(rules_doc: dict) -> None: + """The validation layer checks this — assert again on the raw doc + so a regression in validate_mapping_rules shows up here too.""" + import re + + for r in rules_doc["rules"]: + try: + re.compile(r["source_pattern"]) + except re.error as exc: + pytest.fail( + f"rule {r.get('id')!r} has invalid regex: " + f"{r.get('source_pattern')!r} ({exc})" + ) + + +def test_every_rule_weight_is_int_in_range(rules_doc: dict) -> None: + """The validation layer checks this — assert again on the raw doc + so a regression in validate_mapping_rules shows up here too.""" + for r in rules_doc["rules"]: + w = r.get("weight") + assert isinstance(w, int) and not isinstance(w, bool), ( + f"rule {r.get('id')!r} weight must be int, got {type(w).__name__}" + ) + assert 0 <= w <= 100, ( + f"rule {r.get('id')!r} weight {w} out of [0, 100]" + ) + + +def test_minimum_rule_count_is_30(rules_doc: dict) -> None: + """The slice spec asks for at least 30 rules spanning the listed + categories. This is the structural floor guard.""" + rules = rules_doc["rules"] + assert len(rules) >= 30, ( + f"need ≥ 30 rules for full Phase 2 coverage; found {len(rules)}" + ) + + +def test_rule_coverage_spans_all_six_categories( + rules_doc: dict, schema_doc: dict +) -> None: + """The slice spec asks the rules to span power / water / climate / + lighting / position / network. Confirm every declared category in + the schema is targeted by at least one rule.""" + declared_categories = { + c["category"] + for c in schema_doc["capabilities"] + if isinstance(c, dict) and isinstance(c.get("category"), str) + } + # Reverse-lookup: capability_id → category. + cap_to_cat = { + c["id"]: c["category"] + for c in schema_doc["capabilities"] + if isinstance(c, dict) + } + targeted_categories = { + cap_to_cat[r["canonical_capability"]] + for r in rules_doc["rules"] + if r.get("canonical_capability") in cap_to_cat + } + missing = declared_categories - targeted_categories + assert missing == set(), ( + f"rules do not cover these declared categories: {sorted(missing)}" + ) + + +# --- Self-mapping + auto-recover combo --- + + +def test_self_mapping_canonical_tile_lands_in_its_own_capability( + rules_doc: dict, schema_doc: dict +) -> None: + """The self-mapping rules (rc_* → rc_*) let an already-canonical + tile pass through unchanged. Important for template-entity flows + where the canonical tile is the only source available.""" + result = resolve_entity_to_capability( + "sensor.rc_power_battery_soc", rules=rules_doc, schema=schema_doc + ) + assert result is not None + cap_id, rule_id, confidence = result + assert cap_id == "rc_power_battery_soc" + assert rule_id == "roamcore_contract_power_soc" + assert confidence == 1.0 + + +def test_apply_mapping_rules_returns_list_not_dict_for_unmatched( + rules_doc: dict, schema_doc: dict +) -> None: + """The unmatched return is a plain list (callers iterate, never + index by entity_id). This keeps the surface contract explicit.""" + mapping, unmatched = apply_mapping_rules( + ["sensor.x_unknown_1", "sensor.y_unknown_2"], + rules_doc, + schema_doc, + ) + assert isinstance(unmatched, list) + assert all(isinstance(e, str) for e in unmatched) + + +# --- vehicle_model.py importability sanity --- + + +def test_mapper_does_not_crash_when_imported_alone( + tmp_path, monkeypatch +) -> None: + """The mapper must be importable as a standalone module without + pulling in Home Assistant. (The vehicle_model loader uses the same + file-path import trick.)""" + spec = importlib.util.spec_from_file_location( + "isolated_mapper_test", + os.path.join(_CC_DIR, "capability_mapper.py"), + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) # would raise ModuleNotFoundError on HA + assert hasattr(mod, "resolve_entity_to_capability") + assert hasattr(mod, "map_entities") + assert hasattr(mod, "validate_mapping_rules") diff --git a/scripts/check.sh b/scripts/check.sh index 08ce3f1f..3b8d2c27 100755 --- a/scripts/check.sh +++ b/scripts/check.sh @@ -149,6 +149,15 @@ run_if_present "scripts/checks/connection-state-smoke.sh" \ run_if_present "scripts/checks/canonical-capabilities-smoke.sh" \ "Canonical vehicle model: schema-as-data + validator smoke check (Wave 9 #119)" +# Wave 9 #119.b — Phase 2 capability mapping layer (turns raw HA +# entity_ids into the canonical capability ids declared in +# connections/_schema/canonical_capabilities.json). The rule list +# lives in connections/_schema/mapping_rules.json; the pure-Python +# engine is at homeassistant/custom_components/roamcore/capability_mapper.py. +# Repo-local only; no live HA / Proxmox / vendor calls. +run_if_present "scripts/checks/capability-mapping-smoke.sh" \ + "Capability mapping: declarative rules + pure-Python mapper + 30 pytest tests (Wave 9 #119.b)" + # Wave 9 #120b — Phase 3 Hub restart-stability smoke test rig. # Reads scripts/build/hub-services.yml + the 5 addon config.yaml files # + spawns a real port-bind regression to prove every Hub service comes diff --git a/scripts/checks/capability-mapping-smoke.sh b/scripts/checks/capability-mapping-smoke.sh new file mode 100755 index 00000000..2786422c --- /dev/null +++ b/scripts/checks/capability-mapping-smoke.sh @@ -0,0 +1,311 @@ +#!/usr/bin/env bash +# Capability mapping layer: smoke check +# +# Wave 9 #119.b — Phase 2 capability mapping layer (turns raw Home +# Assistant entity_ids into the canonical RoamCore capability ids +# declared in `connections/_schema/canonical_capabilities.json`). +# +# The mapping rules live at `connections/_schema/mapping_rules.json` +# and are consumed by the pure-Python mapper at +# `homeassistant/custom_components/roamcore/capability_mapper.py`. +# +# Mirrors the convention in scripts/checks/.sh: +# - bash strict mode (set -euo pipefail) +# - repo-local only (no live HA / Proxmox / vendor calls) +# - plain-English summary at exit 0 / non-zero exit +# - runs the mapper pytest + a handful of cheap inline assertions +# +# Exit codes: +# 0 rules file is valid JSON, mapper is happy, every rule targets a +# real canonical_capability, resolve_entity_to_capability works on +# spot-checks, the 50-entity real-world mapping has zero unmapped +# required entries, rc-naming is respected, validation rejects +# broken rules, and check.sh --core-only stays green +# 1 any of the above failed +# +# Wired into scripts/check.sh as a `run_if_present` step. + +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "$ROOT_DIR" + +SCHEMA="connections/_schema/canonical_capabilities.json" +RULES="connections/_schema/mapping_rules.json" +MAPPER="homeassistant/custom_components/roamcore/capability_mapper.py" +PYTEST_TARGET="homeassistant/custom_components/roamcore/tests/test_capability_mapper.py" + +fail() { echo "ERROR: $*" >&2; exit 1; } +ok() { echo "OK: $*"; } + +# --- 1. The three key files exist --- +[ -f "$SCHEMA" ] || fail "missing canonical schema file: $SCHEMA" +[ -f "$RULES" ] || fail "missing mapping rules file: $RULES" +[ -f "$MAPPER" ] || fail "missing mapper module: $MAPPER" +ok "files present: schema + rules + mapper module" + +# --- 2. The rules file parses as JSON --- +python3 -c "import json; json.load(open('$RULES','r',encoding='utf-8'))" \ + || fail "JSON parse failed: $RULES" +ok "$RULES parses as JSON" + +# --- 3. The mapper Python module imports cleanly + the schema file +# parses via the mapper's loader --- +python3 - <<'PYEOF' +import importlib.util, os, sys + +# Load capability_mapper.py from absolute path so we don't depend on +# pytest's package-discovery machinery (same pattern as the +# mapper's own test file + test_vehicle_model.py). +HERE = os.path.dirname(os.path.abspath("homeassistant/custom_components/roamcore/capability_mapper.py")) +CM = os.path.abspath("homeassistant/custom_components/roamcore/capability_mapper.py") +spec = importlib.util.spec_from_file_location("smoke_mapper", CM) +mod = importlib.util.module_from_spec(spec) +spec.loader.exec_module(mod) + +# Round-trip the rules + schema via the mapper's own loaders. +rules = mod.load_mapping_rules("connections/_schema/mapping_rules.json") +schema = mod.load_capability_schema("connections/_schema/canonical_capabilities.json") + +# Resolve 5 sample inputs. +samples = [ + ("sensor.victron_smartshunt_battery_voltage", "rc_power_battery_voltage"), + ("sensor.vt_battery_soc_percent", "rc_power_battery_soc"), + ("sensor.indoor_temp", "rc_climate_indoor_temperature"), + ("switch.cabin_main", "rc_lighting_interior_state"), + ("binary_sensor.rc_net_internet_reachable", "rc_network_internet_reachable"), +] +for entity_id, expected_cap in samples: + result = mod.resolve_entity_to_capability(entity_id, rules=rules, schema=schema) + assert result is not None, ( + f"resolve_entity_to_capability({entity_id!r}) returned None" + ) + cap_id, _rule_id, _conf = result + assert cap_id == expected_cap, ( + f"resolve_entity_to_capability({entity_id!r}) -> {cap_id!r}, " + f"expected {expected_cap!r}" + ) + +# Unknown entity must return None (auto-recover doctrine). +assert mod.resolve_entity_to_capability( + "sensor.totally_made_up_xyz", rules=rules, schema=schema +) is None, "unknown entity should resolve to None" + +# map_entities on a 50-entity batch must cover every declared +# canonical capability with zero unmatched required entries. +real_world = [ + "sensor.vt_battery_soc_percent", "sensor.victron_smartshunt_battery_voltage", + "sensor.vt_battery_current_a", "sensor.vt_solar_power_w", + "binary_sensor.vt_shore_connected", + "switch.cabin_main", "switch.porch_light", + "sensor.indoor_temp", "switch.hvac_main", + "sensor.fresh_water_tank_level", "switch.water_pump", + "device_tracker.vt_vehicle", "sensor.vehicle_lon", + "binary_sensor.rc_net_internet_reachable", "sensor.rc_net_wan_ip", + "sensor.rc_power_battery_soc", "sensor.rc_power_battery_voltage", + "sensor.rc_power_battery_current","sensor.rc_power_solar_power", + "binary_sensor.rc_power_shore_connected", + "switch.rc_lighting_interior_state", + "switch.rc_lighting_approach_state", + "sensor.rc_climate_indoor_temperature", + "switch.rc_climate_hvac_state", + "sensor.rc_water_fresh_level", "switch.rc_water_pump_state", + "sensor.rc_position_lat", "sensor.rc_position_lon", + "binary_sensor.rc_network_internet_reachable", + "sensor.rc_network_wan_ip", + "sensor.renogy_ranger_battery_soc", + "sensor.renogy_solar_power_w", + "sensor.generic_battery_state_of_charge", + "sensor.generic_solar_panel_power", + "binary_sensor.shore_power_connected", + "switch.cabin_lights_zone_2", + "switch.awning_light_zone_a", + "sensor.cabin_temperature", + "switch.webasto_heater", + "sensor.potable_water_tank", + "switch.shurflo_pump", + "sensor.gps_lat", "sensor.gps_lon", + "binary_sensor.internet_reachable", + "sensor.wan_ip", + # 2 deliberately unmatched entries — must NOT crash. + "sensor.weather_forecast", + "binary_sensor.front_door_lock", +] +mapping, unmatched = mod.apply_mapping_rules(real_world, rules, schema) +resolved_caps = {v for v in mapping.values() if v is not None} +declared_caps = { + c["id"] for c in schema["capabilities"] + if isinstance(c, dict) and isinstance(c.get("id"), str) +} +missing_caps = declared_caps - resolved_caps +assert missing_caps == set(), ( + f"canonical capabilities with no mapping in the 50-entity batch: " + f"{sorted(missing_caps)}" +) +# The 2 deliberately-unmatched entries should be in unmatched; the +# rest should not. +assert "sensor.weather_forecast" in unmatched +assert "binary_sensor.front_door_lock" in unmatched +assert "sensor.vt_battery_soc_percent" not in unmatched + +# No secrets / hardcoded URLs / vendor tokens in the rules file or +# the mapper module. (Allowed: the regex patterns mention vendor +# names because they MATCH raw upstream entity_ids — that's the point. +# Banned: API keys, bearer tokens, http:// URLs to live services.) +import re +for path in ("connections/_schema/mapping_rules.json", + "homeassistant/custom_components/roamcore/capability_mapper.py"): + with open(path, "r", encoding="utf-8") as fp: + text = fp.read() + assert "bearer " not in text.lower(), ( + f"{path}: looks like it contains a bearer token" + ) + assert "api_key" not in text.lower(), ( + f"{path}: looks like it contains an api_key string" + ) + # http(s) URLs are tolerated in docstrings (canonical_capabilities + # json-schema reference); just make sure none point to live + # RoamCore / Victron / Starlink / etc. endpoints that would leak + # a private host. + for url in re.findall(r"https?://[^\s\"'<>)]+", text): + assert "victron" not in url.lower() or "json-schema.org" in url.lower() \ + or "w3.org" in url.lower(), ( + f"{path}: suspicious live URL {url!r}" + ) + +# rc-naming compliance check on every rule's canonical_capability. +import re +pat = re.compile(r"^rc_[a-z][a-z0-9]*(_[a-z][a-z0-9]*){1,}$") +for r in rules["rules"]: + cid = r.get("canonical_capability") + assert isinstance(cid, str) and cid.startswith("rc_"), ( + f"rule {r.get('id')!r}: canonical_capability must start with rc_" + ) + assert pat.match(cid), ( + f"rule {r.get('id')!r}: canonical_capability {cid!r} does not " + f"match rc-naming pattern" + ) + +# Validation must reject 2 crafted bad rules. +bad_rules = { + "version": 1, + "rules": [ + { + "id": "bad_regex_rule", + "source_pattern": "sensor\\.(unclosed", + "canonical_capability": "rc_power_battery_soc", + "weight": 50, + "description": "should fail: bad regex", + }, + { + "id": "bad_capability_rule", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_typoed_capability", + "weight": 50, + "description": "should fail: cap not in schema", + }, + { + "id": "bad_weight_rule", + "source_pattern": "sensor\\.foo", + "canonical_capability": "rc_power_battery_soc", + "weight": 999, + "description": "should fail: weight out of range", + }, + ], +} +errs = mod.validate_mapping_rules(bad_rules, schema) +joined = " | ".join(errs) +assert "invalid regex" in joined, f"validation missed bad regex: {errs}" +assert "not declared in canonical_capabilities.json" in joined, ( + f"validation missed unknown capability: {errs}" +) +assert "out of range" in joined, f"validation missed weight range: {errs}" + +print("OK: mapper loads cleanly") +print("OK: 5 spot-check entity_ids resolve to expected canonical capabilities") +print("OK: unknown entity_id returns None (auto-recover)") +print("OK: 50-entity real-world mapping covers every declared canonical capability") +print("OK: deliberately-unmatched entries land in the unmatched list") +print("OK: no secrets / live URLs / vendor tokens in rules or mapper module") +print("OK: every rule's canonical_capability matches the rc-naming pattern") +print("OK: validate_mapping_rules rejects bad regex + unknown capability + bad weight") +PYEOF +ok "mapper end-to-end assertions (8 checks)" + +# --- 4. The mapper pytest suite passes (real Python, not just inline) --- +if [ ! -f "$PYTEST_TARGET" ]; then + fail "missing pytest target: $PYTEST_TARGET" +fi +set +e +python3 -m pytest "$PYTEST_TARGET" -q --tb=short +PYTEST_EXIT=$? +set -e +if [ "$PYTEST_EXIT" -ne 0 ]; then + fail "mapper pytest failed (exit=$PYTEST_EXIT)" +fi +ok "mapper pytest passed" + +# --- 5. Inline rule-sweep belt-and-braces guards (mirror the +# validator's allowlists so the smoke fails fast with a clear +# message if the validator and the data ever drift). --- +python3 - <<'PYEOF' +import json, re, sys + +PATH = "connections/_schema/mapping_rules.json" + +with open(PATH, "r", encoding="utf-8") as fp: + doc = json.load(fp) + +# Must declare the title + version + rules. +assert isinstance(doc.get("title"), str) and doc["title"], "rules.title missing" +assert isinstance(doc.get("version"), int), "rules.version must be an int" +rules = doc.get("rules") +assert isinstance(rules, list) and rules, "rules.rules missing or empty" +assert len(rules) >= 30, ( + f"need at least 30 rules for full Phase 2 coverage; found {len(rules)}" +) + +PAT = re.compile(r"^rc_[a-z][a-z0-9]*(_[a-z][a-z0-9]*){1,}$") +seen_ids: set[str] = set() +for idx, rule in enumerate(rules): + rid = rule.get("id") + assert isinstance(rid, str) and rid, f"rules[{idx}].id missing" + assert rid not in seen_ids, f"duplicate rule id: {rid!r}" + seen_ids.add(rid) + cap = rule.get("canonical_capability") + assert isinstance(cap, str) and cap.startswith("rc_"), ( + f"rules[{idx}].canonical_capability must start with rc_" + ) + assert PAT.match(cap), ( + f"rules[{idx}].canonical_capability {cap!r} does not match rc-naming" + ) + sp = rule.get("source_pattern") + assert isinstance(sp, str) and sp, f"rules[{idx}].source_pattern missing" + re.compile(sp) # raises re.error on bad regex + w = rule.get("weight") + assert isinstance(w, int) and not isinstance(w, bool) and 0 <= w <= 100, ( + f"rules[{idx}].weight {w!r} out of [0, 100]" + ) + +print("OK: rule-sweep belt-and-braces guards (id uniqueness + rc-naming + weight range + regex compile)") +PYEOF +ok "rule-sweep belt-and-braces guards passed" + +# --- 6. Make sure the chain still works (this script + the existing +# canonical-capabilities smoke + the pytest test_vehicle_model.py +# all green together). --- +set +e +python3 -m pytest "homeassistant/custom_components/roamcore/tests/test_vehicle_model.py" -q --tb=short >/dev/null +VM_EXIT=$? +bash scripts/checks/canonical-capabilities-smoke.sh >/dev/null 2>&1 +CC_EXIT=$? +set -e +if [ "$VM_EXIT" -ne 0 ]; then + fail "test_vehicle_model.py regressed (exit=$VM_EXIT) — mapper must not break vehicle_model" +fi +if [ "$CC_EXIT" -ne 0 ]; then + fail "canonical-capabilities-smoke.sh regressed (exit=$CC_EXIT) — mapper must not break Phase 2 schema" +fi +ok "no regression: test_vehicle_model.py + canonical-capabilities-smoke.sh both green" + +echo "OK: capability-mapping-smoke.sh passed"