nickol-knx-mcp/nickol_knx_mcp/device_library.py

"""Device library — per-device group-address decomposition recipes.

A KNX device is not "one GA". Each channel/output of an actuator expands into a set
of communication objects (command, status, dimming, position, mode, …), each with its
own DPT. This module maps a device's ``order_number`` (or a device *type*) to its
**design-time decomposition recipe** — the objects a professional integrator typically
wires per channel — so a spec/ТЗ that says "4-channel dimmer" can be expanded into the
right group-address set.

Provenance: recipes are compiled from KNX manufacturer ETS product databases and public
product documentation (Zennio, ABB), cross-checked against manufacturer "communication
objects" tables. Counts are the *typical wired* set, not the full selectable master menu
(integrators enable a subset — see the ``enablement`` note). DPTs are the canonical KNX
types per function. Where a manufacturer datum is not publicly confirmable it is omitted
rather than guessed.
"""

from __future__ import annotations

import glob
import os
from typing import Any, Optional

# Each recipe: per-channel/per-unit object list [(function, dpt, role)].
# role: cmd | status | param.  These are the *design* objects (typical wired set).
_RECIPES: dict[str, dict[str, Any]] = {
    # ---- Zennio ------------------------------------------------------------
    "switch_output": {
        "aliases": ["ZIO-MB24", "ZIOMB24V2", "ZIO-MB8P", "ZIO-MB16P", "ZIOMB16V3",
                    "SA/S", "switch actuator"],
        "unit": "output",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
        ],
        "note": "Switch output. Zennio master menu ~11 obj/output (adds scene, timer, "
                "lock, logic); ABB labels status 1.001. Typical wired: cmd + status.",
    },
    "dimmer_channel": {
        "aliases": ["ZDIDBDX4", "DIMinBOX", "ZDINDX2", "ZDINDX4", "NarrowDIM",
                    "UD/S", "dimmer"],
        "unit": "channel",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
            ("Relative dimming", "3.007", "cmd"),
            ("Absolute dimming / value", "5.001", "cmd"),
            ("Brightness (status)", "5.001", "status"),
        ],
        "note": "Dimmer channel. Zennio DIMinBOX DX4 master = 23 obj/channel "
                "(adds scenes, timer, alarms, diagnostics). Typical wired = these 5.",
    },
    "led_rgbw": {
        "aliases": ["ZDI-RGBDX4", "Lumento"],
        "unit": "channel",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
            ("Brightness value", "5.001", "cmd"),
            ("Brightness (status)", "5.001", "status"),
            ("Colour RGB(W)", "232.600", "cmd"),
        ],
        "note": "RGB/RGBW LED. Colour DPT 232.600 (RGB) / 251.600 (RGBW). "
                "CCT/tunable-white via a colour-temperature 5.001/7.600 object.",
    },
    "shutter_channel": {
        "aliases": ["ZIO–MBSHU8", "ZIO–MBSHU4", "MAXinBOX SHUTTER", "JRA/S", "JSB/S",
                    "shutter", "blind"],
        "unit": "channel",
        "objects": [
            ("Up/Down", "1.008", "cmd"),
            ("Stop / step", "1.010", "cmd"),
            ("Position", "5.001", "cmd"),
            ("Position (status)", "5.001", "status"),
            ("Lock", "1.003", "param"),
        ],
        "note": "Shutter/blind. ABB uses 1.008 (blind) / 1.007 (shutter) on the move "
                "object; venetian blinds add a slat-position 5.001 object.",
    },
    "floor_heating_zone": {
        "aliases": ["ZCL-8HT230", "ZCL-4HT230", "HeatingBOX", "floor heating"],
        "unit": "zone",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
            ("Setpoint °C", "9.001", "cmd"),
            ("Setpoint (status)", "9.001", "status"),
            ("HVAC mode", "20.102", "cmd"),
            ("HVAC mode (status)", "20.102", "status"),
            ("Valve / contactor", "1.001", "cmd"),
            ("Valve (status)", "1.001", "status"),
        ],
        "note": "Floor-heating zone (8-object recipe). Convector/radiator = 5 "
                "(setpoint·status·mode·status·valve).",
    },
    "ac_gateway": {
        "aliases": ["ZN1CL-KLIC-DI", "KLIC-DI", "KLIC-DA", "DK-AC-KNX", "FCC/S",
                    "FCA/S", "air conditioning"],
        "unit": "unit",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
            ("Mode", "20.105", "cmd"),
            ("Mode (status)", "20.105", "status"),
            ("Setpoint °C", "9.001", "cmd"),
            ("Setpoint (status)", "9.001", "status"),
            ("Fan speed", "5.001", "cmd"),
            ("Ambient temperature", "9.001", "status"),
        ],
        "note": "AC split-unit gateway. Mode 20.105 (heat/cool/fan/dry/auto) or "
                "20.102 (HVAC mode). Fan-coil (FCC/S) uses 20.102.",
    },
    "dali_group": {
        "aliases": ["ZDIDLIV2", "DALI-BOX", "DG/S", "DALI"],
        "unit": "group",
        "objects": [
            ("On/Off", "1.001", "cmd"),
            ("On/Off (status)", "1.001", "status"),
            ("Relative dimming", "3.007", "cmd"),
            ("Brightness value", "5.001", "cmd"),
            ("Brightness (status)", "5.001", "status"),
        ],
        "note": "DALI light group. Gateway addresses up to 64 ballasts / 16 groups; "
                "per-group recipe ≈ dimmer.",
    },
    "presence_detector": {
        "aliases": ["ZPDEZTPVT", "EyeZen", "presence", "motion"],
        "unit": "detector",
        "objects": [
            ("Movement / presence", "1.001", "cmd"),
            ("Movement (status)", "1.001", "status"),
            ("Luminosity (lux)", "9.004", "status"),
            ("Lock", "1.003", "param"),
            ("Scene output", "18.001", "cmd"),
        ],
        "note": "Presence/motion. Zennio master ~10 obj (adds timers, second lock, "
                "test-mode, external-on, threshold). Constant-light uses 5.010.",
    },
    "water_meter": {
        "aliases": ["ZRX-KCI4S0", "KCI", "S0 meter", "meter"],
        "unit": "input",
        "objects": [
            ("Total consumption", "13.013", "status"),
            ("This month", "13.013", "status"),
            ("Last month", "13.013", "status"),
            ("Reset", "1.001", "cmd"),
            ("Start value", "13.013", "param"),
            ("Instant / pulse", "13.013", "status"),
        ],
        "note": "Pulse/S0 meter. DPT depends on medium: energy 13.013, volume 13.x.",
    },
    "leak_sensor": {
        "aliases": ["WS", "leak", "water sensor"],
        "unit": "sensor",
        "objects": [
            ("Leak alarm", "1.005", "status"),
            ("Confirmation", "1.001", "cmd"),
        ],
        "note": "Leak sensor. Pairs with a water-shutoff valve (open/close 1.001 + status).",
    },
    "panel": {
        "aliases": ["ZVIZ50", "ZVIZ35", "Z70", "Z50", "Z35", "Flat", "ZVIF", "panel",
                    "keypad", "touch panel"],
        "unit": "panel",
        "objects": [],
        "note": "Touch panel / keypad. References EXISTING GAs (its buttons drive other "
                "devices' objects); creates ~0 new GAs of its own. Buttons are multi-DPT "
                "(1.001 / 1.008 / 3.007 / 5.010 / 18.001) bound to the target function.",
    },
}


