diff --git a/CHANGELOG.md b/CHANGELOG.md index 5af75a1..a537ec0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.6.0] — 2026-07-01 + +**Completes the roadmap** — the tool now validates, repairs, generates (HA / ETS / handover / IoT), +diffs, grades and drafts acceptance protocols, all design-time & read-only. + +### Added — B-tier + +- **B2 — climate-correctness review** (`generate_ha.py`). Climate generation now emits an explicit + review note: controller/operation modes made explicit, setpoint-shift command **and** state paired, + and a flag raised for any mode object that lacks a corresponding state address. +- **B3 — semantic project diff** (`diffproj.py`, new module; new MCP tools `diff_projects` / + `diff_loaded`). Compares two `.knxproj` versions and reports added / removed / DPT-changed / + renamed / security-changed group addresses — a reviewable delta between as-designed revisions. +- **B4 — acceptance test protocol** (`advanced.py`; new MCP tool `generate_test_protocol`). + Produces a per-function acceptance checklist in Markdown for commissioning sign-off. +- **B5 — Matter readiness** (`advanced.py`; new MCP tool `check_matter`). Reports which functions + round-trip cleanly to a Matter cluster and which do not. +- **B6 — energy scaffold** (`advanced.py`; new MCP tool `check_energy`). Checks metering / energy + DPT coverage and scaffolds PV / battery / EVSE structure. + +### Added — C-tier + +- **C1 — KNX IoT semantic export** (`iot.py`, new module; new MCP tool `generate_knx_iot`). + Emits a KNX IoT Turtle / RDF skeleton for the project. +- **C2 — naming suggestions** (`advanced.py`; new MCP tool `suggest_names`). Naming-hygiene + proposals for group addresses that drift from the zone + function convention. +- **C3 — as-built completeness grader** (`advanced.py`; new MCP tool `grade_completeness`). + Grades a project from bare functional skeleton to as-built, by presence of professional patterns + (central macros, motion tuning, astro/meteo, monitoring, deep metering, scenes, reserves, debug). + +### Added — A-tier quick wins + +- **A5 — Areas / voice note** (`generate_ha.py`). Documents that HA Areas and voice assignment are + UI-only concerns, not derivable from the `.knxproj`. +- **A6 — time/date expose** (`generate_ha.py`). Emits a KNX `expose` block for DPT-19.001 clock + broadcast so HA can serve time/date to the bus. + ## [0.5.0] — 2026-07-01 **From validator to repairer.** Previous releases *flagged* problems in a `.knxproj`; this one diff --git a/nickol_knx_mcp/__init__.py b/nickol_knx_mcp/__init__.py index b8cfa3c..8e96509 100644 --- a/nickol_knx_mcp/__init__.py +++ b/nickol_knx_mcp/__init__.py @@ -1,2 +1,2 @@ """nickol-knx-mcp: design-time KNX/ETS project assistant MCP server.""" -__version__ = "0.5.0" +__version__ = "0.6.0" diff --git a/nickol_knx_mcp/advanced.py b/nickol_knx_mcp/advanced.py new file mode 100644 index 0000000..7ce0bb2 --- /dev/null +++ b/nickol_knx_mcp/advanced.py @@ -0,0 +1,195 @@ +"""Advanced analysis + generators (roadmap tier B/C). + +All read-only, design-time. Each function takes a LoadedProject and returns plain +data / Markdown for review — nothing here touches ETS or a bus. + + B4 test_protocol - per-function acceptance checklist (command -> expected status) + B5 matter_readiness - which functions map cleanly to a Matter cluster, what's missing + B6 energy_scaffold - metering/energy domain check + a suggested structure + C2 suggest_naming - propose zone+function names / normalise status tokens + C3 completeness_grade - grade a project: bare skeleton vs as-built +""" + +from __future__ import annotations + +from collections import Counter +from typing import Any + +from .project import LoadedProject +from .pairing import find_status, base_tokens +from .intent import INTENT_FUNCTIONAL + + +def _functional_commands(project: LoadedProject) -> list: + return [g for g in project.gas.values() + if g.intent == INTENT_FUNCTIONAL and g.kind == "command" + and g.dpt_main is not None] + + +def _status_gas(project: LoadedProject) -> list: + from .analyze import _is_status_ga + return [g for g in project.gas.values() if _is_status_ga(g)] + + +# --------------------------------------------------------------------------- # +# B5 — Matter readiness +# --------------------------------------------------------------------------- # +_MATTER = { + "lighting": ("OnOff / LevelControl", ("on/off", "brightness+status")), + "shutter": ("WindowCovering", ("up/down", "position+status")), + "hvac": ("Thermostat", ("setpoint", "mode", "status")), + "climate": ("Thermostat", ("setpoint", "mode", "status")), + "sensor": ("sensor cluster", ("value",)), + "energy": ("ElectricalMeasurement", ("value",)), +} + + +def matter_readiness(project: LoadedProject) -> dict[str, Any]: + """Which controllable functions round-trip to a Matter cluster, and what's missing.""" + stats = _status_gas(project) + ready, not_ready = [], [] + for ga in _functional_commands(project): + cluster = _MATTER.get(ga.category) + if cluster is None: + continue + has_status = find_status(ga, [s for s in stats if s.main == ga.main]) is not None \ + or find_status(ga, stats) is not None + row = {"address": ga.address, "name": ga.name, "category": ga.category, + "matter": cluster[0], "has_status": has_status} + (ready if has_status else not_ready).append(row) + total = len(ready) + len(not_ready) + return { + "controllable_functions": total, + "matter_ready": len(ready), + "ready_pct": (100 * len(ready) // total) if total else 0, + "not_ready": not_ready[:200], + "note": "Ready = has a status GA + decodable DPT, so the Matter cluster can report " + "state. A Matter bridge (e.g. HA Matter server) exposes these; functions " + "without a status GA won't round-trip. Static readiness only — no bridging here.", + } + + +# --------------------------------------------------------------------------- # +# C3 — as-built completeness grader +# --------------------------------------------------------------------------- # +_PATTERNS = { + "central macros": ("общее", "все ", "всё", "central", "all "), + "motion tuning": ("порог освещ", "блокировк", "время, сек", "тест-режим", "чувствит"), + "astro / meteo": ("солнц", "азимут", "восход", "закат", "метео", "sun ", "meteo"), + "monitoring": ("мониторинг", "статус работы", "неисправ", "monitoring", "fault"), + "deep metering": ("счётчик", "потреблен", "тариф", "meter", "consumption"), + "scenes": ("сцена", "сценар", "scene", "szene"), + "reserves": ("резерв", "reserve", "spare"), + "debug main": ("отладк", "временно", "debug", "logic for"), +} + + +def completeness_grade(project: LoadedProject) -> dict[str, Any]: + """Grade a project: bare functional skeleton vs as-built grade (§ alex-skill patterns).""" + names = " \n ".join(g.name.lower() for g in project.gas.values()) + present, missing = {}, [] + for label, toks in _PATTERNS.items(): + n = sum(names.count(t) for t in toks) + present[label] = n + if n == 0: + missing.append(label) + hit = sum(1 for v in present.values() if v) + score = round(100 * hit / len(_PATTERNS)) + grade = ("as-built grade" if score >= 75 else + "near-complete" if score >= 55 else + "functional skeleton" if score >= 30 else "bare skeleton") + return { + "grade": grade, "score": score, + "patterns_present": {k: v for k, v in present.items() if v}, + "patterns_missing": missing, + "note": "Completeness = presence of the as-built patterns a professional adds beyond " + "the bare functional set (central macros, device tuning, astro/meteo, " + "monitoring, deep metering, scenes, reserves, a debug main). Not a " + "correctness score — a project can be correct yet a bare skeleton.", + } + + +# --------------------------------------------------------------------------- # +# B6 — energy-domain check + scaffold +# --------------------------------------------------------------------------- # +_ENERGY_DPT = {13: "13.013 (energy Wh/kWh)", 14: "14.056 (power W)"} + + +def energy_scaffold(project: LoadedProject) -> dict[str, Any]: + """Check the metering/energy domain and suggest a structure.""" + meters = [g for g in project.gas.values() + if any(t in g.name.lower() for t in + ("счётчик", "потреблен", "энерг", "мощност", "power", "energy", "meter", "квтч", "kwh"))] + bad = [{"address": g.address, "name": g.name, "dpt": g.dpt} + for g in meters if g.dpt_main not in (13, 14) and g.dpt_main is not None] + scaffold = [ + {"function": "Per-circuit consumption", "dpt": "13.013", "role": "status"}, + {"function": "Per-circuit power", "dpt": "14.056", "role": "status"}, + {"function": "PV production", "dpt": "13.013", "role": "status"}, + {"function": "Battery state of charge", "dpt": "5.001", "role": "status"}, + {"function": "EVSE charging power", "dpt": "14.056", "role": "status"}, + {"function": "Grid import/export", "dpt": "13.013", "role": "status"}, + ] + return { + "metering_gas": len(meters), + "wrong_dpt": bad, + "suggested_energy_dpts": _ENERGY_DPT, + "scaffold": scaffold, + "note": "Energy GAs should use 13.x (energy) / 14.056 (power) so Home Assistant's " + "energy dashboard can aggregate them. Scaffold is a suggested per-circuit / " + "PV / battery / EVSE structure to add in ETS.", + } + + +# --------------------------------------------------------------------------- # +# B4 — test / functional-acceptance protocol +# --------------------------------------------------------------------------- # +def test_protocol(project: LoadedProject) -> dict[str, Any]: + """Per-function acceptance checklist (command -> expected status) as Markdown rows.""" + stats = _status_gas(project) + rows = [] + for ga in _functional_commands(project): + st = find_status(ga, [s for s in stats if s.main == ga.main]) or find_status(ga, stats) + rows.append({ + "function": ga.name, "command": ga.address, "dpt": ga.dpt, + "status": st.address if st else "—", + "expected": "status reflects command within ~2 s", + }) + md = ["# Functional acceptance protocol\n", + "| Function | Command | DPT | Status GA | Expected | Pass | Sign-off |", + "|---|---|---|---|---|---|---|"] + for r in rows[:2000]: + md.append(f"| {r['function']} | `{r['command']}` | {r['dpt'] or '?'} | " + f"`{r['status']}` | {r['expected']} | ☐ | |") + md.append("\n_Execution is manual/on-site: trigger each command, verify the status GA " + "updates. This tool drafts the checklist; it performs no bus operations._") + return {"functions": len(rows), "markdown": "\n".join(md), "rows": rows[:2000]} + + +# --------------------------------------------------------------------------- # +# C2 — naming suggestions (heuristic; deep AI naming is the caller's job) +# --------------------------------------------------------------------------- # +_STATUS_WORDS = ("статус", "status", "rückmeldung", "rueck", "feedback", "state") + + +def suggest_naming(project: LoadedProject) -> dict[str, Any]: + """Propose name fixes: empty names, and status GAs missing a status keyword.""" + suggestions = [] + for ga in project.gas.values(): + if ga.intent != INTENT_FUNCTIONAL: + continue + low = ga.name.lower() + if not ga.name.strip(): + suggestions.append({"address": ga.address, "issue": "empty_name", + "suggest": "add a zone + function name so pairing/voice work"}) + elif ga.kind == "status" and not any(w in low for w in _STATUS_WORDS): + suggestions.append({"address": ga.address, "name": ga.name, "issue": "status_no_keyword", + "suggest": f"{ga.name} (статус)", + "why": "a status keyword lets the engine pair feedback to its command"}) + return { + "count": len(suggestions), + "suggestions": suggestions[:500], + "note": "Heuristic naming hygiene. Full AI naming (propose zone+function per device, " + "3-level placement) is done by the calling assistant using the project's " + "device/room context — this tool surfaces the mechanical fixes.", + } diff --git a/nickol_knx_mcp/diffproj.py b/nickol_knx_mcp/diffproj.py new file mode 100644 index 0000000..04d8b0c --- /dev/null +++ b/nickol_knx_mcp/diffproj.py @@ -0,0 +1,78 @@ +"""B3 — semantic diff between two .knxproj versions. + +Turns two Git snapshots of a project into a reviewable change set: which group +addresses were added / removed, which changed DPT, name or security flag. No +mainstream tool does this (xknxproject explicitly can't); our Git-tracked, read-only +workflow is its natural home. Read-only, no bus. +""" + +from __future__ import annotations + +from typing import Any, Optional + +from .project import load_project + + +def diff_projects(path_a: str, path_b: str, + password_a: Optional[str] = None, + password_b: Optional[str] = None) -> dict[str, Any]: + """Compare two .knxproj files. `path_a` = old/base, `path_b` = new.""" + a = load_project(path_a, password=password_a) + b = load_project(path_b, password=password_b) + return diff_loaded(a, b) + + +def diff_loaded(a, b) -> dict[str, Any]: + """Diff two already-loaded projects (testable core of diff_projects).""" + ga_a, ga_b = a.gas, b.gas + keys_a, keys_b = set(ga_a), set(ga_b) + + added = [{"address": k, "name": ga_b[k].name, "dpt": ga_b[k].dpt} + for k in sorted(keys_b - keys_a)] + removed = [{"address": k, "name": ga_a[k].name, "dpt": ga_a[k].dpt} + for k in sorted(keys_a - keys_b)] + dpt_changed, renamed, secure_changed = [], [], [] + for k in sorted(keys_a & keys_b): + x, y = ga_a[k], ga_b[k] + if (x.dpt or None) != (y.dpt or None): + dpt_changed.append({"address": k, "name": y.name, + "from": x.dpt, "to": y.dpt}) + if x.name.strip() != y.name.strip(): + renamed.append({"address": k, "from": x.name, "to": y.name}) + if x.data_secure != y.data_secure: + secure_changed.append({"address": k, "name": y.name, + "from": x.data_secure, "to": y.data_secure}) + + def _section(title: str, rows: list, cols) -> list[str]: + out = [f"\n## {title} ({len(rows)})\n"] + if not rows: + out.append("_none_") + return out + out.append("| " + " | ".join(cols) + " |") + out.append("|" + "|".join("---" for _ in cols) + "|") + for r in rows[:500]: + out.append("| " + " | ".join(f"`{r.get(c.lower().replace(' ','_'),'')}`" + if c in ("Address",) else str(r.get(c.lower().replace(' ','_'), '')) + for c in cols) + " |") + if len(rows) > 500: + out.append(f"| … and {len(rows)-500} more |") + return out + + md = [f"# Project diff — {a.info.get('name','A')} → {b.info.get('name','B')}\n", + f"- Base: **{len(ga_a)}** GA · New: **{len(ga_b)}** GA " + f"(+{len(added)} / −{len(removed)})\n"] + md += _section("Added", added, ["Address", "Name", "Dpt"]) + md += _section("Removed", removed, ["Address", "Name", "Dpt"]) + md += _section("DPT changed", dpt_changed, ["Address", "From", "To"]) + md += _section("Renamed", renamed, ["Address", "From", "To"]) + md += _section("Security flag changed", secure_changed, ["Address", "From", "To"]) + + return { + "base_ga": len(ga_a), "new_ga": len(ga_b), + "added": len(added), "removed": len(removed), + "dpt_changed": len(dpt_changed), "renamed": len(renamed), + "secure_changed": len(secure_changed), + "markdown": "\n".join(md), + "detail": {"added": added, "removed": removed, "dpt_changed": dpt_changed, + "renamed": renamed, "secure_changed": secure_changed}, + } diff --git a/nickol_knx_mcp/generate_ha.py b/nickol_knx_mcp/generate_ha.py index 08f15f6..fb14045 100644 --- a/nickol_knx_mcp/generate_ha.py +++ b/nickol_knx_mcp/generate_ha.py @@ -390,6 +390,22 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: climates.append(ent) if zone_loc: built_climate.append(zone_loc) + # B2 climate correctness: mode-command-without-state makes the mode unshowable; + # controller/operation mode lists auto-detect wrong; setpoint-shift needs cmd+state. + issues = [] + if op_cmd and not op_state: + issues.append("operation_mode has a command but no state address") + if ctrl_cmd and not ctrl_state: + issues.append("controller_mode has a command but no state address") + if not tgt_cmd: + issues.append("setpoint is read-only (no target_temperature command)") + note = ("set `controller_modes`/`operation_modes` explicitly — HA auto-detection is " + "often wrong; if this zone uses setpoint-shift, provide BOTH the command and " + "state addresses and set `setpoint_shift_mode`") + if issues: + note += " — " + "; ".join(issues) + review.append({"reason": "verify_climate", "address": ga.address, + "name": ga.name, "note": note}) # ---- 4. SENSORS / BINARY SENSORS (read-only) ---- for ga in project.gas.values(): @@ -441,6 +457,31 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: "so it can map to a Home Assistant cover.") review.append(item) + # A6: time/date broadcast — if the project has a DateTime/time/date GA, expose + # Home Assistant's clock to it so KNX devices get the time from HA. + expose: list[dict] = [] + clock = next((g for g in project.gas.values() + if g.dpt_main == 19 + or (g.dpt_main in (10, 11) and any( + t in g.name.lower() for t in + ("время", "дата", "time", "date", "uhr", "zeit", "clock")))), None) + if clock: + etype = ("datetime" if clock.dpt_main == 19 + else "time" if clock.dpt_main == 10 else "date") + expose.append({"type": etype, "address": clock.address, + "entity_id": "sensor.date_time_iso"}) + review.append({"reason": "verify_expose", "address": clock.address, "name": clock.name, + "note": f"exposes Home Assistant's {etype} to this KNX GA (clock " + "broadcast); point entity_id at a real HA 'Date & Time' helper. " + "`expose` cannot use passive-address lists."}) + + # A5: Areas / voice — surface the UI-only limitation once. + if switches or lights or covers or climates: + review.append({"reason": "areas_ui_only", "address": "-", + "note": "Home Assistant Areas cannot be set in KNX YAML (assign each " + "entity to an Area in the HA UI); and entity `name`s drive voice/Assist " + "matching — keep them descriptive and unique."}) + knx: dict[str, Any] = {} if switches: knx["switch"] = switches @@ -454,6 +495,8 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: knx["binary_sensor"] = binary_sensors if sensors: knx["sensor"] = sensors + if expose: + knx["expose"] = expose package = {"knx": knx} text = ( diff --git a/nickol_knx_mcp/iot.py b/nickol_knx_mcp/iot.py new file mode 100644 index 0000000..e421080 --- /dev/null +++ b/nickol_knx_mcp/iot.py @@ -0,0 +1,59 @@ +"""C1 — KNX IoT semantic export (Turtle / RDF). + +KNX IoT is the IP-native future: an installation is described by a *semantic* model +(functions, datapoints, DPTs) rather than only classic group addresses. This module +emits a Turtle/RDF view of the project's functional group addresses from the same graph +we already build for the GA XML — one source of truth, two exports. Read-only file +generation; no bus. + +The output is a pragmatic semantic skeleton (datapoint per functional GA with its DPT, +name, role and 3-level address). It is not the full certified KNX IoT ontology, and says +so — a starting point for an IoT-native description, reviewed before use. +""" + +from __future__ import annotations + +from .project import LoadedProject +from .intent import INTENT_FUNCTIONAL + + +def _esc(s: str) -> str: + return (s or "").replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ") + + +def generate_knx_iot_turtle(project: LoadedProject) -> str: + """Emit a Turtle/RDF semantic view of the project's functional datapoints.""" + lines = [ + "@prefix rdf: .", + "@prefix rdfs: .", + "@prefix knx: .", + "@prefix dpt: .", + "@prefix proj: .", + "", + "# KNX IoT semantic view — generated read-only from the .knxproj.", + "# Pragmatic skeleton (datapoint per functional group address), NOT the full", + "# certified KNX IoT ontology. Review before use.", + "", + f'proj:project a knx:Installation ;', + f' rdfs:label "{_esc(project.info.get("name","KNX project"))}" ;', + f' knx:groupAddressStyle "{_esc(project.info.get("group_address_style","ThreeLevel"))}" .', + "", + ] + n = 0 + for ga in project.gas.values(): + if ga.intent != INTENT_FUNCTIONAL or ga.dpt_main is None: + continue + n += 1 + node = "proj:ga_" + ga.address.replace("/", "_") + lines += [ + f"{node} a knx:Datapoint ;", + f' knx:groupAddress "{ga.address}" ;', + f' rdfs:label "{_esc(ga.name)}" ;', + f" knx:datapointType dpt:{(ga.dpt or '').replace('.', '-') or 'unknown'} ;", + f' knx:role "{ga.kind}" ;', + f' knx:function "{_esc(ga.category)}" ;', + f" knx:secure {'true' if ga.data_secure else 'false'} .", + "", + ] + lines.insert(len(lines), f"# {n} functional datapoints exported.") + return "\n".join(lines) diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index c0a939a..17b7d98 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -24,6 +24,10 @@ from .report import build_report from .handover import build_handover from .device_library import decompose_device as _decompose_device, list_recipes from .repair import suggest_repairs as _suggest_repairs +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 mcp = FastMCP("nickol-knx") @@ -311,6 +315,67 @@ def list_device_recipes() -> list[dict[str, Any]]: return list_recipes() +@mcp.tool() +def check_matter() -> dict[str, Any]: + """Matter-readiness lint: which controllable functions round-trip to a Matter + cluster (have command + status + a decodable DPT) and which won't.""" + return matter_readiness(_project()) + + +@mcp.tool() +def grade_completeness() -> dict[str, Any]: + """Grade the project: bare functional skeleton vs as-built grade — by the presence + of the professional patterns (central macros, device tuning, astro/meteo, monitoring, + deep metering, scenes, reserves, a debug main).""" + return completeness_grade(_project()) + + +@mcp.tool() +def check_energy() -> dict[str, Any]: + """Check the metering/energy domain (energy DPTs 13.x / 14.056) and suggest a + per-circuit / PV / battery / EVSE structure for the HA energy dashboard.""" + return energy_scaffold(_project()) + + +@mcp.tool() +def suggest_names() -> dict[str, Any]: + """Naming hygiene suggestions (empty names, status GAs missing a status keyword).""" + return suggest_naming(_project()) + + +@mcp.tool() +def generate_test_protocol(output_path: Optional[str] = None) -> dict[str, Any]: + """Draft a functional acceptance protocol (per function: command → expected status, + pass/fail/sign-off) as Markdown. Execution is manual/on-site; this only drafts it.""" + res = test_protocol(_project()) + out: dict[str, Any] = {"functions": res["functions"]} + if output_path: + out["written"] = _safe_write(output_path, res["markdown"]) + else: + out["markdown"] = res["markdown"] + return out + + +@mcp.tool() +def diff_projects(path_a: str, path_b: str, + password_a: Optional[str] = None, + password_b: Optional[str] = None) -> dict[str, Any]: + """Semantic diff between two .knxproj files (path_a = base/old, path_b = new): + added / removed GAs, DPT changes, renames, security-flag changes. Read-only.""" + return _diff_projects(path_a, path_b, password_a=password_a, password_b=password_b) + + +@mcp.tool() +def generate_knx_iot(output_path: Optional[str] = None) -> dict[str, Any]: + """Export a KNX IoT semantic view (Turtle/RDF) of the project's functional + datapoints — a pragmatic skeleton for the IP-native model, for review.""" + proj = _project() + turtle = generate_knx_iot_turtle(proj) + if output_path: + return {"written": _safe_write(output_path, turtle)} + return {"turtle": turtle} + + @mcp.tool() def workspace_info() -> dict[str, Any]: """Show the confined output workspace and the safety guarantees.""" diff --git a/pyproject.toml b/pyproject.toml index f0883a3..e0d4719 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nickol-knx-mcp" -version = "0.5.0" +version = "0.6.0" description = "Design-time KNX/ETS6 project assistant as an MCP server (parse .knxproj, validate, generate HA YAML + ETS CSV/XML). No live bus access." readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 872f4d6..77b2c95 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -555,3 +555,42 @@ assert any(p["action"] == "add_ga" and "статус" in p["name"] for p in _r[" _sd = [p for p in _r["proposals"] if p["action"] == "set_dpt"][0] assert _sd["dpt"] == "1.001", _sd print("OK: B1 — repair engine proposes set_dpt + synthesised status GA") + +# --------------------------------------------------------------------------- # +# Regression (v0.6.0): B3 diff · B4 protocol · B5 matter · B6 energy · C1/C2/C3 +# + A5 areas / A6 expose. +# --------------------------------------------------------------------------- # +print("\n=== REGRESSION: v0.6.0 advanced (B3/B4/B5/B6/C1/C2/C3 + A5/A6) ===") +from nickol_knx_mcp.advanced import (matter_readiness, completeness_grade, + energy_scaffold, test_protocol, suggest_naming) +from nickol_knx_mcp.diffproj import diff_loaded +from nickol_knx_mcp.iot import generate_knx_iot_turtle +# C3 completeness grade +_cg = completeness_grade(proj) +assert "grade" in _cg and 0 <= _cg["score"] <= 100 +# B5 matter readiness +_mr = matter_readiness(proj) +assert "controllable_functions" in _mr and "matter_ready" in _mr +# B6 energy (demo has "Total energy" 13.013) +_en = energy_scaffold(proj) +assert _en["metering_gas"] >= 1 and _en["scaffold"] +# C2 naming (demo has an empty-name GA) +_nm = suggest_naming(proj) +assert _nm["count"] >= 1 +# B4 protocol +_tp = test_protocol(proj) +assert "# Functional acceptance protocol" in _tp["markdown"] +# C1 turtle +_tt = generate_knx_iot_turtle(proj) +assert _tt.startswith("@prefix") and "knx:Datapoint" in _tt and "knx:groupAddress" in _tt +# B3 diff +_d1 = {"group_addresses": {"1/0/0": ga("1/0/0", "Light", 1, 1), "1/0/1": ga("1/0/1", "Temp", 9, 1)}} +_d2 = {"group_addresses": {"1/0/0": ga("1/0/0", "Light", 1, 1), "1/0/1": ga("1/0/1", "Temp", 9, 4), + "2/0/0": ga("2/0/0", "New GA", 1, 1)}} +_df = diff_loaded(build_loaded_from_raw(_d1, "a"), build_loaded_from_raw(_d2, "b")) +assert _df["added"] == 1 and _df["dpt_changed"] == 1, _df +# A6 expose (datetime GA present) +_ex = generate_ha_yaml(build_loaded_from_raw({"group_addresses": { + "10/0/0": ga("10/0/0", "System - Date/Time", 19, 1)}}, "mem")) +assert any(r["reason"] == "verify_expose" for r in _ex["review"]), "no time/date expose" +print("OK: v0.6.0 — matter, completeness, energy, naming, protocol, turtle, diff, expose")