nickol-knx-mcp/nickol_knx_mcp/project.py

"""Load and parse a .knxproj file and build an enriched in-memory model.

This module is the ONLY place that touches the ETS project file. It is strictly
read-only: it opens the .knxproj archive, never modifies it, and never opens any
KNX/IP connection. The custom MCP server has no bus connectivity by design.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from collections.abc import Mapping
from typing import Any, Optional

from xknxproject import XKNXProj
from xknxproject.models import KNXProject

from .dpt_map import classify_dpt, dpt_key, is_exact_dpt, CATEGORY_UNKNOWN
from .intent import classify_intent, INTENT_FUNCTIONAL
from .safexml import preflight_archive

CATEGORY_UNKNOWN_STR = CATEGORY_UNKNOWN


# Multilingual keyword sets (EN / DE / RU) used by the heuristic fallbacks.
STATUS_KEYWORDS = [
    "status", "state", "stat", "fb", "feedback", "rueck", "rück", "rm ",
    "rm_", "статус", "состоян", "обратн", "сост.",
]
COMMAND_KEYWORDS = [
    "switch", "schalt", "dimm", "control", "steuer", "befehl", "set",
    "soll", "вкл", "выкл", "упр", "команд", "задан",
]

# Standalone status abbreviations that sit as a whole trailing token ("… RM").
# The substring keywords above carry "rm "/"rm_" (a following delimiter) and so
# MISS a name that *ends* in the bare token, and base_tokens() drops "rm"/"fb"
# as stopwords, so token-overlap pairing can't see them either — a DPT-5.001
# shutter feedback named "… RM" then falls to the lighting default (issue #12,
# field data by Kris1166). Region conventions vary; RM/FB are the German-market
# standard for Rückmeldung / Feedback.
_STATUS_ABBREV = frozenset({"rm", "fb"})


def name_is_status(name: str) -> bool:
    """True if a GA name marks it a status/feedback object — via the substring
    keywords OR a standalone status abbreviation token (RM/FB) that both the
    substring form and the tokenizer miss."""
    low = (name or "").lower()
    if any(k in low for k in STATUS_KEYWORDS):
        return True
    for ch in "/-_.,()[]:":
        low = low.replace(ch, " ")
    return any(t in _STATUS_ABBREV for t in low.split())

# Domain terms by name (used to decide a GA's functional domain as ONE signal,
# combined with the DPT and the group-range context — see _classify_category).
# Checked in this priority order (most specific first; lighting is the generic
# last resort). Matching is by WORD BOUNDARY, not bare substring, so short tokens
# like "ac" don't fire inside "terrace" and "led" doesn't fire inside "ledge".
_DOMAIN_TERMS: dict[str, list[str]] = {
    "diagnostics": ["alarm", "fault", "error", "diag", "leak", "smoke",
                    "online", "offline", "heartbeat", "watchdog",
                    "тревог", "ошибк", "диагност", "утечк", "дым", "авари", "неисправ",
                    "связь"],
    "energy": ["energy", "power", "consum", "meter", "kwh", "watt", "энерг",
               "мощност", "потребл", "счётчик", "счетчик", "тариф"],
    "scene": ["scene", "preset", "mood", "сцен", "пресет"],
    "shutter": ["blind", "shutter", "jalousie", "roll", "rollo", "marqui",
                "awning", "curtain", "behang", "lamelle", "raffstore", "markise",
                "auf/ab", "ab/auf", "штор", "жалюзи", "рольставн", "ролет", "ролл",
                "позиц", "вверх/вниз", "ламел", "position", "shade"],
    # sensor BEFORE hvac: a name with an explicit sensor word (weather station,
    # датчик, motion) is a sensor even when it measures temperature — otherwise
    # "Метеостанция - Температура" lands in hvac via the "температур" term.
    "sensor": ["sensor", "motion", "presence", "occupancy", "brightness sens", "lux",
               "humidity", "weather", "meteo", "illuminance", "датчик", "движени",
               "присутств", "влажн", "люкс", "метео", "погод",
               "освещённост", "освещенност"],  # illuminance ≠ "освещение" (checked first)
    "hvac": ["hvac", "climate", "thermostat", "heat", "cool", "ac", "a/c", "aircon",
             "air con", "air-con", "conditioner", "ventilation", "radiator", "boiler",
             "underfloor", "fancoil", "fan coil", "fan-coil", "fan", "valve", "setpoint",
             # ventilation / airing (F5 fix): "Airing step" was 'unknown' — an airing
             # stage is HVAC. lüftung/lüften carry the DE umlaut and an ASCII fallback.
             "airing", "lüftung", "lüften", "luftung", "luften",
             "климат", "отопл", "тёпл", "тепл", "конвектор", "вентил", "проветр",
             "клапан", "кондиц", "котёл", "радиатор", "фанкойл", "уставк",
             "а/с", "сплит", "вытяжк"],
    # NOTE: no bare "температур"/"temperature" term — a temperature GA takes its
    # domain from context (уставк/кондиц/тёпл name words, or its main group):
    # "Гостиная температура" in a Sensors main is a sensor, not an HVAC actuator.
    "lighting": ["light", "lamp", "dimm", "led", "spot", "sconce", "chandelier",
                 "rgb", "rgbw", "colour", "color", "xyy",
                 "свет", "лампа", "подсветк", "освещ", "люстра", "диммер", "бра", "торшер",
                 "яркост"],  # RU only: EN "brightness" also means weather lux — too ambiguous
}

# Domains a group-range name may legitimately pin for an actuator command whose
# DPT is domain-soft: a switch/scaling in a "Lighting"/"Shutters"/"HVAC" range is
# that load. A range named "Scenes"/"Sensors"/"Energy" does NOT retype a 1-bit
# command into a scene/sensor/energy datapoint — those carry their own strong DPTs.
_RANGE_ASSIGNABLE = {"lighting", "shutter", "hvac"}

# Short/ambiguous tokens need a boundary on BOTH sides (whole word), else they
# fire inside unrelated words ("ac" in "terrace", "led" in "ledge"). "а/с" is the
# Cyrillic air-con abbreviation. "spot" is NOT here: prefix-matching lets "spots"
# hit, and no KNX name plausibly embeds it mid-word.
_AMBIGUOUS_TERMS = {"ac", "a/c", "а/с", "led", "fan", "co2", "uv", "rgb", "lux", "roll"}


def _compile_domain(terms: list[str]) -> "re.Pattern":
    parts = []
    for t in terms:
        esc = re.escape(t)
        parts.append(rf"\b{esc}\b" if (t in _AMBIGUOUS_TERMS or len(t) <= 2) else rf"\b{esc}")
    return re.compile("|".join(parts))


_DOMAIN_RE = {dom: _compile_domain(terms) for dom, terms in _DOMAIN_TERMS.items()}


# Location phrases that CONTAIN a domain word but describe WHERE, not WHAT:
# "Boiler room light" is a light in the boiler room, not an HVAC function.
_LOCATION_STOP_RE = re.compile(r"\b(boiler\s*room|котельн\w*|laundry\s*room)\b")


def _domain_from_text(text: str) -> Optional[str]:
    """Return the functional domain a piece of text (a GA or range name) implies,
    or None. Word-boundary matching over _DOMAIN_TERMS, first domain by priority;
    known location phrases are stripped first so they don't vote."""
    low = (text or "").lower()
    if not low:
        return None
    low = _LOCATION_STOP_RE.sub(" ", low)
    for dom, rx in _DOMAIN_RE.items():
        if rx.search(low):
            return dom
    return None