def _lookup(key: str) -> Optional[str]:
    """Resolve an order_number / type / alias to a recipe key (case-insensitive)."""
    k = (key or "").strip().lower()
    if not k:
        return None
    if k in _RECIPES:
        return k
    for name, rec in _RECIPES.items():
        for a in rec.get("aliases", []):
            al = a.lower()
            if k == al or al in k or k in al:
                return name
    return None


# ---------------------------------------------------------------------------
# Local exact catalog (opt-in) — harvested vendor app-program models.
#
# When the env var ``NICKOL_KNX_CATALOG`` points at a YAML file OR a directory of
# YAML files (device-library schema — see library-schema.md), ``decompose_device``
# prefers the EXACT vendor object model over the generic recipe. The catalog is
# NOT shipped with the package (it is vendor-catalog data kept locally); with the
# env unset the tool behaves exactly as before (generic recipes only).
# ---------------------------------------------------------------------------

_CATALOG_INDEX: Optional[dict[str, tuple[dict[str, Any], Any]]] = None


def _norm(s: Any) -> str:
    """Normalise an order number / name for matching (case- and space-insensitive)."""
    return "".join(str(s).lower().split())


def _catalog_paths() -> list[str]:
    p = os.environ.get("NICKOL_KNX_CATALOG")
    if not p:
        return []
    if os.path.isdir(p):
        return sorted(glob.glob(os.path.join(p, "*.yaml")) + glob.glob(os.path.join(p, "*.yml")))
    if os.path.isfile(p):
        return [p]
    return []


def _load_catalog(force: bool = False) -> dict[str, tuple[dict[str, Any], Any]]:
    """Build (and cache) an index ``normalised order/name -> (device, manufacturer)``.

    Never raises: a missing/invalid catalog yields an empty index (recipe-only mode).
    """
    global _CATALOG_INDEX
    if _CATALOG_INDEX is not None and not force:
        return _CATALOG_INDEX
    index: dict[str, tuple[dict[str, Any], Any]] = {}
    try:
        import yaml  # PyYAML is a package dependency; guard anyway
    except Exception:
        _CATALOG_INDEX = index
        return index
    for path in _catalog_paths():
        try:
            with open(path, encoding="utf-8") as fh:
                data = yaml.safe_load(fh)
        except Exception:
            continue
        if not isinstance(data, dict):
            continue
        top_manufacturer = data.get("manufacturer")
        for dev in data.get("devices", []) or []:
            if not isinstance(dev, dict):
                continue
            manufacturer = dev.get("manufacturer") or top_manufacturer
            for field in ("order_number", "name"):
                val = dev.get(field)
                if isinstance(val, str) and val.strip():
                    index.setdefault(_norm(val), (dev, manufacturer))
    _CATALOG_INDEX = index
    return index


