diff --git a/CHANGELOG.md b/CHANGELOG.md index ccda8fb..5f4d3b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Cross-device parameter QA** (`param_check.py`, new MCP tool `check_device_parameters`). + Reads per-device `ParameterInstanceRef` values straight from the `.knxproj` project part + (data xknxproject does not expose), groups identical devices by application program, and + flags the odd one out: `clear_outliers` (a strong majority with a small minority — e.g. one + thermostat with a different setpoint/hysteresis, one presence detector with a different + detection time) and `split_configs` (balanced 2+ variants — review, often two zones). + Parameter names resolved from the app-program. Read-only; no ETS/bus; encrypted projects are + skipped honestly. Validated on real 42–275-device projects; synthetic `tests/test_param_check.py`. + Community-driven (asked for in Discussions). Tool count 25 → 26. + - **`skills/ha-git-backup`** — an ops-companion skill for the engineer package: a two-circuit Home Assistant backup system (real git in `/config` with a deploy key and a pre-commit secret scanner + age-encrypted full backups in GitHub Releases), with install/sync/scan/offsite/restore diff --git a/README.md b/README.md index d38f904..e1fb304 100644 --- a/README.md +++ b/README.md @@ -250,7 +250,7 @@ keyring handling, and the recommended workflow). --- -## MCP tools (25) +## MCP tools (26) **Read** | Tool | Purpose | @@ -279,6 +279,7 @@ keyring handling, and the recommended workflow). | `decompose_device(order_number, channels?)` | device → GA decomposition: **exact vendor model** from a local catalog (`NICKOL_KNX_CATALOG`), or generic recipe | | `list_device_recipes()` | the built-in device library (Zennio + ABB families) | | `parse_devices_from_project(path, output_path?, password?)` | extract **exact device object models** from the app-programs inside a `.knxproj`/`.knxprod` → device-library YAML (feeds the local catalog) | +| `check_device_parameters(path, password?, min_group?)` | **cross-device parameter QA**: find the device whose ETS parameters differ from its N identical siblings (the odd thermostat/sensor out) — `clear_outliers` (likely mistake) + `split_configs` (balanced variants, review) | | `grade_completeness()` | grade a project: bare skeleton vs as-built | | `diff_projects(path_a, path_b, …)` | semantic diff between two `.knxproj` versions | diff --git a/nickol_knx_mcp/param_check.py b/nickol_knx_mcp/param_check.py new file mode 100644 index 0000000..ab6eb52 --- /dev/null +++ b/nickol_knx_mcp/param_check.py @@ -0,0 +1,180 @@ +"""Cross-device parameter consistency — find the device whose ETS parameter +settings differ from its N identical siblings (the odd thermostat/sensor out). + +Reads per-device ``ParameterInstanceRef`` values straight from the ``.knxproj`` +project part (``P-*/0.xml``) — data that xknxproject does not expose — groups +devices by their application program (identical devices share the same +``Hardware2ProgramRefId``), and reports two kinds of finding: + + * ``clear_outlier`` — a strong majority value with a small minority + (likely a configuration mistake: 19 thermostats at 0.5 K, one at 1.0 K); + * ``split_config`` — the group splits into 2+ balanced variants (probably + two zones/roles — surfaced for review, NOT an error). + +Parameter RefIds are resolved to human names from the device application +program (``ParameterRef`` → ``Parameter.Name``); module-definition parameters +that don't resolve keep a readable fallback and are counted honestly. + +Read-only; no ETS, no bus. A password-protected ``.knxproj`` is encrypted and +cannot be read. +""" +from __future__ import annotations + +import re +import zipfile +import xml.etree.ElementTree as ET +from collections import Counter, defaultdict +from typing import Any, Optional + +_NUMERIC_RE = re.compile(r"^-?\d+$") + + +def _localname(tag: str) -> str: + return tag.rsplit("}", 1)[-1] + + +def _app_of(refid: str) -> Optional[str]: + m = re.match(r"(M-[0-9A-Za-z]+)_(A-[0-9A-Za-z-]+?)_", refid) + return f"{m.group(1)}_{m.group(2)}" if m else None + + +def _resolve_names(zf: zipfile.ZipFile, refids: set[str]) -> dict[str, str]: + """RefId -> human parameter name, best-effort from the application program.""" + by_app: dict[str, set[str]] = defaultdict(set) + for r in refids: + app = _app_of(r) + if app: + by_app[app].add(r) + out: dict[str, str] = {} + for app, rids in by_app.items(): + cand = [n for n in zf.namelist() if n.endswith(f"{app}.xml")] + if not cand: + continue + try: + axml = ET.fromstring(zf.read(cand[0])) + except Exception: + continue + pref: dict[str, str] = {} + pname: dict[str, str] = {} + for e in axml.iter(): + ln = _localname(e.tag) + if ln == "ParameterRef": + pref[e.get("Id", "")] = e.get("RefId", "") + elif ln == "Parameter": + pname[e.get("Id", "")] = e.get("Name") or e.get("Text") or "" + for r in rids: + nm = pname.get(pref.get(r, ""), "") + if nm: + out[r] = nm + return out + + +def check_device_parameters(path: str, password: Optional[str] = None, + min_group: int = 3, majority: float = 0.70, + minority_frac: float = 0.25, + max_findings: int = 40) -> dict[str, Any]: + """Find parameter outliers across identical devices in a `.knxproj`. + + Returns clear outliers (a device whose value differs from its siblings) and + balanced split-configs (review), with resolved parameter names. + """ + try: + zf = zipfile.ZipFile(path) + except Exception as e: # noqa: BLE001 + return {"error": f"cannot open .knxproj: {e}"} + + proj0 = [n for n in zf.namelist() if re.match(r"P-[^/]+/0\.xml$", n)] + if not proj0: + return {"error": "no P-*/0.xml (project part) found in archive"} + try: + root = ET.fromstring(zf.read(proj0[0])) + except ET.ParseError: + return {"error": "project part is not plain XML — the .knxproj is likely " + "password-protected/encrypted; parameters cannot be read."} + + # devices -> group by app-program, collect {param_refid: value} + devices: list[dict[str, Any]] = [] + for el in root.iter(): + if _localname(el.tag) != "DeviceInstance": + continue + hp = el.get("Hardware2ProgramRefId") + if not hp: + continue + params = {s.get("RefId"): s.get("Value") + for s in el.iter() if _localname(s.tag) == "ParameterInstanceRef" + and s.get("RefId") and s.get("Value") is not None} + devices.append({ + "group": hp, + "product": (el.get("ProductRefId") or "").split("_P-")[-1] or hp, + "addr": el.get("Address") or "?", + "name": el.get("Name") or "", + "params": params, + }) + + groups: dict[str, list[dict[str, Any]]] = defaultdict(list) + for d in devices: + groups[d["group"]].append(d) + + clear: list[dict[str, Any]] = [] + splits: list[dict[str, Any]] = [] + refids_needed: set[str] = set() + + for gkey, devs in groups.items(): + if len(devs) < min_group: + continue + all_rids: set[str] = set().union(*(set(d["params"]) for d in devs)) if devs else set() + for rid in all_rids: + present = [(d, d["params"][rid]) for d in devs if rid in d["params"]] + if len(present) < min_group: + continue + counter = Counter(v for _, v in present) + if len(counter) == 1: + continue + (maj_val, maj_n), = counter.most_common(1) + minority = [(d, v) for d, v in present if v != maj_val] + product = present[0][0]["product"] + numeric = _NUMERIC_RE.match(maj_val) is not None + rec = { + "group_product": product, "refid": rid, "numeric": numeric, + "total": len(present), "majority_value": maj_val, + } + if maj_n / len(present) >= majority and len(minority) <= max(1, int(len(present) * minority_frac)): + refids_needed.add(rid) + rec["odd_devices"] = [{"address": d["addr"], "name": d["name"], "value": v} + for d, v in minority] + clear.append(rec) + elif len(counter) >= 2: + refids_needed.add(rid) + rec["variants"] = [{"value": v, "count": c} for v, c in counter.most_common()] + splits.append(rec) + + names = _resolve_names(zf, refids_needed) + + def _decorate(rec: dict[str, Any]) -> dict[str, Any]: + rec["parameter"] = names.get(rec["refid"], rec["refid"]) + rec["name_resolved"] = rec["refid"] in names + return rec + + # numeric config outliers first (time/setpoint/hysteresis-like), then the rest + clear = [_decorate(r) for r in clear] + splits = [_decorate(r) for r in splits] + clear.sort(key=lambda r: (not r["numeric"], len(r["odd_devices"]), -r["total"])) + splits.sort(key=lambda r: (not r["numeric"], -r["total"])) + + grp_sizes = sorted((len(v) for v in groups.values() if len(v) >= min_group), reverse=True) + return { + "devices": len(devices), + "identical_device_groups": len(grp_sizes), + "largest_groups": grp_sizes[:8], + "clear_outliers_count": len(clear), + "split_configs_count": len(splits), + "clear_outliers": clear[:max_findings], + "split_configs": splits[:max_findings], + "names_unresolved": sum(1 for r in (clear + splits) if not r["name_resolved"]), + "note": "clear_outliers = a device whose value differs from its N identical " + "siblings (likely a mistake). split_configs = the group splits into " + "balanced variants (review — often two zones/roles, not an error). " + "Numeric config parameters (times/setpoints/hysteresis) are listed first. " + "Read-only; parameter values come from P-*/0.xml (xknxproject does not " + "expose them). Some module-definition parameter names may stay as RefIds.", + } diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index 26bcd4d..d87d928 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -30,6 +30,7 @@ from .advanced import (matter_readiness, completeness_grade, energy_scaffold, test_protocol, suggest_naming) from .diffproj import diff_projects as _diff_projects from .iot import generate_knx_iot_turtle +from .param_check import check_device_parameters as _check_device_parameters mcp = FastMCP("nickol-knx") @@ -345,6 +346,23 @@ def parse_devices_from_project(path: str, output_path: Optional[str] = None, return out +@mcp.tool() +def check_device_parameters(path: str, password: Optional[str] = None, + min_group: int = 3) -> dict[str, Any]: + """Find the device whose ETS **parameter** settings differ from its N identical + siblings — the odd thermostat/sensor out (e.g. one thermostat with a different + setpoint/hysteresis, one presence detector with a different detection time). + + Reads per-device parameter values straight from the `.knxproj` project part + (data xknxproject does not expose), groups identical devices by application + program, and returns `clear_outliers` (a strong majority with a small minority — + likely a mistake) and `split_configs` (balanced 2+ variants — review, often two + zones). Numeric config parameters are listed first; names are resolved from the + device application program. Read-only, no ETS/bus. Give a real `.knxproj` `path` + (a password-protected/encrypted project cannot be read).""" + return _check_device_parameters(path, password=password, min_group=min_group) + + @mcp.tool() def check_matter() -> dict[str, Any]: """Matter-readiness lint: which controllable functions round-trip to a Matter diff --git a/tests/test_param_check.py b/tests/test_param_check.py new file mode 100644 index 0000000..d87142c --- /dev/null +++ b/tests/test_param_check.py @@ -0,0 +1,94 @@ +"""Synthetic test for cross-device parameter outlier detection (check_device_parameters). + +Builds an in-memory .knxproj (a ZIP with a P-*/0.xml project part and a tiny +application program) containing 5 identical devices where one has an odd value, +one parameter that is balanced-split, and one that is uniform — then asserts the +tool flags the outlier, resolves its name, classifies the split, and stays quiet +on the uniform one. +""" +import io +import zipfile + +from nickol_knx_mcp.param_check import check_device_parameters + +HP = "M-TEST_H-X-1_HP-1111" + + +def _device(addr, hyst, mode, split): + pirs = "".join([ + f'', + f'', + f'', + ]) + return (f'' + f'{pirs}') + + +def _make_knxproj() -> str: + # 5 identical devices: hysteresis 5,5,5,5,10 (dev .10 is the outlier); + # mode uniform (2); split 1,1,1,2,2 (balanced 3/2) + devs = [ + _device(1, 5, 2, 1), _device(2, 5, 2, 1), _device(3, 5, 2, 1), + _device(4, 5, 2, 2), _device(10, 10, 2, 2), + ] + proj0 = ('' + '' + '' + + "".join(devs) + + '') + app = ('' + '' + '' + '' + '' + '' + '' + '' + '' + '' + '') + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as z: + z.writestr("P-TEST/0.xml", proj0) + z.writestr("M-TEST/M-TEST_A-1111.xml", app) + path = "/tmp/_paramcheck_test.knxproj" + with open(path, "wb") as f: + f.write(buf.getvalue()) + return path + + +def main(): + path = _make_knxproj() + r = check_device_parameters(path, min_group=3) + assert "error" not in r, r + assert r["devices"] == 5, r["devices"] + assert r["identical_device_groups"] == 1, r + + # clear outlier: hysteresis 4x5 vs 1x10, resolved name, odd device addr 10 + hits = [o for o in r["clear_outliers"] if o["refid"].endswith("R-1")] + assert hits, f"hysteresis outlier not found: {r['clear_outliers']}" + h = hits[0] + assert h["parameter"] == "Hysteresis (K)", h["parameter"] + assert h["name_resolved"] is True + assert h["numeric"] is True + assert h["majority_value"] == "5" + assert [d["value"] for d in h["odd_devices"]] == ["10"], h["odd_devices"] + assert h["odd_devices"][0]["address"] == "10" + + # uniform param (mode, all "2") must NOT appear anywhere + assert not any(o["refid"].endswith("R-2") for o in r["clear_outliers"] + r["split_configs"]) + + # balanced split (1,1,1,2,2) -> split_configs, name resolved + splits = [o for o in r["split_configs"] if o["refid"].endswith("R-3")] + assert splits, f"split not found: {r['split_configs']}" + assert splits[0]["parameter"] == "Zone role" + variants = {v["value"]: v["count"] for v in splits[0]["variants"]} + assert variants == {"1": 3, "2": 2}, variants + + print("test_param_check: OK — outlier flagged (Hysteresis 4x5 vs 1x10 @ addr 10), " + "uniform param quiet, balanced split classified, names resolved.") + + +if __name__ == "__main__": + main()