def _refine_category(name: str, current: str) -> str:
    """Name-domain wins over the given category, else keep it. Kept for callers
    that only have a name + a starting category (explain, tests)."""
    return _domain_from_text(name) or current


# DPTs whose category is a *soft* default because the type does not, by itself,
# pin a domain — the name / range decides. A 1-bit switch is a light as easily as
# a pump or an AC; a 5.001 scaling is a brightness OR a shutter position. For soft
# DPTs a contradicting name is NOT a conflict (it disambiguates), unlike a strong,
# domain-encoding DPT (a shutter 1.008, an HVAC-mode 20.102, a temperature 9.001).
# 1.010 start/stop is generic (vent timers too); 9.001 temperature is hvac by
# default but a weather/sensor name legitimately re-domains it without conflict.
_SOFT_DPT = {(1, 1), (1, 11), (1, 10), (5, 1), (9, 1)}
# The subset that is domain-AGNOSTIC end to end: with no name and no range signal
# these stay 'unknown' — we never infer 'lighting' from a bare 1-bit DPT. Other
# soft DPTs (5.001) keep their sensible default (brightness) when unsignalled.
_AGNOSTIC_NO_DEFAULT = {(1, 1), (1, 11)}


def _classify_category(name: str, main_name: str, middle_name: str,
                       main: Optional[int], sub: Optional[int], dpt_cat: str,
                       dpt_kind: str = "unknown") -> str:
    """Combine signals into a domain: explicit name > strong DPT > range context.

    * an explicit name domain wins — unless it *contradicts* a strong (domain-
      encoding) DPT, in which case the result is 'unknown' (a genuine conflict,
      not a silent pick — surfaced by explain_ga as 'contested');
    * a strong DPT keeps its domain;
    * for a soft DPT with no name domain, the group-range name decides;
    * with no signal at all, a bare 1-bit DPT is 'unknown' (never guessed
      'lighting'); a 5.001 falls back to its brightness default.
    """
    # Soft = the DPT does not pin the domain: the listed agnostic types, an
    # unknown category, or a category that came from a whole-main-group FALLBACK
    # (a guess by construction) rather than an exact (main, sub) table entry.
    soft = ((main, sub) in _SOFT_DPT or dpt_cat == CATEGORY_UNKNOWN_STR
            or not is_exact_dpt(main, sub))
    name_dom = _domain_from_text(name)
    if name_dom:
        if soft or name_dom == dpt_cat:
            return name_dom
        return CATEGORY_UNKNOWN_STR  # strong DPT vs explicit name -> honest unknown
    if not soft:
        return dpt_cat
    # Domain context comes from the MAIN group only — per the 3-level convention the
    # main group is the function domain, while the middle group is a sub-function
    # (Switch / Status / Parameters / "motion-detector settings") whose name carries
    # misleading domain words. Fall back to the middle name only if the main is unnamed.
    range_dom = _domain_from_text(main_name) or _domain_from_text(middle_name)
    if range_dom in _RANGE_ASSIGNABLE:
        return range_dom
    # Passive mains constrain passive GAs: a measurement sitting in a Sensors/
    # Energy/Diagnostics range is that domain (a 9.001 in "Sensors" is a sensor,
    # not a default hvac) — but a COMMAND is never retyped by a passive range.
    if dpt_kind != "command" and range_dom in ("sensor", "energy", "diagnostics"):
        return range_dom
    if (main, sub) in _AGNOSTIC_NO_DEFAULT:
        return CATEGORY_UNKNOWN_STR
    return dpt_cat


