From bba8befcde2d6ce744445218352cb127e6e0e5d9 Mon Sep 17 00:00:00 2001 From: Nikolay Miroshnichenko Date: Sun, 28 Jun 2026 09:25:57 +0200 Subject: [PATCH] =?UTF-8?q?nickol-knx-mcp=20v0.1.0=20=E2=80=94=20design-ti?= =?UTF-8?q?me=20KNX/ETS6=20MCP=20server=20(public=20beta)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .github/ISSUE_TEMPLATE/bug_report.yml | 46 ++++ .github/ISSUE_TEMPLATE/config.yml | 5 + .github/ISSUE_TEMPLATE/feature_request.yml | 20 ++ .github/ISSUE_TEMPLATE/real_project_test.yml | 55 ++++ .github/PULL_REQUEST_TEMPLATE.md | 13 + .github/workflows/ci.yml | 37 +++ .gitignore | 30 +++ CHANGELOG.md | 33 +++ CLAUDE.md | 70 +++++ CONTRIBUTING.md | 70 +++++ LICENSE | 21 ++ README.md | 228 +++++++++++++++++ README.ru.md | 162 ++++++++++++ SECURITY.md | 29 +++ examples/claude_desktop_config.json | 25 ++ nickol_knx_mcp/__init__.py | 2 + nickol_knx_mcp/analyze.py | 253 +++++++++++++++++++ nickol_knx_mcp/dpt_map.py | 123 +++++++++ nickol_knx_mcp/generate_ets.py | 127 ++++++++++ nickol_knx_mcp/generate_ha.py | 159 ++++++++++++ nickol_knx_mcp/pairing.py | 62 +++++ nickol_knx_mcp/project.py | 214 ++++++++++++++++ nickol_knx_mcp/report.py | 118 +++++++++ nickol_knx_mcp/server.py | 252 ++++++++++++++++++ pyproject.toml | 41 +++ tests/test_pipeline.py | 128 ++++++++++ 26 files changed, 2323 insertions(+) create mode 100644 .github/ISSUE_TEMPLATE/bug_report.yml create mode 100644 .github/ISSUE_TEMPLATE/config.yml create mode 100644 .github/ISSUE_TEMPLATE/feature_request.yml create mode 100644 .github/ISSUE_TEMPLATE/real_project_test.yml create mode 100644 .github/PULL_REQUEST_TEMPLATE.md create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 CLAUDE.md create mode 100644 CONTRIBUTING.md create mode 100644 LICENSE create mode 100644 README.md create mode 100644 README.ru.md create mode 100644 SECURITY.md create mode 100644 examples/claude_desktop_config.json create mode 100644 nickol_knx_mcp/__init__.py create mode 100644 nickol_knx_mcp/analyze.py create mode 100644 nickol_knx_mcp/dpt_map.py create mode 100644 nickol_knx_mcp/generate_ets.py create mode 100644 nickol_knx_mcp/generate_ha.py create mode 100644 nickol_knx_mcp/pairing.py create mode 100644 nickol_knx_mcp/project.py create mode 100644 nickol_knx_mcp/report.py create mode 100644 nickol_knx_mcp/server.py create mode 100644 pyproject.toml create mode 100644 tests/test_pipeline.py 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]))