commit bba8befcde2d6ce744445218352cb127e6e0e5d9 Author: Nikolay Miroshnichenko Date: Sun Jun 28 09:25:57 2026 +0200 nickol-knx-mcp v0.1.0 — design-time KNX/ETS6 MCP server (public beta) Design-time MCP server that reads .knxproj (read-only), validates naming/DPT/status, and generates Home Assistant KNX YAML + ETS-importable group addresses (XML/CSV). No live bus access — confined-workspace writes only. Includes: 12 MCP tools, end-to-end smoke test, MIT license, English-first README (+ Russian), CONTRIBUTING with a real-project test call, SECURITY policy, CHANGELOG, GitHub Actions CI (Python 3.10–3.12), and issue/PR templates. Co-Authored-By: Claude Opus 4.8 diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml new file mode 100644 index 0000000..baa2e5d --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug_report.yml @@ -0,0 +1,46 @@ +name: "🐛 Bug report" +description: "Something is broken or behaves incorrectly." +title: "[bug] " +labels: ["bug"] +body: + - type: textarea + id: what + attributes: + label: What happened? + description: What you did, what you expected, what you got. + validations: + required: true + - type: textarea + id: repro + attributes: + label: Steps to reproduce + placeholder: | + 1. load_project on ... + 2. call analyze_all + 3. ... + validations: + required: true + - type: input + id: tool + attributes: + label: Which MCP tool / command + placeholder: "generate_ha_package" + - type: textarea + id: env + attributes: + label: Environment + placeholder: "OS, Python version (python3 --version), nickol-knx-mcp version, Claude Desktop/Code" + validations: + required: true + - type: textarea + id: logs + attributes: + label: Full error / stack trace + render: shell + - type: checkboxes + id: nopii + attributes: + label: Confirmation + options: + - label: "I have not attached a real `.knxproj`/`.knxkeys` (redacted snippets only)." + required: true diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml new file mode 100644 index 0000000..2dfce6b --- /dev/null +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -0,0 +1,5 @@ +blank_issues_enabled: true +contact_links: + - name: 💬 Questions & discussion + url: https://github.com/NickoScope/nickol-knx-mcp/discussions + about: General questions, ideas, and show-and-tell. For bugs and test reports, please use an issue template. diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml new file mode 100644 index 0000000..a597d20 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature_request.yml @@ -0,0 +1,20 @@ +name: "💡 Feature request" +description: "Suggest an improvement or a new capability." +title: "[feat] " +labels: ["enhancement"] +body: + - type: textarea + id: problem + attributes: + label: Problem / use case + description: What are you trying to do that's hard today? + validations: + required: true + - type: textarea + id: proposal + attributes: + label: Proposed solution + - type: textarea + id: alternatives + attributes: + label: Alternatives considered diff --git a/.github/ISSUE_TEMPLATE/real_project_test.yml b/.github/ISSUE_TEMPLATE/real_project_test.yml new file mode 100644 index 0000000..d966e51 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/real_project_test.yml @@ -0,0 +1,55 @@ +name: "🧪 Real-project test report" +description: "You ran nickol-knx-mcp on a real ETS project — tell us how it went (this is the most valuable feedback)." +title: "[test] " +labels: ["test-report"] +body: + - type: markdown + attributes: + value: | + Thank you for testing! The tool is read-only and never touches a bus, so this is safe. + **Please do NOT attach your actual `.knxproj`** — a redacted snippet or a screenshot of the + report is plenty. + - type: input + id: ets + attributes: + label: ETS version + placeholder: "ETS6 (6.3.0) / ETS5 ..." + validations: + required: true + - type: input + id: size + attributes: + label: Project size + placeholder: "~120 group addresses, ~30 devices, 3-level GA" + validations: + required: true + - type: dropdown + id: parsed + attributes: + label: Did load_project parse it? + options: + - "Yes, fully" + - "Yes, but with warnings" + - "No, it crashed" + validations: + required: true + - type: textarea + id: right + attributes: + label: What did it get RIGHT? + placeholder: "Correctly flagged missing status GAs for lights; HA covers looked good..." + - type: textarea + id: wrong + attributes: + label: What did it get WRONG? + placeholder: "False missing-status on X; wrong category for Y; bad DPT call on Z; HA YAML issue..." + - type: textarea + id: env + attributes: + label: Environment + placeholder: "OS, Python version, Claude Desktop or Claude Code" + - type: textarea + id: logs + attributes: + label: Any crash / stack trace + render: shell diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md new file mode 100644 index 0000000..bb091ec --- /dev/null +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -0,0 +1,13 @@ + + +## What does this PR do? + +## Checklist +- [ ] `python tests/test_pipeline.py` passes locally +- [ ] Safety invariant preserved: `project.py` stays the only `.knxproj` reader and stays read-only; no networking/bus libraries added +- [ ] Generators remain conservative (ambiguous → `review`, not guessed) +- [ ] New classification/behavior is covered by a case in `tests/test_pipeline.py` +- [ ] Docs updated if user-facing (README / CHANGELOG) + +## Related issues +Closes # diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1a54372 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,37 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + smoke-test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11", "3.12"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + + - name: Install package + run: | + python -m pip install --upgrade pip + pip install -e . + + - name: End-to-end smoke test (synthetic 16-GA project) + run: python tests/test_pipeline.py + + - name: Console script is installed + run: | + python -c "import importlib.metadata as m; print('entry points:', [e.name for e in m.entry_points(group='console_scripts') if e.name == 'nickol-knx-mcp'])" + command -v nickol-knx-mcp diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6977844 --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +.eggs/ +build/ +dist/ +*.egg + +# Virtualenvs +.venv/ +venv/ +env/ + +# Test / type / lint caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ + +# Workspace / generated artifacts (never commit live project exports blindly) +knx-workspace/ +*.knxproj +*.knxkeys +*.keyring + +# OS / editors +.DS_Store +.idea/ +.vscode/ +*.swp diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d0e0276 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,33 @@ +# Changelog + +All notable changes to this project are documented here. +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +## [0.1.0] — 2026-06-28 + +Initial public beta. + +### Added +- Design-time MCP server with **12 tools**: `load_project`, `list_group_addresses`, + `get_devices`, `get_topology`, `check_naming`, `check_missing_status`, `check_dpt`, + `analyze_all`, `generate_ha_package`, `generate_ets_group_addresses`, `project_report`, + `workspace_info`. +- Read-only `.knxproj` parsing via `xknxproject` (ETS5/ETS6, password-protected supported). +- GA classification (category + kind) from DPT and multilingual (EN/DE/RU) name keywords. +- Naming, missing-status, and DPT validation checks. +- Home Assistant KNX YAML generation with a conservative `review` list for ambiguous items. +- ETS-importable group-address export in XML (`knx.org/xml/ga-export/01`) and CSV. +- Markdown project report. +- Confined-workspace write guarantee (`NICKOL_KNX_WORKSPACE`); no bus access by design. +- `CLAUDE.md` design playbook, end-to-end smoke test (synthetic 16-GA project), example + Claude Desktop config. + +### Known limitations +- Tested end-to-end on a synthetic project only; real-world `.knxproj` testing is ongoing + (see the call for testers in the README). + +[Unreleased]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.1.0...HEAD +[0.1.0]: https://github.com/NickoScope/nickol-knx-mcp/releases/tag/v0.1.0 diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..df47652 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,70 @@ +# ETS / KNX Assistant — project rules + +This file is the design playbook for working on this KNX project with the +`nickol-knx` MCP server. Claude Code loads it automatically as project context. +Apply these rules when reading the `.knxproj`, generating Home Assistant config, +or producing ETS group-address exports. + +## Hard safety rules +- **Never** propose writing to a live KNX bus. The `nickol-knx` server has no bus + access; keep it that way. Live control happens only through the Home Assistant + layer, and only when the user explicitly asks. +- **Always** generate a Markdown report (`project_report`) and let the user read + it **before** any import into ETS or deploy into Home Assistant. +- All generated files go into the Git-tracked workspace. Commit before and after + changes so every step is reversible. + +## Two-layer architecture +- **KNX (ETS)** owns foundational logic and runs autonomously. The integrator + programs it. Treat the `.knxproj` as the source of truth for addresses + DPTs. +- **Home Assistant** is the upper smart layer (automations, templates). HA must + read *real device state*, never assume it. + +## Group address structure (3-level Main/Middle/Sub) +- Main group = function domain. Suggested split: + `0 Central/Scenes · 1 Lighting · 2 Shutters/Blinds · 3 HVAC · 4 Sensors · + 5 Energy · 6 Diagnostics · 7 Reserve`. +- Middle group = sub-function or zone (e.g. Switch / Dimming / Status / Position). +- Keep **commands and status in distinct, predictable middle groups** (e.g. + command in `.../0/...`, feedback in `.../4/...`). The pairing engine matches by + name tokens, so consistent naming matters more than adjacency. +- Reserve address space in every range; never pack ranges 100%. + +## Command / status pairs (the most important rule) +- **Every controllable actuator must have a status object.** A switch needs a + state GA; a dimmer needs a brightness-state GA; a blind needs position + position + state. `check_missing_status` flags anything without one. +- HA entities must have a `state_address` (or `*_state_address`) wherever the KNX + device can report it. No status = HA shows stale/guessed state. + +## DPT discipline +- Every GA must have a DPT set. Missing DPT is a hard error (`check_dpt`): HA + cannot decode it. +- Same logical name → same DPT. Inconsistent DPTs across same-named GAs are a bug. +- Common DPTs: switch `1.001`, status `1.011`, up/down `1.008`, stop `1.010`, + dimming `3.007`, brightness/position `5.001`, temperature `9.001`, humidity + `9.007`, CO₂/ppm `9.008`, lux `9.004`, energy kWh `13.013`, power W `14.056`, + scene `17.001`/`18.001`, HVAC mode `20.102`. + +## Naming +- Names encode **zone + function** (e.g. "Kitchen ceiling light switch", + "Bedroom blind position status"). The status keyword (`status`/`state`/`статус`/ + `Rückmeldung`) is what lets the engine pair feedback to its command. +- No empty names, no duplicates. Main and middle ranges get descriptive names. + +## Category separation +Keep these domains in their own ranges and HA platforms: lighting, dimming, +shutters, HVAC, sensors, scenes, energy, diagnostics. + +## KNX Secure +- When KNX Data Secure / IP Secure is enabled, secure tunneling needs the **ETS + Keyring export** (`.knxkeys`). Exports from this tool carry the `Security` flag + per GA; the keyring itself is handled in ETS/HA, never by this server. + +## Typical workflow +1. `load_project` the `.knxproj`. +2. `analyze_all` → fix 🔴 errors in ETS, add missing status GAs. +3. `project_report` → human review. +4. `generate_ets_group_addresses` (xml) → import into ETS for any new GAs. +5. `generate_ha_package` → review YAML → deploy to Home Assistant. +6. Commit every artifact to Git. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..ae57510 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,70 @@ +# Contributing to nickol-knx-mcp + +Thanks for helping! This project is in **public beta**, and the single most valuable contribution +right now is **testing against real ETS projects**. + +## 🧪 The #1 ask: test on your real `.knxproj` + +The tool is **read-only and never connects to a KNX bus**, so running it against a production +project is safe. Here is the fastest way to help: + +```bash +git clone https://github.com/NickoScope/nickol-knx-mcp.git +cd nickol-knx-mcp +python3 -m venv .venv && source .venv/bin/activate +pip install -e . +python tests/test_pipeline.py # should pass +``` + +Then run it from Claude (Desktop or Code) and ask it to: + +1. `load_project` your `.knxproj` (with the ETS password if it's protected), +2. `analyze_all`, and +3. `project_report`. + +Open a **[Real-project test report](https://github.com/NickoScope/nickol-knx-mcp/issues/new?template=real_project_test.yml)** +issue and tell us: + +- ETS version (5 / 6) and roughly how many group addresses / devices, +- what the report got **right**, +- what it got **wrong** (false missing-status, wrong category, wrong DPT call, bad HA YAML), +- any crash or stack trace. + +**Please do not attach your actual `.knxproj`** (it can contain personal/topology data). A redacted +snippet or a screenshot of the report is plenty. If a parse crash needs a sample, we'll coordinate a +minimal redacted project privately. + +## 🐛 Bugs & 💡 features + +Use the [issue templates](.github/ISSUE_TEMPLATE). For bugs, include OS, Python version, the exact +tool call, and the full error. + +## 🔧 Code contributions + +- Keep the **safety invariant**: `project.py` is the only module allowed to read `.knxproj`, and it + must stay read-only. No networking / bus libraries may enter the dependency tree. +- Keep generators **conservative**: when in doubt, push an item to the `review` list rather than + emitting a guessed entity. +- Run the smoke test before opening a PR: + ```bash + python tests/test_pipeline.py + ``` +- Match the existing module boundaries (one responsibility per file) and naming style. +- New classification rules should be backed by a case in `tests/test_pipeline.py`. + +## Dev setup + +```bash +pip install -e . +python tests/test_pipeline.py +``` + +CI runs the smoke test on Python 3.10–3.12 for every push and PR. + +## Code of conduct + +Be kind and constructive. This is a hobby/community project; assume good faith. + +## License + +By contributing, you agree your contributions are licensed under the [MIT License](LICENSE). diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..cc058a7 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Nikolay Miroshnichenko + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..75dca19 --- /dev/null +++ b/README.md @@ -0,0 +1,228 @@ +# nickol-knx-mcp + +**A design-time KNX / ETS6 assistant exposed as an [MCP](https://modelcontextprotocol.io) server.** + +It reads your `.knxproj`, analyzes group addresses / DPTs / topology, generates Home Assistant KNX YAML and ETS-importable group-address files (XML/CSV), and produces human-readable reports — **without ever touching the live KNX bus.** + +[![CI](https://github.com/NickoScope/nickol-knx-mcp/actions/workflows/ci.yml/badge.svg)](https://github.com/NickoScope/nickol-knx-mcp/actions/workflows/ci.yml) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) +[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/) +[![Status: beta](https://img.shields.io/badge/status-beta-orange.svg)](#-status--call-for-testers) + +🇷🇺 **Русская версия:** [README.ru.md](README.ru.md) + +--- + +## 🧪 Status & call for testers + +This is a **public beta**. The full pipeline passes an end-to-end smoke test on a synthetic +16-group-address project, but it has had **limited testing against real-world `.knxproj` files** — +and real ETS projects are wonderfully messy and diverse. + +**👉 If you have an ETS5/ETS6 project, please try it and tell us what happens.** Open a +[Real-project test report](https://github.com/NickoScope/nickol-knx-mcp/issues/new?template=real_project_test.yml) +issue. The tool is read-only and never connects to a bus, so testing is safe (see +[Safety model](#-safety-model)). See [CONTRIBUTING.md](CONTRIBUTING.md) for details. + +--- + +## Why this exists + +As of mid-2026 there is **no off-the-shelf ETS6 ↔ Claude / MCP tool**. The KNX community has +been explicitly asking for an integration that can inspect and help modify projects (adding / +renaming devices and group addresses) through an AI/CLI workflow. This package fills exactly the +**design-time** layer — the missing one. + +The recommended full setup is four layers; only one needs to be built from scratch: + +| Layer | Purpose | What to use | Build it? | +|-------|---------|-------------|-----------| +| 1. Live | states, control, debugging a running house | **official Home Assistant MCP Server** + KNX (XKNX) integration | No, already exists | +| 2. **Design-time** | parse `.knxproj`, validate DPT/naming/status, generate HA YAML & ETS XML/CSV | **`nickol-knx-mcp` (this package)** | **YES — this is the gap** | +| 3. Files + Git | YAML/CSV/XML, versioning the address schema | standard filesystem + git MCP servers | No, already exists | +| 4. Skill | design rules (GA structure, naming, DPT, scenes) | `CLAUDE.md` in this package | No, included | + +> **Safety by design:** layer 2 (this server) **physically cannot** connect to a bus. It has no +> network/bus dependency at all — it only reads `.knxproj` and writes files into a confined +> workspace. The "never write to a live bus" requirement is enforced **structurally**, not by +> promise. Any real interaction with the house goes only through layer 1 (Home Assistant). + +--- + +## What the server does + +- **Parses** password-protected ETS5/ETS6 `.knxproj` files via [`xknxproject`](https://github.com/XKNX/xknxproject) (3.9.x). +- **Extracts** group addresses, DPTs, devices, topology, descriptions, and ETS Functions. +- **Classifies** every GA: category (lighting / shutter / hvac / sensor / scene / energy / + diagnostics) and kind (command / status / sensor) — from the DPT plus multilingual (EN/DE/RU) + keywords in the name. +- **Validates naming** against a 3-level structure and a configurable regex. +- **Finds missing status addresses** — primarily from ETS Function roles, falling back to + name-token pairing (command in `…/0/…`, feedback in `…/4/…` is common, so it matches by name + tokens rather than by middle-group adjacency). +- **Catches DPT problems**: missing DPT, mismatch between a Communication Object and its GA, and + the same logical name carrying different DPTs. +- **Generates Home Assistant KNX YAML** — category by category, conservatively: covers → lights → + switches → sensors/binary. Ambiguous items (e.g. DPT 5.001 — brightness vs blind position) are + **not guessed**; they go into a `review` list instead. +- **Generates ETS-importable** group addresses in **XML** (the recommended `knx.org/xml/ga-export/01` + schema) and **CSV** (ETS's native layout). +- **Writes a Markdown report** (inventory + 🔴🟡🔵 findings + HA-mapping preview + next steps) for + human review **before** any import. + +All writes go only into the workspace directory (`NICKOL_KNX_WORKSPACE`, default `./knx-workspace`); +writes outside it are rejected. + +--- + +## Installation + +Requires **Python 3.10+**. + +```bash +git clone https://github.com/NickoScope/nickol-knx-mcp.git +cd nickol-knx-mcp +python3 -m venv .venv && source .venv/bin/activate +pip install -e . +``` + +Dependencies: `mcp>=1.10`, `xknxproject>=3.8`, `PyYAML>=6.0`. + +> On Debian/Ubuntu, if pip complains about an externally-managed environment, use a venv (as above) +> or `pip install -e . --break-system-packages`. If `PyJWT` conflicts, run +> `pip install mcp --ignore-installed PyJWT` first. + +Verify: + +```bash +python tests/test_pipeline.py # synthetic 16-GA project, end-to-end smoke test +nickol-knx-mcp # start the MCP server (stdio) +``` + +--- + +## Connecting to Claude + +### Claude Desktop + +`examples/claude_desktop_config.json` wires up nickol-knx + filesystem + git + home-assistant. +Minimal fragment (macOS config path: `~/Library/Application Support/Claude/claude_desktop_config.json`): + +```json +{ + "mcpServers": { + "nickol-knx": { + "command": "nickol-knx-mcp", + "env": { "NICKOL_KNX_WORKSPACE": "/path/to/your/knx-workspace" } + } + } +} +``` + +### Claude Code + +```bash +claude mcp add nickol-knx \ + -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" \ + -- /absolute/path/to/.venv/bin/nickol-knx-mcp +``` + +Then drop `CLAUDE.md` into your project root — it acts as an ETS Assistant skill (design rules, +safety rules, 3-level GA structure, command/status pairing, DPT discipline, naming, KNX Secure +keyring handling, and the recommended workflow). + +--- + +## MCP tools (12) + +| Tool | Purpose | +|------|---------| +| `load_project(path, password?, language?)` | parse a `.knxproj` (read-only) and cache it | +| `list_group_addresses(category?, kind?)` | list GAs with classification and filters | +| `get_devices()` | devices + their communication objects | +| `get_topology()` | topology (areas / lines / devices) | +| `check_naming(name_regex?)` | validate naming / structure | +| `check_missing_status()` | actuators lacking a status object | +| `check_dpt()` | missing / inconsistent DPTs | +| `analyze_all(name_regex?)` | run every check at once | +| `generate_ha_package(output_path?)` | HA KNX YAML + review list | +| `generate_ets_group_addresses(fmt="xml"\|"csv", output_path?)` | ETS-importable GAs | +| `project_report(output_path?, name_regex?)` | Markdown report | +| `workspace_info()` | workspace path + safety guarantees | + +--- + +## Typical workflow + +1. `load_project` → point it at your `.knxproj` (+ password if protected). +2. `analyze_all` or `project_report` → read the findings; **human review first**. +3. Fix naming/DPT/status in ETS (by importing generated GAs or manually). +4. `generate_ets_group_addresses(fmt="xml")` → import the missing GAs into ETS. +5. `generate_ha_package` → place the YAML into Home Assistant; resolve `review` items by hand. +6. Keep everything (`.knxproj` export, HA configs, address schema) in Git. +7. Touch the live house only through the Home Assistant MCP (layer 1). + +--- + +## Limitations (honest) + +- command/status and category classification is a **heuristic** (DPT + names + ETS Functions). On + messy projects with no Functions and non-standard names, false negatives/positives are possible — + which is why the report is always for human review, and ambiguity goes to `review`, not into config. +- DPT 5.001 is structurally ambiguous (brightness vs position); it's disambiguated by keywords — + double-check with non-standard naming. +- The HA generator is conservative: it would rather defer an item to `review` than emit a wrong entity. +- The server never writes to the bus and never talks to ETS directly — ETS exchange is file + import/export of GAs only. +- **Tested only on a synthetic project so far.** Real `.knxproj` files vary a lot — hence the + [call for testers](#-status--call-for-testers). + +--- + +## 🔒 Safety model + +- **No bus access, structurally.** There is no networking or bus library in the dependency tree. + `workspace_info()` reports `bus_access: false`. +- **Read-only on your project.** `project.py` is the only module that touches `.knxproj`, and it + only reads. +- **Confined writes.** All output is constrained to `NICKOL_KNX_WORKSPACE`; paths outside it are rejected. +- **Human-in-the-loop.** Generate a `project_report` and review it **before** importing into ETS or + deploying into Home Assistant. + +Found a security issue? See [SECURITY.md](SECURITY.md). + +--- + +## Package layout + +``` +nickol-knx-mcp/ +├── nickol_knx_mcp/ +│ ├── dpt_map.py # DPT → category / kind / HA platform / value_type +│ ├── project.py # the ONLY module that reads .knxproj (read-only) +│ ├── pairing.py # command↔status pairing by name tokens +│ ├── analyze.py # naming / missing-status / DPT checks +│ ├── generate_ha.py # Home Assistant KNX YAML generation +│ ├── generate_ets.py # ETS XML + CSV generation +│ ├── report.py # Markdown report +│ └── server.py # FastMCP server, 12 tools, confined writes +├── tests/test_pipeline.py +├── examples/claude_desktop_config.json +├── CLAUDE.md # ETS Assistant skill / playbook +├── pyproject.toml +└── README.md +``` + +--- + +## Contributing + +Testers and contributors are very welcome — especially **real-project test reports**. See +[CONTRIBUTING.md](CONTRIBUTING.md) and the [issue templates](.github/ISSUE_TEMPLATE). + +## License + +[MIT](LICENSE) © 2026 Nikolay Miroshnichenko + +> Not affiliated with or endorsed by the KNX Association. "KNX" and "ETS" are trademarks of the +> KNX Association cc. This is an independent, community tool. diff --git a/README.ru.md b/README.ru.md new file mode 100644 index 0000000..1f62075 --- /dev/null +++ b/README.ru.md @@ -0,0 +1,162 @@ +> 🌍 **English version:** [README.md](README.md) · Русская версия ниже. + +# nickol-knx-mcp + +**Design-time ассистент KNX/ETS6 в виде MCP-сервера.** +Читает `.knxproj`, анализирует group addresses / DPT / топологию, генерирует Home Assistant KNX YAML и ETS-импортируемые group-address файлы (XML/CSV), делает человекочитаемые отчёты — **никогда не подключаясь к живой шине KNX.** + +> ⚠️ **Статус: BETA.** Сервис проверен на синтетическом проекте (16 GA) и проходит end-to-end тест, но **на реальных `.knxproj` пока тестировался ограниченно**. Нужны тестировщики — см. [CONTRIBUTING.md](CONTRIBUTING.md). + +--- + +## 1. Зачем это и где оно в общей схеме + +На июнь 2026 готового официального **ETS6 ↔ Claude / MCP** инструмента не существует. +KNX Community в мае 2026 прямо просит такую интеграцию (изменение проектов, добавление/переименование устройств и group addresses через Claude/CLI). Этот пакет закрывает именно **design-time** слой — самый недостающий. + +Полная рекомендованная схема — четыре слоя, и собирать с нуля нужно только один: + +| Слой | Назначение | Что использовать | Собирать? | +|------|-----------|------------------|-----------| +| 1. Live | состояния, управление, отладка автоматизаций живого дома | **официальный Home Assistant MCP Server** + KNX (XKNX) integration | нет, уже есть | +| 2. **Design-time** | парсинг `.knxproj`, проверка DPT/именования/статусов, генерация HA YAML и ETS XML/CSV | **`nickol-knx-mcp` (этот пакет)** | **ДА — это и есть пробел** | +| 3. Files + Git | YAML/CSV/XML, версионирование схемы адресов | стандартные filesystem + git MCP | нет, уже есть | +| 4. Skill | правила проектирования (структура GA, naming, DPT, сцены) | `CLAUDE.md` из этого пакета | нет, готов | + +> **Принцип безопасности:** слой 2 (этот сервер) **физически не умеет** подключаться к шине. У него нет ни одной сетевой/bus-зависимости — только чтение `.knxproj` и запись файлов в изолированный workspace. Требование «никогда не писать в живую шину» выполнено **структурно**, а не «честным словом». Любое реальное взаимодействие с домом идёт только через слой 1 (Home Assistant). + +--- + +## 2. Что умеет сервер + +- **Парсит** запароленные ETS5/ETS6 `.knxproj` через `xknxproject` (3.9.x). +- **Извлекает** group addresses, DPT, устройства, топологию, описания, функции (Functions). +- **Классифицирует** каждый GA: категория (lighting / shutter / hvac / sensor / scene / energy / diagnostics) и вид (command / status / sensor) — по DPT + многоязычным (EN/DE/RU) ключевым словам в имени. +- **Проверяет именование** по 3-уровневой структуре и регэкспу. +- **Находит отсутствующие статусные адреса** — приоритетно по ролям из ETS Functions, как fallback — по парности имён (token-overlap). +- **Ловит проблемы DPT**: отсутствующий DPT, рассогласование DPT между Communication Object и GA, одинаковое имя с разными DPT. +- **Генерирует Home Assistant KNX YAML** — категорийно, консервативно. Неоднозначные элементы уходят в список `review`, а не угадываются вслепую. +- **Генерирует ETS-импортируемые** group addresses: **XML** (схема `knx.org/xml/ga-export/01`) и **CSV** (родная раскладка ETS). +- **Пишет Markdown-отчёт** (инвентаризация + находки 🔴🟡🔵 + превью HA-маппинга + следующие шаги). + +Все записи идут только в каталог workspace (`NICKOL_KNX_WORKSPACE`, по умолчанию `./knx-workspace`); запись за его пределы отклоняется. + +--- + +## 3. Установка + +Требуется Python 3.10+. + +```bash +git clone https://github.com/NickoScope/nickol-knx-mcp.git +cd nickol-knx-mcp +python3 -m venv .venv && source .venv/bin/activate +pip install -e . +``` + +Зависимости: `mcp>=1.10`, `xknxproject>=3.8`, `PyYAML>=6.0`. + +> Если на Debian/Ubuntu система ругается на externally-managed окружение — используйте venv, либо `pip install -e . --break-system-packages`. При конфликте `PyJWT` помогает `pip install mcp --ignore-installed PyJWT`. + +Проверка: + +```bash +python tests/test_pipeline.py # синтетический проект из 16 GA, end-to-end smoke test +nickol-knx-mcp # запустить MCP-сервер (stdio) +``` + +--- + +## 4. Подключение к Claude + +### Claude Desktop + +`examples/claude_desktop_config.json` уже сводит вместе nickol-knx + filesystem + git + home-assistant. Минимальный фрагмент: + +```json +{ + "mcpServers": { + "nickol-knx": { + "command": "nickol-knx-mcp", + "env": { "NICKOL_KNX_WORKSPACE": "/path/to/your/knx-workspace" } + } + } +} +``` + +### Claude Code + +```bash +claude mcp add nickol-knx -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" -- /abs/path/to/.venv/bin/nickol-knx-mcp +``` + +Положите `CLAUDE.md` в корень проекта — он работает как ETS Assistant skill (правила проектирования, safety-rules, 3-уровневая структура GA, command/status, DPT-дисциплина, naming, KNX Secure keyring, рабочий процесс). + +--- + +## 5. Инструменты MCP (12) + +| Инструмент | Назначение | +|-----------|-----------| +| `load_project(path, password?, language?)` | загрузить и распарсить `.knxproj` (read-only), закэшировать | +| `list_group_addresses(category?, kind?)` | список GA с классификацией, фильтры | +| `get_devices()` | устройства + их communication objects | +| `get_topology()` | топология (areas / lines / devices) | +| `check_naming(name_regex?)` | проверка именования/структуры | +| `check_missing_status()` | актуаторы без статусного объекта | +| `check_dpt()` | отсутствующие/несогласованные DPT | +| `analyze_all(name_regex?)` | все проверки разом | +| `generate_ha_package(output_path?)` | HA KNX YAML + список review | +| `generate_ets_group_addresses(fmt="xml"\|"csv", output_path?)` | ETS-импортируемые GA | +| `project_report(output_path?, name_regex?)` | Markdown-отчёт | +| `workspace_info()` | путь и содержимое workspace | + +--- + +## 6. Типовой рабочий процесс + +1. `load_project` → указать `.knxproj` (+ пароль, если запаролен). +2. `analyze_all` или `project_report` → прочитать находки, **сначала ревью человеком**. +3. Исправить именование/DPT/статусы в ETS (импортом сгенерированных GA или вручную). +4. `generate_ets_group_addresses(fmt="xml")` → импортировать в ETS как недостающие GA. +5. `generate_ha_package` → положить YAML в Home Assistant; разобрать `review`-элементы руками. +6. Всё (экспорт `.knxproj`, HA-конфиги, схема адресов) держать в Git. +7. Живой дом — только через Home Assistant MCP (слой 1). + +--- + +## 7. Ограничения (честно) + +- Классификация command/status и категорий — **эвристика** (DPT + имена + ETS Functions). На «грязных» проектах без Functions и с нестандартными именами возможны пропуски/ложные срабатывания — поэтому отчёт всегда для ревью человеком. +- DPT 5.001 структурно неоднозначен (яркость vs позиция) — разводится по ключевым словам; при нестандартном нейминге проверяйте вручную. +- Генератор HA консервативен: лучше отдать элемент в review, чем сгенерировать неверную сущность. +- Сервер не пишет в шину и не подключается к ETS напрямую — обмен с ETS только через файловый импорт/экспорт GA. +- **Тест пока только на синтетическом проекте.** Реальные `.knxproj` очень разнообразны — поэтому и нужны тестировщики. + +--- + +## 8. Структура пакета + +``` +nickol-knx-mcp/ +├── nickol_knx_mcp/ +│ ├── dpt_map.py # DPT → категория/вид/HA-платформа/value_type +│ ├── project.py # ЕДИНСТВЕННЫЙ модуль, читающий .knxproj (read-only) +│ ├── pairing.py # парность command↔status по токенам имени +│ ├── analyze.py # naming / missing-status / DPT проверки +│ ├── generate_ha.py # генерация HA KNX YAML +│ ├── generate_ets.py # генерация ETS XML + CSV +│ ├── report.py # Markdown-отчёт +│ └── server.py # FastMCP сервер, 12 инструментов, confined writes +├── tests/test_pipeline.py +├── examples/claude_desktop_config.json +├── CLAUDE.md # ETS Assistant skill / playbook +├── pyproject.toml +└── README.md +``` + +--- + +## Лицензия + +[MIT](LICENSE) © 2026 Nikolay Miroshnichenko diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..41b97cd --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,29 @@ +# Security Policy + +## The safety model + +`nickol-knx-mcp` is a **design-time** tool with a deliberately small attack surface: + +- **No bus access.** There is no KNX/IP or other networking/bus library in the dependency tree. + The server cannot reach a live KNX installation. `workspace_info()` reports `bus_access: false`. +- **Read-only on `.knxproj`.** Only `project.py` reads the project, and it never writes to it. +- **Confined writes.** All generated files are constrained to the `NICKOL_KNX_WORKSPACE` directory; + writes outside it are rejected. + +## Handling project data + +A `.knxproj` and an ETS keyring (`.knxkeys`) can contain sensitive information (topology, device +addresses, secure keys). This tool reads the project locally and writes only into your workspace — +nothing is uploaded anywhere. **Do not commit real `.knxproj` / `.knxkeys` files** to a public +repository; the provided `.gitignore` excludes them by default. + +## Reporting a vulnerability + +If you find a security issue (e.g. a path-escape past the workspace confinement, or any way the +server could touch a bus), please **do not open a public issue**. Instead use GitHub's +[private vulnerability reporting](https://github.com/NickoScope/nickol-knx-mcp/security/advisories/new) +for this repository. We'll acknowledge within a reasonable time and coordinate a fix and disclosure. + +## Supported versions + +This is a beta; security fixes target the latest `main` and the most recent release. diff --git a/examples/claude_desktop_config.json b/examples/claude_desktop_config.json new file mode 100644 index 0000000..390bffa --- /dev/null +++ b/examples/claude_desktop_config.json @@ -0,0 +1,25 @@ +{ + "_comment": "Full KNX toolchain. Layer 1 (live HA) is optional and only needed once the house is running. Layers 2-4 are the design-time core. macOS path for Claude Desktop config: ~/Library/Application Support/Claude/claude_desktop_config.json. For Claude Code, put the mcpServers block in .mcp.json at your project root.", + "mcpServers": { + "nickol-knx": { + "command": "uv", + "args": ["--directory", "/ABSOLUTE/PATH/TO/nickol-knx-mcp", "run", "nickol-knx-mcp"], + "env": { + "NICKOL_KNX_WORKSPACE": "/ABSOLUTE/PATH/TO/your-knx-repo" + } + }, + "filesystem": { + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", "/ABSOLUTE/PATH/TO/your-knx-repo"] + }, + "git": { + "command": "uvx", + "args": ["mcp-server-git", "--repository", "/ABSOLUTE/PATH/TO/your-knx-repo"] + }, + "home-assistant": { + "_comment": "Optional live layer. Official HA MCP Server add-on exposes an SSE endpoint; point an SSE-capable client at http://HA_HOST:8123/mcp_server/sse with a long-lived token. Replace below with your client's SSE config if it differs.", + "url": "http://HA_HOST:8123/mcp_server/sse", + "headers": { "Authorization": "Bearer YOUR_LONG_LIVED_TOKEN" } + } + } +} diff --git a/nickol_knx_mcp/__init__.py b/nickol_knx_mcp/__init__.py new file mode 100644 index 0000000..c4ba04d --- /dev/null +++ b/nickol_knx_mcp/__init__.py @@ -0,0 +1,2 @@ +"""nickol-knx-mcp: design-time KNX/ETS project assistant MCP server.""" +__version__ = "0.1.0" diff --git a/nickol_knx_mcp/analyze.py b/nickol_knx_mcp/analyze.py new file mode 100644 index 0000000..86e6417 --- /dev/null +++ b/nickol_knx_mcp/analyze.py @@ -0,0 +1,253 @@ +"""Validation and analysis passes over a LoadedProject. + +Three independent checks, each returning a list of structured findings: + * validate_naming - 3-level structure, empty/duplicate names, missing DPT + * detect_missing_status - command GAs lacking a status/feedback counterpart + * detect_dpt_issues - missing, inconsistent or mismatched DPTs + +Findings are plain dicts so they serialize straight to JSON for the MCP client. +""" + +from __future__ import annotations + +import re +from collections import defaultdict +from typing import Any, Optional + +from .project import LoadedProject, GARecord, STATUS_KEYWORDS +from .pairing import find_status + +SEVERITY_ERROR = "error" +SEVERITY_WARN = "warning" +SEVERITY_INFO = "info" + + +def _finding(severity: str, code: str, address: str, message: str, + **extra: Any) -> dict[str, Any]: + f = {"severity": severity, "code": code, "address": address, "message": message} + f.update(extra) + return f + + +# --------------------------------------------------------------------------- # +# Naming validation +# --------------------------------------------------------------------------- # +def validate_naming(project: LoadedProject, + name_regex: Optional[str] = None, + min_name_len: int = 3) -> list[dict[str, Any]]: + """Check naming conventions and 3-level structure.""" + findings: list[dict[str, Any]] = [] + + style = (project.style or "").lower() + if "three" not in style: + findings.append(_finding( + SEVERITY_WARN, "ga_style_not_three_level", "-", + f"Group address style is '{project.style or 'unknown'}', expected ThreeLevel. " + "The agreed convention is a 3-level Main/Middle/Sub structure.", + )) + + pattern = re.compile(name_regex) if name_regex else None + seen_names: dict[str, list[str]] = defaultdict(list) + + for addr, ga in project.gas.items(): + name = ga.name.strip() + if not name: + findings.append(_finding( + SEVERITY_ERROR, "empty_name", addr, + "Group address has no name.", + )) + continue + + seen_names[name.lower()].append(addr) + + if len(name) < min_name_len: + findings.append(_finding( + SEVERITY_WARN, "name_too_short", addr, + f"Name '{name}' is shorter than {min_name_len} characters.", + name=name, + )) + + if pattern and not pattern.search(name): + findings.append(_finding( + SEVERITY_WARN, "name_pattern_mismatch", addr, + f"Name '{name}' does not match the required pattern.", + name=name, + )) + + if ga.main is None: + findings.append(_finding( + SEVERITY_WARN, "not_three_level_address", addr, + f"Address '{addr}' is not a 3-level address.", + )) + + for low, addrs in seen_names.items(): + if len(addrs) > 1: + findings.append(_finding( + SEVERITY_WARN, "duplicate_name", ", ".join(addrs), + f"Name used by {len(addrs)} group addresses: {addrs}.", + addresses=addrs, + )) + + # Main-group names present? + main_named: dict[int, str] = {} + for ga in project.gas.values(): + if ga.main is not None and ga.main not in main_named: + main_named[ga.main] = ga.main_name + for main, mname in sorted(main_named.items()): + if not mname: + findings.append(_finding( + SEVERITY_INFO, "main_group_unnamed", f"{main}/-/-", + f"Main group {main} has no descriptive name.", + )) + + return findings + + +# --------------------------------------------------------------------------- # +# Missing status detection +# --------------------------------------------------------------------------- # +def _function_role_status(project: LoadedProject) -> list[dict[str, Any]]: + """Use ETS Functions (GA roles) as the primary signal. + + A function that has at least one command-role GA but no status/info-role GA + is flagged. Role strings vary by manufacturer, so we match generously. + """ + findings: list[dict[str, Any]] = [] + status_tokens = ("info", "status", "state", "feedback", "rueck", "rück") + for fid, fn in project.functions.items(): + roles = fn.get("group_addresses", {}) or {} + if not roles: + continue + has_command = False + has_status = False + cmd_addrs: list[str] = [] + for ga_addr, ref in roles.items(): + role = (ref.get("role") or "").lower() + addr = ref.get("address", ga_addr) + if any(t in role for t in status_tokens): + has_status = True + else: + # treat non-status roles that look writable as commands + has_command = True + cmd_addrs.append(addr) + if has_command and not has_status: + findings.append(_finding( + SEVERITY_WARN, "function_missing_status", + ", ".join(cmd_addrs) or "-", + f"Function '{fn.get('name', fid)}' ({fn.get('function_type', '?')}) " + "has command GAs but no status/feedback GA.", + function=fn.get("name", fid), + addresses=cmd_addrs, + )) + return findings + + +def _is_status_ga(ga: GARecord) -> bool: + if ga.kind == "status": + return True + low = ga.name.lower() + return any(k in low for k in STATUS_KEYWORDS) + + +# command DPT main -> acceptable status DPT mains +_STATUS_COMPAT = { + 1: {1}, # switch command -> 1.x status + 3: {1, 5}, # relative dim -> on/off or brightness status + 5: {5}, # scaling setpoint -> scaling status + 9: {9}, # float setpoint -> float status +} + + +def detect_missing_status(project: LoadedProject) -> list[dict[str, Any]]: + """Detect controllable GAs that lack a status counterpart. + + Strategy: + 1. ETS Functions roles (authoritative when present). + 2. Heuristic per middle-group sibling search for command GAs not covered + by any function. + """ + findings = _function_role_status(project) + + # addresses already explained by a function finding -> skip in heuristic + covered: set[str] = set() + for f in findings: + covered.update(f.get("addresses", [])) + + # All status-like GAs become pairing candidates (project-wide). + status_gas = [ga for ga in project.gas.values() if _is_status_ga(ga)] + + for addr, ga in project.gas.items(): + if addr in covered: + continue + if ga.kind != "command": + continue + if ga.dpt_main is None: + continue # DPT issue handled elsewhere + # prefer same-main-group candidates, fall back to whole project + same_main = [s for s in status_gas if s.main == ga.main] + match = find_status(ga, same_main) or find_status(ga, status_gas) + if match is None: + findings.append(_finding( + SEVERITY_WARN, "missing_status_address", addr, + f"Command '{ga.name}' (DPT {ga.dpt or '?'}, {ga.label}) has no " + "status/feedback GA. Home Assistant cannot read real state.", + name=ga.name, dpt=ga.dpt, category=ga.category, + )) + + return findings + + +# --------------------------------------------------------------------------- # +# DPT consistency +# --------------------------------------------------------------------------- # +def detect_dpt_issues(project: LoadedProject) -> list[dict[str, Any]]: + findings: list[dict[str, Any]] = [] + + # 1. missing DPT + for addr, ga in project.gas.items(): + if ga.dpt_main is None: + findings.append(_finding( + SEVERITY_ERROR, "missing_dpt", addr, + f"'{ga.name}' has no DPT assigned. Home Assistant requires a DPT " + "to decode this group address.", + name=ga.name, + )) + + # 2. CO <-> GA dpt mismatch + cos = project.raw.get("communication_objects", {}) + for addr, ga in project.gas.items(): + if ga.dpt_main is None: + continue + for co_id in ga.co_ids: + co = cos.get(co_id) + if not co: + continue + co_dpts = co.get("dpts") or [] + if not co_dpts: + continue + mains = {d.get("main") for d in co_dpts} + if ga.dpt_main not in mains: + findings.append(_finding( + SEVERITY_WARN, "dpt_mismatch_co", addr, + f"GA DPT main {ga.dpt_main} differs from linked communication " + f"object '{co.get('name','?')}' DPT main(s) {sorted(m for m in mains if m is not None)}.", + name=ga.name, + )) + break + + # 3. same normalized name, different DPT + by_name: dict[str, list[GARecord]] = defaultdict(list) + for ga in project.gas.values(): + if ga.name.strip(): + by_name[ga.name.strip().lower()].append(ga) + for low, recs in by_name.items(): + dpts = {r.dpt for r in recs if r.dpt} + if len(dpts) > 1: + findings.append(_finding( + SEVERITY_WARN, "inconsistent_dpt", ", ".join(r.address for r in recs), + f"Group addresses sharing name '{recs[0].name}' use different DPTs: " + f"{sorted(dpts)}.", + addresses=[r.address for r in recs], dpts=sorted(dpts), + )) + + return findings diff --git a/nickol_knx_mcp/dpt_map.py b/nickol_knx_mcp/dpt_map.py new file mode 100644 index 0000000..9bbd9e9 --- /dev/null +++ b/nickol_knx_mcp/dpt_map.py @@ -0,0 +1,123 @@ +"""DPT classification and Home Assistant KNX platform mapping. + +Central knowledge base translating a KNX Datapoint Type into: + * a functional *category* (lighting / shutter / hvac / sensor / scene / energy / diagnostics) + * a *kind* (command vs status vs sensor) + * a suggested Home Assistant KNX platform + value_type + +Everything here is heuristic but conservative: when in doubt we mark the +datapoint for manual review rather than guessing an entity wrong. +""" + +from __future__ import annotations + +from typing import Optional, TypedDict + + +class DptInfo(TypedDict): + category: str # lighting|shutter|hvac|sensor|scene|energy|diagnostics|unknown + kind: str # command|status|sensor|unknown + ha_platform: str # switch|light|cover|sensor|binary_sensor|climate|number|scene|button|unknown + value_type: Optional[str] # HA value_type for sensor/number, None otherwise + label: str # human-readable description + + +CATEGORY_LIGHTING = "lighting" +CATEGORY_SHUTTER = "shutter" +CATEGORY_HVAC = "hvac" +CATEGORY_SENSOR = "sensor" +CATEGORY_SCENE = "scene" +CATEGORY_ENERGY = "energy" +CATEGORY_DIAG = "diagnostics" +CATEGORY_UNKNOWN = "unknown" + +KIND_COMMAND = "command" +KIND_STATUS = "status" +KIND_SENSOR = "sensor" +KIND_UNKNOWN = "unknown" + + +def dpt_key(main: Optional[int], sub: Optional[int]) -> str: + """Return a normalized 'main.sub' / 'main' string, '' if no DPT.""" + if main is None: + return "" + if sub is None: + return str(main) + return f"{main}.{sub:03d}" + + +def dpt_ets_token(main: Optional[int], sub: Optional[int]) -> str: + """ETS GA-export 'DPTs' attribute token, e.g. DPST-1-1 or DPT-1.""" + if main is None: + return "" + if sub is None: + return f"DPT-{main}" + return f"DPST-{main}-{sub}" + + +# Exact (main, sub) overrides. None sub = applies to whole main group as fallback. +_EXACT: dict[tuple[int, Optional[int]], DptInfo] = { + (1, 1): {"category": CATEGORY_LIGHTING, "kind": KIND_COMMAND, "ha_platform": "switch", "value_type": None, "label": "Switch on/off"}, + (1, 2): {"category": CATEGORY_UNKNOWN, "kind": KIND_COMMAND, "ha_platform": "switch", "value_type": None, "label": "Boolean"}, + (1, 3): {"category": CATEGORY_UNKNOWN, "kind": KIND_COMMAND, "ha_platform": "switch", "value_type": None, "label": "Enable"}, + (1, 5): {"category": CATEGORY_DIAG, "kind": KIND_SENSOR, "ha_platform": "binary_sensor", "value_type": None, "label": "Alarm"}, + (1, 8): {"category": CATEGORY_SHUTTER, "kind": KIND_COMMAND, "ha_platform": "cover", "value_type": None, "label": "Up/Down"}, + (1, 9): {"category": CATEGORY_UNKNOWN, "kind": KIND_SENSOR, "ha_platform": "binary_sensor", "value_type": None, "label": "Open/Close"}, + (1, 10): {"category": CATEGORY_SHUTTER, "kind": KIND_COMMAND, "ha_platform": "cover", "value_type": None, "label": "Start/Stop"}, + (1, 11): {"category": CATEGORY_LIGHTING, "kind": KIND_STATUS, "ha_platform": "binary_sensor", "value_type": None, "label": "State (status)"}, + (1, 18): {"category": CATEGORY_DIAG, "kind": KIND_SENSOR, "ha_platform": "binary_sensor", "value_type": None, "label": "Occupancy"}, + (1, 19): {"category": CATEGORY_DIAG, "kind": KIND_SENSOR, "ha_platform": "binary_sensor", "value_type": None, "label": "Window/Door"}, + (3, 7): {"category": CATEGORY_LIGHTING, "kind": KIND_COMMAND, "ha_platform": "light", "value_type": None, "label": "Dimming control (relative)"}, + (3, 8): {"category": CATEGORY_SHUTTER, "kind": KIND_COMMAND, "ha_platform": "cover", "value_type": None, "label": "Blinds control (relative)"}, + (5, 1): {"category": CATEGORY_LIGHTING, "kind": KIND_COMMAND, "ha_platform": "light", "value_type": "percent", "label": "Scaling 0-100% (brightness/position)"}, + (5, 3): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "angle", "label": "Angle 0-360°"}, + (5, 10): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "pulse", "label": "Counter pulses"}, + (7, 12): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "electric_current", "label": "Current (mA)"}, + (7, 13): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "illuminance", "label": "Brightness (lux)"}, + (9, 1): {"category": CATEGORY_HVAC, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "temperature", "label": "Temperature (°C)"}, + (9, 4): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "illuminance", "label": "Illuminance (lux)"}, + (9, 5): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "wind_speed_ms", "label": "Wind speed (m/s)"}, + (9, 7): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "humidity", "label": "Humidity (%)"}, + (9, 8): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "ppm", "label": "Air quality (ppm)"}, + (9, 20): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "voltage", "label": "Voltage (mV)"}, + (9, 21): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "current", "label": "Current (mA)"}, + (12, 1): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "pulse_4_byte", "label": "Counter (4-byte unsigned)"}, + (13, 10): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "active_energy", "label": "Active energy (Wh)"}, + (13, 13): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "active_energy_kwh", "label": "Active energy (kWh)"}, + (14, 19): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "electric_current", "label": "Electric current (A)"}, + (14, 27): {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "temperature", "label": "Temperature (K → °C)"}, + (14, 56): {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "power", "label": "Power (W)"}, + (17, 1): {"category": CATEGORY_SCENE, "kind": KIND_COMMAND, "ha_platform": "scene", "value_type": None, "label": "Scene number"}, + (18, 1): {"category": CATEGORY_SCENE, "kind": KIND_COMMAND, "ha_platform": "scene", "value_type": None, "label": "Scene control"}, + (20, 102): {"category": CATEGORY_HVAC, "kind": KIND_COMMAND, "ha_platform": "climate", "value_type": None, "label": "HVAC mode"}, +} + +# Whole-main-group fallbacks when an exact (main, sub) is not known. +_MAIN_FALLBACK: dict[int, DptInfo] = { + 1: {"category": CATEGORY_UNKNOWN, "kind": KIND_COMMAND, "ha_platform": "switch", "value_type": None, "label": "1-bit boolean"}, + 3: {"category": CATEGORY_LIGHTING, "kind": KIND_COMMAND, "ha_platform": "light", "value_type": None, "label": "4-bit relative control"}, + 5: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "1byte_unsigned", "label": "1-byte unsigned"}, + 6: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "1byte_signed", "label": "1-byte signed"}, + 7: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "2byte_unsigned", "label": "2-byte unsigned"}, + 8: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "2byte_signed", "label": "2-byte signed"}, + 9: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "2byte_float", "label": "2-byte float"}, + 12: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "4byte_unsigned", "label": "4-byte unsigned"}, + 13: {"category": CATEGORY_ENERGY, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "4byte_signed", "label": "4-byte signed"}, + 14: {"category": CATEGORY_SENSOR, "kind": KIND_SENSOR, "ha_platform": "sensor", "value_type": "4byte_float", "label": "4-byte float"}, + 17: {"category": CATEGORY_SCENE, "kind": KIND_COMMAND, "ha_platform": "scene", "value_type": None, "label": "Scene number"}, + 18: {"category": CATEGORY_SCENE, "kind": KIND_COMMAND, "ha_platform": "scene", "value_type": None, "label": "Scene control"}, + 20: {"category": CATEGORY_HVAC, "kind": KIND_COMMAND, "ha_platform": "climate", "value_type": None, "label": "1-byte HVAC enum"}, +} + + +def classify_dpt(main: Optional[int], sub: Optional[int]) -> DptInfo: + """Classify a DPT into category / kind / HA platform.""" + if main is None: + return {"category": CATEGORY_UNKNOWN, "kind": KIND_UNKNOWN, + "ha_platform": "unknown", "value_type": None, "label": "No DPT assigned"} + if (main, sub) in _EXACT: + return dict(_EXACT[(main, sub)]) # copy + if main in _MAIN_FALLBACK: + return dict(_MAIN_FALLBACK[main]) + return {"category": CATEGORY_UNKNOWN, "kind": KIND_UNKNOWN, + "ha_platform": "unknown", "value_type": None, "label": f"DPT {main}"} diff --git a/nickol_knx_mcp/generate_ets.py b/nickol_knx_mcp/generate_ets.py new file mode 100644 index 0000000..f5256e0 --- /dev/null +++ b/nickol_knx_mcp/generate_ets.py @@ -0,0 +1,127 @@ +"""Generate ETS-importable Group Address exports (CSV + XML). + +XML uses the official ETS GA-export schema (http://knx.org/xml/ga-export/01), +which is the most reliable import path. CSV uses the native ETS group-address +layout (group name + X/Y/Z address with -/- range placeholders). + +Neither export touches a live bus. The intended flow is: generate -> review the +human-readable report -> import into ETS via 'Import Group Addresses'. +""" + +from __future__ import annotations + +import csv +import io +from typing import Any, Optional +from xml.sax.saxutils import escape + +from .project import LoadedProject +from .dpt_map import dpt_ets_token + + +def _hierarchy(project: LoadedProject) -> dict[int, dict[str, Any]]: + """Build {main: {name, middles: {middle: {name, gas: [rec]}}}}. + + Range names come from raw group_ranges when available, otherwise synthesized. + """ + # raw range names indexed by (main) and (main, middle) + main_names: dict[int, str] = {} + mid_names: dict[tuple[int, int], str] = {} + + def walk(rng: dict[str, Any]) -> None: + start = rng.get("address_start") + name = rng.get("name", "") + if isinstance(start, int): + main = (start >> 11) & 0x1F + middle = (start >> 8) & 0x07 + if start % 2048 == 0 and not rng.get("group_addresses"): + main_names.setdefault(main, name) + else: + mid_names.setdefault((main, middle), name) + for child in rng.get("group_ranges", {}).values(): + walk(child) + + for rng in project.raw.get("group_ranges", {}).values(): + walk(rng) + + tree: dict[int, dict[str, Any]] = {} + for ga in project.gas.values(): + if ga.main is None or ga.middle is None: + continue + m = tree.setdefault(ga.main, { + "name": main_names.get(ga.main) or ga.main_name or f"Main group {ga.main}", + "middles": {}, + }) + mid = m["middles"].setdefault(ga.middle, { + "name": mid_names.get((ga.main, ga.middle)) or ga.middle_name + or f"Middle group {ga.main}/{ga.middle}", + "gas": [], + }) + mid["gas"].append(ga) + return tree + + +def _security(ga) -> str: + return "On" if ga.data_secure else "Auto" + + +def generate_ets_csv(project: LoadedProject) -> str: + """Native ETS GA CSV (semicolon-separated, with range placeholder rows).""" + buf = io.StringIO() + w = csv.writer(buf, delimiter=";", quoting=csv.QUOTE_ALL, lineterminator="\n") + w.writerow(["Group name", "Address", "Central", "Unfiltered", + "Description", "DatapointType", "Security"]) + tree = _hierarchy(project) + for main in sorted(tree): + m = tree[main] + w.writerow([m["name"], f"{main}/-/-", "", "", "", "", ""]) + for middle in sorted(m["middles"]): + mid = m["middles"][middle] + w.writerow([mid["name"], f"{main}/{middle}/-", "", "", "", "", ""]) + for ga in sorted(mid["gas"], key=lambda g: g.sub or 0): + w.writerow([ + ga.name, ga.address, "", "", + ga.description or ga.comment or "", + dpt_ets_token(ga.dpt_main, ga.dpt_sub), + _security(ga), + ]) + return buf.getvalue() + + +def generate_ets_xml(project: LoadedProject) -> str: + """ETS GA-export/01 XML.""" + tree = _hierarchy(project) + lines = [ + '', + '', + ] + for main in sorted(tree): + m = tree[main] + m_start = main << 11 + m_end = m_start | 0x7FF + lines.append( + f' ' + ) + for middle in sorted(m["middles"]): + mid = m["middles"][middle] + mid_start = (main << 11) | (middle << 8) + mid_end = mid_start | 0xFF + lines.append( + f' ' + ) + for ga in sorted(mid["gas"], key=lambda g: g.sub or 0): + dpts = dpt_ets_token(ga.dpt_main, ga.dpt_sub) + dpt_attr = f' DPTs="{dpts}"' if dpts else "" + desc = ga.description or ga.comment or "" + desc_attr = f' Description="{escape(desc, {chr(34): """})}"' if desc else "" + sec_attr = ' Security="On"' if ga.data_secure else "" + lines.append( + f' ' + ) + lines.append(' ') + lines.append(' ') + lines.append('') + return "\n".join(lines) + "\n" diff --git a/nickol_knx_mcp/generate_ha.py b/nickol_knx_mcp/generate_ha.py new file mode 100644 index 0000000..65d641e --- /dev/null +++ b/nickol_knx_mcp/generate_ha.py @@ -0,0 +1,159 @@ +"""Generate a Home Assistant KNX package (YAML) from the parsed project. + +Conservative by design: emits switch / binary_sensor / sensor / light / cover +entities only when command+status pairing is reasonably certain, and routes +everything ambiguous (climate, scenes, multi-GA fixtures) to a 'review' list the +caller surfaces in the report. Pairing is name-token based (see pairing.py), so +status GAs in a separate middle group are still matched. +""" + +from __future__ import annotations + +from typing import Any + +import yaml + +from .project import LoadedProject, GARecord +from .analyze import _is_status_ga +from .pairing import find_status, base_tokens + + +def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: + """Return {'yaml': str, 'review': [...], 'counts': {...}}.""" + status_gas = [g for g in project.gas.values() if _is_status_ga(g)] + + def status_for(cmd: GARecord): + same_main = [s for s in status_gas if s.main == cmd.main] + return find_status(cmd, same_main) or find_status(cmd, status_gas) + + def same_main_gas(cmd: GARecord): + return [g for g in project.gas.values() if g.main == cmd.main] + + switches: list[dict] = [] + binary_sensors: list[dict] = [] + sensors: list[dict] = [] + lights: list[dict] = [] + covers: list[dict] = [] + review: list[dict[str, Any]] = [] + consumed: set[str] = set() + + # ---- 1. COVERS first (they own 5.001 position) ---- + for ga in project.gas.values(): + if ga.address in consumed: + continue + if ga.category == "shutter" and ga.dpt_main == 1 and ga.dpt_sub == 8: + entity = {"name": ga.name, "move_long_address": ga.address} + for sib in same_main_gas(ga): + if sib.address in consumed or sib.address == ga.address: + continue + if sib.category != "shutter": + continue + if sib.dpt_main == 1 and sib.dpt_sub == 10: + entity["move_short_address"] = sib.address + consumed.add(sib.address) + elif sib.dpt_main == 5 and sib.kind == "command": + entity["position_address"] = sib.address + consumed.add(sib.address) + elif sib.dpt_main == 5 and _is_status_ga(sib): + entity["position_state_address"] = sib.address + consumed.add(sib.address) + covers.append(entity) + consumed.add(ga.address) + review.append({"reason": "verify_cover_mapping", "address": ga.address, + "name": ga.name}) + + # ---- 2. LIGHTS (brightness 5.001 command, lighting category) ---- + for ga in project.gas.values(): + if ga.address in consumed: + continue + if ga.category == "lighting" and ga.dpt_main == 5 and ga.kind == "command": + entity = {"name": ga.name, "brightness_address": ga.address} + st = status_for(ga) + if st and st.dpt_main == 5: + entity["brightness_state_address"] = st.address + consumed.add(st.address) + for sib in same_main_gas(ga): + if sib.address in consumed: + continue + if sib.category == "lighting" and sib.dpt_main == 1 and sib.kind == "command" \ + and len(base_tokens(ga.name) & base_tokens(sib.name)) >= 2: + entity["address"] = sib.address + s2 = status_for(sib) + if s2 and s2.dpt_main == 1: + entity["state_address"] = s2.address + consumed.add(s2.address) + consumed.add(sib.address) + break + lights.append(entity) + consumed.add(ga.address) + + # ---- 3. SWITCHES (1.001 command) ---- + for ga in project.gas.values(): + if ga.address in consumed: + continue + if ga.dpt_main == 1 and ga.dpt_sub == 1 and ga.kind == "command" \ + and ga.category in ("lighting", "unknown"): + if not ga.name.strip(): + review.append({"reason": "switch_unnamed", "address": ga.address, "name": ""}) + consumed.add(ga.address) + continue + entity = {"name": ga.name, "address": ga.address} + st = status_for(ga) + if st and st.dpt_main == 1: + entity["state_address"] = st.address + consumed.add(st.address) + else: + review.append({"reason": "switch_without_status", + "address": ga.address, "name": ga.name}) + switches.append(entity) + consumed.add(ga.address) + + # ---- 4. SENSORS / BINARY SENSORS (read-only) ---- + for ga in project.gas.values(): + if ga.address in consumed: + continue + if ga.ha_platform == "sensor": + entity = {"name": ga.name, "state_address": ga.address} + if ga.value_type: + entity["type"] = ga.value_type + else: + review.append({"reason": "sensor_without_value_type", + "address": ga.address, "name": ga.name}) + sensors.append(entity) + consumed.add(ga.address) + elif ga.ha_platform == "binary_sensor": + binary_sensors.append({"name": ga.name, "state_address": ga.address}) + consumed.add(ga.address) + elif ga.ha_platform in ("climate", "scene", "number"): + review.append({"reason": f"manual_{ga.ha_platform}", "address": ga.address, + "name": ga.name, "dpt": ga.dpt}) + elif ga.ha_platform == "unknown" and ga.dpt_main is not None: + review.append({"reason": "unmapped_dpt", "address": ga.address, + "name": ga.name, "dpt": ga.dpt}) + + knx: dict[str, Any] = {} + if switches: + knx["switch"] = switches + if lights: + knx["light"] = lights + if covers: + knx["cover"] = covers + if binary_sensors: + knx["binary_sensor"] = binary_sensors + if sensors: + knx["sensor"] = sensors + + package = {"knx": knx} + text = ( + "# Home Assistant KNX package generated by nickol-knx-mcp\n" + f"# Source project: {project.info.get('name', '?')}\n" + "# REVIEW before use: command/status pairing is inferred heuristically.\n" + "# Multi-GA fixtures (light/cover/climate) and scenes need manual verification.\n\n" + + yaml.safe_dump(package, allow_unicode=True, sort_keys=False, default_flow_style=False) + ) + counts = { + "switch": len(switches), "light": len(lights), "cover": len(covers), + "binary_sensor": len(binary_sensors), "sensor": len(sensors), + "review": len(review), + } + return {"yaml": text, "review": review, "counts": counts} diff --git a/nickol_knx_mcp/pairing.py b/nickol_knx_mcp/pairing.py new file mode 100644 index 0000000..481826d --- /dev/null +++ b/nickol_knx_mcp/pairing.py @@ -0,0 +1,62 @@ +"""Name-token based command/status pairing. + +In real KNX projects the status/feedback GA almost never sits in the same middle +group as its command (commands in .../0/..., feedback in .../4/...). So pairing +must be driven by *name similarity within the same main group*, not by address +proximity. This module centralizes that logic so analysis and HA generation agree. +""" + +from __future__ import annotations + +from typing import Iterable, Optional + +from .project import GARecord, STATUS_KEYWORDS, COMMAND_KEYWORDS + +_STOP = set(STATUS_KEYWORDS) | set(COMMAND_KEYWORDS) | { + "the", "of", "and", "und", "der", "die", "das", "и", "в", "на", + "rm", "fb", +} + +# command DPT main -> acceptable status DPT mains +STATUS_COMPAT = { + 1: {1, 5}, # switch/up-down -> 1.x state or 5.x position feedback + 3: {1, 5}, # relative dim -> on/off or brightness feedback + 5: {5}, # scaling setpoint -> scaling feedback + 9: {9}, # float setpoint -> float feedback +} + + +def base_tokens(name: str) -> set[str]: + low = name.lower() + for ch in "/-_.,()[]:": + low = low.replace(ch, " ") + return {t for t in low.split() if t and t not in _STOP and len(t) > 1} + + +def find_status(command: GARecord, candidates: Iterable[GARecord]) -> Optional[GARecord]: + """Return the best status GA matching `command`, or None. + + A candidate matches when (a) token overlap with the command base name is at + least min(2, len(command tokens)), and (b) category matches OR DPT main is + compatible. Same-main-group candidates are preferred. + """ + cmd_tokens = base_tokens(command.name) + if not cmd_tokens: + return None + need = min(2, len(cmd_tokens)) + compat = STATUS_COMPAT.get(command.dpt_main or -1, {command.dpt_main}) + + best: Optional[GARecord] = None + best_score = -1 + for c in candidates: + if c.address == command.address: + continue + inter = cmd_tokens & base_tokens(c.name) + if len(inter) < need: + continue + if c.category != command.category and (c.dpt_main not in compat): + continue + score = len(inter) + (2 if c.main == command.main else 0) + if score > best_score: + best, best_score = c, score + return best diff --git a/nickol_knx_mcp/project.py b/nickol_knx_mcp/project.py new file mode 100644 index 0000000..169ef1d --- /dev/null +++ b/nickol_knx_mcp/project.py @@ -0,0 +1,214 @@ +"""Load and parse a .knxproj file and build an enriched in-memory model. + +This module is the ONLY place that touches the ETS project file. It is strictly +read-only: it opens the .knxproj archive, never modifies it, and never opens any +KNX/IP connection. The custom MCP server has no bus connectivity by design. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any, Optional + +from xknxproject import XKNXProj +from xknxproject.models import KNXProject + +from .dpt_map import classify_dpt, dpt_key + + +# Multilingual keyword sets (EN / DE / RU) used by the heuristic fallbacks. +STATUS_KEYWORDS = [ + "status", "state", "stat", "fb", "feedback", "rueck", "rück", "rm ", + "rm_", "статус", "состоян", "обратн", "сост.", +] +COMMAND_KEYWORDS = [ + "switch", "schalt", "dimm", "control", "steuer", "befehl", "set", + "soll", "вкл", "выкл", "упр", "команд", "задан", +] + +# Category disambiguation by name (used for DPTs ambiguous between domains, +# e.g. 5.001 = brightness OR shutter position; 1.001 = light OR generic). +_CATEGORY_KEYWORDS: dict[str, list[str]] = { + "shutter": ["blind", "shutter", "jalousie", "roll", "rollo", "marqui", + "awning", "curtain", "position", "штор", "жалюзи", "рольставн", + "ролет", "ролл", "позиц"], + "lighting": ["light", "lamp", "dimm", "led", "spot", "свет", "лампа", + "подсветк", "освещ", "люстра", "диммер"], + "hvac": ["heat", "cool", "climate", "thermostat", "hvac", "valve", "fan", + "отопл", "климат", "тепл", "конвектор", "вентил", "клапан", + "кондиц", "тёпл"], + "energy": ["energy", "power", "consum", "meter", "kwh", "watt", "энерг", + "мощност", "потребл", "счётчик", "счетчик"], + "scene": ["scene", "scene", "сцен", "preset", "пресет"], + "diagnostics": ["alarm", "fault", "error", "diag", "leak", "smoke", + "тревог", "ошибк", "диагност", "утечк", "дым"], +} + + +def _refine_category(name: str, current: str) -> str: + low = name.lower() + for cat, words in _CATEGORY_KEYWORDS.items(): + if any(w in low for w in words): + return cat + return current + + +@dataclass +class GARecord: + """Enriched group-address record used by all analysis tools.""" + address: str + name: str + description: str + comment: str + dpt_main: Optional[int] + dpt_sub: Optional[int] + data_secure: bool + co_ids: list[str] + category: str + kind: str + ha_platform: str + value_type: Optional[str] + label: str + # 3-level decomposition + main: Optional[int] = None + middle: Optional[int] = None + sub: Optional[int] = None + main_name: str = "" + middle_name: str = "" + + @property + def dpt(self) -> str: + return dpt_key(self.dpt_main, self.dpt_sub) + + @property + def middle_key(self) -> str: + """Identity of the parent middle group (for sibling search).""" + if self.main is not None and self.middle is not None: + return f"{self.main}/{self.middle}" + return "" + + +@dataclass +class LoadedProject: + path: str + info: dict[str, Any] + gas: dict[str, GARecord] # address -> record + raw: KNXProject = field(repr=False) + devices: dict[str, Any] = field(default_factory=dict, repr=False) + functions: dict[str, Any] = field(default_factory=dict, repr=False) + topology: dict[str, Any] = field(default_factory=dict, repr=False) + + @property + def style(self) -> str: + return self.info.get("group_address_style", "") + + +def _split_three_level(address: str) -> tuple[Optional[int], Optional[int], Optional[int]]: + parts = address.split("/") + if len(parts) == 3: + try: + return int(parts[0]), int(parts[1]), int(parts[2]) + except ValueError: + return None, None, None + return None, None, None + + +def _override_kind_by_name(name: str, kind: str) -> str: + """Refine command/status using name keywords (helps when DPT is generic).""" + low = name.lower() + if any(k in low for k in STATUS_KEYWORDS): + return "status" + if any(k in low for k in COMMAND_KEYWORDS): + # only upgrade unknown -> command; never overwrite explicit sensor + if kind in ("unknown", "command", "status"): + return "command" + return kind + + +def _build_range_name_map(raw: KNXProject) -> dict[str, str]: + """Map address-prefix -> human range name from group_ranges. + + Returns keys like '1' (main) and '1/1' (middle) -> name. + """ + out: dict[str, str] = {} + + def walk(rng: dict[str, Any]) -> None: + start = rng.get("address_start") + name = rng.get("name", "") + if isinstance(start, int): + # main range: start aligned to 0x0800 boundaries; derive main number + main = (start >> 11) & 0x1F + middle = (start >> 8) & 0x07 + # heuristic: if start is multiple of 2048 -> a main range, else middle + if start % 2048 == 0: + out[str(main)] = name + else: + out[f"{main}/{middle}"] = name + for child in rng.get("group_ranges", {}).values(): + walk(child) + + for rng in raw.get("group_ranges", {}).values(): + walk(rng) + return out + + +def load_project(path: str, password: Optional[str] = None, + language: Optional[str] = None) -> LoadedProject: + """Parse a .knxproj file into an enriched, read-only model.""" + kwargs: dict[str, Any] = {"path": path} + if password: + kwargs["password"] = password + if language: + kwargs["language"] = language + raw: KNXProject = XKNXProj(**kwargs).parse() + return build_loaded_from_raw(raw, path) + + +def build_loaded_from_raw(raw: KNXProject, path: str) -> LoadedProject: + """Build an enriched LoadedProject from an already-parsed KNXProject dict.""" + range_names = _build_range_name_map(raw) + + gas: dict[str, GARecord] = {} + for addr, ga in raw.get("group_addresses", {}).items(): + dpt = ga.get("dpt") + main = dpt.get("main") if dpt else None + sub = dpt.get("sub") if dpt else None + info = classify_dpt(main, sub) + kind = _override_kind_by_name(ga.get("name", ""), info["kind"]) + category = _refine_category(ga.get("name", ""), info["category"]) + ha_platform = info["ha_platform"] + # If the name says shutter but DPT mapped it to light (5.001/1.001), + # correct the HA platform so the generator builds a cover, not a light. + if category == "shutter" and ha_platform in ("light", "switch"): + ha_platform = "cover" + m, mid, s = _split_three_level(ga.get("address", addr)) + rec = GARecord( + address=ga.get("address", addr), + name=ga.get("name", ""), + description=ga.get("description", "") or "", + comment=ga.get("comment", "") or "", + dpt_main=main, + dpt_sub=sub, + data_secure=bool(ga.get("data_secure", False)), + co_ids=list(ga.get("communication_object_ids", []) or []), + category=category, + kind=kind, + ha_platform=ha_platform, + value_type=info["value_type"], + label=info["label"], + main=m, middle=mid, sub=s, + main_name=range_names.get(str(m), "") if m is not None else "", + middle_name=range_names.get(f"{m}/{mid}", "") if m is not None and mid is not None else "", + ) + gas[rec.address] = rec + + return LoadedProject( + path=path, + info=dict(raw.get("info", {})), + gas=gas, + raw=raw, + devices=dict(raw.get("devices", {})), + functions=dict(raw.get("functions", {})), + topology=dict(raw.get("topology", {})), + ) diff --git a/nickol_knx_mcp/report.py b/nickol_knx_mcp/report.py new file mode 100644 index 0000000..127a7b7 --- /dev/null +++ b/nickol_knx_mcp/report.py @@ -0,0 +1,118 @@ +"""Human-readable Markdown report. + +Per the safety requirement, a report like this is produced BEFORE any ETS import +or HA deployment, so a human can review what will change. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from typing import Any + +from .project import LoadedProject +from .analyze import validate_naming, detect_missing_status, detect_dpt_issues +from .generate_ha import generate_ha_yaml + + +_SEV_ICON = {"error": "🔴", "warning": "🟡", "info": "🔵"} + + +def _section(title: str, findings: list[dict[str, Any]]) -> str: + if not findings: + return f"### {title}\n\n_No issues found._\n" + out = [f"### {title} ({len(findings)})\n"] + by_sev = defaultdict(list) + for f in findings: + by_sev[f["severity"]].append(f) + for sev in ("error", "warning", "info"): + for f in by_sev.get(sev, []): + out.append(f"- {_SEV_ICON.get(sev,'')} `{f['address']}` — {f['message']}") + return "\n".join(out) + "\n" + + +def build_report(project: LoadedProject, + name_regex: str | None = None) -> dict[str, Any]: + """Return {'markdown': str, 'summary': {...}}.""" + naming = validate_naming(project, name_regex=name_regex) + status = detect_missing_status(project) + dpts = detect_dpt_issues(project) + ha = generate_ha_yaml(project) + + info = project.info + gas = project.gas + + cat_counts = Counter(ga.category for ga in gas.values()) + kind_counts = Counter(ga.kind for ga in gas.values()) + no_dpt = sum(1 for ga in gas.values() if ga.dpt_main is None) + secure = sum(1 for ga in gas.values() if ga.data_secure) + + all_findings = naming + status + dpts + sev_counts = Counter(f["severity"] for f in all_findings) + + md: list[str] = [] + md.append(f"# KNX Project Report — {info.get('name', '?')}\n") + md.append( + f"- **Source:** `{project.path}`\n" + f"- **GA style:** {info.get('group_address_style', '?')}\n" + f"- **ETS tool version:** {info.get('tool_version', '?')} " + f"(xknxproject {info.get('xknxproject_version', '?')})\n" + f"- **Last modified:** {info.get('last_modified', '?')}\n" + ) + + md.append("\n## 1. Inventory\n") + md.append( + f"- Group addresses: **{len(gas)}**\n" + f"- Devices: **{len(project.devices)}**\n" + f"- Functions: **{len(project.functions)}**\n" + f"- Without DPT: **{no_dpt}**\n" + f"- KNX Data Secure GAs: **{secure}**\n" + ) + md.append("\n**By category:** " + + ", ".join(f"{k}={v}" for k, v in sorted(cat_counts.items())) + "\n") + md.append("**By kind:** " + + ", ".join(f"{k}={v}" for k, v in sorted(kind_counts.items())) + "\n") + + md.append("\n## 2. Findings\n") + md.append( + f"Totals: 🔴 errors **{sev_counts.get('error',0)}**, " + f"🟡 warnings **{sev_counts.get('warning',0)}**, " + f"🔵 info **{sev_counts.get('info',0)}**\n" + ) + md.append("\n" + _section("2.1 Naming & structure", naming)) + md.append("\n" + _section("2.2 Missing status addresses", status)) + md.append("\n" + _section("2.3 DPT consistency", dpts)) + + md.append("\n## 3. Home Assistant mapping preview\n") + c = ha["counts"] + md.append( + f"Entities that can be generated now: switch **{c['switch']}**, " + f"light **{c['light']}**, cover **{c['cover']}**, " + f"binary_sensor **{c['binary_sensor']}**, sensor **{c['sensor']}**.\n" + ) + if ha["review"]: + md.append(f"\n**Needs manual review ({len(ha['review'])}):**\n") + for r in ha["review"][:50]: + md.append(f"- `{r.get('address','-')}` {r.get('name','')} — {r['reason']}") + if len(ha["review"]) > 50: + md.append(f"- … and {len(ha['review']) - 50} more") + md.append("") + + md.append("\n## 4. Next steps\n") + md.append( + "1. Resolve 🔴 errors (missing DPT, empty names) in ETS first.\n" + "2. Add status/feedback GAs for every flagged command.\n" + "3. Re-run this report until errors are clear.\n" + "4. Generate ETS CSV/XML and HA YAML, commit to Git, then import into ETS " + "and deploy to Home Assistant.\n" + ) + + summary = { + "ga_count": len(gas), + "errors": sev_counts.get("error", 0), + "warnings": sev_counts.get("warning", 0), + "info": sev_counts.get("info", 0), + "missing_status": len(status), + "ha_entities": {k: v for k, v in c.items() if k != "review"}, + "ha_review": c["review"], + } + return {"markdown": "\n".join(md), "summary": summary} diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py new file mode 100644 index 0000000..5cb2a96 --- /dev/null +++ b/nickol_knx_mcp/server.py @@ -0,0 +1,252 @@ +"""nickol-knx-mcp — MCP server for design-time KNX/ETS project work. + +Exposes read + analysis + generation tools over a parsed .knxproj. The server +has NO KNX/IP bus connectivity of any kind: it only reads the project archive and +writes output files into a confined workspace. It can never write to a live bus. + +Run (stdio): python -m nickol_knx_mcp.server +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Optional + +from mcp.server.fastmcp import FastMCP + +from .project import load_project, LoadedProject +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 + +mcp = FastMCP("nickol-knx") + +# --------------------------------------------------------------------------- # +# State + safety helpers +# --------------------------------------------------------------------------- # +_STATE: dict[str, Optional[LoadedProject]] = {"project": None} + +# Output writes are confined to this directory (default: ./knx-workspace). +_WORKSPACE = Path(os.environ.get("NICKOL_KNX_WORKSPACE", "./knx-workspace")).resolve() + + +def _project() -> LoadedProject: + p = _STATE["project"] + if p is None: + raise ValueError("No project loaded. Call load_project(path) first.") + return p + + +def _safe_write(rel_or_abs_path: str, content: str) -> str: + """Write inside the workspace only. Returns the absolute path written.""" + _WORKSPACE.mkdir(parents=True, exist_ok=True) + target = Path(rel_or_abs_path) + if not target.is_absolute(): + target = _WORKSPACE / target + target = target.resolve() + if _WORKSPACE not in target.parents and target != _WORKSPACE: + raise ValueError( + f"Refusing to write outside workspace {_WORKSPACE}. " + "Set NICKOL_KNX_WORKSPACE to change it." + ) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return str(target) + + +# --------------------------------------------------------------------------- # +# Read tools +# --------------------------------------------------------------------------- # +@mcp.tool() +def load_project(path: str, password: Optional[str] = None, + language: Optional[str] = None) -> dict[str, Any]: + """Parse a .knxproj file (read-only) and cache it for the session. + + Args: + path: Path to the .knxproj file. + password: Project password, if the .knxproj is protected. + language: Optional language code (e.g. 'de-DE', 'ru-RU'). + """ + proj = load_project(path, password=password, language=language) + _STATE["project"] = proj + return { + "loaded": True, + "name": proj.info.get("name"), + "ga_style": proj.style, + "group_addresses": len(proj.gas), + "devices": len(proj.devices), + "functions": len(proj.functions), + "ets_tool_version": proj.info.get("tool_version"), + } + + +@mcp.tool() +def list_group_addresses(category: Optional[str] = None, + kind: Optional[str] = None, + missing_dpt_only: bool = False, + limit: int = 500) -> list[dict[str, Any]]: + """List parsed group addresses with classification. + + Filters: category (lighting/shutter/hvac/sensor/scene/energy/diagnostics), + kind (command/status/sensor), missing_dpt_only. + """ + proj = _project() + out = [] + for ga in proj.gas.values(): + if category and ga.category != category: + continue + if kind and ga.kind != kind: + continue + if missing_dpt_only and ga.dpt_main is not None: + continue + out.append({ + "address": ga.address, "name": ga.name, "dpt": ga.dpt, + "category": ga.category, "kind": ga.kind, + "ha_platform": ga.ha_platform, "secure": ga.data_secure, + "description": ga.description, + }) + if len(out) >= limit: + break + return out + + +@mcp.tool() +def get_devices() -> list[dict[str, Any]]: + """List devices: individual address, name, order number, manufacturer.""" + proj = _project() + return [{ + "individual_address": d.get("individual_address"), + "name": d.get("name"), + "order_number": d.get("order_number"), + "manufacturer": d.get("manufacturer_name"), + "communication_objects": len(d.get("communication_object_ids", []) or []), + } for d in proj.devices.values()] + + +@mcp.tool() +def get_topology() -> dict[str, Any]: + """Return the area/line/device topology tree.""" + proj = _project() + tree: dict[str, Any] = {} + for aid, area in proj.topology.items(): + lines = {} + for lid, line in area.get("lines", {}).items(): + lines[lid] = {"name": line.get("name"), + "medium": line.get("medium_type"), + "devices": line.get("devices", [])} + tree[aid] = {"name": area.get("name"), "lines": lines} + return tree + + +# --------------------------------------------------------------------------- # +# Analysis tools +# --------------------------------------------------------------------------- # +@mcp.tool() +def check_naming(name_regex: Optional[str] = None) -> list[dict[str, Any]]: + """Validate naming conventions and 3-level structure.""" + return validate_naming(_project(), name_regex=name_regex) + + +@mcp.tool() +def check_missing_status() -> list[dict[str, Any]]: + """Detect controllable GAs lacking a status/feedback counterpart.""" + return detect_missing_status(_project()) + + +@mcp.tool() +def check_dpt() -> list[dict[str, Any]]: + """Detect missing, inconsistent or mismatched DPTs.""" + return detect_dpt_issues(_project()) + + +@mcp.tool() +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) + return { + "summary": rep["summary"], + "naming": validate_naming(proj, name_regex=name_regex), + "missing_status": detect_missing_status(proj), + "dpt": detect_dpt_issues(proj), + } + + +# --------------------------------------------------------------------------- # +# Generation tools (write only into the confined workspace) +# --------------------------------------------------------------------------- # +@mcp.tool() +def generate_ha_package(output_path: Optional[str] = None) -> dict[str, Any]: + """Generate a Home Assistant KNX package YAML. + + If output_path is given, the YAML is written into the workspace and the path + returned; otherwise the YAML text is returned inline. + """ + proj = _project() + res = generate_ha_yaml(proj) + out: dict[str, Any] = {"counts": res["counts"], "review": res["review"]} + if output_path: + out["written"] = _safe_write(output_path, res["yaml"]) + else: + out["yaml"] = res["yaml"] + return out + + +@mcp.tool() +def generate_ets_group_addresses(fmt: str = "xml", + output_path: Optional[str] = None) -> dict[str, Any]: + """Generate an ETS-importable Group Address export. + + Args: + fmt: 'xml' (ga-export/01, recommended) or 'csv' (native ETS layout). + output_path: optional file inside the workspace. + """ + proj = _project() + if fmt == "csv": + content = generate_ets_csv(proj) + elif fmt == "xml": + content = generate_ets_xml(proj) + else: + raise ValueError("fmt must be 'xml' or 'csv'") + out: dict[str, Any] = {"format": fmt} + if output_path: + out["written"] = _safe_write(output_path, content) + else: + out["content"] = content + return out + + +@mcp.tool() +def project_report(output_path: Optional[str] = None, + name_regex: Optional[str] = None) -> dict[str, Any]: + """Produce the human-readable Markdown report (review before any import).""" + proj = _project() + rep = build_report(proj, name_regex=name_regex) + out: dict[str, Any] = {"summary": rep["summary"]} + if output_path: + out["written"] = _safe_write(output_path, rep["markdown"]) + else: + out["markdown"] = rep["markdown"] + return out + + +@mcp.tool() +def workspace_info() -> dict[str, Any]: + """Show the confined output workspace and the safety guarantees.""" + return { + "workspace": str(_WORKSPACE), + "bus_access": False, + "note": "This server never connects to a KNX/IP bus. It only reads the " + ".knxproj and writes files inside the workspace. Use a Git MCP / " + "filesystem MCP to version the outputs.", + } + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6abef51 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,41 @@ +[project] +name = "nickol-knx-mcp" +version = "0.1.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" +license = { text = "MIT" } +authors = [{ name = "Nikolay Miroshnichenko" }] +keywords = ["knx", "ets", "ets6", "home-assistant", "mcp", "model-context-protocol", "smart-home", "home-automation"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Home Automation", + "Topic :: Software Development :: Libraries", +] +dependencies = [ + "mcp>=1.10", + "xknxproject>=3.8", + "PyYAML>=6.0", +] + +[project.urls] +Homepage = "https://github.com/NickoScope/nickol-knx-mcp" +Repository = "https://github.com/NickoScope/nickol-knx-mcp" +Issues = "https://github.com/NickoScope/nickol-knx-mcp/issues" +Changelog = "https://github.com/NickoScope/nickol-knx-mcp/blob/main/CHANGELOG.md" + +[project.scripts] +nickol-knx-mcp = "nickol_knx_mcp.server:main" + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +packages = ["nickol_knx_mcp"] diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..2b1060a --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,128 @@ +"""Smoke test the full pipeline against a synthetic KNX project.""" +import sys, os +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from nickol_knx_mcp.project import build_loaded_from_raw +from nickol_knx_mcp.analyze import validate_naming, detect_missing_status, detect_dpt_issues +from nickol_knx_mcp.generate_ha import generate_ha_yaml +from nickol_knx_mcp.generate_ets import generate_ets_csv, generate_ets_xml +from nickol_knx_mcp.report import build_report + + +def ga(addr, name, main, sub, secure=False, desc="", co_ids=None): + return { + "name": name, "identifier": f"GA-{addr}", "raw_address": 0, + "address": addr, "project_uid": None, + "dpt": ({"main": main, "sub": sub} if main is not None else None), + "data_secure": secure, "communication_object_ids": co_ids or [], + "description": desc, "comment": "", + } + + +raw = { + "info": { + "project_id": "P-1", "name": "Nikolay Test House", + "last_modified": "2026-06-01T10:00:00Z", "group_address_style": "ThreeLevel", + "guid": "x", "created_by": "ETS6", "schema_version": "21", + "tool_version": "6.3.0", "xknxproject_version": "3.9.0", "language_code": "ru-RU", + }, + "communication_objects": { + "CO-1": {"name": "Switch", "number": 1, "text": "", "function_text": "", + "description": "", "device_address": "1.1.1", "device_application": None, + "module": None, "channel": None, "dpts": [{"main": 1, "sub": 1}], + "object_size": "1 bit", "group_address_links": ["1/0/0"], + "flags": {"read": False, "write": True, "communication": True, + "transmit": False, "update": False, "read_on_init": False}, + "dpas": None}, + }, + "devices": { + "1.1.1": {"name": "MDT Switch", "hardware_name": "AKK", "order_number": "AKK-0416.03", + "description": "", "manufacturer_name": "MDT", "individual_address": "1.1.1", + "application": None, "project_uid": None, + "communication_object_ids": ["CO-1"], "channels": {}}, + }, + "topology": { + "1": {"name": "Area 1", "description": None, "lines": { + "1.1": {"name": "Line 1", "medium_type": "TP", "description": None, + "devices": ["1.1.1"]}}} + }, + "locations": {}, + "group_addresses": { + # Lighting: command 1.001 WITH status -> ok + "1/0/0": ga("1/0/0", "Kitchen light switch", 1, 1, co_ids=["CO-1"]), + "1/4/0": ga("1/4/0", "Kitchen light status", 1, 11), + # Lighting: command WITHOUT status -> should flag (different middle from status) + "1/0/1": ga("1/0/1", "Hall light switch", 1, 1), + # Dimming light: brightness command + status + "1/1/0": ga("1/1/0", "Living dimmer brightness", 5, 1), + "1/4/1": ga("1/4/1", "Living dimmer brightness status", 5, 1), + # Shutter up/down + stop + position + "2/0/0": ga("2/0/0", "Bedroom blind up/down", 1, 8), + "2/0/1": ga("2/0/1", "Bedroom blind stop", 1, 10), + "2/1/0": ga("2/1/0", "Bedroom blind position", 5, 1), + "2/4/0": ga("2/4/0", "Bedroom blind position status", 5, 1), + # Sensors + "4/0/0": ga("4/0/0", "Living temperature", 9, 1), + "4/0/1": ga("4/0/1", "Living CO2", 9, 8), + # Missing DPT -> error + "5/0/0": ga("5/0/0", "Mystery object", None, None), + # Empty name -> error + "5/0/1": ga("5/0/1", "", 1, 1), + # Duplicate name + inconsistent DPT + "6/0/0": ga("6/0/0", "Sensor X", 9, 1), + "6/0/1": ga("6/0/1", "Sensor X", 9, 4), + # Energy + "7/0/0": ga("7/0/0", "Total energy", 13, 13), + }, + "group_ranges": { + "1": {"name": "Lighting", "address_start": 2048, "address_end": 4095, + "comment": "", "group_addresses": [], "group_ranges": { + "1/0": {"name": "Switch", "address_start": 2048, "address_end": 2303, + "comment": "", "group_addresses": ["1/0/0", "1/0/1"], "group_ranges": {}}, + }}, + }, + "functions": { + "F-1": {"function_type": "SwitchableLight", + "group_addresses": { + "1/0/1": {"address": "1/0/1", "name": "Hall light switch", + "project_uid": None, "role": "SwitchOnOff"}, + }, + "identifier": "F-1", "name": "Hall Light", "project_uid": None, + "space_id": "S-1", "usage_text": ""}, + }, +} + +proj = build_loaded_from_raw(raw, "/tmp/test.knxproj") +print("=== LOADED ===") +print("GAs:", len(proj.gas), "style:", proj.style) +for a, r in list(proj.gas.items())[:4]: + print(f" {a} {r.name!r} dpt={r.dpt} cat={r.category} kind={r.kind} ha={r.ha_platform} mid={r.middle_key}") + +print("\n=== NAMING ===") +for f in validate_naming(proj): + print(" ", f["severity"], f["code"], f["address"], "-", f["message"][:70]) + +print("\n=== MISSING STATUS ===") +for f in detect_missing_status(proj): + print(" ", f["severity"], f["code"], f["address"], "-", f["message"][:70]) + +print("\n=== DPT ISSUES ===") +for f in detect_dpt_issues(proj): + print(" ", f["severity"], f["code"], f["address"], "-", f["message"][:70]) + +print("\n=== HA YAML ===") +ha = generate_ha_yaml(proj) +print("counts:", ha["counts"]) +print(ha["yaml"]) + +print("=== ETS CSV ===") +print(generate_ets_csv(proj)) + +print("=== ETS XML ===") +print(generate_ets_xml(proj)) + +print("=== REPORT SUMMARY ===") +rep = build_report(proj) +print(rep["summary"]) +print("\n--- report head ---") +print("\n".join(rep["markdown"].splitlines()[:30]))