diff --git a/CHANGELOG.md b/CHANGELOG.md index cf81176..e7d3db0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,12 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -**Noise-reduction refinements from a 3646-GA multi-vendor villa.** Validated the tool on a -second real signed Zennio project (5× larger, no ETS Functions). Three de-noising rules, -each downgrading a false alarm to INFO without hiding any real problem — on that project -errors fell **36 → 32** and warnings **179 → 139**. +**Project handover pack + noise-reduction refinements**, both driven by a second real signed +Zennio project (a 3646-GA multi-vendor villa, 5× larger, no ETS Functions). ### Added +- **`generate_handover_pack` — the as-built commissioning deliverable** (Track B). From a + read-only `.knxproj` it assembles `handover.md` (equipment inventory by manufacturer, + group-address map by domain, command/status feedback coverage %, KNX Secure scope, QA state), + a standalone **`topology.svg`** area/line/device diagram, the full `group-addresses.csv` and + the `ha-package.yaml`. On the villa: 275 devices across 8 manufacturers, 14 domains, 93 % + feedback coverage — one command produces the whole handover bundle. - **Divider/separator scratch detection** — commissioning placeholder names made only of punctuation (`---`, `=====`) or a marker wrapped in it (`-----addition------`) now classify as `scratch` intent, so a missing DPT on them is an INFO note, not a 🔴 error. diff --git a/nickol_knx_mcp/handover.py b/nickol_knx_mcp/handover.py new file mode 100644 index 0000000..4b4f69c --- /dev/null +++ b/nickol_knx_mcp/handover.py @@ -0,0 +1,256 @@ +"""Project handover pack — the deliverable an integrator hands over at commissioning. + +From a read-only ``.knxproj`` this assembles the "as-built" picture the next +engineer (or the client) needs: an **equipment inventory** by manufacturer, the +**group-address map by domain**, **command/status feedback coverage**, the +**KNX Secure scope**, and the current **QA state** — plus a standalone **SVG +topology diagram**. It reuses the existing analysis passes so the numbers match +``project_report``; it adds the packaging, structure and diagram that turn a bag +of findings into a handover document. + +No bus access, no writes outside the caller-provided path. Pure text + SVG. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from html import escape +from typing import Any + +from .project import LoadedProject +from .analyze import validate_naming, detect_missing_status, detect_dpt_issues +from .intent import INTENT_FUNCTIONAL + + +# --------------------------------------------------------------------------- # +# Structure helpers +# --------------------------------------------------------------------------- # +def _domain_map(project: LoadedProject) -> dict[int, dict[str, Any]]: + """main -> {name, count, middles: {middle -> {name, count}}}, sorted-ready.""" + mains: dict[int, dict[str, Any]] = {} + for ga in project.gas.values(): + if ga.main is None: + continue + m = mains.setdefault(ga.main, {"name": ga.main_name or "", "count": 0, "middles": {}}) + m["count"] += 1 + if not m["name"] and ga.main_name: + m["name"] = ga.main_name + mid = m["middles"].setdefault( + ga.middle if ga.middle is not None else -1, + {"name": ga.middle_name or "", "count": 0}, + ) + mid["count"] += 1 + if not mid["name"] and ga.middle_name: + mid["name"] = ga.middle_name + return mains + + +def _device_inventory(project: LoadedProject) -> dict[str, list[dict[str, Any]]]: + """manufacturer -> [ {ia, name, order, cos} ], each list sorted by IA.""" + by_mfr: dict[str, list[dict[str, Any]]] = defaultdict(list) + for d in project.devices.values(): + by_mfr[d.get("manufacturer_name") or "?"].append({ + "ia": d.get("individual_address") or "?", + "name": d.get("name") or "?", + "order": d.get("order_number") or "?", + "cos": len(d.get("communication_object_ids", []) or []), + }) + for lst in by_mfr.values(): + lst.sort(key=lambda x: x["ia"]) + return by_mfr + + +def _feedback_coverage(project: LoadedProject, + missing: list[dict[str, Any]]) -> dict[str, int]: + """How many functional command GAs have a status vs are missing one.""" + commands = [ga for ga in project.gas.values() + if ga.intent == INTENT_FUNCTIONAL and ga.kind == "command" + and ga.dpt_main is not None] + gaps = sum(1 for f in missing if f["code"] == "missing_status_address") + total = len(commands) + return {"commands": total, "with_status": max(total - gaps, 0), "missing": gaps} + + +# --------------------------------------------------------------------------- # +# SVG topology diagram (standalone, no external CSS) +# --------------------------------------------------------------------------- # +def build_topology_svg(project: LoadedProject) -> str: + """A simple, valid, standalone SVG of areas -> lines -> device counts.""" + areas = project.topology or {} + W = 820 + pad, row_h, gap = 20, 34, 10 + rows: list[tuple[str, str, str, int]] = [] # (kind, id, label, devices) + for aid, area in areas.items(): + lines = area.get("lines", {}) or {} + adev = sum(len(l.get("devices", []) or []) for l in lines.values()) + rows.append(("area", str(aid), area.get("name") or f"Area {aid}", adev)) + for lid, line in lines.items(): + n = len(line.get("devices", []) or []) + label = f"{lid} {line.get('name') or ''}".strip() + rows.append(("line", str(lid), label, n)) + if not rows: + rows = [("area", "-", "No topology in project", 0)] + + max_dev = max((n for _, _, _, n in rows if n), default=1) or 1 + H = pad * 2 + len(rows) * (row_h + gap) + 40 + out: list[str] = [ + f'', + f'', + f'KNX topology — {escape(project.info.get("name","?"))}', + ] + y = pad + 28 + for kind, rid, label, n in rows: + if kind == "area": + out.append( + f'' + f'▣ {escape(label)} · {n} devices' + ) + else: + bar = int((W - 2 * pad - 340) * (n / max_dev)) if n else 0 + out.append( + f'' + f'' + f'{escape(label)}' + f'' + f'{n} dev' + ) + y += row_h + gap + out.append('') + return "\n".join(out) + + +# --------------------------------------------------------------------------- # +# Handover document (Markdown) +# --------------------------------------------------------------------------- # +def build_handover(project: LoadedProject, + name_regex: str | None = None) -> dict[str, Any]: + """Return {'markdown': str, 'svg': str, 'summary': {...}}.""" + naming = validate_naming(project, name_regex=name_regex) + missing = detect_missing_status(project) + dpts = detect_dpt_issues(project) + all_f = naming + missing + dpts + sev = Counter(f["severity"] for f in all_f) + + gas = project.gas + info = project.info + inv = _device_inventory(project) + dmap = _domain_map(project) + cover = _feedback_coverage(project, missing) + secure = [ga for ga in gas.values() if ga.data_secure] + intent_counts = Counter(ga.intent for ga in gas.values()) + + md: list[str] = [] + md.append(f"# KNX Handover Pack — {info.get('name','?')}\n") + md.append( + "_As-built handover document generated read-only from the ETS project. " + "Review before commissioning sign-off; this server never touches the bus._\n" + ) + md.append( + f"- **Source project:** `{info.get('name','?')}`\n" + f"- **GA style:** {info.get('group_address_style','?')} · " + f"**ETS tool:** {info.get('tool_version','?')}\n" + f"- **Last modified:** {info.get('last_modified','?')}\n" + f"- **Group addresses:** {len(gas)} · **Devices:** {len(project.devices)} · " + f"**Lines:** {sum(len(a.get('lines',{})) for a in project.topology.values())}\n" + ) + + # 1. Topology + md.append("\n## 1. Topology\n") + md.append("See `topology.svg` for the diagram. Lines and device counts:\n") + md.append("| Line | Name | Medium | Devices |") + md.append("|---|---|---|---|") + for aid, area in project.topology.items(): + for lid, line in area.get("lines", {}).items(): + md.append(f"| `{lid}` | {line.get('name','')} | {line.get('medium_type','')} " + f"| {len(line.get('devices',[]) or [])} |") + + # 2. Equipment inventory (Спецификация оборудования) + md.append("\n## 2. Equipment inventory\n") + for mfr in sorted(inv): + devs = inv[mfr] + md.append(f"\n**{mfr}** — {len(devs)} device(s)\n") + md.append("| IA | Device | Order № | Comm-objects |") + md.append("|---|---|---|---|") + for d in devs: + md.append(f"| `{d['ia']}` | {d['name']} | `{d['order']}` | {d['cos']} |") + + # 3. Group-address map by domain + md.append("\n## 3. Group-address map by domain\n") + for main in sorted(dmap): + m = dmap[main] + md.append(f"\n**[{main}] {m['name'] or '(unnamed)'}** — {m['count']} GA\n") + for mid in sorted(m["middles"]): + sub = m["middles"][mid] + label = f".{mid}" if mid >= 0 else ".-" + md.append(f"- `{label}` {sub['name'] or '(unnamed)'} — {sub['count']} GA") + + # 4. Feedback coverage + md.append("\n## 4. Command / status coverage\n") + pct = (100 * cover["with_status"] // cover["commands"]) if cover["commands"] else 0 + md.append( + f"- Functional command GAs: **{cover['commands']}**\n" + f"- With a status/feedback GA: **{cover['with_status']}** ({pct}%)\n" + f"- Missing status: **{cover['missing']}** — Home Assistant cannot read real " + "state for these until a feedback GA is added.\n" + ) + + # 5. KNX Secure + md.append("\n## 5. KNX Secure scope\n") + if secure: + md.append( + f"**{len(secure)}** group address(es) carry the KNX Data Secure flag. " + "The next engineer needs the **ETS Keyring export (`.knxkeys`)** to " + "commission or read these — it is handled in ETS/HA, never by this tool.\n" + ) + md.append("\n
Secured group addresses\n") + for ga in secure[:200]: + md.append(f"- `{ga.address}` {ga.name}") + if len(secure) > 200: + md.append(f"- … and {len(secure)-200} more") + md.append("
\n") + else: + md.append("No KNX Data Secure group addresses in this project.\n") + + # 6. QA state at handover + md.append("\n## 6. QA state at handover\n") + md.append( + f"Totals: 🔴 errors **{sev.get('error',0)}**, " + f"🟡 warnings **{sev.get('warning',0)}**, 🔵 info **{sev.get('info',0)}**.\n" + ) + nonfunc = ", ".join(f"{k}={v}" for k, v in sorted(intent_counts.items()) if k != INTENT_FUNCTIONAL) + if nonfunc: + md.append(f"\nNon-functional GAs (excluded from error checks): {nonfunc}.\n") + md.append( + "\nSee `project_report` (or `analyze_all`) for the itemised findings. Resolve " + "🔴 errors in ETS before sign-off.\n" + ) + + # 7. Pack contents + md.append("\n## 7. Pack contents\n") + md.append( + "- `handover.md` — this document.\n" + "- `topology.svg` — area/line/device diagram.\n" + "- `group-addresses.csv` — full GA list (ETS-native export).\n" + "- `ha-package.yaml` — Home Assistant KNX package (optional deploy).\n" + ) + + summary = { + "ga_count": len(gas), + "devices": len(project.devices), + "manufacturers": len(inv), + "lines": sum(len(a.get("lines", {})) for a in project.topology.values()), + "domains": len(dmap), + "feedback_coverage_pct": pct, + "secure_gas": len(secure), + "errors": sev.get("error", 0), + "warnings": sev.get("warning", 0), + } + return {"markdown": "\n".join(md), "svg": build_topology_svg(project), + "summary": summary} diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index 2b8fe8d..f4c8cbd 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -20,6 +20,7 @@ from .analyze import validate_naming, detect_missing_status, detect_dpt_issues from .generate_ha import generate_ha_yaml from .generate_ets import generate_ets_csv, generate_ets_xml from .report import build_report +from .handover import build_handover mcp = FastMCP("nickol-knx") @@ -232,6 +233,37 @@ def project_report(output_path: Optional[str] = None, return out +@mcp.tool() +def generate_handover_pack(output_dir: Optional[str] = None) -> dict[str, Any]: + """Generate a project handover pack (as-built deliverable for commissioning). + + Assembles an equipment inventory, group-address map by domain, command/status + coverage, KNX Secure scope and QA state into ``handover.md``, plus a + ``topology.svg`` diagram, the full ``group-addresses.csv`` and the + ``ha-package.yaml``. When ``output_dir`` is given (a folder inside the + workspace) all files are written there and the paths returned; otherwise the + handover markdown + SVG are returned inline. + """ + proj = _project() + pack = build_handover(proj) + out: dict[str, Any] = {"summary": pack["summary"]} + if output_dir: + d = output_dir.rstrip("/") + written = { + "handover": _safe_write(f"{d}/handover.md", pack["markdown"]), + "topology_svg": _safe_write(f"{d}/topology.svg", pack["svg"]), + "group_addresses_csv": _safe_write(f"{d}/group-addresses.csv", + generate_ets_csv(proj)), + "ha_package": _safe_write(f"{d}/ha-package.yaml", + generate_ha_yaml(proj)["yaml"]), + } + out["written"] = written + else: + out["markdown"] = pack["markdown"] + out["svg"] = pack["svg"] + return out + + @mcp.tool() def workspace_info() -> dict[str, Any]: """Show the confined output workspace and the safety guarantees.""" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index a23d9df..eae67e4 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -438,3 +438,33 @@ assert _cl["command_value_state_address"] == "3/6/1" assert any(r["reason"] == "manual_climate" and r["address"] == "3/3/2" for r in _haT["review"]), \ "mode-only zone should go to review, not emit an invalid climate" print("OK: full zone -> valid climate; mode-only zone -> review") + +# --------------------------------------------------------------------------- # +# Regression (Track B): handover pack assembles a document + valid SVG diagram. +# --------------------------------------------------------------------------- # +print("\n=== REGRESSION: handover pack (Track B) ===") +from nickol_knx_mcp.handover import build_handover, build_topology_svg + +_hp = build_handover(proj) +_hmd = _hp["markdown"] +# all seven sections present +for _sec in ("# KNX Handover Pack", "## 1. Topology", "## 2. Equipment inventory", + "## 3. Group-address map by domain", "## 4. Command / status coverage", + "## 5. KNX Secure scope", "## 6. QA state at handover", "## 7. Pack contents"): + assert _sec in _hmd, f"handover missing section: {_sec}" +# equipment inventory picked up the synthetic MDT device + its order number +assert "MDT" in _hmd and "AKK-0416.03" in _hmd, "device inventory not rendered" +# summary shape +_hs = _hp["summary"] +for _k in ("ga_count", "devices", "manufacturers", "lines", "domains", + "feedback_coverage_pct", "secure_gas", "errors", "warnings"): + assert _k in _hs, f"handover summary missing {_k}" +assert _hs["devices"] == 1 and _hs["manufacturers"] == 1, _hs +# valid, standalone SVG +_svg = _hp["svg"] +assert _svg.startswith(""), "SVG malformed" +assert "KNX topology" in _svg +# it parses as XML (well-formed) +import xml.dom.minidom as _mdom +_mdom.parseString(_svg) +print("OK: handover pack — 7 sections, inventory, summary, well-formed SVG")