mirror of
https://github.com/NickoScope/nickol-knx-mcp.git
synced 2026-09-29 19:31:12 +02:00
Nikolay1andClaude Opus 4.8
1e924ede6f
Rank findings by significance: config_value (setpoint/hysteresis/time/threshold — an odd device is usually a real mistake) vs config_flag (mode/type) vs label (per-room text, often intentional). New focus list surfaces the config-value clear outliers first. On a real 275-device project this turns 422 raw outliers into a 9-item focus that includes the thermostats whose init-setpoint differs from their siblings. Multilingual name hints (EN/DE/RU). Test updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
215 lines
9.3 KiB
Python
215 lines
9.3 KiB
Python
"""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+$")
|
|
|
|
# Parameter-name hints for a meaningful numeric CONFIG VALUE — the kind of setting
|
|
# where an odd value is usually a real mistake (a different hysteresis / setpoint /
|
|
# timeout on one device). Multilingual (EN / DE / RU). Everything numeric but not a
|
|
# value is a "config_flag" (mode/type/enable); non-numeric is a "label" (often a
|
|
# deliberate per-room name, lowest priority).
|
|
_VALUE_HINTS = (
|
|
"setpoint", "hyster", "temperatur", "threshold", "delay", "duration", "timeout",
|
|
"time", "period", "level", "speed", "offset", "value", "brightness", "scaling",
|
|
"уставк", "гистерез", "температур", "порог", "задержк", "время", "период",
|
|
"длительн", "уровень", "скорост", "яркост", "значени",
|
|
)
|
|
|
|
|
|
def _significance(name: str, numeric: bool) -> str:
|
|
if not numeric:
|
|
return "label"
|
|
low = (name or "").lower()
|
|
return "config_value" if any(h in low for h in _VALUE_HINTS) else "config_flag"
|
|
|
|
|
|
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)
|
|
|
|
_SIG_RANK = {"config_value": 0, "config_flag": 1, "label": 2}
|
|
|
|
def _decorate(rec: dict[str, Any]) -> dict[str, Any]:
|
|
rec["parameter"] = names.get(rec["refid"], rec["refid"])
|
|
rec["name_resolved"] = rec["refid"] in names
|
|
rec["significance"] = _significance(rec["parameter"], rec["numeric"])
|
|
return rec
|
|
|
|
# most meaningful first: config_value (setpoint/hysteresis/time…) > flag > label
|
|
clear = [_decorate(r) for r in clear]
|
|
splits = [_decorate(r) for r in splits]
|
|
clear.sort(key=lambda r: (_SIG_RANK[r["significance"]], len(r["odd_devices"]), -r["total"]))
|
|
splits.sort(key=lambda r: (_SIG_RANK[r["significance"]], -r["total"]))
|
|
|
|
def _by_sig(items: list[dict[str, Any]]) -> dict[str, int]:
|
|
c = Counter(r["significance"] for r in items)
|
|
return {k: c.get(k, 0) for k in ("config_value", "config_flag", "label")}
|
|
|
|
# the short list an integrator should actually look at first
|
|
focus = [r for r in clear if r["significance"] == "config_value"]
|
|
|
|
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),
|
|
"focus_count": len(focus),
|
|
"focus": focus[:max_findings],
|
|
"clear_outliers_by_significance": _by_sig(clear),
|
|
"split_configs_by_significance": _by_sig(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": "START WITH `focus` — clear outliers on a config VALUE parameter "
|
|
"(setpoint/hysteresis/time/threshold…), where an odd device is usually a "
|
|
"real mistake. `clear_outliers` also includes config_flag (mode/type) and "
|
|
"label (per-room text, often intentional) — see *_by_significance. "
|
|
"split_configs = the group splits into balanced variants (review — often two "
|
|
"zones, not an error). Read-only; values come from P-*/0.xml (xknxproject does "
|
|
"not expose them). Some module-definition parameter names may stay as RefIds.",
|
|
}
|