load_ga_export, entity naming, climate setpoint shift; fix suggest_repairs crash; CI runs every test

Three things found by auditing a TapPlan export, plus one regression found on the way.

load_ga_export(path): new tool. Reads an ETS ga-export/01 XML (ETS "Export Group
Addresses", or the import file a planning tool writes) into a project without
devices, through safe_fromstring, capped at 50 MB and 8 levels of range nesting.
Invalid or duplicate addresses and unknown DPT tokens go to import_warnings, never
silently. Round trip with our own generate_ets_xml is covered by a test.

Entity naming: lights, covers and climates are named after the common word prefix
of their member names, cutting only function words. The first version of the rule
turned "01. <room> - All Blinds - Move" into "01" on a real project, so anything not
in the function vocabulary now stays. On six real projects 214 of 1189 entities got
a shorter name, each rename reviewed, no address mapping changed. Regenerated
packages show different names; noted in the changelog.

Setpoint shift: 9.002 / 6.010 with a shift word maps to setpoint_shift_address,
setpoint_shift_state_address and setpoint_shift_mode (keys checked against the HA
KNX climate docs). It used to become a plain sensor.

Fixed: suggest_repairs raised UnboundLocalError on any project with a missing status
GA. My local rename in the 12.09 typing cleanup left two references to the old name.
test_council_fixes covers it but was not in CI, so it shipped. CI now runs all 21
test files instead of a hand-picked 10; the 11 added ones pass from a clean clone.

Verified: all 21 tests, ruff, mypy with the package installed, corpus guard no drift.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Nikolay Miroshnichenko
2026-09-14 00:11:45 +02:00
co-authored by Claude Opus 5
parent 29bfba7684
commit ec63bcf106
8 changed files with 583 additions and 8 deletions
+36
View File
@@ -76,6 +76,42 @@ jobs:
- name: Stable ordering + cursor paging for large tool results
run: python tests/test_paging.py
- name: GA-export intake, entity naming, climate setpoint shift
run: python tests/test_ga_export.py
- name: Repair proposals and council-review fixes
run: python tests/test_council_fixes.py
- name: Domain classifier (name + DPT + range context)
run: python tests/test_domain_classifier.py
- name: explain_ga provenance
run: python tests/test_explain.py
- name: Aggregates carry their evidence
run: python tests/test_explainable_aggregates.py
- name: Device parameter checks
run: python tests/test_param_check.py
- name: Value-status completeness (brightness / position)
run: python tests/test_role_completeness.py
- name: Room template library
run: python tests/test_room_library.py
- name: Root-cause regressions
run: python tests/test_root_cause.py
- name: Hardened archive and XML parsing
run: python tests/test_safexml.py
- name: Theben reference projects
run: python tests/test_theben_corrections.py
- name: Topology limits
run: python tests/test_topology.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'])"
+33
View File
@@ -6,8 +6,41 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **`load_ga_export(path)`: audit a group-address list without a `.knxproj`.** Reads an ETS
ga-export/01 XML (ETS "Export Group Addresses", or the import file a planning tool such as TapPlan
writes) into a project without devices: names, addresses, DPTs (`DPST-x-y`, `DPT-x`), descriptions,
the `Security` flag and the range tree. GA-level checks and both generators work on it; device,
Function and topology tools have nothing to read and say so. Parsed through the same hardened XML
reader as `.knxproj` members, size-capped at 50 MB. Invalid or duplicate addresses and unknown DPT
tokens are skipped with an entry in `import_warnings`, never silently.
- **Climate: setpoint shift.** A DPT 9.002 or 6.010 group address whose name says shift
(shift / Sollwertverschiebung / смещение / сдвиг) in a climate zone now maps to
`setpoint_shift_address` / `setpoint_shift_state_address` with `setpoint_shift_mode` `DPT9002` or
`DPT6010` taken from the DPT (keys per the Home Assistant KNX climate docs). It used to fall through
to a plain sensor.
### Changed
- **Multi-address entities are named after what their addresses share.** A light built from
"Kitchen Spots On-Off", "Kitchen Spots B.Value" and their feedbacks is now "Kitchen Spots", not
"Kitchen Spots B.Value". Lights, covers and climates take the longest common word prefix of their
member names, but only when it still carries an identity token and does not drop a word containing a
digit; two entities that would collide keep their original names. Found by auditing a TapPlan export.
Only function words are ever cut (value, brightness, up/down, mode, Яркость, Абсолютное диммирование,
Движение, Вкл…); a room, device type, channel or anything else stays. On the maintainer's six real
projects 214 of 1189 entities got a shorter name and no address mapping changed. **A regenerated
package will therefore show different entity names than one generated before** — review the diff
before replacing a package that is already deployed.
### Fixed
- **`suggest_repairs` crashed on any project with a missing status GA** (`UnboundLocalError` in
`repair.py`). Introduced by a local-variable rename in the 12.09 typing cleanup; the test that covers
it was not part of CI, which is why it shipped. CI now runs every test file in `tests/` (11 were
missing), not a hand-picked subset.
- **`suggest.py`: multi-output actuators are split by their vendor object marker.** Zennio-style
devices (Lumento DX4, MAXinBOX, KLIC-DI…) put every output of a device into ONE ETS channel and
separate them only in the object text (`[1] Switch On/Off`, `[2] On/Off (Status)`). The channel
+2 -1
View File
@@ -275,6 +275,7 @@ keyring handling, and the recommended workflow).
| Tool | Purpose |
|------|---------|
| `load_project(path, password?, language?)` | parse a `.knxproj` (read-only) and cache it |
| `load_ga_export(path)` | load an ETS group-address export (ga-export/01 XML, e.g. from ETS or TapPlan) as a project without devices |
| `list_group_addresses(category?, kind?)` | list GAs with classification and filters |
| `get_devices()` | devices + their communication objects |
| `get_topology()` | topology (areas / lines / devices) |
@@ -384,7 +385,7 @@ nickol-knx-mcp/
│ ├── report.py # Markdown report
│ ├── room_library.py # Room Library R1 — compose a new project from templates
│ ├── room_templates/ # built-in room YAML templates + SCHEMA.md (public contract)
│ └── server.py # FastMCP server, 31 tools, confined writes
│ └── server.py # FastMCP server, 32 tools, confined writes
├── tests/test_pipeline.py
├── examples/claude_desktop_config.json
├── skills/
+170
View File
@@ -0,0 +1,170 @@
"""Read an ETS group-address export (ga-export/01 XML) as a project without devices.
A full ``.knxproj`` is not always what people have. ETS can export just the group
addresses, and planning tools (TapPlan and others) produce the same format for
import into ETS. That file carries names, addresses, DPTs, descriptions, the
security flag and the range tree, which is everything the GA-level checks and the
Home Assistant / ETS generators work on.
The export is turned into the same raw shape xknxproject produces for the group
address part, then goes through the normal ``build_loaded_from_raw``. Devices,
communication objects, ETS Functions and topology are simply empty, so the
device-level tools have nothing to report rather than guessing.
The file is untrusted input: size-capped and parsed through ``safe_fromstring``
(no DTD, no entities, no external fetches).
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from typing import Any, Optional, cast
from .project import LoadedProject, build_loaded_from_raw
from .safexml import SafeXmlError, safe_fromstring
MAX_EXPORT_BYTES = 50 * 1024 * 1024 # a 15 000-GA export is a few MB; this only bounds abuse
MAX_RANGE_DEPTH = 8 # ETS nests main/middle (2); planning tools rarely go deeper
_DPT_RE = re.compile(r"^DPS?T-(\d+)(?:-(\d+))?$", re.IGNORECASE)
class GaExportError(ValueError):
"""The file is not a readable ETS group-address export."""
def _local(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
def parse_dpt(value: Optional[str]) -> Optional[dict[str, Optional[int]]]:
"""'DPST-1-1' -> {main 1, sub 1}; 'DPT-9' -> {main 9, sub None}; several -> the first."""
if not value:
return None
first = re.split(r"[\s,;]+", value.strip())[0]
m = _DPT_RE.match(first)
if not m:
return None
return {"main": int(m.group(1)), "sub": int(m.group(2)) if m.group(2) else None}
def _raw_address(address: str) -> Optional[int]:
try:
parts = [int(x) for x in address.split("/")]
except ValueError:
return None
if len(parts) == 3 and parts[0] <= 31 and parts[1] <= 7 and parts[2] <= 255:
return (parts[0] << 11) | (parts[1] << 8) | parts[2]
if len(parts) == 2 and parts[0] <= 31 and parts[1] <= 2047:
return (parts[0] << 11) | parts[1]
if len(parts) == 1 and 0 <= parts[0] <= 65535:
return parts[0]
return None
def _style(addresses: list[str]) -> str:
depths = {a.count("/") for a in addresses}
if depths == {2}:
return "ThreeLevel"
if depths == {1}:
return "TwoLevel"
if depths == {0}:
return "Free"
return "Mixed" if depths else ""
def read_ga_export_bytes(data: bytes, name: str = "ga-export") -> dict[str, Any]:
"""Parse export bytes into the raw project dict (no devices). Raises GaExportError."""
try:
root = safe_fromstring(data)
except SafeXmlError as e:
raise GaExportError(str(e)) from e
if _local(root.tag) != "GroupAddress-Export":
raise GaExportError(f"not an ETS group-address export (root element is <{_local(root.tag)}>)")
gas: dict[str, dict[str, Any]] = {}
warnings: list[str] = []
def walk_range(el, depth: int = 1) -> dict[str, Any]:
if depth > MAX_RANGE_DEPTH:
raise GaExportError(f"group ranges nested deeper than {MAX_RANGE_DEPTH} levels — refused")
start = el.get("RangeStart")
end = el.get("RangeEnd")
rng: dict[str, Any] = {
"name": el.get("Name", ""),
"address_start": int(start) if start and start.isdigit() else None,
"address_end": int(end) if end and end.isdigit() else None,
"comment": el.get("Description", ""),
"group_addresses": [],
"group_ranges": {},
}
for child in el:
tag = _local(child.tag)
if tag == "GroupRange":
sub = walk_range(child, depth + 1)
rng["group_ranges"][f"{sub['name']}@{sub['address_start']}"] = sub
elif tag == "GroupAddress":
addr = add_ga(child)
if addr:
rng["group_addresses"].append(addr)
return rng
def add_ga(el) -> Optional[str]:
address = (el.get("Address") or "").strip()
raw = _raw_address(address)
if raw is None:
warnings.append(f"skipped group address with invalid Address {address!r}")
return None
if address in gas:
warnings.append(f"duplicate address {address} — kept the first entry")
return None
dpts = el.get("DPTs")
dpt = parse_dpt(dpts)
if dpts and dpt is None:
warnings.append(f"{address}: DPT {dpts!r} not understood, left unset")
gas[address] = {
"name": el.get("Name", ""),
"identifier": f"GA-{raw}",
"raw_address": raw,
"address": address,
"project_uid": None,
"dpt": dpt,
"data_secure": (el.get("Security") or "").strip().lower() == "on",
"communication_object_ids": [],
"description": el.get("Description", "") or "",
"comment": "",
}
return address
ranges: dict[str, Any] = {}
for child in root:
tag = _local(child.tag)
if tag == "GroupRange":
rng = walk_range(child)
ranges[f"{rng['name']}@{rng['address_start']}"] = rng
elif tag == "GroupAddress":
add_ga(child)
info = {
"name": name,
"source": "ga-export",
"group_address_style": _style(list(gas)),
"tool_version": None,
"import_warnings": warnings,
}
return {"info": info, "group_addresses": gas, "group_ranges": ranges, "devices": {},
"communication_objects": {}, "functions": {}, "locations": {}, "topology": {}}
def load_ga_export(path: str) -> LoadedProject:
"""Load an ETS ga-export/01 XML file as a read-only project without devices."""
if not os.path.isfile(path):
raise GaExportError(f"file not found: {path}")
size = os.path.getsize(path)
if size > MAX_EXPORT_BYTES:
raise GaExportError(f"file is {size} bytes, over the {MAX_EXPORT_BYTES}-byte limit")
with open(path, "rb") as fh:
data = fh.read(MAX_EXPORT_BYTES + 1)
raw = read_ga_export_bytes(data, name=Path(path).stem)
return build_loaded_from_raw(cast(Any, raw), path)
+112 -6
View File
@@ -164,6 +164,72 @@ def _has(name: str, words) -> bool:
return any(w in low for w in words)
# Setpoint shift (HA climate `setpoint_shift_address`, DPT 6.010 or 9.002). The DPT
# alone is not enough — 9.002 is any temperature difference — so a shift needs the
# word as well. German "Sollwertverschiebung", Russian "смещение/сдвиг уставки".
_SHIFT_WORDS = ("shift", "verschieb", "смещ", "сдвиг")
_SHIFT_MODE = {(9, 2): "DPT9002", (6, 10): "DPT6010"}
def _is_shift(g: GARecord) -> bool:
return (g.dpt_main, g.dpt_sub) in _SHIFT_MODE and _has(g.name, _SHIFT_WORDS)
_NAME_TRIM = " \t-–—:;,./|_"
_NAME_SPLIT = "/-_.,()[]:;|–—"
# Words that name what an address DOES, never which device it is. Only these may be
# cut off the end of an entity name. Anything else — a room, "дверь", "А/С", "ТП",
# a channel letter — is identity and stays.
_NAME_FUNC_WORDS = frozenset(_FUNC_WORDS | _DIR_WORDS | {
"operation", "mode", "hvac", "absolute", "relative", "open", "close", "switch",
"switching", "b.value", "bewegen", "fahren", "wert", "helligkeit",
"режим", "абсолютное", "относительное", "димм", "открыть", "закрыть",
"управление", "команда",
})
def _name_words(text: str) -> list[str]:
low = text.lower().replace("b.value", " value ")
for ch in _NAME_SPLIT:
low = low.replace(ch, " ")
return [w for w in low.split() if w]
def _entity_name(anchor: str, member_names: list[str]) -> str:
"""Name a multi-GA entity after what its addresses share, not after one of them.
A light built from "Kitchen Spots On-Off", "Kitchen Spots B.Value" and their
feedbacks should be "Kitchen Spots", not whichever address anchored it. Takes the
longest common word prefix of the member names, and only accepts it when every
word cut from the anchor is a function word (value, brightness, up/down, mode,
Движение, Яркость…). If the names diverge earlier — a different room or device
word, a channel number — the anchor name is kept, which is exactly the behaviour
before this change. The candidate must still carry an identity token.
"""
names = [n for n in member_names if n and n.strip()]
if len(names) < 2:
return anchor
norm = [[w.strip(_NAME_TRIM).lower() for w in n.split()] for n in names]
common = 0
for i in range(min(len(ws) for ws in norm)):
if len({ws[i] for ws in norm}) != 1:
break
common = i + 1
anchor_words = anchor.split()
if common == 0 or common >= len(anchor_words):
return anchor
if [w.strip(_NAME_TRIM).lower() for w in anchor_words[:common]] != norm[0][:common]:
return anchor
dropped = _name_words(" ".join(anchor_words[common:]))
if any(w not in _NAME_FUNC_WORDS for w in dropped):
return anchor
candidate = " ".join(anchor_words[:common]).strip(_NAME_TRIM)
if not candidate or not _pair_ident(candidate):
return anchor
return candidate
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)]
@@ -454,6 +520,8 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]:
ctrl_cmd = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 105 and g.kind == "command")
ctrl_state = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 105 and _is_status_ga(g))
valve = _pick(lambda g: g.dpt_main == 5 and _is_status_ga(g))
shift_cmd = _pick(lambda g: _is_shift(g) and not _is_status_ga(g))
shift_state = _pick(lambda g: _is_shift(g) and _is_status_ga(g))
if not (cur and tgt_state):
review.append({"reason": "manual_climate", "address": ga.address,
@@ -478,7 +546,15 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]:
ent["controller_mode_state_address"] = ctrl_state.address
if valve:
ent["command_value_state_address"] = valve.address
for m in (cur, tgt_state, tgt_cmd, op_cmd, op_state, ctrl_cmd, ctrl_state, valve):
if shift_cmd:
ent["setpoint_shift_address"] = shift_cmd.address
if shift_state:
ent["setpoint_shift_state_address"] = shift_state.address
shift_ref = shift_cmd or shift_state
if shift_ref:
ent["setpoint_shift_mode"] = _SHIFT_MODE[(shift_ref.dpt_main, shift_ref.dpt_sub)]
for m in (cur, tgt_state, tgt_cmd, op_cmd, op_state, ctrl_cmd, ctrl_state, valve,
shift_cmd, shift_state):
if m:
consumed.add(m.address)
climates.append(ent)
@@ -491,11 +567,20 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]:
issues.append("operation_mode has a command but no state address")
if ctrl_cmd and not ctrl_state:
issues.append("controller_mode has a command but no state address")
if not tgt_cmd:
issues.append("setpoint is read-only (no target_temperature command)")
note = ("set `controller_modes`/`operation_modes` explicitly — HA auto-detection is "
"often wrong; if this zone uses setpoint-shift, provide BOTH the command and "
"state addresses and set `setpoint_shift_mode`")
if not tgt_cmd and not shift_cmd:
issues.append("setpoint is read-only (no target_temperature or setpoint shift command)")
if shift_cmd and not shift_state:
issues.append("setpoint shift has a command but no state address")
if shift_state and not shift_cmd:
issues.append("setpoint shift has a state but no command address")
if shift_ref:
note = ("set `controller_modes`/`operation_modes` explicitly — HA auto-detection is "
f"often wrong; setpoint shift mapped as {ent['setpoint_shift_mode']} from the "
"DPT, check it matches the thermostat's parameter")
else:
note = ("set `controller_modes`/`operation_modes` explicitly — HA auto-detection is "
"often wrong; if this zone uses setpoint-shift, provide BOTH the command and "
"state addresses and set `setpoint_shift_mode`")
if issues:
note += " — " + "; ".join(issues)
review.append({"reason": "verify_climate", "address": ga.address,
@@ -576,6 +661,27 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]:
"entity to an Area in the HA UI); and entity `name`s drive voice/Assist "
"matching — keep them descriptive and unique."})
# Multi-GA entities are named after what their addresses share (see _entity_name).
# Two entities that would end up with the same name keep their anchor names, so
# the rename never merges two things into one Home Assistant name.
multi = [e for group in (lights, covers, climates) for e in group]
proposed: dict[int, str] = {}
for e in multi:
members = [project.gas[v].name for k, v in e.items()
if k.endswith("address") and isinstance(v, str) and v in project.gas]
proposed[id(e)] = _entity_name(e["name"], members)
taken: dict[str, int] = {}
for e in multi:
key = proposed[id(e)].lower()
taken[key] = taken.get(key, 0) + 1
for e in switches + sensors + binary_sensors:
key = (e.get("name") or "").lower()
taken[key] = taken.get(key, 0) + 1
for e in multi:
new = proposed[id(e)]
if new != e["name"] and taken[new.lower()] == 1:
e["name"] = new
knx: dict[str, Any] = {}
if switches:
knx["switch"] = switches
+1 -1
View File
@@ -131,7 +131,7 @@ def suggest_repairs(project: LoadedProject) -> dict[str, Any]:
new = _next_free(used, sga.main if sga.main is not None else 1, prefer_middle=4)
proposals.append({
"code": "missing_status", "action": "add_ga", "address": new, "for": addr,
"name": f"{ga.name}{_suffix(ga.name, ' (статус)', ' (status)')}", "dpt": sdpt,
"name": f"{sga.name}{_suffix(sga.name, ' (статус)', ' (status)')}", "dpt": sdpt,
"rationale": "status/feedback GA so Home Assistant reads real state",
})
+36
View File
@@ -17,6 +17,7 @@ from typing import Any, Optional
from mcp.server.fastmcp import FastMCP
from .project import load_project as load_project_file, LoadedProject
from .ga_export import load_ga_export as load_ga_export_file
from .analyze import (validate_naming, detect_missing_status, detect_dpt_issues,
detect_topology_issues, secure_posture)
from .generate_ha import generate_ha_yaml
@@ -126,6 +127,41 @@ def load_project(path: str, password: Optional[str] = None,
}
@mcp.tool()
def load_ga_export(path: str) -> dict[str, Any]:
"""Load an ETS group-address export (ga-export/01 XML) instead of a full .knxproj.
For when you only have the GA list: an ETS "Export Group Addresses" file, or the
ETS import file a planning tool produces (TapPlan and similar). Names, addresses,
DPTs, descriptions, the security flag and the range tree are read; the result
replaces the loaded project for every other tool.
Works: check_naming, check_missing_status, check_dpt, check_policy, check_secure,
analyze_all, suggest_repairs, project_report, generate_ha_package,
generate_ets_group_addresses. Nothing to read (the export has no devices, ETS
Functions or topology): get_devices, get_topology, check_topology,
decompose_device, check_device_parameters, parse_devices_from_project. Pairing
relies on names only, since there are no ETS Function roles.
Args:
path: Path to the exported .xml file.
"""
proj = load_ga_export_file(path)
_STATE["project"] = proj
with_dpt = sum(1 for g in proj.gas.values() if g.dpt_main is not None)
return {
"loaded": True,
"source": "ga-export",
"name": proj.info.get("name"),
"ga_style": proj.style,
"group_addresses": len(proj.gas),
"with_dpt": with_dpt,
"devices": 0,
"import_warnings": proj.info.get("import_warnings", []),
"note": ("GA-only project: device, function and topology tools have nothing to read; "
"command/status pairing uses names only."),
}
@mcp.tool()
def list_group_addresses(category: Optional[str] = None,
kind: Optional[str] = None,
+193
View File
@@ -0,0 +1,193 @@
"""GA-export intake, multi-address entity naming, climate setpoint shift.
Fixture mirrors a real TapPlan export (2026-09-13): ga-export/01 namespace, main 0
range starting at 1, DPST tokens, block templates for a dimmable light, a roller
and an RTC. Extended with a setpoint-shift status and the DPT/security variants an
ETS export can carry.
"""
import os
import sys
import tempfile
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import yaml
from nickol_knx_mcp.ga_export import (GaExportError, load_ga_export, parse_dpt,
read_ga_export_bytes)
from nickol_knx_mcp.project import build_loaded_from_raw
from nickol_knx_mcp.generate_ha import generate_ha_yaml, _entity_name
from nickol_knx_mcp.generate_ets import generate_ets_xml
from nickol_knx_mcp.analyze import validate_naming, detect_topology_issues, detect_missing_status
XML = b'''<?xml version="1.0" encoding="utf-8" standalone="yes"?>
<GroupAddress-Export xmlns="http://knx.org/xml/ga-export/01">
<GroupRange Name="Lighting" RangeStart="1" RangeEnd="2047">
<GroupRange Name="Ground Floor" RangeStart="1" RangeEnd="255">
<GroupAddress Name="Kitchen Spots On-Off" Address="0/0/6" DPTs="DPST-1-1" />
<GroupAddress Name="Kitchen Spots On-Off Feedback" Address="0/0/7" DPTs="DPST-1-11" />
<GroupAddress Name="Kitchen Spots Dimm" Address="0/0/8" DPTs="DPST-3-7" />
<GroupAddress Name="Kitchen Spots B.Value" Address="0/0/9" DPTs="DPST-5-1" />
<GroupAddress Name="Kitchen Spots B.Value Feedback" Address="0/0/10" DPTs="DPST-5-1" />
</GroupRange>
</GroupRange>
<GroupRange Name="Shading" RangeStart="2048" RangeEnd="4095">
<GroupRange Name="Ground Floor" RangeStart="2048" RangeEnd="2303">
<GroupAddress Name="Bedroom Blind Long Operation" Address="1/0/3" DPTs="DPST-1-8" />
<GroupAddress Name="Bedroom Blind Short Operation" Address="1/0/4" DPTs="DPST-1-7" />
<GroupAddress Name="Bedroom Blind Position" Address="1/0/5" DPTs="DPST-5-1" />
<GroupAddress Name="Bedroom Blind Position Feedback" Address="1/0/6" DPTs="DPST-5-1" />
</GroupRange>
</GroupRange>
<GroupRange Name="Climate" RangeStart="4096" RangeEnd="6143">
<GroupRange Name="First Floor" RangeStart="4096" RangeEnd="4351">
<GroupAddress Name="Bedroom Climate Setpoint" Address="2/0/2" DPTs="DPST-9-1" />
<GroupAddress Name="Bedroom Climate Setpoint Shift" Address="2/0/3" DPTs="DPST-9-2" />
<GroupAddress Name="Bedroom Climate Setpoint Status" Address="2/0/4" DPTs="DPST-9-1" />
<GroupAddress Name="Bedroom Climate Actual Temp" Address="2/0/5" DPTs="DPST-9-1" />
<GroupAddress Name="Bedroom Climate Mode" Address="2/0/6" DPTs="DPST-20-102" />
<GroupAddress Name="Bedroom Climate Mode Status" Address="2/0/7" DPTs="DPST-20-102" />
<GroupAddress Name="Bedroom Climate Setpoint Shift Status" Address="2/0/8" DPTs="DPST-9-2" />
<GroupAddress Name="Hall Secure Temp" Address="2/0/9" DPTs="DPT-9" Security="On" Description="door sensor" />
</GroupRange>
</GroupRange>
</GroupAddress-Export>
'''
def _load(data=XML):
with tempfile.NamedTemporaryFile(suffix=".xml", delete=False) as fh:
fh.write(data)
path = fh.name
try:
return load_ga_export(path)
finally:
os.unlink(path)
def _knx(proj):
return yaml.safe_load(generate_ha_yaml(proj)["yaml"])["knx"]
# 1. Intake: every GA, DPTs, style, range names, security, description.
p = _load()
assert len(p.gas) == 17, len(p.gas)
assert p.style == "ThreeLevel", p.style
assert p.info["source"] == "ga-export" and p.info["import_warnings"] == []
assert (p.gas["0/0/7"].dpt_main, p.gas["0/0/7"].dpt_sub) == (1, 11)
assert (p.gas["2/0/9"].dpt_main, p.gas["2/0/9"].dpt_sub) == (9, None)
assert p.gas["2/0/9"].data_secure and p.gas["2/0/9"].description == "door sensor"
assert not p.gas["0/0/6"].data_secure
assert not p.devices and not p.functions
print("OK: intake — 17 GAs, ThreeLevel, DPST/DPT tokens, Security, Description")
# 2. Device-less project runs the GA-level checks without inventing device findings.
codes = {f["code"] for f in validate_naming(p)}
assert "ga_style_not_three_level" not in codes, codes
assert detect_topology_issues(p) == []
assert not [f for f in detect_missing_status(p) if f["address"] in ("0/0/9", "1/0/5")]
print("OK: GA-level checks run; no style or topology noise on a GA-only project")
# 3. Naming: entities are named after the shared part of their member names.
knx = _knx(p)
light = knx["light"][0]
cover = knx["cover"][0]
climate = knx["climate"][0]
assert light["name"] == "Kitchen Spots", light
assert (light["address"], light["state_address"], light["brightness_address"],
light["brightness_state_address"]) == ("0/0/6", "0/0/7", "0/0/9", "0/0/10"), light
assert cover["name"] == "Bedroom Blind", cover
assert climate["name"] == "Bedroom Climate", climate
print("OK: naming — 'Kitchen Spots', 'Bedroom Blind', 'Bedroom Climate'")
# 4. Setpoint shift lands in the climate, with the mode from the DPT, not as a sensor.
assert climate["setpoint_shift_address"] == "2/0/3", climate
assert climate["setpoint_shift_state_address"] == "2/0/8", climate
assert climate["setpoint_shift_mode"] == "DPT9002", climate
assert climate["target_temperature_state_address"] == "2/0/4"
assert climate["temperature_address"] == "2/0/5"
sensor_addrs = {s["state_address"] for s in knx.get("sensor", [])}
assert not sensor_addrs & {"2/0/3", "2/0/8"}, sensor_addrs
print("OK: setpoint shift -> climate setpoint_shift_address/state, mode DPT9002")
# 4b. 6.010 shift gives DPT6010; a 9.002 without the word stays out of the climate.
x610 = XML.replace(b'Address="2/0/3" DPTs="DPST-9-2"', b'Address="2/0/3" DPTs="DPST-6-10"') \
.replace(b'Address="2/0/8" DPTs="DPST-9-2"', b'Address="2/0/8" DPTs="DPST-6-10"')
assert _knx(_load(x610))["climate"][0]["setpoint_shift_mode"] == "DPT6010"
noword = XML.replace(b"Setpoint Shift Status", b"Delta Status").replace(b"Setpoint Shift", b"Delta")
c2 = _knx(_load(noword))["climate"][0]
assert "setpoint_shift_address" not in c2 and "setpoint_shift_mode" not in c2, c2
print("OK: 6.010 -> DPT6010; a bare 9.002 without a shift word is not taken as a shift")
# 5. Naming guards.
assert _entity_name("Spot 1 Brightness", ["Spot 1 Brightness", "Spot 12 Status"]) == "Spot 1 Brightness"
assert _entity_name("Licht Küche Wert", ["Licht Küche Wert", "RM Licht Küche"]) == "Licht Küche Wert"
assert _entity_name("Kitchen Light", ["Kitchen Light"]) == "Kitchen Light"
assert _entity_name("Kanal A Helligkeit", ["Kanal A Helligkeit", "Kanal A Schalten"]) == "Kanal A"
assert _entity_name("Status: Wert", ["Status: Wert", "Status: Schalten"]) == "Status: Wert"
# Shapes seen on real projects (anonymised). Trim only function words; anything that
# names the room, the device or its type stays, even when the names diverge early.
assert _entity_name("01. Hall - All Blinds - Move", ["01. Hall - All Blinds - Move", "01. Hall blinds stop"]) \
== "01. Hall - All Blinds - Move"
assert _entity_name("ch1 long door", ["ch1 long door", "ch1 short door"]) == "ch1 long door"
assert _entity_name("06. Bath/Convector - Mode", ["06. Bath/Convector - Mode", "06. Bath temperature"]) \
== "06. Bath/Convector - Mode"
assert _entity_name("28. Living - A/C - Mode", ["28. Living - A/C - Mode", "28. Living - A/C - Mode status"]) \
== "28. Living - A/C - Mode"
assert _entity_name("01. Hall - FH - HVAC", ["01. Hall - FH - HVAC", "01. Hall temperature"]) == "01. Hall - FH - HVAC"
assert _entity_name("Kitchen ceiling - Absolute dimming", ["Kitchen ceiling - Absolute dimming",
"Kitchen ceiling - On"]) == "Kitchen ceiling"
assert _entity_name("Garage gate left - Open/Close", ["Garage gate left - Open/Close",
"Garage gate left - Stop"]) == "Garage gate left"
assert _entity_name("Living room blind up/down", ["Living room blind up/down",
"Living room blind position"]) == "Living room blind"
# Collision: two lights whose common prefix would be the same name keep their anchors.
two = XML.replace(b"</GroupRange>\n </GroupRange>\n <GroupRange Name=\"Shading\"", b'''
<GroupAddress Name="Kitchen Spots B.Value Left" Address="0/0/20" DPTs="DPST-5-1" />
<GroupAddress Name="Kitchen Spots On-Off Left" Address="0/0/21" DPTs="DPST-1-1" />
</GroupRange>
</GroupRange>
<GroupRange Name="Shading"''', 1)
names = sorted(e["name"] for e in _knx(_load(two))["light"])
assert len(names) == len(set(n.lower() for n in names)), names
print("OK: naming guards — digit identity kept, no common prefix, no identity, collisions")
# 6. Round trip: our own ETS export reads back to the same GAs and DPTs.
back = build_loaded_from_raw(read_ga_export_bytes(generate_ets_xml(p).encode("utf-8"), "rt"), "rt")
assert set(back.gas) == set(p.gas)
for a, g in p.gas.items():
assert (back.gas[a].dpt_main, back.gas[a].dpt_sub, back.gas[a].name) == (g.dpt_main, g.dpt_sub, g.name), a
print("OK: generate_ets_xml -> load_ga_export round trip keeps every GA, name and DPT")
# 7. Hostile and malformed input is refused with a plain error, and warnings are never silent.
for bad, why in [
(b'<?xml version="1.0"?><!DOCTYPE x [<!ENTITY a "b">]><GroupAddress-Export/>', "DTD"),
(b"<Project/>", "root"),
(b"<GroupAddress-Export", "malformed"),
(b'<GroupAddress-Export xmlns="http://knx.org/xml/ga-export/01">' + b'<GroupRange Name="x">' * 2000
+ b'</GroupRange>' * 2000 + b"</GroupAddress-Export>", "nesting bomb"),
]:
try:
_load(bad)
except GaExportError:
pass
else:
raise AssertionError(f"accepted {why}")
w = _load(b'''<GroupAddress-Export xmlns="http://knx.org/xml/ga-export/01">
<GroupAddress Name="a" Address="1/0/1" DPTs="DPST-1-1" />
<GroupAddress Name="dup" Address="1/0/1" DPTs="DPST-1-1" />
<GroupAddress Name="bad" Address="99/9/999" />
<GroupAddress Name="weird" Address="1/0/2" DPTs="foo" />
</GroupAddress-Export>''')
assert set(w.gas) == {"1/0/1", "1/0/2"}
assert len(w.info["import_warnings"]) == 3, w.info["import_warnings"]
assert parse_dpt("DPST-1-1 DPST-1-11") == {"main": 1, "sub": 1} and parse_dpt(None) is None
try:
load_ga_export("/nonexistent/x.xml")
except GaExportError:
pass
else:
raise AssertionError("missing file accepted")
print("OK: DTD, wrong root, malformed, nesting bomb and missing files refused; duplicates/invalid/unknown DPT warned")
print("\nALL GA-EXPORT / NAMING / SETPOINT-SHIFT TESTS PASSED")