def _obj_row(o: dict[str, Any]) -> dict[str, Any]:
    return {"number": o.get("number"), "name": o.get("name"),
            "dpt": o.get("dpt"), "role": o.get("role"), "size_bits": o.get("size_bits")}


def _catalog_response(query: str, entry: dict[str, Any], manufacturer: Any,
                      channels: int) -> dict[str, Any]:
    """Normalise a catalog device entry (either schema variant) into a response."""
    blocks: list[dict[str, Any]] = []
    # Variant A: exact-from-app-program (repeating_blocks + first-instance objects)
    for b in entry.get("repeating_blocks", []) or []:
        if not isinstance(b, dict):
            continue
        objs = [_obj_row(o) for o in b.get("objects", []) or [] if isinstance(o, dict)]
        blocks.append({"unit": b.get("unit"), "instances": b.get("instances"),
                       "objects_per_instance": b.get("objects_per_instance"),
                       "stride": b.get("stride"), "first_instance_objects": objs})
    # Variant B: older decomposition_recipe (fn/dpt/role gas list)
    for b in entry.get("decomposition_recipe", []) or []:
        if not isinstance(b, dict):
            continue
        gas = [{"function": g.get("fn") or g.get("function"), "dpt": g.get("dpt"),
                "role": g.get("role")} for g in b.get("gas", []) or [] if isinstance(g, dict)]
        blocks.append({"unit": b.get("unit"),
                       "objects_per_instance": b.get("objects_per_unit"),
                       "first_instance_objects": gas})

    counts = entry.get("object_counts") or {}
    total_master = counts.get("master_catalog_total")
    if total_master is None and isinstance(entry.get("comm_objects"), list):
        total_master = len(entry["comm_objects"])

    app = entry.get("application_program") or {}
    lf = entry.get("logic_functions_block") or {}
    return {
        "matched": True,
        "source": "catalog-exact",
        "query": query,
        "order_number": entry.get("order_number"),
        "name": entry.get("name"),
        "manufacturer": manufacturer,
        "category": entry.get("category"),
        "application_program": {"name": app.get("name"), "version": app.get("version"),
                                "app_id": app.get("app_id")},
        "channels_native": entry.get("channels"),
        "channels_requested": channels,
        "total_master_objects": total_master,
        "general_objects": counts.get("general_objects"),
        "logic_functions": ({"present": lf.get("present"),
                             "total_objects": lf.get("total_objects"),
                             "function_results": lf.get("function_results")} if lf else None),
        "blocks": blocks,
        "note": "Exact vendor model from the local catalog (app-program-resolved). "
                "The master catalog is a SUPERSET; ETS parameters enable a subset per config. "
                "DPT 'unverified' = the vendor app-program declares no DatapointType (never guessed).",
        "provenance": entry.get("source_ref") or "local device-library (vendor app-program, ETS-resolved)",
    }


def decompose_device(order_number: str, channels: int = 1) -> dict[str, Any]:
    """Return the group-address decomposition recipe for a device.

    Args:
        order_number: manufacturer order number, device type, or alias
            (e.g. 'ZIO-MB24', 'dimmer', 'JRA/S', 'presence detector').
        channels: number of channels/outputs/zones to expand (default 1).

    Returns a recipe with per-unit objects and the total GA count for ``channels``.
    When a local catalog is configured (``NICKOL_KNX_CATALOG``) and the device is
    found in it, the EXACT vendor object model is returned instead (``source:
    catalog-exact``); otherwise a generic recipe is used (``source: recipe-approximate``).
    """
    hit = _load_catalog().get(_norm(order_number))
    if hit is not None:
        return _catalog_response(order_number, hit[0], hit[1], channels)

    key = _lookup(order_number)
    if key is None:
        return {
            "matched": False,
            "source": "recipe-approximate",
            "query": order_number,
            "note": "No catalog entry or recipe found. Known recipe types: "
                    + ", ".join(sorted(_RECIPES)),
        }
    rec = _RECIPES[key]
    objs = [{"function": f, "dpt": d, "role": r} for (f, d, r) in rec["objects"]]
    return {
        "matched": True,
        "source": "recipe-approximate",
        "query": order_number,
        "type": key,
        "unit": rec["unit"],
        "objects_per_unit": len(objs),
        "objects": objs,
        "channels": channels,
        "total_ga": len(objs) * max(channels, 1),
        "note": rec["note"],
        "provenance": "compiled from KNX manufacturer ETS databases + public docs; "
                      "typical-wired set (integrators enable a subset of the master menu)",
    }


def list_recipes() -> list[dict[str, Any]]:
    """List all device decomposition recipes (type, unit, objects/unit)."""
    return [{"type": k, "unit": v["unit"], "objects_per_unit": len(v["objects"]),
             "aliases": v["aliases"]} for k, v in sorted(_RECIPES.items())]