# ETS Function roles that unambiguously mark a shutter/blind object. Matched as
# substrings of the (case-folded) role, so "MoveUpDown"/"StopStepUpDown" both hit.
# Provenance: the exact role strings "MoveUpDown" and "StopStepUpDown" are taken
# from Kris1166's real ETS-6 sunblind Function dump (issue #11 field data via
# explain_ga) — not invented. The list may not be exhaustive for every ETS
# FunctionType; extend it if another real project surfaces a shutter role that
# isn't covered (source over assumption).
_SHUTTER_ROLE_TOKENS = ("updown", "moveup", "movedown", "stepupdown",
                        "stopstep", "stepstop", "slat")


@dataclass
class GARecord:
    """Enriched group-address record used by all analysis tools."""
    address: str
    name: str
    description: str
    comment: str
    dpt_main: Optional[int]
    dpt_sub: Optional[int]
    data_secure: bool
    co_ids: list[str]
    category: str
    kind: str
    ha_platform: str
    value_type: Optional[str]
    label: str
    # Purpose of the GA: functional / reserve / logic / scratch. Non-functional
    # GAs are intentional noise (spares, internal logic, leftovers) and the
    # checks reclassify them instead of raising false errors/warnings.
    intent: str = INTENT_FUNCTIONAL
    # 3-level decomposition
    main: Optional[int] = None
    middle: Optional[int] = None
    sub: Optional[int] = None
    main_name: str = ""
    middle_name: str = ""

    @property
    def dpt(self) -> str:
        return dpt_key(self.dpt_main, self.dpt_sub)

    @property
    def middle_key(self) -> str:
        """Identity of the parent middle group (for sibling search)."""
        if self.main is not None and self.middle is not None:
            return f"{self.main}/{self.middle}"
        return ""


@dataclass
class LoadedProject:
    path: str
    info: dict[str, Any]
    gas: dict[str, GARecord]              # address -> record
    raw: KNXProject = field(repr=False)
    devices: dict[str, Any] = field(default_factory=dict, repr=False)
    functions: dict[str, Any] = field(default_factory=dict, repr=False)
    topology: dict[str, Any] = field(default_factory=dict, repr=False)

    @property
    def style(self) -> str:
        return self.info.get("group_address_style", "")


