diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1953478..4c69019 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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'])" diff --git a/CHANGELOG.md b/CHANGELOG.md index 877ee67..a98fb9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md index da6bbe5..39e357b 100644 --- a/README.md +++ b/README.md @@ -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/ diff --git a/nickol_knx_mcp/ga_export.py b/nickol_knx_mcp/ga_export.py new file mode 100644 index 0000000..0a9fd66 --- /dev/null +++ b/nickol_knx_mcp/ga_export.py @@ -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) diff --git a/nickol_knx_mcp/generate_ha.py b/nickol_knx_mcp/generate_ha.py index 67b8051..0b817d5 100644 --- a/nickol_knx_mcp/generate_ha.py +++ b/nickol_knx_mcp/generate_ha.py @@ -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 diff --git a/nickol_knx_mcp/repair.py b/nickol_knx_mcp/repair.py index 48f534a..403ea07 100644 --- a/nickol_knx_mcp/repair.py +++ b/nickol_knx_mcp/repair.py @@ -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", }) diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index 1b685d0..f4abd44 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -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, diff --git a/tests/test_ga_export.py b/tests/test_ga_export.py new file mode 100644 index 0000000..78b2930 --- /dev/null +++ b/tests/test_ga_export.py @@ -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''' + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +''' + + +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"\n \n + + + + 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']>', "DTD"), + (b"", "root"), + (b"' + b'' * 2000 + + b'' * 2000 + b"", "nesting bomb"), +]: + try: + _load(bad) + except GaExportError: + pass + else: + raise AssertionError(f"accepted {why}") +w = _load(b''' + + + + +''') +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")