diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ad5a22..b6e6fb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`check_topology()` — topology & individual-address sanity, grounded in the KNX standard** (tool + count **30 → 31**; `analyze.py`, `server.py`). Flags devices-per-line over the TP1 limits (info at + >64 per segment, warning at >256 per line — KNX Handbook p.36/40/55), individual addresses that + don't parse as a valid `A.L.D` (area 0-15, line 0-15, device 0-255 — confirmed against the xknx + `IndividualAddress` class), duplicate individual addresses (error), and multi-line TP projects with + no coupler (device `.0`) on a line (info). Empty topology / device-less synthetic projects return + no findings. Folded into `analyze_all` under a `topology` key and its severity totals. + `tests/test_topology.py`. + ### Security - **Hardened parsing of untrusted `.knxproj` / `.knxprod`** (`safexml.py`). A project file is a diff --git a/README.md b/README.md index 5134c82..5512a86 100644 --- a/README.md +++ b/README.md @@ -266,7 +266,7 @@ keyring handling, and the recommended workflow). --- -## MCP tools (30) +## MCP tools (31) **Read** | Tool | Purpose | @@ -283,6 +283,7 @@ keyring handling, and the recommended workflow). | `check_naming(name_regex?)` | validate naming / 3-level structure | | `check_missing_status()` | actuators lacking a status object | | `check_dpt()` | missing / inconsistent DPTs **+ sub-DPT sanity** (temp→9.001, power→14.056…) | +| `check_topology()` | topology capacity + individual-address validity (TP1 64/segment, 256/line, valid & unique `A.L.D`, coupler presence — KNX Handbook) | | `check_secure()` | KNX Data Secure posture + keyring handover checklist | | `check_matter()` | Matter-readiness lint (which functions round-trip to a Matter cluster) | | `check_energy()` | metering/energy DPT check + PV/battery/EVSE scaffold | @@ -380,7 +381,7 @@ nickol-knx-mcp/ │ ├── report.py # Markdown report │ ├── room_library.py # Room Library R1 — compose a new project from templates │ ├── room_templates/ # built-in room YAML templates + SCHEMA.md (public contract) -│ └── server.py # FastMCP server, 30 tools, confined writes +│ └── server.py # FastMCP server, 31 tools, confined writes ├── tests/test_pipeline.py ├── examples/claude_desktop_config.json ├── skills/ diff --git a/README.ru.md b/README.ru.md index f35cd4c..88ecd42 100644 --- a/README.ru.md +++ b/README.ru.md @@ -224,11 +224,11 @@ claude mcp add nickol-knx -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" -- /abs/ --- -## 5. Инструменты MCP (30) +## 5. Инструменты MCP (31) **Чтение:** `load_project` · `list_group_addresses` · `get_devices` · `get_topology` · `explain_ga` -**Валидация:** `check_naming` · `check_missing_status` · `check_dpt` (+ **sub-DPT** проверка) · `check_secure` (KNX Secure posture + keyring-чеклист) · `check_matter` (Matter-готовность) · `check_energy` (энергодомен) · `analyze_all` · `check_policy` (Project Policy Profile — *ваша* конвенция) +**Валидация:** `check_naming` · `check_missing_status` · `check_dpt` (+ **sub-DPT** проверка) · `check_topology` (топология/адресация: TP1 64/сегмент, 256/линию, валидность и уникальность `A.L.D`, каплеры — KNX Handbook) · `check_secure` (KNX Secure posture + keyring-чеклист) · `check_matter` (Matter-готовность) · `check_energy` (энергодомен) · `analyze_all` · `check_policy` (Project Policy Profile — *ваша* конвенция) **Починка и дизайн:** `suggest_repairs` (**предлагает фиксы, а не только флагает**) · `suggest_names` · `decompose_device` (устройство → декомпозиция: **точная вендорская модель** из локального каталога или generic-рецепт) · `list_device_recipes` (device-library: Zennio + ABB) · `parse_devices_from_project` (**точные модели устройств** из app-programs `.knxproj`/`.knxprod` → YAML каталога) · `check_device_parameters` (**кросс-девайс QA параметров** — «неправильное» устройство среди одинаковых) · `grade_completeness` (скелет vs as-built) · `diff_projects` (семантический дифф двух версий) @@ -248,6 +248,7 @@ claude mcp add nickol-knx -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" -- /abs/ | `check_naming(name_regex?)` | проверка именования/структуры | | `check_missing_status()` | актуаторы без статусного объекта | | `check_dpt()` | отсутствующие/несогласованные DPT + sub-DPT | +| `check_topology()` | ёмкость топологии + валидность индив. адресов (TP1 64/сегмент, 256/линию, валидный и уникальный `A.L.D`, наличие каплеров — KNX Handbook) | | `check_secure()` | KNX Data Secure posture + keyring-чеклист | | `check_matter()` | Matter-готовность функций | | `check_energy()` | метеринг/энергодомен | @@ -311,7 +312,7 @@ nickol-knx-mcp/ │ ├── report.py # Markdown-отчёт │ ├── room_library.py # Room Library R1 — сборка нового проекта из шаблонов │ ├── room_templates/ # YAML-шаблоны комнат + SCHEMA.md (публичный контракт) -│ └── server.py # FastMCP сервер, 30 инструментов, confined writes +│ └── server.py # FastMCP сервер, 31 инструмент, confined writes ├── tests/test_pipeline.py ├── examples/claude_desktop_config.json ├── CLAUDE.md # ETS Assistant skill / playbook diff --git a/nickol_knx_mcp/analyze.py b/nickol_knx_mcp/analyze.py index 5814aec..b54d7f1 100644 --- a/nickol_knx_mcp/analyze.py +++ b/nickol_knx_mcp/analyze.py @@ -514,6 +514,154 @@ def detect_dpt_issues(project: LoadedProject) -> list[dict[str, Any]]: return findings +# --------------------------------------------------------------------------- # +# Topology / individual-address sanity, grounded in the KNX standard. +# +# GROUNDING (verified, not invented): +# * The individual (physical) address is area(4 bits, 0-15).line(4 bits, 0-15). +# device(1 byte, 0-255), and device number 0 denotes a line/area coupler. +# Confirmed against the xknx `IndividualAddress` class via deepwiki +# (repo XKNX/xknx): MAX area = 15, MAX line = 15, MAX device = 255, and the +# `is_line` property is True exactly when the device part is 0 (a coupler). +# * The 64-devices-per-TP1-segment and 256-devices-per-line (4 segments via +# repeaters) capacity figures are NOT encoded in the xknx source — they come +# from the KNX Handbook and are cited to their pages inline below. Treat the +# page citations as the authority for the capacity numbers. +# --------------------------------------------------------------------------- # +def _parse_individual_address(ia: str) -> Optional[tuple[int, int, int]]: + """Parse 'A.L.D' -> (area, line, device) or None if not a valid triple. + + Valid ranges: area 0-15, line 0-15, device 0-255 (confirmed via xknx + IndividualAddress: MAX_AREA=15, MAX_MAIN=15, MAX_LINE=255).""" + parts = (ia or "").strip().split(".") + if len(parts) != 3: + return None + try: + a, ln, d = int(parts[0]), int(parts[1]), int(parts[2]) + except ValueError: + return None + if 0 <= a <= 15 and 0 <= ln <= 15 and 0 <= d <= 255: + return (a, ln, d) + return None + + +def detect_topology_issues(project: LoadedProject) -> list[dict[str, Any]]: + """Check topology capacity + individual-address validity/uniqueness + couplers. + + Passes (each grounded — see the module comment above): + * devices per line vs the TP1 segment (64) / line (256) capacity; + * every device individual address parses as a valid A.L.D triple; + * individual addresses are unique across devices; + * every multi-line project has a coupler (device .0) per TP line. + + An empty topology / a project with zero devices returns [] (synthetic demo + projects legitimately carry no devices — this must never raise). + """ + findings: list[dict[str, Any]] = [] + topo = project.topology or {} + devices = project.devices or {} + + # Flatten the topology to (line_id, line_name, medium, [device_addr_str,...]). + # The line key in xknxproject is already the full 'area.line' (e.g. '1.1'). + lines: list[tuple[str, str, str, list[str]]] = [] + for area in topo.values(): + for lid, line in (area.get("lines", {}) or {}).items(): + lines.append(( + str(lid), + line.get("name") or "", + (line.get("medium_type") or "").upper(), + list(line.get("devices", []) or []), + )) + + # Empty topology / zero devices -> nothing to check (do not raise). + total_topo_devices = sum(len(devs) for _, _, _, devs in lines) + if total_topo_devices == 0 and not devices: + return findings + + n_lines = len(lines) + + # 1. Devices per line vs TP1 capacity (KNX Handbook). >256 is the harder cap, + # so it wins over the >64 segment note for the same line. + for lid, lname, medium, devs in lines: + # 64/256 are TP1-specific limits (KNX Handbook TP1). Other media + # (IP/PL/RF) have different capacities — apply only to Twisted Pair lines. + # xknxproject reports the full medium string ('Twisted Pair (TP)'), so + # match the '(TP)' tag, not a bare 'TP'. + if "(TP)" not in medium: + continue + count = len(devs) + label = f"Line {lid}" + (f" ('{lname}')" if lname else "") + if count > 256: + findings.append(_finding( + SEVERITY_WARN, "topology_line_overflow", lid, + f"{label} has {count} devices — exceeds the TP1 line maximum of 256 " + "(4 segments of 64 via repeaters). KNX Handbook p.55.", + line=lid, device_count=count, + )) + elif count > 64: + findings.append(_finding( + SEVERITY_INFO, "topology_segment_limit", lid, + f"{label} has {count} devices — a single TP1 segment holds max 64 " + "(KNX Handbook p.36/40); split into further segments/repeaters and " + "ensure sufficient bus power (≤640 mA per segment).", + line=lid, device_count=count, + )) + + # 2. Individual-address validity — parse each device's address as A.L.D. + for dev in devices.values(): + ia = (dev.get("individual_address") or "").strip() + if _parse_individual_address(ia) is None: + findings.append(_finding( + SEVERITY_WARN, "invalid_individual_address", ia or "-", + f"Device '{dev.get('name', '?')}' has individual address " + f"'{ia or '(none)'}' — not a valid A.L.D where area 0-15, line 0-15, " + "device 0-255 (confirmed via xknx IndividualAddress).", + name=dev.get("name"), + )) + + # 3. Uniqueness — duplicate individual addresses across devices. Iterate by the + # address FIELD (not the devices-dict key) so a genuine duplicate is caught + # even if the dict happens to key devices differently. + by_addr: dict[str, list[str]] = defaultdict(list) + for dev in devices.values(): + ia = (dev.get("individual_address") or "").strip() + if ia: + by_addr[ia].append(dev.get("name", "?")) + for ia, names in by_addr.items(): + if len(names) > 1: + findings.append(_finding( + SEVERITY_ERROR, "duplicate_individual_address", ia, + f"Individual address '{ia}' is used by {len(names)} devices: {names}. " + "Every KNX device must have a unique individual address.", + names=names, + )) + + # 4. Coupler presence (soft). Only for multi-line projects, and only on TP + # lines — a single-line or IP-only design needs no per-line TP coupler, so + # those must not be flagged (avoid noise). A coupler carries device 0. + if n_lines > 1: + for lid, lname, medium, devs in lines: + # Coupler-on-.0 is a TP-topology concept; skip IP/PL/RF/Unknown lines. + # xknxproject reports the full medium string ('KNXnet/IP (IP)', etc.), + # so match the '(TP)' tag rather than a bare 'IP'. + if not devs or "(TP)" not in medium: + continue + has_coupler = any( + (_parse_individual_address(str(da)) or (None, None, None))[2] == 0 + for da in devs + ) + if not has_coupler: + label = f"Line {lid}" + (f" ('{lname}')" if lname else "") + findings.append(_finding( + SEVERITY_INFO, "line_without_coupler", lid, + f"{label} has devices but none at .0 — a line/area coupler uses " + "device 0; check coupler addressing. KNX Handbook p.43.", + line=lid, + )) + + return findings + + # --------------------------------------------------------------------------- # # KNX Secure posture + keyring handover checklist (A4). # Report-only: this server never handles key material — it only summarises the diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index a833503..644554b 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -10,6 +10,7 @@ Run (stdio): python -m nickol_knx_mcp.server from __future__ import annotations import os +from collections import Counter from pathlib import Path from typing import Any, Optional @@ -17,7 +18,7 @@ from mcp.server.fastmcp import FastMCP from .project import load_project as load_project_file, LoadedProject from .analyze import (validate_naming, detect_missing_status, detect_dpt_issues, - secure_posture) + detect_topology_issues, secure_posture) from .generate_ha import generate_ha_yaml from .generate_ets import generate_ets_csv, generate_ets_xml from .report import build_report @@ -176,6 +177,16 @@ def check_dpt() -> list[dict[str, Any]]: return detect_dpt_issues(_project()) +@mcp.tool() +def check_topology() -> list[dict[str, Any]]: + """Check topology capacity and individual-address validity (KNX Handbook). + + Flags lines over the TP1 segment (64) / line (256) limits, invalid or + duplicate individual addresses, and multi-line projects missing a coupler. + """ + return detect_topology_issues(_project()) + + @mcp.tool() def suggest_repairs() -> dict[str, Any]: """Propose concrete fixes for the project's findings — repair, don't just flag. @@ -206,11 +217,20 @@ def analyze_all(name_regex: Optional[str] = None) -> dict[str, Any]: """Run every check and return the report summary plus all findings.""" proj = _project() rep = build_report(proj, name_regex=name_regex) + topology = detect_topology_issues(proj) + # build_report counts only naming+status+dpt severities; fold in topology so + # the errors/warnings/info totals cover every check the summary reports on. + summary = dict(rep["summary"]) + topo_sev = Counter(f["severity"] for f in topology) + summary["errors"] = summary.get("errors", 0) + topo_sev.get("error", 0) + summary["warnings"] = summary.get("warnings", 0) + topo_sev.get("warning", 0) + summary["info"] = summary.get("info", 0) + topo_sev.get("info", 0) return { - "summary": rep["summary"], + "summary": summary, "naming": validate_naming(proj, name_regex=name_regex), "missing_status": detect_missing_status(proj), "dpt": detect_dpt_issues(proj), + "topology": topology, } diff --git a/tests/test_topology.py b/tests/test_topology.py new file mode 100644 index 0000000..b5587f4 --- /dev/null +++ b/tests/test_topology.py @@ -0,0 +1,131 @@ +"""Topology / individual-address checks (detect_topology_issues). + +Grounded in the KNX standard: individual address = area(0-15).line(0-15).device +(0-255), device 0 = coupler (confirmed via xknx IndividualAddress, deepwiki +XKNX/xknx); TP1 segment holds 64 devices, a line up to 256 (KNX Handbook). +""" +from nickol_knx_mcp.project import build_loaded_from_raw +from nickol_knx_mcp.analyze import detect_topology_issues + + +def _dev(ia, name="Dev"): + return {"name": name, "individual_address": ia, "order_number": "X", + "manufacturer_name": "M", "hardware_name": "", "description": "", + "application": None, "project_uid": None, + "communication_object_ids": [], "channels": {}} + + +def _proj(devices, topology): + raw = {"info": {"group_address_style": "ThreeLevel", "schema_version": "21"}, + "group_addresses": {}, "communication_objects": {}, + "devices": devices, "functions": {}, "topology": topology, + "group_ranges": {}} + return build_loaded_from_raw(raw, "t.knxproj") + + +def _codes(findings): + return {f["code"] for f in findings} + + +def test_segment_limit_over_64_is_info(): + devs = [f"1.1.{i}" for i in range(1, 71)] # 70 devices on one line + devices = {ia: _dev(ia) for ia in devs} + topology = {"1": {"name": "Area 1", + "lines": {"1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", + "devices": devs}}}} + findings = detect_topology_issues(_proj(devices, topology)) + seg = [f for f in findings if f["code"] == "topology_segment_limit"] + assert seg and seg[0]["severity"] == "info", findings + assert seg[0]["device_count"] == 70 + # 70 <= 256 so it must NOT also raise a line overflow + assert "topology_line_overflow" not in _codes(findings) + + +def test_invalid_individual_address_is_warning(): + # area 16 is out of range (max 15) + devices = {"bad": _dev("16.0.1", "Rogue"), "1.1.1": _dev("1.1.1", "Ok")} + topology = {"1": {"name": "Area 1", + "lines": {"1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", + "devices": ["1.1.1"]}}}} + findings = detect_topology_issues(_proj(devices, topology)) + inv = [f for f in findings if f["code"] == "invalid_individual_address"] + assert len(inv) == 1 and inv[0]["severity"] == "warning", findings + assert inv[0]["address"] == "16.0.1" + + +def test_duplicate_individual_address_is_error(): + devices = {"a": _dev("1.1.5", "First"), "b": _dev("1.1.5", "Second")} + topology = {"1": {"name": "Area 1", + "lines": {"1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", + "devices": ["1.1.5"]}}}} + findings = detect_topology_issues(_proj(devices, topology)) + dup = [f for f in findings if f["code"] == "duplicate_individual_address"] + assert len(dup) == 1 and dup[0]["severity"] == "error", findings + assert dup[0]["address"] == "1.1.5" + + +def test_valid_small_project_is_clean(): + # single line, valid addresses, well under capacity -> no findings at all + devs = ["1.1.1", "1.1.2", "1.1.3"] + devices = {ia: _dev(ia) for ia in devs} + topology = {"1": {"name": "Area 1", + "lines": {"1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", + "devices": devs}}}} + findings = detect_topology_issues(_proj(devices, topology)) + assert findings == [], findings + + +def test_empty_topology_and_no_devices_returns_empty(): + assert detect_topology_issues(_proj({}, {})) == [] + + +def test_multiline_without_coupler_is_info_but_singleline_is_clean(): + # two TP lines, neither has a .0 coupler device -> info per line + devices = {"1.1.1": _dev("1.1.1"), "1.2.1": _dev("1.2.1")} + topology = {"1": {"name": "Area 1", "lines": { + "1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", "devices": ["1.1.1"]}, + "1.2": {"name": "Line 2", "medium_type": "Twisted Pair (TP)", "devices": ["1.2.1"]}, + }}} + findings = detect_topology_issues(_proj(devices, topology)) + coupler = [f for f in findings if f["code"] == "line_without_coupler"] + assert len(coupler) == 2 and all(f["severity"] == "info" for f in coupler), findings + + # a line WITH a .0 coupler is not flagged + devices2 = {"1.1.0": _dev("1.1.0", "Coupler"), "1.1.1": _dev("1.1.1"), + "1.2.0": _dev("1.2.0", "Coupler2"), "1.2.1": _dev("1.2.1")} + topology2 = {"1": {"name": "Area 1", "lines": { + "1.1": {"name": "Line 1", "medium_type": "Twisted Pair (TP)", "devices": ["1.1.0", "1.1.1"]}, + "1.2": {"name": "Line 2", "medium_type": "Twisted Pair (TP)", "devices": ["1.2.0", "1.2.1"]}, + }}} + findings2 = detect_topology_issues(_proj(devices2, topology2)) + assert "line_without_coupler" not in _codes(findings2), findings2 + + +def test_ip_line_is_not_flagged_regression(): + """xknxproject reports the full medium string ('KNXnet/IP (IP)'), not 'IP'. + An IP backbone/main line must NOT get TP1 findings: no segment/line-capacity + note and no line_without_coupler (couplers-on-.0 is a TP concept).""" + # IP line with 100 devices and no .0 coupler + a TP line over 64 + ip_devs = [f"1.0.{i}" for i in range(1, 101)] # 100 on the IP main line + tp_devs = [f"1.1.{i}" for i in range(1, 71)] # 70 on a TP line + devices = {ia: _dev(ia) for ia in ip_devs + tp_devs} + topology = {"1": {"name": "Area 1", "lines": { + "1.0": {"name": "Main", "medium_type": "KNXnet/IP (IP)", "devices": ip_devs}, + "1.1": {"name": "TP line", "medium_type": "Twisted Pair (TP)", "devices": tp_devs}, + }}} + findings = detect_topology_issues(_proj(devices, topology)) + # The IP line (1.0) must appear in NO topology finding. + for f in findings: + assert f.get("line") != "1.0", f + # The TP line (1.1) still gets its segment note + coupler note. + assert "topology_segment_limit" in _codes(findings) + seg = [f for f in findings if f["code"] == "topology_segment_limit"] + assert all(f["line"] == "1.1" for f in seg), seg + assert "line_without_coupler" in _codes(findings) # 1.1 has no .0 coupler + + +if __name__ == "__main__": + for _name, _fn in list(globals().items()): + if _name.startswith("test_") and callable(_fn): + _fn() + print("test_topology: OK")