def _split_three_level(address: str) -> tuple[Optional[int], Optional[int], Optional[int]]:
    parts = address.split("/")
    if len(parts) == 3:
        try:
            return int(parts[0]), int(parts[1]), int(parts[2])
        except ValueError:
            return None, None, None
    return None, None, None


def _override_kind_by_name(name: str, kind: str) -> str:
    """Refine command/status using name keywords (helps when DPT is generic)."""
    low = name.lower()
    if name_is_status(name):
        return "status"
    if any(k in low for k in COMMAND_KEYWORDS):
        # only upgrade unknown -> command; never overwrite explicit sensor
        if kind in ("unknown", "command", "status"):
            return "command"
    return kind


def _build_range_name_map(raw: KNXProject) -> dict[str, str]:
    """Map address-prefix -> human range name from group_ranges.

    Returns keys like '1' (main) and '1/1' (middle) -> name.
    """
    out: dict[str, str] = {}

    def walk(rng: Mapping[str, Any], depth: int) -> None:
        start = rng.get("address_start")
        name = rng.get("name", "")
        if isinstance(start, int):
            main = (start >> 11) & 0x1F
            middle = (start >> 8) & 0x07
            # Depth decides main vs middle, NOT address arithmetic: a middle group
            # 0 starts at the same address as its main (start % 2048 == 0 too), so
            # the old modulo heuristic let "middle 0" overwrite the main's name.
            if depth == 0:
                out[str(main)] = name
            else:
                out[f"{main}/{middle}"] = name
        for child in rng.get("group_ranges", {}).values():
            walk(child, depth + 1)

    for rng in raw.get("group_ranges", {}).values():
        walk(rng, 0)
    return out


def load_project(path: str, password: Optional[str] = None,
                 language: Optional[str] = None) -> LoadedProject:
    """Parse a .knxproj file into an enriched, read-only model."""
    preflight_archive(path)  # reject zip-bomb / oversize / traversal before xknxproject reads it
    kwargs: dict[str, Any] = {"path": path}
    if password:
        kwargs["password"] = password
    if language:
        kwargs["language"] = language
    raw: KNXProject = XKNXProj(**kwargs).parse()
    return build_loaded_from_raw(raw, path)


