mirror of
https://github.com/NickoScope/nickol-knx-mcp.git
synced 2026-09-29 19:31:12 +02:00
Nikolay MiroshnichenkoandClaude Opus 5
a3f436fc74
Found on a real 1312-GA house: a room with floor heating, a convector and an AC unit became
ONE climate (convector setpoint, AC controller mode, AC fan speed as the valve). Room 1.09
matched 2.09 and a central "09." GA, "Kids room 1" took "Kids room 2 temperature", and valves
came from light brightness and blind position statuses. The zone was the room alone: device
words were stripped as qualifiers, "А/С" vanishes in the tokenizer, the floor digit of "1.09"
is dropped.
Now a climate is assembled per room AND device type (floor, wall, radiator/convector, fan
coil, AC; RU/EN/DE words, removed from the zone as whole words):
- room code "N.NN" only at the start of a name, and it must match on both sides;
- standalone numbers of the anchor must be in a device member; a shared room sensor's
numbers must be among the anchor's; digits in dotted codes and values do not count;
- an untyped GA is shared room data: any role if the room has one device, otherwise only
the current temperature, which several devices may reuse; with a room code the room's
untyped sensor qualifies even without the device's extra words;
- a word that sets another device of the room apart keeps its GAs away;
- control roles must be unambiguous: more than one candidate -> review climate_ambiguous;
- two mode GAs that cannot be told apart -> review climate_duplicate_anchor, not dropped;
- only 20.102 / 20.105 anchor; valve needs a valve word; AC never gets a valve;
- stable address order, so results never depend on parse order.
Two gates. Own audit on seven real projects, every changed entity inspected: no control GA
used by two climates, no address lost; house 16 -> 33 climates (one per device), manual
review 33 -> 12; flat 7 -> 9; HDL 3 -> 5; demo 6 -> 13; villa 23 -> 21 with the 3 it cannot
disambiguate sent to review instead of guessed. LLM council (three models + devil's
advocate, sanitised packet): accept with hardening; its P0s (same-type devices, silent
first pick) and P1s (mid-name codes, dotted digits, non-102/105 anchors, generic valve
words) are all in this commit, with tests E and F covering its failure catalogue.
Out of scope, recorded: identically named rooms of different flats without room codes, and
lexical identity in general -> device-channel identity ("lever 0").
Also works around a mypy 2.3.1 parse error on a comment right after a compound if.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
938 lines
48 KiB
Python
938 lines
48 KiB
Python
"""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
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
import yaml
|
||
|
||
from .project import LoadedProject, GARecord
|
||
from .analyze import _is_status_ga
|
||
from .pairing import find_status, base_tokens, function_status_pairs, self_reporting
|
||
|
||
# Venetian-blind slat (tilt) detection: a slat GA is the tilt of its parent
|
||
# blind, not a standalone cover.
|
||
_SLAT_WORDS = ("lamelle", "lamel", "ламел", "slat", "louver", "louvre", "tilt")
|
||
_DIR_WORDS = {"auf", "ab", "up", "down", "hoch", "runter", "move", "step",
|
||
"long", "short", "open", "close", "вверх", "вниз"}
|
||
_BLIND_GENERIC = {"behang", "blind", "blinds", "shutter", "jalousie", "rollo",
|
||
"roller", "store", "markise", "roll", "cover", "curtain",
|
||
"штора", "жалюзи", "ролета", "ролл"}
|
||
|
||
|
||
def _is_slat(name: str) -> bool:
|
||
low = name.lower()
|
||
return any(w in low for w in _SLAT_WORDS)
|
||
|
||
|
||
def _ident_tokens(name: str) -> set:
|
||
"""Identity/zone tokens for matching a slat to its blind (drop direction,
|
||
slat and generic-blind words so only the zone/name identity remains)."""
|
||
return {t for t in base_tokens(name)
|
||
if t not in _DIR_WORDS and t not in _SLAT_WORDS and t not in _BLIND_GENERIC}
|
||
|
||
|
||
_UPDOWN_PHRASES = ("up/down", "auf/ab", "ab/auf", "updown", "up-down", "вверх/вниз")
|
||
_STOP_WORDS = ("stop", "stopp", "стоп")
|
||
|
||
# Tokens that mark a bare 1.007/1.010 as belonging to ANOTHER domain, so it must
|
||
# never be admitted as a cover's step/stop even if its name says "stop" — a
|
||
# "stop" is an OPERATION signal, not a DOMAIN one (council review, issue #11).
|
||
_FOREIGN_DOMAIN_TOKENS = frozenset({
|
||
"light", "licht", "lamp", "lampe", "dim", "dimm", "dimmen", "dimming", "led",
|
||
"socket", "steckdose", "outlet", "heiz", "heizung", "heating", "hvac", "klima",
|
||
"свет", "розетка", "диммер"})
|
||
# NB: ventilation words (vent/Lüftung/fan) are intentionally NOT here — a roof-window
|
||
# or ventilation-flap opener is a legitimate cover, so those stay admissible.
|
||
# Central/collective tokens: a house-wide "Alle Stopp" 1.007 is not one cover's own
|
||
# step/stop and must not be greedily stolen by the first cover (council review). NB:
|
||
# "master" is intentionally NOT here — it is a common room qualifier (Master Bedroom)
|
||
# and excluding it dropped legitimate covers (gate-1 audit); real macros use all/zentral.
|
||
_CENTRAL_TOKENS = frozenset({
|
||
"all", "alle", "zentral", "zentrale", "central", "global", "gesamt",
|
||
"universal", "все", "центр", "общий"})
|
||
|
||
|
||
def _is_updown(name: str) -> bool:
|
||
"""A shutter move (long) command: 'up/down', 'auf/ab', or both directions."""
|
||
low = name.lower()
|
||
if any(p in low for p in _UPDOWN_PHRASES):
|
||
return True
|
||
toks = set(base_tokens(low))
|
||
return ("up" in toks and "down" in toks) or ("auf" in toks and "ab" in toks)
|
||
|
||
|
||
def _is_stop(name: str) -> bool:
|
||
low = name.lower()
|
||
return any(w in low for w in _STOP_WORDS)
|
||
|
||
|
||
def _is_shutter_control(g, slat_addrs) -> bool:
|
||
"""A bare step/stop object is a shutter control even when the classifier left
|
||
its category 'unknown' for want of a name keyword or Function role — the
|
||
function-less collective case (issue #11 group C). DPT 1.007 (step) / 1.010
|
||
(start-stop), or a slat object. Position 5.001 still requires a shutter
|
||
category (it classifies reliably), so it is intentionally NOT admitted here.
|
||
|
||
The bare-DPT admission REQUIRES a shutter-ish name signal (stop / up-down / a
|
||
generic blind word): DPT 1.007/1.010 alone is domain-agnostic, so admitting it
|
||
on the DPT alone would let a foreign 1.007 (e.g. a lighting relative-dim step)
|
||
sharing only a zone token be mis-paired as a cover's step/stop — audit finding
|
||
on the issue #11 fix. A slat address is already shutter-derived, so it stands."""
|
||
if g.address in slat_addrs:
|
||
return True
|
||
if g.dpt_main == 1 and g.dpt_sub in (7, 10):
|
||
toks = set(base_tokens(g.name or ""))
|
||
# A foreign-domain step (lighting dim-stop, HVAC) or a central/collective
|
||
# "Alle Stopp" is not a cover's own step/stop — reject before the name
|
||
# operation signal (council review: operation != domain; no master-steal).
|
||
if toks & _FOREIGN_DOMAIN_TOKENS or toks & _CENTRAL_TOKENS:
|
||
return False
|
||
low = (g.name or "").lower()
|
||
return (_is_stop(g.name) or _is_updown(g.name)
|
||
or any(w in low for w in _BLIND_GENERIC))
|
||
return False
|
||
|
||
|
||
# Function words that differ between a command and its status (on/off, value,
|
||
# state, brightness, ...). Stripping them leaves the device/zone identity, so a
|
||
# command pairs to its feedback even when the identity is a single token
|
||
# (e.g. "HaloSpotLeft.A.VALUE" <-> "HaloSpotLeft.A.STATE%").
|
||
_FUNC_WORDS = {
|
||
"on", "off", "onoff", "value", "val", "dim", "dimming", "bright", "brightness",
|
||
"state", "status", "stat", "fb", "feedback", "rm", "pct", "percent",
|
||
"up", "down", "move", "stop", "step", "pos", "position", "set", "control",
|
||
"schalt", "schalten", "steuer", "wert", "helligkeit", "rueck", "rück", "soll",
|
||
# colour function words: ignored so a colour GA pairs to its zone's light
|
||
"colour", "color", "rgb", "rgbw", "rgbww", "hue", "saturation", "sat",
|
||
"white", "warm", "cct", "kelvin", "temp", "temperature", "xyy", "farbe",
|
||
# Russian function words — identity is the zone/device, not the verb
|
||
"вкл", "выкл", "статус", "состояние", "яркость", "значение", "диммирование",
|
||
"цвет", "позиция", "движение", "стоп", "вверх", "вниз",
|
||
}
|
||
|
||
|
||
def _pair_ident(name: str) -> set:
|
||
"""Device/zone identity tokens for command<->status pairing."""
|
||
return {t for t in base_tokens(name)
|
||
if t not in _FUNC_WORDS and not t.endswith("%")}
|
||
|
||
|
||
def _identity_match(a_name: str, b_name: str) -> bool:
|
||
"""True when two names share the SAME device/zone identity.
|
||
|
||
One identity must contain the other — a single shared zone token (e.g.
|
||
"kitchen") is NOT enough. This stops a command from borrowing a sibling's
|
||
status when it has none of its own (bug B1: "worktop LED" must not grab
|
||
"island pendants" status just because both are "kitchen")."""
|
||
a, b = _pair_ident(a_name), _pair_ident(b_name)
|
||
if not a or not b:
|
||
return False
|
||
return a <= b or b <= a
|
||
|
||
|
||
# Colour command DPT -> (HA address key, HA state-address key).
|
||
_COLOUR_DPT: dict[tuple[int, int], tuple[str, str]] = {
|
||
(232, 600): ("color_address", "color_state_address"), # RGB
|
||
(251, 600): ("rgbw_address", "rgbw_state_address"), # RGBW
|
||
(242, 600): ("xyy_address", "xyy_state_address"), # xyY
|
||
}
|
||
|
||
# Words marking a target/setpoint temperature (vs the current room temperature).
|
||
_TARGET_WORDS = ("target", "setpoint", "soll", "sollwert", "уставк", "задан",
|
||
"целев", "зад.", "устав")
|
||
|
||
# HVAC function qualifiers stripped from a climate anchor's name to leave only
|
||
# the zone/location tokens — climate members (current temp, target, mode, valve)
|
||
# share the LOCATION, not the full identity, so they are gathered by location.
|
||
_CLIMATE_QUALIFIERS = {
|
||
"hvac", "mode", "controller", "operation", "heat", "cool", "heating",
|
||
"cooling", "climate", "thermostat", "ac", "тп", "режим", "отопл", "климат",
|
||
"термостат", "конвектор", "тёплый", "теплый", "пол",
|
||
}
|
||
|
||
|
||
def _has(name: str, words) -> bool:
|
||
low = name.lower()
|
||
return any(w in low for w in words)
|
||
|
||
|
||
# ---- Climate zone identity -------------------------------------------------------------
|
||
# A room often holds several climate devices: floor heating, a radiator or convector, an AC
|
||
# unit, sometimes wall heating or a second floor loop. Each is its own Home Assistant climate.
|
||
# The zone used to be the room alone (device words were stripped as qualifiers, and "А/С"
|
||
# vanishes in the tokenizer), so all of them were merged into one entity. Found on a real
|
||
# 1312-GA house; hardened after an LLM-council review (same-type devices, ambiguous picks).
|
||
|
||
# "1.09" style room code at the START of a name. The tokenizer drops the floor digit
|
||
# ("1.09" -> "09"), which made 1.09 match 2.09 and a central "09. ..." GA. Mid-name dotted
|
||
# numbers are device tags ("ДД 34.1"), dates ("16.10") or values ("21.5 °C"), not rooms.
|
||
_ROOM_CODE_RE = re.compile(r"^\W*(\d{1,2})\s*\.\s*(\d{1,2})(?![\d.])")
|
||
|
||
|
||
def _room_code(name: str) -> str | None:
|
||
m = _ROOM_CODE_RE.search(name or "")
|
||
return f"{int(m.group(1))}.{int(m.group(2)):02d}" if m else None
|
||
|
||
|
||
_CLIMATE_DEVICE_PATTERNS: tuple[tuple[str, re.Pattern], ...] = (
|
||
# bare "floor" only as a delimited name part ("Floor-Living-RealTemp") or right before
|
||
# valve/loop/circuit/heating, so a level ("1st floor", "ground floor") does not count
|
||
("floor", re.compile(r"тепл\w*\s+пол|водян\w*\s+пол|\bтп\b|fu(?:ss|ß)boden|floor\s*heat|underfloor"
|
||
r"|(?:^|[-_/])floor(?=[-_/]|$)|\bfloor(?=\s+(?:valve|loop|circuit|heating)\b)")),
|
||
("wall", re.compile(r"тепл\w*\s+стен|\bстена\b|wandheiz|wall\s*heat")),
|
||
("radiator", re.compile(r"радиатор|конвектор|батаре|radiator|convector|heizk(?:ö|oe)rper")),
|
||
# a fan coil is hydronic, not an AC unit; its own type keeps it apart from both
|
||
("fancoil", re.compile(r"fan\s*-?\s*coil|фан\s*-?\s*койл|\bfcu\b")),
|
||
# bare German "Klima" is deliberately absent: it also just means "climate"
|
||
("ac", re.compile(r"(?<![a-zа-я])(?:а/с|a/c|ac)(?![a-zа-я])|кондиционер|сплит|split|klimaanlage|air\s*con"
|
||
r"|\bvr[vf]\b|\bврв\b")),
|
||
)
|
||
|
||
|
||
# Standalone numbers in a name ("Kids room 1", "2. Гостиная"). The tokenizer drops single
|
||
# digits, so "Kids room 1" matched "Kids room 2". Digits glued to a word ("Статус_1") or
|
||
# part of a dotted code or value ("1.09", "D.1.2", "21.5") are not counted.
|
||
_NUMBER_RE = re.compile(r"(?<![\w.])\d+(?![\w]|\.\d)")
|
||
|
||
|
||
def _numbers(name: str) -> frozenset:
|
||
return frozenset(int(n) for n in _NUMBER_RE.findall(name or ""))
|
||
|
||
|
||
def _without_device_words(name: str) -> str:
|
||
"""The name with every climate device word removed, so the room identity is the same for
|
||
"Convector - Mode", "Конвектор - Режим" and a shared "Air temperature" in that room."""
|
||
low = (name or "").lower().replace("ё", "е")
|
||
blank = [False] * len(low)
|
||
for _kind, rx in _CLIMATE_DEVICE_PATTERNS:
|
||
for m in rx.finditer(low):
|
||
# widen to whole words: "floor heat" inside "floor heating" must not leave "ing"
|
||
start, end = m.start(), m.end()
|
||
while start > 0 and low[start - 1].isalnum():
|
||
start -= 1
|
||
while end < len(low) and low[end].isalnum():
|
||
end += 1
|
||
for i in range(start, end):
|
||
blank[i] = True
|
||
return "".join(" " if b else ch for ch, b in zip(low, blank))
|
||
|
||
|
||
def _climate_types(name: str) -> frozenset:
|
||
low = (name or "").lower().replace("ё", "е")
|
||
return frozenset(kind for kind, rx in _CLIMATE_DEVICE_PATTERNS if rx.search(low))
|
||
|
||
|
||
def _addr_key(address: str) -> tuple:
|
||
try:
|
||
return (0, tuple(int(x) for x in (address or "").split("/")))
|
||
except ValueError:
|
||
return (1, (address or "",))
|
||
|
||
|
||
# A 5.x status is a heating valve only when the name says so; an AC fan speed (5.001
|
||
# "Вентилятор") or a generic "control value" is not.
|
||
_VALVE_WORDS = ("клапан", "valve", "stellwert", "stellgr")
|
||
|
||
|
||
|
||
# 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)]
|
||
fpairs = function_status_pairs(project) # authoritative cmd-addr -> status-addr
|
||
slat_addrs = {g.address for g in project.gas.values()
|
||
if g.category == "shutter" and _is_slat(g.name)}
|
||
|
||
# Map each GA address to the ETS Function(s) that own it. A cover's step/stop
|
||
# and position/status must never be paired across an ETS Function boundary:
|
||
# in a room with both a window and an awning the names share a zone token
|
||
# ("Room A"), so name-token matching alone grabs the wrong shutter's step/stop
|
||
# (issue #11). ETS Function membership is the authoritative grouping.
|
||
addr_to_fns: dict[str, set[str]] = {}
|
||
for fkey, fn in (project.functions or {}).items():
|
||
for a_key, ref in (fn.get("group_addresses", {}) or {}).items():
|
||
a = ref.get("address") or a_key # dict key is the address fallback (cf. pairing.py)
|
||
if a:
|
||
addr_to_fns.setdefault(a, set()).add(fkey)
|
||
|
||
def status_for(cmd: GARecord):
|
||
# 1. ETS Function role pairing is authoritative (names not needed).
|
||
paired = fpairs.get(cmd.address)
|
||
if paired and paired in project.gas:
|
||
return project.gas[paired]
|
||
# 2. fall back to name-token pairing.
|
||
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 status_for_dpt(cmd: GARecord, want_main: int):
|
||
"""Find a status GA for cmd of a given DPT main, matched by device/zone
|
||
identity. Lets a dimmable light find BOTH its on/off status (1.x) and
|
||
brightness status (5.x), and pairs even when identity is a single token."""
|
||
paired = fpairs.get(cmd.address)
|
||
if paired and paired in project.gas and project.gas[paired].dpt_main == want_main:
|
||
return project.gas[paired]
|
||
cid = _pair_ident(cmd.name)
|
||
if not cid:
|
||
cands = [s for s in status_gas
|
||
if s.dpt_main == want_main and s.address not in consumed]
|
||
same_main = [s for s in cands if s.main == cmd.main]
|
||
return find_status(cmd, same_main) or find_status(cmd, cands)
|
||
best, best_score = None, 0
|
||
for s in status_gas:
|
||
if s.dpt_main != want_main or s.address == cmd.address:
|
||
continue
|
||
if s.address in consumed: # a status maps to exactly one entity
|
||
continue
|
||
# B1: require a full identity match, not just a shared zone token,
|
||
# so a command never borrows a sibling's status.
|
||
if not _identity_match(cmd.name, s.name):
|
||
continue
|
||
sid = _pair_ident(s.name)
|
||
# EXACT identity outranks any subset overlap: "RGBW подсветка"
|
||
# must take "RGBW подсветка статус" over "камин подсветка статус"
|
||
# even though both share {zone, подсветка}.
|
||
score = len(cid & sid) + (2 if s.main == cmd.main else 0) \
|
||
+ (3 if sid == cid else 0)
|
||
if score > best_score:
|
||
best, best_score = s, score
|
||
return best
|
||
|
||
def same_main_gas(cmd: GARecord):
|
||
return [g for g in project.gas.values() if g.main == cmd.main]
|
||
|
||
def attach_colour(entity: dict, ref: GARecord) -> None:
|
||
"""Attach RGB / RGBW / xyY / colour-temperature GAs (and their statuses)
|
||
of the same zone identity to a light entity."""
|
||
for sib in same_main_gas(ref):
|
||
if sib.address in consumed or sib.kind != "command":
|
||
continue
|
||
if not _identity_match(ref.name, sib.name):
|
||
continue
|
||
key = (sib.dpt_main, sib.dpt_sub)
|
||
if key in _COLOUR_DPT:
|
||
a_key, s_key = _COLOUR_DPT[key]
|
||
if a_key in entity:
|
||
continue
|
||
entity[a_key] = sib.address
|
||
consumed.add(sib.address)
|
||
cst = status_for_dpt(sib, sib.dpt_main)
|
||
if cst:
|
||
entity[s_key] = cst.address
|
||
consumed.add(cst.address)
|
||
elif key == (7, 600) and "color_temperature_address" not in entity:
|
||
entity["color_temperature_address"] = sib.address
|
||
entity["color_temperature_mode"] = "absolute"
|
||
consumed.add(sib.address)
|
||
cst = status_for_dpt(sib, 7)
|
||
if cst:
|
||
entity["color_temperature_state_address"] = cst.address
|
||
consumed.add(cst.address)
|
||
|
||
switches: list[dict] = []
|
||
binary_sensors: list[dict] = []
|
||
sensors: list[dict] = []
|
||
lights: list[dict] = []
|
||
covers: list[dict] = []
|
||
climates: list[dict] = []
|
||
review: list[dict[str, Any]] = []
|
||
consumed: set[str] = set()
|
||
|
||
# ---- 1. COVERS first (they own 5.001 position) ----
|
||
# A cover's "move long" is a shutter command that is up/down: canonical DPT
|
||
# 1.008, OR a 1.x command whose name says up/down (real projects use 1.001).
|
||
for ga in project.gas.values():
|
||
if ga.address in consumed:
|
||
continue
|
||
is_move_long = (ga.category == "shutter" and ga.kind == "command"
|
||
and ga.dpt_main == 1 and ga.address not in slat_addrs
|
||
and (ga.dpt_sub == 8 or _is_updown(ga.name)))
|
||
if not is_move_long:
|
||
continue
|
||
entity = {"name": ga.name, "move_long_address": ga.address}
|
||
ptoks = _ident_tokens(ga.name)
|
||
my_fns = addr_to_fns.get(ga.address, set())
|
||
# Gather the eligible siblings once, then pick the BEST candidate per role
|
||
# rather than first-match — so a step/stop that shares the move's type AND
|
||
# zone token beats one that shares the zone alone (a "West Side Roller
|
||
# Shutters" move must take its own step/stop, not the "West Side Awnings"
|
||
# one) — issue #11 group C.
|
||
cands = [] # (sib, same_fn, overlap)
|
||
for sib in same_main_gas(ga):
|
||
if sib.address in consumed or sib.address == ga.address:
|
||
continue
|
||
# Admit shutter-category siblings AND bare step/stop DPTs the
|
||
# classifier left 'unknown' (function-less collectives — issue #11).
|
||
if sib.category != "shutter" and not _is_shutter_control(sib, slat_addrs):
|
||
continue
|
||
sib_fns = addr_to_fns.get(sib.address, set())
|
||
same_fn = bool(sib_fns & my_fns)
|
||
# Never cross an ETS Function boundary: a sibling owned by a DIFFERENT
|
||
# function is another shutter's GA even when the zone token matches
|
||
# (a window vs an awning in the same room) — issue #11 group B.
|
||
if sib_fns and not same_fn:
|
||
continue
|
||
overlap = len(ptoks & _ident_tokens(sib.name))
|
||
# A same-function sibling is authoritative (names not needed);
|
||
# otherwise require a shared zone/type identity as before.
|
||
if not same_fn and ptoks and _ident_tokens(sib.name) and overlap == 0:
|
||
continue
|
||
cands.append((sib, same_fn, overlap))
|
||
|
||
def _rank(item):
|
||
# same ETS Function first (authoritative), then the strongest name
|
||
# overlap (type+zone beats zone alone), then the nearest sub index, then
|
||
# a stable lexicographic address tiebreak so the pick is DETERMINISTIC
|
||
# across parses on a full tie (council review; a true semantic tie is a
|
||
# known limitation — determinism here is not a correctness claim).
|
||
sib, same_fn, overlap = item
|
||
return (1 if same_fn else 0, overlap, -abs((sib.sub or 0) - (ga.sub or 0)),
|
||
-(sib.main or 0), -(sib.middle or 0), -(sib.sub or 0))
|
||
|
||
def _take(pred, key):
|
||
pool = [c for c in cands
|
||
if c[0].address not in consumed and pred(c[0])]
|
||
best = max(pool, key=_rank, default=None)
|
||
if best is not None:
|
||
entity[key] = best[0].address
|
||
consumed.add(best[0].address)
|
||
|
||
# step/stop or slat -> the short move; then position command; then status.
|
||
_take(lambda s: (s.dpt_main == 1 and s.dpt_sub in (7, 10, 17))
|
||
or _is_stop(s.name) or s.address in slat_addrs, "move_short_address")
|
||
_take(lambda s: s.dpt_main == 5 and s.kind == "command", "position_address")
|
||
_take(lambda s: s.dpt_main == 5 and _is_status_ga(s), "position_state_address")
|
||
covers.append(entity)
|
||
consumed.add(ga.address)
|
||
# A3: surface the actuator-dependent flags that are NOT in the .knxproj —
|
||
# the three invert flags (position / up-down / angle) and travel times —
|
||
# so the installer sets them; and whether position lacks its state address.
|
||
note = ("set actuator-dependent flags not in the .knxproj: "
|
||
"invert_position / invert_updown / invert_angle, and "
|
||
"travelling_time_up / travelling_time_down")
|
||
if "position_address" in entity and "position_state_address" not in entity:
|
||
note += " — position has NO position_state_address (HA will guess position)"
|
||
review.append({"reason": "verify_cover_invert", "address": ga.address,
|
||
"name": ga.name, "note": note})
|
||
|
||
# ---- 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":
|
||
# two-pass sibling pick: EXACT identity first, subset only as a
|
||
# fallback — else "RGBW подсветка яркость" ({living, подсветка})
|
||
# grabs "камин подсветка" ({living, камин, подсветка}) by subset
|
||
# while its true on/off sits one address further.
|
||
sibs = [s for s in same_main_gas(ga)
|
||
if s.address not in consumed and s.dpt_main == 1
|
||
and s.kind == "command" and s.category in ("lighting", "unknown")]
|
||
my_ident = _pair_ident(ga.name)
|
||
sib = next((s for s in sibs if _pair_ident(s.name) == my_ident), None) \
|
||
or next((s for s in sibs if _identity_match(ga.name, s.name)), None)
|
||
if sib is None:
|
||
# Home Assistant requires `address` on a KNX light, so a brightness GA
|
||
# with no on/off GA in its zone would be invalid YAML. On real projects
|
||
# these are mostly device parameters on 5.001 (motion-detector
|
||
# sensitivity, "daytime command"), not lights. Fail closed.
|
||
review.append({"reason": "light_without_switch", "address": ga.address,
|
||
"name": ga.name, "dpt": ga.dpt,
|
||
"hint": "5.001 lighting GA with no on/off GA in its zone. A Home "
|
||
"Assistant light needs `address`; attach the on/off GA "
|
||
"manually, or it is a device parameter, not a light."})
|
||
consumed.add(ga.address)
|
||
continue
|
||
entity = {"name": ga.name, "brightness_address": ga.address}
|
||
st5 = status_for_dpt(ga, 5) # brightness status (5.x)
|
||
if st5:
|
||
entity["brightness_state_address"] = st5.address
|
||
consumed.add(st5.address)
|
||
elif self_reporting(ga, project):
|
||
entity["brightness_state_address"] = ga.address
|
||
if sib is not None:
|
||
entity["address"] = sib.address
|
||
s1 = status_for_dpt(sib, 1) # on/off status (1.x)
|
||
if s1:
|
||
entity["state_address"] = s1.address
|
||
consumed.add(s1.address)
|
||
elif self_reporting(sib, project):
|
||
entity["state_address"] = sib.address
|
||
consumed.add(sib.address)
|
||
attach_colour(entity, ga) # RGB/RGBW/xyY/colour-temp of this zone
|
||
lights.append(entity)
|
||
consumed.add(ga.address)
|
||
|
||
# ---- 2b. Colour lights that have no separate 5.001 brightness channel ----
|
||
for ga in project.gas.values():
|
||
if ga.address in consumed:
|
||
continue
|
||
if ga.category != "lighting" or ga.kind != "command":
|
||
continue
|
||
if (ga.dpt_main, ga.dpt_sub) not in _COLOUR_DPT and (ga.dpt_main, ga.dpt_sub) != (7, 600):
|
||
continue
|
||
onoff = next((s for s in same_main_gas(ga)
|
||
if s.address not in consumed and s.dpt_main == 1 and s.dpt_sub == 1
|
||
and s.kind == "command" and _identity_match(ga.name, s.name)), None)
|
||
if onoff is None:
|
||
review.append({"reason": "color_light_without_switch", "address": ga.address,
|
||
"name": ga.name, "dpt": ga.dpt,
|
||
"hint": "Colour GA with no on/off switch in its zone — attach it "
|
||
"to a light's color/rgbw address manually."})
|
||
consumed.add(ga.address)
|
||
continue
|
||
entity = {"name": ga.name, "address": onoff.address}
|
||
s1 = status_for_dpt(onoff, 1)
|
||
if s1:
|
||
entity["state_address"] = s1.address
|
||
consumed.add(s1.address)
|
||
consumed.add(onoff.address)
|
||
attach_colour(entity, ga)
|
||
lights.append(entity)
|
||
|
||
# ---- 3. ON/OFF LIGHTS and 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", "hvac"):
|
||
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}
|
||
# Lighting on/off is a light in Home Assistant, not a switch: the KNX light
|
||
# platform takes a plain on/off light with `address` + `state_address`
|
||
# ("Simple light" in the HA KNX docs), and light entities are what Assist
|
||
# and "all lights" targeting act on.
|
||
is_light = ga.category == "lighting"
|
||
st = status_for_dpt(ga, 1)
|
||
if st:
|
||
entity["state_address"] = st.address
|
||
consumed.add(st.address)
|
||
elif self_reporting(ga, project):
|
||
# The actuator's status object (Read + Transmit) is linked to the command GA
|
||
# itself, so the command GA is its own state. detect_missing_status already
|
||
# treats these as satisfied; the generator now agrees.
|
||
entity["state_address"] = ga.address
|
||
else:
|
||
review.append({"reason": "light_without_status" if is_light else "switch_without_status",
|
||
"address": ga.address, "name": ga.name})
|
||
(lights if is_light else switches).append(entity)
|
||
consumed.add(ga.address)
|
||
|
||
# ---- 3b. CLIMATE — anchored on an HVAC mode (DPT 20.102/20.105). Zone
|
||
# members are matched project-wide by identity (the current temperature often
|
||
# lives in a different main group than the mode). Emitted only when the HA-
|
||
# required minimum is present (current temp + target-temp status); otherwise
|
||
# the zone goes to review so we never write an invalid climate entity. ----
|
||
built_climate: set = set()
|
||
# only the two mode DPTs a climate entity uses; stable order so results never depend on
|
||
# how the project happened to parse
|
||
climate_anchors = sorted((g for g in project.gas.values()
|
||
if g.category == "hvac" and g.kind == "command"
|
||
and (g.dpt_main, g.dpt_sub) in ((20, 102), (20, 105))),
|
||
key=lambda g: _addr_key(g.address))
|
||
|
||
def _zone_loc(g: GARecord) -> set:
|
||
return _pair_ident(_without_device_words(g.name)) - _CLIMATE_QUALIFIERS
|
||
|
||
def _zone_key(g: GARecord):
|
||
return _room_code(g.name) or frozenset(_zone_loc(g))
|
||
|
||
shared_current: set[str] = set() # untyped room sensors already used as a current temperature
|
||
room_types: dict = {} # room -> set of anchor device-type sets
|
||
room_anchor_locs: dict = {} # room -> [(anchor address, its zone tokens)]
|
||
for a in climate_anchors:
|
||
room_types.setdefault(_zone_key(a), set()).add(_climate_types(a.name))
|
||
room_anchor_locs.setdefault(_zone_key(a), []).append((a.address, frozenset(_zone_loc(a))))
|
||
|
||
for ga in climate_anchors:
|
||
if ga.address in consumed:
|
||
continue
|
||
zone_loc = _zone_loc(ga)
|
||
anchor_code = _room_code(ga.name)
|
||
anchor_types = _climate_types(ga.name)
|
||
anchor_numbers = _numbers(ga.name)
|
||
zkey = (_zone_key(ga), anchor_types, frozenset(zone_loc), anchor_numbers)
|
||
if zone_loc and zkey in built_climate:
|
||
# same room, device type and name identity as an entity already built: two mode
|
||
# GAs we cannot tell apart. Never drop it silently, never merge it.
|
||
review.append({"reason": "climate_duplicate_anchor", "address": ga.address,
|
||
"name": ga.name, "dpt": ga.dpt,
|
||
"hint": "Another HVAC mode GA with the same room, device type and name "
|
||
"already produced a climate entity — map this device manually."})
|
||
consumed.add(ga.address)
|
||
continue
|
||
typed_kinds = {t for t in room_types.get(_zone_key(ga), set()) if t}
|
||
typed_kinds_flat = frozenset().union(*typed_kinds) if typed_kinds else frozenset()
|
||
ambiguous_room = len(typed_kinds) >= 2
|
||
# words that set OTHER climate devices of this room apart ("душ", "холл"): a member
|
||
# carrying one of them belongs to that device, not to this one
|
||
foreign_words: set = set()
|
||
for other_addr, other_loc in room_anchor_locs.get(_zone_key(ga), []):
|
||
if other_addr != ga.address:
|
||
foreign_words |= set(other_loc) - zone_loc
|
||
|
||
def _member_ok(g: GARecord, role: str) -> bool:
|
||
member_code = _room_code(g.name)
|
||
if anchor_code and member_code != anchor_code:
|
||
return False
|
||
if not anchor_code and member_code:
|
||
return False
|
||
gt = _climate_types(g.name)
|
||
member_numbers = _numbers(g.name)
|
||
if role == "current" and not gt:
|
||
# a shared room sensor names the room, not the device ("Room 1 Temperature" for
|
||
# "Room 1 Floor heating 2"), so its numbers must be among the anchor's
|
||
if not member_numbers <= anchor_numbers:
|
||
return False
|
||
elif not anchor_numbers <= member_numbers:
|
||
return False
|
||
if foreign_words & _pair_ident(_without_device_words(g.name)) \
|
||
and not (role == "current" and not gt and anchor_code):
|
||
return False
|
||
if gt == anchor_types:
|
||
return True
|
||
# The anchor names no device type and no other device in the room has the member's
|
||
# type, so the typed member ("floor valve") belongs to this untyped device.
|
||
unique_type_for_untyped = bool(gt) and not anchor_types and not (gt & typed_kinds_flat)
|
||
if unique_type_for_untyped:
|
||
return True
|
||
if not gt:
|
||
# an untyped GA is shared room data; with several devices in the room it
|
||
# can only be the measured temperature, never a device's setpoint or mode
|
||
return role == "current" or not ambiguous_room
|
||
return False
|
||
|
||
# a shared, untyped room temperature sensor may serve every climate device in the room
|
||
zone_all = sorted((g for g in project.gas.values()
|
||
if (g.address not in consumed or g.address in shared_current)
|
||
and zone_loc and (zone_loc <= _pair_ident(g.name)
|
||
or zone_loc <= _pair_ident(_without_device_words(g.name)))),
|
||
key=lambda g: _addr_key(g.address))
|
||
# prefer members of the anchor's own device type over shared, untyped ones
|
||
zone = ([g for g in zone_all if _climate_types(g.name) == anchor_types]
|
||
+ [g for g in zone_all if _climate_types(g.name) != anchor_types])
|
||
# With a room code the room is already certain, so an untyped room sensor with the same
|
||
# code may be the current temperature even if it lacks the device's extra words
|
||
# ("1.09 Bath air temperature" for "1.09 Bath - Floor heating shower").
|
||
room_sensors = [] if not anchor_code else sorted(
|
||
(g for g in project.gas.values()
|
||
if (g.address not in consumed or g.address in shared_current)
|
||
and g not in zone and _room_code(g.name) == anchor_code and not _climate_types(g.name)),
|
||
key=lambda g: _addr_key(g.address))
|
||
ambiguous_roles: dict[str, list[str]] = {}
|
||
|
||
def _pick(pred, role: str):
|
||
pool = zone + room_sensors if role == "current" else zone
|
||
cands = [g for g in pool if pred(g) and _member_ok(g, role)
|
||
and (role == "current" or g.address not in shared_current)]
|
||
if role == "current" or len(cands) <= 1:
|
||
# a current temperature only feeds the display; for it the preferred first
|
||
# candidate is enough. Control roles must be unambiguous.
|
||
return cands[0] if cands else None
|
||
own = [g for g in cands if _climate_types(g.name) == anchor_types]
|
||
if len(own) == 1:
|
||
return own[0]
|
||
ambiguous_roles[role] = [g.address for g in cands]
|
||
return None
|
||
|
||
# Temperature GAs must be DPT 9.001 specifically — DPT main 9 also covers
|
||
# humidity (9.007), CO2 (9.008), lux (9.004); never treat those as a setpoint.
|
||
cur = _pick(lambda g: (g.dpt_main, g.dpt_sub) == (9, 1) and g.kind == "sensor"
|
||
and not _has(g.name, _TARGET_WORDS), "current")
|
||
tgt_state = _pick(lambda g: (g.dpt_main, g.dpt_sub) == (9, 1) and _is_status_ga(g)
|
||
and _has(g.name, _TARGET_WORDS), "target_state")
|
||
tgt_cmd = _pick(lambda g: (g.dpt_main, g.dpt_sub) == (9, 1) and not _is_status_ga(g)
|
||
and _has(g.name, _TARGET_WORDS), "target")
|
||
op_cmd = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 102 and g.kind == "command",
|
||
"operation_mode")
|
||
op_state = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 102 and _is_status_ga(g),
|
||
"operation_mode_state")
|
||
ctrl_cmd = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 105 and g.kind == "command",
|
||
"controller_mode")
|
||
ctrl_state = _pick(lambda g: g.dpt_main == 20 and g.dpt_sub == 105 and _is_status_ga(g),
|
||
"controller_mode_state")
|
||
# an AC unit has no heating valve; a valve GA in the room belongs to another device
|
||
valve = None if "ac" in anchor_types else \
|
||
_pick(lambda g: g.dpt_main == 5 and _is_status_ga(g) and _has(g.name, _VALVE_WORDS),
|
||
"valve")
|
||
shift_cmd = _pick(lambda g: _is_shift(g) and not _is_status_ga(g), "setpoint_shift")
|
||
shift_state = _pick(lambda g: _is_shift(g) and _is_status_ga(g), "setpoint_shift_state")
|
||
|
||
if ambiguous_roles:
|
||
review.append({"reason": "climate_ambiguous", "address": ga.address,
|
||
"name": ga.name, "dpt": ga.dpt, "candidates": ambiguous_roles,
|
||
"hint": "More than one GA fits a control role of this climate device "
|
||
"(several devices share the naming) — pick the right ones manually."})
|
||
consumed.add(ga.address)
|
||
continue
|
||
|
||
if not (cur and tgt_state):
|
||
review.append({"reason": "manual_climate", "address": ga.address,
|
||
"name": ga.name, "dpt": ga.dpt,
|
||
"hint": "HVAC mode found, but no clean current-temp + target-temp-"
|
||
"status pair in this zone — map the climate entity manually."})
|
||
consumed.add(ga.address)
|
||
continue
|
||
|
||
ent = {"name": ga.name,
|
||
"temperature_address": cur.address,
|
||
"target_temperature_state_address": tgt_state.address}
|
||
if tgt_cmd:
|
||
ent["target_temperature_address"] = tgt_cmd.address
|
||
if op_cmd:
|
||
ent["operation_mode_address"] = op_cmd.address
|
||
if op_state:
|
||
ent["operation_mode_state_address"] = op_state.address
|
||
if ctrl_cmd:
|
||
ent["controller_mode_address"] = ctrl_cmd.address
|
||
if ctrl_state:
|
||
ent["controller_mode_state_address"] = ctrl_state.address
|
||
if valve:
|
||
ent["command_value_state_address"] = valve.address
|
||
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)]
|
||
if cur is not None and not _climate_types(cur.name):
|
||
shared_current.add(cur.address)
|
||
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)
|
||
if zone_loc:
|
||
built_climate.add(zkey)
|
||
# B2 climate correctness: mode-command-without-state makes the mode unshowable;
|
||
# controller/operation mode lists auto-detect wrong; setpoint-shift needs cmd+state.
|
||
issues = []
|
||
if op_cmd and not op_state:
|
||
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 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,
|
||
"name": ga.name, "note": note})
|
||
|
||
# ---- 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", "datetime", "text"):
|
||
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})
|
||
|
||
# ---- 5. Catch-all: never drop a GA silently ("no silent caps") ----
|
||
accounted: set[str] = set(consumed)
|
||
accounted.update(r["address"] for r in review if r.get("address"))
|
||
for addr, ga in project.gas.items():
|
||
if addr in accounted:
|
||
continue
|
||
item = {
|
||
"reason": "not_mapped",
|
||
"address": addr, "name": ga.name, "dpt": ga.dpt,
|
||
"category": ga.category, "kind": ga.kind,
|
||
}
|
||
# A slat/tilt GA that wasn't matched to a blind cover by name.
|
||
if ga.address in slat_addrs:
|
||
item["reason"] = "shutter_slat_unattached"
|
||
item["hint"] = ("Slat/tilt control of a venetian blind; could not be matched "
|
||
"to a blind cover by name. Attach it manually as the cover's "
|
||
"tilt/move_short address.")
|
||
# A shutter-looking command that didn't become a cover almost always has
|
||
# an incomplete DPT (e.g. bare "1" instead of 1.008 up/down). Make that
|
||
# actionable instead of an opaque drop.
|
||
elif ga.category == "shutter" and ga.kind == "command" and ga.dpt_sub is None:
|
||
item["reason"] = "shutter_incomplete_dpt"
|
||
item["hint"] = ("Looks like a shutter control but the DPT has no sub-type. "
|
||
"Set it in ETS (1.008 up/down, 1.010 stop, 5.001 position) "
|
||
"so it can map to a Home Assistant cover.")
|
||
review.append(item)
|
||
|
||
# A6: time/date broadcast — if the project has a DateTime/time/date GA, expose
|
||
# Home Assistant's clock to it so KNX devices get the time from HA.
|
||
expose: list[dict] = []
|
||
clock = next((g for g in project.gas.values()
|
||
if g.dpt_main == 19
|
||
or (g.dpt_main in (10, 11) and any(
|
||
t in g.name.lower() for t in
|
||
("время", "дата", "time", "date", "uhr", "zeit", "clock")))), None)
|
||
if clock:
|
||
etype = ("datetime" if clock.dpt_main == 19
|
||
else "time" if clock.dpt_main == 10 else "date")
|
||
expose.append({"type": etype, "address": clock.address,
|
||
"entity_id": "sensor.date_time_iso"})
|
||
review.append({"reason": "verify_expose", "address": clock.address, "name": clock.name,
|
||
"note": f"exposes Home Assistant's {etype} to this KNX GA (clock "
|
||
"broadcast); point entity_id at a real HA 'Date & Time' helper. "
|
||
"`expose` cannot use passive-address lists."})
|
||
|
||
# A5: Areas / voice — surface the UI-only limitation once.
|
||
if switches or lights or covers or climates:
|
||
review.append({"reason": "areas_ui_only", "address": "-",
|
||
"note": "Home Assistant Areas cannot be set in KNX YAML (assign each "
|
||
"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
|
||
if lights:
|
||
knx["light"] = lights
|
||
if covers:
|
||
knx["cover"] = covers
|
||
if climates:
|
||
knx["climate"] = climates
|
||
if binary_sensors:
|
||
knx["binary_sensor"] = binary_sensors
|
||
if sensors:
|
||
knx["sensor"] = sensors
|
||
if expose:
|
||
knx["expose"] = expose
|
||
|
||
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"
|
||
f"# {len(review)} group address(es) need manual review (see the 'review' list); "
|
||
"nothing is dropped silently.\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),
|
||
"climate": len(climates),
|
||
"binary_sensor": len(binary_sensors), "sensor": len(sensors),
|
||
"review": len(review),
|
||
}
|
||
return {"yaml": text, "review": review, "counts": counts}
|