mirror of
https://github.com/NickoScope/nickol-knx-mcp.git
synced 2026-09-30 03:41:58 +02:00
feat: check_device_parameters — cross-device parameter outlier QA (25->26 tools)
New module param_check.py + MCP tool check_device_parameters: reads per-device ParameterInstanceRef values from the .knxproj project part (xknxproject does not expose them), 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) and split_configs (balanced variants → review). Parameter names resolved from the app-program; encrypted projects skipped honestly. Read-only, no ETS/bus. Validated on real 42-275-device projects; synthetic test_param_check.py. Answers a community feature request. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Nikolay1
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
546d71ec2b
commit
4c36bc182e
@@ -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
|
||||
|
||||
@@ -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 |
|
||||
|
||||
|
||||
@@ -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.",
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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'<ParameterInstanceRef RefId="M-TEST_A-1111_P-1_R-1" Value="{hyst}" />',
|
||||
f'<ParameterInstanceRef RefId="M-TEST_A-1111_P-2_R-2" Value="{mode}" />',
|
||||
f'<ParameterInstanceRef RefId="M-TEST_A-1111_P-3_R-3" Value="{split}" />',
|
||||
])
|
||||
return (f'<DeviceInstance Id="P-TEST-0_DI-{addr}" Address="{addr}" Name="Th{addr}" '
|
||||
f'Hardware2ProgramRefId="{HP}" ProductRefId="M-TEST_H-X-1_P-TVALVE">'
|
||||
f'{pirs}</DeviceInstance>')
|
||||
|
||||
|
||||
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 = ('<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<KNX xmlns="http://knx.org/xml/project/20"><Project><Installations>'
|
||||
'<Installation><Topology><Area><Line>'
|
||||
+ "".join(devs) +
|
||||
'</Line></Area></Topology></Installation></Installations></Project></KNX>')
|
||||
app = ('<?xml version="1.0" encoding="utf-8"?><KNX><ManufacturerData><Manufacturer>'
|
||||
'<ApplicationPrograms><ApplicationProgram Id="M-TEST_A-1111">'
|
||||
'<ParameterRefs>'
|
||||
'<ParameterRef Id="M-TEST_A-1111_P-1_R-1" RefId="M-TEST_A-1111_P-1" />'
|
||||
'<ParameterRef Id="M-TEST_A-1111_P-3_R-3" RefId="M-TEST_A-1111_P-3" />'
|
||||
'</ParameterRefs>'
|
||||
'<Parameters>'
|
||||
'<Parameter Id="M-TEST_A-1111_P-1" Name="Hysteresis (K)" />'
|
||||
'<Parameter Id="M-TEST_A-1111_P-3" Name="Zone role" />'
|
||||
'</Parameters>'
|
||||
'</ApplicationProgram></ApplicationPrograms></Manufacturer></ManufacturerData></KNX>')
|
||||
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()
|
||||
Reference in New Issue
Block a user