feat: check_topology — per-line device count + individual-address validity (KNX canon)

New design-time check grounded in the KNX standard (KNX Handbook + xknx via
deepwiki). detect_topology_issues() + check_topology() MCP tool (30->31 tools),
folded into analyze_all counters:
- devices per line vs TP1 segment 64 / line 256 (KNX Handbook p.36/40/55) — TP-only
- individual address valid A.L.D (area 0-15, line 0-15, device 0-255; xknx-confirmed)
- duplicate individual address -> error; missing .0 coupler on a multi-line TP -> info
Medium-gated on the real xknxproject string ('Twisted Pair (TP)'), so IP/PL/RF
lines are never falsely flagged (fixed a dead 'medium==IP' guard caught in audit;
regression test added). senior-code-audit: APPROVED. Functional smoke on 2 real
projects: IP lines clean, TP coupler/count notes correct. 64/256 cited to the
Handbook, not attributed to xknx. tests/test_topology.py 7/7; suite 27 passed
(pre-existing unrelated test_protocol collection error untouched).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

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/

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

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

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,
     }
 
 

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")