def build_loaded_from_raw(raw: KNXProject, path: str) -> LoadedProject:
    """Build an enriched LoadedProject from an already-parsed KNXProject dict."""
    range_names = _build_range_name_map(raw)

    gas: dict[str, GARecord] = {}
    for addr, ga in raw.get("group_addresses", {}).items():
        dpt = ga.get("dpt")
        main = dpt.get("main") if dpt else None
        sub = dpt.get("sub") if dpt else None
        info = classify_dpt(main, sub)
        kind = _override_kind_by_name(ga.get("name", ""), info["kind"])
        m, mid, s = _split_three_level(ga.get("address", addr))
        main_name = range_names.get(str(m), "") if m is not None else ""
        middle_name = range_names.get(f"{m}/{mid}", "") if m is not None and mid is not None else ""
        # Domain is a COMBINATION of signals (name > strong DPT > range context),
        # not the DPT alone — a 1-bit switch is domain-agnostic, so an "AC on/off"
        # is HVAC, not lighting, and a truly context-less switch is 'unknown'.
        category = _classify_category(ga.get("name", ""), main_name, middle_name,
                                      main, sub, info["category"], info["kind"])
        ha_platform = info["ha_platform"]
        # If the name says shutter but DPT mapped it to light (5.001/1.001),
        # correct the HA platform so the generator builds a cover, not a light.
        if category == "shutter" and ha_platform in ("light", "switch"):
            ha_platform = "cover"
        # A 1-bit diagnostics GA (wind/frost/rain/smoke/leak alarm, fault) is a
        # read-only input, not a command switch -> binary_sensor.
        if category == "diagnostics" and main == 1 and ha_platform == "switch":
            ha_platform = "binary_sensor"
            kind = "sensor"
        rec = GARecord(
            address=ga.get("address", addr),
            name=ga.get("name", ""),
            description=ga.get("description", "") or "",
            comment=ga.get("comment", "") or "",
            dpt_main=main,
            dpt_sub=sub,
            data_secure=bool(ga.get("data_secure", False)),
            co_ids=list(ga.get("communication_object_ids", []) or []),
            category=category,
            kind=kind,
            ha_platform=ha_platform,
            value_type=info["value_type"],
            label=info["label"],
            intent=classify_intent(ga.get("name", "")),
            main=m, middle=mid, sub=s,
            main_name=main_name,
            middle_name=middle_name,
        )
        gas[rec.address] = rec

    # Authoritative shutter classification from ETS Function MEMBERSHIP.
    # A Function that carries a shutter role (MoveUpDown / StopStepUpDown / slat)
    # on ANY of its members is a shutter Function — so every GA inside it is a
    # shutter object, even one with a blank role, a domain-agnostic bare DPT
    # (1.007 step, 1.010 start/stop) and no shutter keyword in its name:
    #   * the step/stop a name-only classifier leaves 'unknown' (issue #11), and
    #   * a DPT-5.001 position feedback named only "… RM" (Rückmeldung) that
    #     otherwise falls to the lighting brightness default (issue #12).
    # The Function outranks DPT and name (explain_ga's hierarchy). Never demotes;
    # only promotes a non-shutter GA whose DPT is plausibly a shutter object
    # (1.x control / 5.x position) — never a 9.x temperature / 13.x energy GA.
    # An ETS Function is a free grouping, not proof of a single domain (an
    # integrator can drop lighting, shutters and a scene recall into one "Floor 1"
    # Function). So promote a member to shutter only on POSITIVE shutter evidence,
    # not blanket membership (LLM-council review of the issue #12 fix):
    #   * the member is itself a step/stop control (DPT 1.007 / 1.010), or a
    #     DPT-5.001 POSITION STATUS (a feedback — never a 5.001 command, which could
    #     be a scene recall / dimming value), AND
    #   * the Function actually carries an up/down MOVE (DPT 1.008) — the sibling
    #     that makes it a real shutter, not just a shutter-role substring, AND
    #   * the member's name carries no EXPLICIT non-shutter domain (defense in depth).
    # This rescues the issue #11 step/stop and the issue #12 "… RM" position feedback
    # while leaving a scene recall / dimmer feedback / boolean lock in a mixed
    # Function untouched.
    def _promote_to_shutter(rec: Optional[GARecord], has_move: bool) -> None:
        if rec is None or rec.category == "shutter" or not has_move:
            return
        # 1.017 (trigger) is what several actuator templates emit for step/stop; the
        # cover builder already treats it as a stop, so promotion must agree.
        is_step = rec.dpt_main == 1 and rec.dpt_sub in (7, 10, 17)
        is_pos_status = rec.dpt_main == 5 and rec.dpt_sub == 1 and rec.kind == "status"
        if not (is_step or is_pos_status):
            return
        dom = _domain_from_text(rec.name)
        if dom and dom != "shutter":
            return
        rec.category = "shutter"
        if rec.ha_platform in ("light", "switch", "unknown"):
            rec.ha_platform = "cover"

    for fn in (raw.get("functions", {}) or {}).values():
        members = fn.get("group_addresses", {}) or {}
        recs = [gas.get(ref.get("address") or a_key) for a_key, ref in members.items()]
        has_shutter_role = any(
            any(t in (ref.get("role") or "").lower() for t in _SHUTTER_ROLE_TOKENS)
            for ref in members.values())
        # The licensing move must be an UNCONTESTED shutter move: a 1.008 whose name
        # pins another domain classifies 'unknown' and must not license its neutral
        # siblings (gate-1 audit) — positive evidence only.
        has_move = any(r is not None and r.category == "shutter"
                       and r.dpt_main == 1 and r.dpt_sub == 8 for r in recs)
        # A shutter Function is recognised by a shutter ROLE token OR by an up/down
        # MOVE member (DPT 1.008 is shutter-specific). Roles are often blank or a raw
        # GUID on actuator-level Functions, so role strings alone silently skipped
        # 15/29 real position feedbacks (issue #12 re-test by Kris1166).
        if not (has_shutter_role or has_move):
            continue
        for member in recs:
            _promote_to_shutter(member, has_move)

    return LoadedProject(
        path=path,
        info=dict(raw.get("info", {})),
        gas=gas,
        raw=raw,
        devices=dict(raw.get("devices", {})),
        functions=dict(raw.get("functions", {})),
        topology=dict(raw.get("topology", {})),
    )