feat: exact device models — local catalog for decompose_device + parse_devices_from_project (v0.7.0)

- device_library: NICKOL_KNX_CATALOG env points at a local device-library YAML file/dir; decompose_device returns the exact vendor object model (source: catalog-exact) and falls back to generic recipes (source: recipe-approximate). Env unset = behaviour unchanged. - appprog_parser (new) + MCP tool parse_devices_from_project: deterministic extraction of exact comm-object models from M-* application programs in a .knxproj/.knxprod (order number via nested <Product>, DPST-x-y -> x.00y, per-channel block/stride detection, coverage manifest). Read-only, PII-safe: never reads the client P-*/0.xml. Now 25 MCP tools. - tests: test_device_catalog.py + test_appprog_parser.py (synthetic, self-contained) - docs: README/README.ru/docs site/announcements synced to v0.7.0, 25 tools Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-09-29 19:31:12 +02:00 · 2026-07-02 06:48:19 +02:00
parent 224f44a9c5
commit 5651be2930
12 changed files with 677 additions and 17 deletions
@@ -6,6 +6,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 ## [Unreleased]

+## [0.7.0] — 2026-07-02
+
+### Added
+
+- **Exact device decomposition from a local catalog** (`device_library.py`). When the
+  `NICKOL_KNX_CATALOG` env var points at a device-library YAML file or directory (schema:
+  `library-schema.md`), `decompose_device` now returns the **exact vendor object model**
+  (`source: catalog-exact`) — real per-channel blocks, object counts, app-program version and
+  first-instance objects with their true DPTs — instead of the generic recipe. Falls back to the
+  built-in recipes (`source: recipe-approximate`) for any device not in the catalog, so behaviour
+  is unchanged when the env is unset. The catalog itself is vendor-catalog data kept **local** and
+  is not shipped with the package. Objects the vendor app-program leaves without a DatapointType
+  stay `dpt: null` — never guessed.
+- **`parse_devices_from_project` (new MCP tool + `appprog_parser.py`)** — deterministic parser that
+  extracts the exact vendor comm-object model from the `M-*` application programs embedded in a
+  `.knxproj` / `.knxprod`: per device it reports order number, app-program version, object counts and
+  detected per-channel blocks (unit · objects-per-instance · stride), converting `DPST-x-y` → `x.00y`.
+  Read-only and PII-safe (reads only vendor catalog data, never the client `P-*/0.xml`). With
+  `output_path` it writes a device-library YAML into the workspace — feeding the local catalog above,
+  so `parse → catalog → decompose_device (catalog-exact)` is a closed loop. Now **25 MCP tools**.
+
 ## [0.6.0] — 2026-07-01

 **Completes the roadmap** — the tool now validates, repairs, generates (HA / ETS / handover / IoT),
@@ -272,7 +293,8 @@ Initial public beta.
 - Tested end-to-end on a synthetic project only; real-world `.knxproj` testing is ongoing
  (see the call for testers in the README).

-[Unreleased]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.6.0...HEAD
+[Unreleased]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.7.0...HEAD
+[0.7.0]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.6.0...v0.7.0
 [0.6.0]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.5.0...v0.6.0
 [0.5.0]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.4.0...v0.5.0
 [0.4.0]: https://github.com/NickoScope/nickol-knx-mcp/compare/v0.3.0...v0.4.0
@@ -2,7 +2,7 @@

 **A design-time KNX / ETS6 assistant exposed as an [MCP](https://modelcontextprotocol.io) server.**

-It reads your `.knxproj` and **validates** it (naming · DPT & sub-DPT · command↔status · KNX Secure · Matter-readiness), **repairs** it (proposes concrete fixes — infers DPTs, synthesises missing status GAs), **decomposes devices** into their group-address recipes, **diffs** two project versions, **grades** completeness, and **generates** Home Assistant YAML, ETS-importable exports (XML/CSV), an as-built **handover pack**, an acceptance test protocol and a KNX IoT semantic export — all **without ever touching the live KNX bus.**
+It reads your `.knxproj` and **validates** it (naming · DPT & sub-DPT · command↔status · KNX Secure · Matter-readiness), **repairs** it (proposes concrete fixes — infers DPTs, synthesises missing status GAs), **decomposes devices** into their group-address recipes (or the **exact vendor object model**, parsed straight from the ETS application programs into a local device catalog), **diffs** two project versions, **grades** completeness, and **generates** Home Assistant YAML, ETS-importable exports (XML/CSV), an as-built **handover pack**, an acceptance test protocol and a KNX IoT semantic export — all **without ever touching the live KNX bus.**

 [![CI](https://github.com/NickoScope/nickol-knx-mcp/actions/workflows/ci.yml/badge.svg)](https://github.com/NickoScope/nickol-knx-mcp/actions/workflows/ci.yml)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
@@ -201,7 +201,7 @@ keyring handling, and the recommended workflow).

 ---

-## MCP tools (24)
+## MCP tools (25)

 **Read**
 | Tool | Purpose |
@@ -227,8 +227,9 @@ keyring handling, and the recommended workflow).
 |------|---------|
 | `suggest_repairs()` | **propose fixes, not just flag** — infer DPTs, synthesise status/brightness GAs |
 | `suggest_names()` | naming-hygiene suggestions |
-| `decompose_device(order_number, channels?)` | device → group-address decomposition recipe |
+| `decompose_device(order_number, channels?)` | device → GA decomposition: **exact vendor model** from a local catalog (`NICKOL_KNX_CATALOG`), or generic recipe |
 | `list_device_recipes()` | the built-in device library (Zennio + ABB families) |
+| `parse_devices_from_project(path, output_path?, password?)` | extract **exact device object models** from the app-programs inside a `.knxproj`/`.knxprod` → device-library YAML (feeds the local catalog) |
 | `grade_completeness()` | grade a project: bare skeleton vs as-built |
 | `diff_projects(path_a, path_b, …)` | semantic diff between two `.knxproj` versions |

@@ -298,7 +299,7 @@ nickol-knx-mcp/
 │   ├── generate_ha.py    # Home Assistant KNX YAML generation
 │   ├── generate_ets.py   # ETS XML + CSV generation
 │   ├── report.py         # Markdown report
-│   └── server.py         # FastMCP server, 24 tools, confined writes
+│   └── server.py         # FastMCP server, 25 tools, confined writes
 ├── tests/test_pipeline.py
 ├── examples/claude_desktop_config.json
 ├── CLAUDE.md             # ETS Assistant skill / playbook
@@ -3,7 +3,7 @@
 # nickol-knx-mcp

 **Design-time ассистент KNX/ETS6 в виде MCP-сервера.**
-Читает `.knxproj` и **валидирует** его (именование · DPT и sub-DPT · команда↔статус · KNX Secure · Matter-готовность), **чинит** (предлагает конкретные фиксы — выводит DPT, синтезирует недостающие status-адреса), **раскладывает устройства** на рецепты групповых адресов, **диффит** две версии проекта, **грейдит** полноту, и **генерирует** Home Assistant YAML, ETS-импортируемые экспорты (XML/CSV), **пакет сдачи** (as-built), протокол приёмки и KNX IoT-экспорт — **никогда не подключаясь к живой шине KNX.**
+Читает `.knxproj` и **валидирует** его (именование · DPT и sub-DPT · команда↔статус · KNX Secure · Matter-готовность), **чинит** (предлагает конкретные фиксы — выводит DPT, синтезирует недостающие status-адреса), **раскладывает устройства** на рецепты групповых адресов (или на **точную вендорскую модель объектов**, распарсенную прямо из application programs ETS в локальный каталог устройств), **диффит** две версии проекта, **грейдит** полноту, и **генерирует** Home Assistant YAML, ETS-импортируемые экспорты (XML/CSV), **пакет сдачи** (as-built), протокол приёмки и KNX IoT-экспорт — **никогда не подключаясь к живой шине KNX.**

 [![nickol-knx-mcp MCP server](https://glama.ai/mcp/servers/NickoScope/nickol-knx-mcp/badges/score.svg)](https://glama.ai/mcp/servers/NickoScope/nickol-knx-mcp)
 [![Кейс](https://img.shields.io/badge/📐_кейс-ТЗ_PDF_→_KNX_(96%25)-0b3d2e)](docs/case-study.ru.md)
@@ -136,13 +136,13 @@ claude mcp add nickol-knx -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" -- /abs/

 ---

-## 5. Инструменты MCP (24)
+## 5. Инструменты MCP (25)

 **Чтение:** `load_project` · `list_group_addresses` · `get_devices` · `get_topology`

 **Валидация:** `check_naming` · `check_missing_status` · `check_dpt` (+ **sub-DPT** проверка) · `check_secure` (KNX Secure posture + keyring-чеклист) · `check_matter` (Matter-готовность) · `check_energy` (энергодомен) · `analyze_all`

-**Починка и дизайн:** `suggest_repairs` (**предлагает фиксы, а не только флагает**) · `suggest_names` · `decompose_device` (устройство → рецепт декомпозиции) · `list_device_recipes` (device-library: Zennio + ABB) · `grade_completeness` (скелет vs as-built) · `diff_projects` (семантический дифф двух версий)
+**Починка и дизайн:** `suggest_repairs` (**предлагает фиксы, а не только флагает**) · `suggest_names` · `decompose_device` (устройство → декомпозиция: **точная вендорская модель** из локального каталога или generic-рецепт) · `list_device_recipes` (device-library: Zennio + ABB) · `parse_devices_from_project` (**точные модели устройств** из app-programs `.knxproj`/`.knxprod` → YAML каталога) · `grade_completeness` (скелет vs as-built) · `diff_projects` (семантический дифф двух версий)

 **Генерация:** `generate_ha_package` (цвет + климат + expose) · `generate_ets_group_addresses` · `generate_handover_pack` (пакет сдачи) · `generate_test_protocol` (протокол приёмки) · `generate_knx_iot` (Turtle/RDF) · `project_report` · `workspace_info`

@@ -163,8 +163,9 @@ claude mcp add nickol-knx -e NICKOL_KNX_WORKSPACE="$HOME/knx-workspace" -- /abs/
 | `analyze_all(name_regex?)` | все проверки разом |
 | `suggest_repairs()` | предложить фиксы для находок |
 | `suggest_names()` | гигиена именования |
-| `decompose_device(order_number, channels?)` | устройство → рецепт декомпозиции GA |
+| `decompose_device(order_number, channels?)` | устройство → декомпозиция GA: **точная вендорская модель** из локального каталога (`NICKOL_KNX_CATALOG`) или generic-рецепт |
 | `list_device_recipes()` | встроенная device-library |
+| `parse_devices_from_project(path, output_path?, password?)` | извлечь **точные модели объектов устройств** из app-programs внутри `.knxproj`/`.knxprod` → YAML device-library (питает локальный каталог) |
 | `grade_completeness()` | грейд полноты (скелет vs as-built) |
 | `diff_projects(path_a, path_b, …)` | семантический дифф двух `.knxproj` |
 | `generate_ha_package(output_path?)` | HA KNX YAML + список review |
@@ -213,7 +214,7 @@ nickol-knx-mcp/
 │   ├── generate_ha.py    # генерация HA KNX YAML
 │   ├── generate_ets.py   # генерация ETS XML + CSV
 │   ├── report.py         # Markdown-отчёт
-│   └── server.py         # FastMCP сервер, 24 инструмента, confined writes
+│   └── server.py         # FastMCP сервер, 25 инструментов, confined writes
 ├── tests/test_pipeline.py
 ├── examples/claude_desktop_config.json
 ├── CLAUDE.md             # ETS Assistant skill / playbook
@@ -2,7 +2,7 @@

 Repo: https://github.com/NickoScope/nickol-knx-mcp · Site: https://nickoscope.github.io/nickol-knx-mcp/ · Case study: https://github.com/NickoScope/nickol-knx-mcp/blob/main/docs/case-study.md

-**Current: v0.6.0** — roadmap complete: **24 MCP tools** that validate (naming · DPT/sub-DPT · status · KNX Secure · Matter), **repair** (propose fixes), **decompose devices**, **diff** two versions, **grade** completeness, and **generate** Home Assistant YAML (colour + climate), ETS exports, an as-built **handover pack** & KNX IoT — all read-only. Plus a spec-PDF → **96%-match** GA-structure case study. The posts below still lead with the colour/climate assembly milestone.
+**Current: v0.7.0** — **25 MCP tools** that validate (naming · DPT/sub-DPT · status · KNX Secure · Matter), **repair** (propose fixes), **decompose devices** (incl. exact vendor models parsed from ETS app-programs into a local catalog), **diff** two versions, **grade** completeness, and **generate** Home Assistant YAML (colour + climate), ETS exports, an as-built **handover pack** & KNX IoT — all read-only. Plus a spec-PDF → **96%-match** GA-structure case study. The posts below still lead with the colour/climate assembly milestone.

 Etiquette reminder: disclose you're the author, lead with value, be online to answer for a few hours after posting. Don't cross-post everything in one hour — space it out (home base → targeted forums → Reddit/Discord → social).

@@ -4,7 +4,7 @@
 <meta charset="UTF-8" />
 <meta name="viewport" content="width=device-width, initial-scale=1.0" />
 <title>nickol-knx-mcp — design-time KNX/ETS6 assistant (MCP server)</title>
-<meta name="description" content="A design-time KNX/ETS6 assistant exposed as an MCP server: validate (naming/DPT/status/Secure/Matter), repair (propose fixes), decompose devices, diff versions, and generate Home Assistant YAML, ETS exports, an as-built handover pack & KNX IoT. No live bus access. 24 tools. Includes a full demo house + smart Home Assistant brain." />
+<meta name="description" content="A design-time KNX/ETS6 assistant exposed as an MCP server: validate (naming/DPT/status/Secure/Matter), repair (propose fixes), decompose devices (exact vendor models from ETS app-programs), diff versions, and generate Home Assistant YAML, ETS exports, an as-built handover pack & KNX IoT. No live bus access. 25 tools. Includes a full demo house + smart Home Assistant brain." />
 <style>
  :root{
    --bg:#0d0f12;--panel:#14171c;--card:#181c22;--card2:#1c2027;--line:#242a33;
@@ -150,7 +150,7 @@
 <header class="hero"><div class="wrap">
  <div class="pills">
    <span class="pill mit">MIT</span>
-    <span class="pill beta">public beta · v0.6.0</span>
+    <span class="pill beta">public beta · v0.7.0</span>
    <span class="pill">Python 3.10+</span>
    <span class="pill">Model Context Protocol</span>
  </div>
@@ -177,7 +177,7 @@
  </div>
  <div class="grid g3" style="margin-top:18px">
    <div class="card"><h3>🔒 No bus access</h3><p>Zero networking/bus libraries in the dependency tree. <code>bus_access: false</code> — structural, not a promise.</p></div>
-    <div class="card"><h3>🧩 24 MCP tools</h3><p><b>Validate</b> (naming · DPT &amp; sub-DPT · status · KNX Secure · Matter) · <b>repair</b> (propose fixes, not just flag) · <b>decompose devices</b> into GA recipes · <b>diff</b> two versions · <b>grade</b> completeness · <b>generate</b> HA YAML (<b>colour + climate</b>), ETS XML/CSV, an as-built <b>handover pack</b> &amp; KNX IoT. All read-only.</p></div>
+    <div class="card"><h3>🧩 25 MCP tools</h3><p><b>Validate</b> (naming · DPT &amp; sub-DPT · status · KNX Secure · Matter) · <b>repair</b> (propose fixes, not just flag) · <b>decompose devices</b> into GA recipes or <b>exact vendor models</b> (parsed from ETS app-programs into a local catalog) · <b>diff</b> two versions · <b>grade</b> completeness · <b>generate</b> HA YAML (<b>colour + climate</b>), ETS XML/CSV, an as-built <b>handover pack</b> &amp; KNX IoT. All read-only.</p></div>
    <div class="card"><h3>🌍 EN / DE / RU</h3><p>Classifies by DPT + multilingual name keywords. Validated on ETS 4.2 / 5.0 / 5.5 / 6 fixtures, a real <b>signed ETS6 round-trip</b>, and a <b>685-GA Zennio</b> project (false errors there: 29 → 6).</p></div>
  </div>
 </div></section>
@@ -1,2 +1,2 @@
 """nickol-knx-mcp: design-time KNX/ETS project assistant MCP server."""
-__version__ = "0.6.0"
+__version__ = "0.7.0"
@@ -0,0 +1,309 @@
+"""Parse exact device object models from an ETS ``.knxproj`` / ``.knxprod``.
+
+Deterministic extraction of the vendor communication-object model — the SUPERSET of
+objects a device can expose — from the ``M-*/`` application programs embedded in a
+``.knxproj`` (used devices) or a manufacturer ``.knxprod`` product database. The result
+feeds the local device catalog that ``decompose_device`` consumes (``catalog-exact``).
+
+Read-only, and PII-safe: it reads ONLY the manufacturer application-program XML
+(``M-*/Hardware.xml`` + ``M-*/M-*_A-*.xml`` — object number / name / size / DPT / flags).
+It never opens the client project (``P-*/0.xml``); group addresses, project names and
+device placement are not read.
+
+Parsing note: this reads the base ``<ComObject>`` elements, where most vendors
+(Zennio, ABB, EOS, Hugo Müller) publish the full object model. A few vendors (STEINEL,
+Hörmann, some Intesis) publish names/DPTs on ``<ComObjectRef>`` instead; those devices
+yield object *counts* but sparser per-object detail — flagged in the coverage manifest.
+"""
+from __future__ import annotations
+
+import os
+import re
+import zipfile
+from typing import Any, Optional
+
+_ATTR_RE = re.compile(r'(\w+)="([^"]*)"')
+_COMOBJ_RE = re.compile(r"<ComObject\b([^>]*?)/?>")
+_HW_SPLIT_RE = re.compile(r"<Hardware\b")
+_APPREF_RE = re.compile(r'<ApplicationProgramRef\s+RefId="([^"]+)"')
+_APPVER_RE = re.compile(r'ApplicationVersion="(\d+)"')
+_SIZE_RE = re.compile(r"(\d+)\s*(Bit|Byte)", re.I)
+_CH_TOKEN_RE = re.compile(r"\[([A-Za-z]{1,3}\d+)\]")          # display token: [C1] [O12] [T3]
+_LF_RE = re.compile(r"^\s*\[LF\]")
+
+# KNX manufacturer id (hex in the M-code) -> vendor name (common ones; extend freely)
+_MANUFACTURERS = {
+    "M-0001": "Siemens", "M-0002": "ABB", "M-0008": "Insta/Gira", "M-0083": "Berker",
+    "M-000C": "Merten", "M-0064": "Zennio(alt)", "M-0071": "Zennio", "M-0077": "Intesis (HMS)",
+    "M-008E": "STEINEL", "M-00C5": "Ekinex", "M-00FC": "Hugo Müller", "M-01F6": "EOS",
+    "M-0201": "Hörmann", "M-0083b": "Berker",
+}
+
+
+def _attrs(s: str) -> dict[str, str]:
+    return dict(_ATTR_RE.findall(s))
+
+
+def _size_bits(s: Optional[str]) -> Optional[int]:
+    if not s:
+        return None
+    m = _SIZE_RE.search(s)
+    if not m:
+        return None
+    n = int(m.group(1))
+    return n if m.group(2).lower() == "bit" else n * 8
+
+
+def _dpt(s: Optional[str]) -> Optional[str]:
+    """'DPST-1-2' -> '1.002'; 'DPT-1' -> '1'; missing -> None (never guessed)."""
+    if not s:
+        return None
+    m = re.match(r"DPST-(\d+)-(\d+)$", s)
+    if m:
+        return f"{int(m.group(1))}.{int(m.group(2)):03d}"
+    m = re.match(r"DPT-(\d+)$", s)
+    if m:
+        return str(int(m.group(1)))
+    return s  # unrecognised format: keep verbatim rather than drop
+
+
+def _role(write: bool, transmit: bool, read: bool, name: str) -> str:
+    n = (name or "").lower()
+    if any(k in n for k in ("status", "(status)", "state", "rückmeldung", "статус")):
+        return "status"
+    if write and not transmit:
+        return "cmd"
+    if transmit and read and not write:
+        return "status"
+    if write:
+        return "cmd"
+    return "status" if transmit else "param"
+
+
+def _channel_token(text: str, name: str) -> Optional[str]:
+    """Return the repeating-block key for an object, e.g. 'C', 'O', 'T', or None (general)."""
+    m = _CH_TOKEN_RE.search(text or "")
+    if m:
+        return re.match(r"([A-Za-z]{1,3})", m.group(1)).group(1)
+    m = re.search(r"\bch\[(\d+)\]", name or "")   # internal 'oX.ch[0].y' form
+    return "ch" if m else None
+
+
+def _parse_comobjects(xml_text: str) -> list[dict[str, Any]]:
+    objs: list[dict[str, Any]] = []
+    for m in _COMOBJ_RE.finditer(xml_text):
+        a = _attrs(m.group(1))
+        if "Number" not in a:
+            continue
+        w = a.get("WriteFlag", "").lower() == "enabled"
+        t = a.get("TransmitFlag", "").lower() == "enabled"
+        r = a.get("ReadFlag", "").lower() == "enabled"
+        text = a.get("Text") or a.get("Name") or ""
+        try:
+            number = int(a["Number"])
+        except ValueError:
+            continue
+        objs.append({
+            "number": number,
+            "name": text,
+            "internal_name": a.get("Name"),
+            "function": a.get("FunctionText"),
+            "size_bits": _size_bits(a.get("ObjectSize")),
+            "dpt": _dpt(a.get("DatapointType")),
+            "flags": {"C": a.get("CommunicationFlag", "").lower() == "enabled",
+                      "R": r, "W": w, "T": t,
+                      "U": a.get("UpdateFlag", "").lower() == "enabled"},
+            "role": _role(w, t, r, text),
+        })
+    objs.sort(key=lambda o: o["number"])
+    return objs
+
+
+def _detect_blocks(objs: list[dict[str, Any]]) -> dict[str, Any]:
+    """Best-effort per-channel block/stride + general/[LF] split from object names."""
+    general, lf, chan = [], [], {}
+    for o in objs:
+        text = o["name"] or ""
+        if _LF_RE.match(text):
+            lf.append(o)
+            continue
+        tok = _channel_token(text, o.get("internal_name") or "")
+        if tok:
+            chan.setdefault(tok, []).append(o)
+        else:
+            general.append(o)
+
+    blocks = []
+    for tok, group in chan.items():
+        # instances = distinct channel indices seen in the display tokens
+        idxs = sorted({int(mm.group(1)) for o in group
+                       for mm in [re.search(r"\[[A-Za-z]{1,3}(\d+)\]", o["name"] or "")] if mm})
+        instances = len(idxs) or 1
+        per = max(1, len(group) // instances)
+        nums = sorted(o["number"] for o in group)
+        stride = None
+        if instances > 1 and len(nums) >= per + 1:
+            stride = nums[per] - nums[0]
+        first = [o for o in group if re.search(rf"\[{tok}?0*1\]", o["name"] or "")] or group[:per]
+        blocks.append({
+            "unit": tok, "instances": instances, "objects_per_instance": per,
+            "stride": stride, "first_instance_objects": first,
+        })
+    return {"blocks": blocks, "general_objects": general, "logic_function_objects": lf}
+
+
+def _hardware_map(xml_text: str) -> list[dict[str, Any]]:
+    """From a Hardware.xml, list {order_number, name, app_refs[]} (latest app last).
+
+    ``OrderNumber`` lives on the nested ``<Product>`` element, not on ``<Hardware>``
+    (which carries ``SerialNumber``); fall back to ``SerialNumber`` if absent.
+    """
+    out = []
+    for chunk in _HW_SPLIT_RE.split(xml_text)[1:]:
+        body = chunk.split("</Hardware>")[0]
+        head = _attrs(chunk[:chunk.find(">")] if ">" in chunk else chunk)
+        m = re.search(r'OrderNumber="([^"]+)"', body)
+        order = m.group(1) if m else head.get("SerialNumber")
+        if not order:
+            continue
+        refs = _APPREF_RE.findall(body)
+        out.append({"order_number": order, "name": head.get("Name"), "app_refs": refs})
+    return out
+
+
+def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]:
+    """Parse device object models from a ``.knxproj``/``.knxprod`` archive.
+
+    Returns ``{"devices": [...], "coverage": {...}}``. Each device carries its order
+    number, application-program id/version, object counts, detected per-channel blocks
+    and the full comm-object list. Reads only ``M-*`` manufacturer data.
+    """
+    if not os.path.isfile(path):
+        return {"error": f"file not found: {path}", "devices": [], "coverage": {}}
+
+    pwd = password.encode() if password else None
+    devices: list[dict[str, Any]] = []
+    seen_mfr, app_cache, hw_count = set(), {}, 0
+    with zipfile.ZipFile(path) as z:
+        names = z.namelist()
+        hardware_files = [n for n in names if re.search(r"/Hardware\.xml$", n) and n.startswith("M-")]
+        appfiles = {n for n in names if re.match(r"M-[0-9A-Fa-f]+/M-[0-9A-Fa-f]+_A-[^/]+\.xml$", n)}
+
+        for hw in hardware_files:
+            mcode = hw.split("/", 1)[0]
+            seen_mfr.add(mcode)
+            try:
+                hw_xml = z.read(hw, pwd=pwd).decode("utf-8", "replace")
+            except Exception:
+                continue
+            for entry in _hardware_map(hw_xml):
+                hw_count += 1
+                app_ref = entry["app_refs"][-1] if entry["app_refs"] else None
+                app_path = f"{mcode}/{app_ref}.xml" if app_ref else None
+                if not app_path or app_path not in appfiles:
+                    devices.append({
+                        "order_number": entry["order_number"], "name": entry["name"],
+                        "manufacturer": _MANUFACTURERS.get(mcode, mcode),
+                        "application_program": {"app_id": app_ref, "version": None},
+                        "object_counts": {"master_catalog_total": 0},
+                        "blocks": [], "comm_objects": [],
+                        "note": "no application-program XML present for this order number",
+                    })
+                    continue
+                if app_path not in app_cache:
+                    xml_text = z.read(app_path, pwd=pwd).decode("utf-8", "replace")
+                    ver = _APPVER_RE.search(xml_text[:8000])
+                    app_cache[app_path] = (_parse_comobjects(xml_text),
+                                           ver.group(1) if ver else None)
+                objs, ver = app_cache[app_path]
+                bd = _detect_blocks(objs)
+                no_dpt = sum(1 for o in objs if o["dpt"] is None)
+                devices.append({
+                    "order_number": entry["order_number"], "name": entry["name"],
+                    "manufacturer": _MANUFACTURERS.get(mcode, mcode),
+                    "application_program": {"app_id": app_ref,
+                                            "version": f"v{ver}" if ver else None},
+                    "object_counts": {
+                        "master_catalog_total": len(objs),
+                        "general_objects": len(bd["general_objects"]),
+                        "logic_function_objects": len(bd["logic_function_objects"]),
+                        "objects_without_declared_dpt": no_dpt,
+                    },
+                    "blocks": [{k: v for k, v in b.items() if k != "first_instance_objects"}
+                               | {"first_instance_objects": b["first_instance_objects"]}
+                               for b in bd["blocks"]],
+                    "comm_objects": objs,
+                })
+
+    total_obj = sum(d["object_counts"]["master_catalog_total"] for d in devices)
+    total_nodpt = sum(d["object_counts"].get("objects_without_declared_dpt", 0) for d in devices)
+    coverage = {
+        "archive": os.path.basename(path),
+        "manufacturers_seen": sorted(seen_mfr),
+        "hardware_entries": hw_count,
+        "devices_parsed": len(devices),
+        "app_programs_parsed": len(app_cache),
+        "objects_total": total_obj,
+        "objects_without_dpt": total_nodpt,
+        "note": "Object model = vendor master SUPERSET (all objects a device CAN expose); "
+                "ETS parameters enable a subset per config. DPT null = vendor declared none "
+                "(never guessed). Base <ComObject> only; ref-level vendors yield counts but "
+                "sparser detail.",
+    }
+    return {"devices": devices, "coverage": coverage}
+
+
+def _yaml_row(o: dict[str, Any]) -> dict[str, Any]:
+    return {"number": o["number"], "name": o["name"], "function": o.get("function"),
+            "size_bits": o.get("size_bits"), "dpt": o.get("dpt") or "unverified",
+            "flags": o.get("flags"), "role": o.get("role")}
+
+
+def summary(result: dict[str, Any]) -> dict[str, Any]:
+    """Trim a parse result to a per-device summary (no full object lists) + coverage."""
+    devs = []
+    for d in result.get("devices", []):
+        oc = d.get("object_counts", {})
+        devs.append({
+            "order_number": d.get("order_number"), "name": d.get("name"),
+            "manufacturer": d.get("manufacturer"),
+            "app_version": (d.get("application_program") or {}).get("version"),
+            "master_objects": oc.get("master_catalog_total"),
+            "objects_without_dpt": oc.get("objects_without_declared_dpt"),
+            "blocks": [{"unit": b.get("unit"), "objects_per_instance": b.get("objects_per_instance"),
+                        "instances": b.get("instances"), "stride": b.get("stride")}
+                       for b in d.get("blocks", [])],
+        })
+    return {"coverage": result.get("coverage", {}), "devices": devs}
+
+
+def to_catalog_yaml(result: dict[str, Any]) -> str:
+    """Render a parse result as device-library YAML (library-schema.md), catalog-exact-ready.
+
+    Per-channel blocks are collapsed to their first instance (+ stride); the full object
+    list is not duplicated, matching the harvested-catalog convention.
+    """
+    import yaml
+    devs = []
+    for d in result.get("devices", []):
+        blocks = [{"unit": b.get("unit"), "instances": b.get("instances"),
+                   "objects_per_instance": b.get("objects_per_instance"),
+                   "stride": b.get("stride"),
+                   "objects": [_yaml_row(o) for o in b.get("first_instance_objects", [])]}
+                  for b in d.get("blocks", [])]
+        devs.append({
+            "order_number": d.get("order_number"), "name": d.get("name"),
+            "manufacturer": d.get("manufacturer"),
+            "application_program": d.get("application_program"),
+            "object_counts": d.get("object_counts"),
+            "repeating_blocks": blocks,
+        })
+    top = {
+        "schema_version": "0.1",
+        "manufacturer": "(multiple — see per-device)",
+        "source_note": "Parsed from ETS application programs by parse_devices_from_project. "
+                       "Master catalog SUPERSET; DPT 'unverified' = vendor declared none. "
+                       "No client project (P-*/0.xml) data.",
+        "devices": devs,
+    }
+    return yaml.safe_dump(top, allow_unicode=True, sort_keys=False, width=100)
@@ -17,6 +17,8 @@ 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)].
@@ -183,6 +185,129 @@ def _lookup(key: str) -> Optional[str]:
    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.

@@ -192,18 +317,28 @@ def decompose_device(order_number: str, channels: int = 1) -> dict[str, Any]:
        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 recipe found. Known types: " + ", ".join(sorted(_RECIPES)),
+            "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"],
@@ -23,6 +23,8 @@ from .generate_ets import generate_ets_csv, generate_ets_xml
 from .report import build_report
 from .handover import build_handover
 from .device_library import decompose_device as _decompose_device, list_recipes
+from .appprog_parser import (parse_project as _parse_project,
+                             summary as _appprog_summary, to_catalog_yaml as _appprog_to_yaml)
 from .repair import suggest_repairs as _suggest_repairs
 from .advanced import (matter_readiness, completeness_grade, energy_scaffold,
                       test_protocol, suggest_naming)
@@ -315,6 +317,34 @@ def list_device_recipes() -> list[dict[str, Any]]:
    return list_recipes()


+@mcp.tool()
+def parse_devices_from_project(path: str, output_path: Optional[str] = None,
+                               password: Optional[str] = None) -> dict[str, Any]:
+    """Extract exact device object models from a .knxproj / .knxprod application programs.
+
+    Reads the manufacturer application programs (M-*) embedded in an ETS `.knxproj`
+    (devices actually used) or a `.knxprod` product database, and returns each device's
+    order number, app-program version, object counts and detected per-channel blocks —
+    the EXACT vendor comm-object model, not a generic recipe. Read-only and PII-safe: it
+    reads only vendor catalog data, never the client project (P-*/0.xml).
+
+    Use this to build/grow the local device catalog that `decompose_device` consumes
+    (set NICKOL_KNX_CATALOG to the catalog dir). If `output_path` is given, the full
+    catalog is written into the workspace as device-library YAML; the return value is
+    always a compact per-device summary + coverage manifest (the full object lists are
+    not inlined). DPT `unverified` = the vendor app-program declares none (never guessed).
+    """
+    result = _parse_project(path, password=password)
+    if "error" in result:
+        return result
+    out = _appprog_summary(result)
+    if output_path:
+        out["written"] = _safe_write(output_path, _appprog_to_yaml(result))
+        out["written_note"] = ("Local catalog file. Point NICKOL_KNX_CATALOG at its "
+                               "directory to make decompose_device return catalog-exact.")
+    return out
+
+
@mcp.tool()
 def check_matter() -> dict[str, Any]:
    """Matter-readiness lint: which controllable functions round-trip to a Matter
@@ -1,6 +1,6 @@
 [project]
 name = "nickol-knx-mcp"
-version = "0.6.0"
+version = "0.7.0"
 description = "Design-time KNX/ETS6 project assistant as an MCP server (parse .knxproj, validate, generate HA YAML + ETS CSV/XML). No live bus access."
 readme = "README.md"
 requires-python = ">=3.10"
@@ -0,0 +1,84 @@
+"""appprog_parser: extract exact device object models from a (synthetic) .knxprod.
+
+Self-contained: builds a minimal ETS-shaped archive in memory — no real project needed.
+"""
+import sys, os, io, zipfile, tempfile
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from nickol_knx_mcp.appprog_parser import parse_project, to_catalog_yaml, summary
+import nickol_knx_mcp.device_library as dl
+
+HARDWARE = """<?xml version="1.0" encoding="utf-8"?>
+<KNX><ManufacturerData><Manufacturer RefId="M-0071"><Hardware>
+<Hardware Id="M-0071_H-TST-1" Name="Test Dimmer 2CH" SerialNumber="TST-DIM2">
+<Products><Product Id="M-0071_H-TST-1_P-1" OrderNumber="TST-DIM2" /></Products>
+<Hardware2Programs><Hardware2Program>
+<ApplicationProgramRef RefId="M-0071_A-TEST-1-0" />
+</Hardware2Program></Hardware2Programs>
+</Hardware>
+</Hardware></Manufacturer></ManufacturerData></KNX>"""
+
+APPPROG = """<?xml version="1.0" encoding="utf-8"?>
+<KNX><ManufacturerData><Manufacturer><ApplicationPrograms>
+<ApplicationProgram Id="M-0071_A-TEST-1-0" ApplicationVersion="7"><Static><ComObjectTable>
+<ComObject Id="O-1" Name="oX.ch[0].onOff" Text="[C1] On/Off" Number="1" FunctionText="on/off" ObjectSize="1 Bit" ReadFlag="Disabled" WriteFlag="Enabled" CommunicationFlag="Enabled" TransmitFlag="Disabled" UpdateFlag="Disabled" DatapointType="DPST-1-1" />
+<ComObject Id="O-2" Name="oX.ch[0].dim" Text="[C1] Relative Dimming" Number="2" ObjectSize="4 Bit" WriteFlag="Enabled" CommunicationFlag="Enabled" DatapointType="DPST-3-7" />
+<ComObject Id="O-3" Name="oX.ch[1].onOff" Text="[C2] On/Off" Number="3" ObjectSize="1 Bit" WriteFlag="Enabled" CommunicationFlag="Enabled" DatapointType="DPST-1-1" />
+<ComObject Id="O-4" Name="oX.ch[1].dim" Text="[C2] Relative Dimming" Number="4" ObjectSize="4 Bit" WriteFlag="Enabled" CommunicationFlag="Enabled" DatapointType="DPST-3-7" />
+<ComObject Id="O-5" Name="oX.pattern" Text="[LF] Pattern" Number="5" ObjectSize="1 Byte" WriteFlag="Enabled" CommunicationFlag="Enabled" />
+</ComObjectTable></Static></ApplicationProgram>
+</ApplicationPrograms></Manufacturer></ManufacturerData></KNX>"""
+
+
+def _make_archive(path):
+    with zipfile.ZipFile(path, "w") as z:
+        z.writestr("P-1/0.xml", "<KNX/>")  # a 'client project' the parser must NOT read
+        z.writestr("M-0071/Hardware.xml", HARDWARE)
+        z.writestr("M-0071/M-0071_A-TEST-1-0.xml", APPPROG)
+
+
+with tempfile.TemporaryDirectory() as d:
+    arc = os.path.join(d, "sample.knxprod")
+    _make_archive(arc)
+
+    res = parse_project(arc)
+    cov = res["coverage"]
+    assert cov["devices_parsed"] == 1, cov
+    assert cov["objects_total"] == 5, cov
+    assert cov["objects_without_dpt"] == 1, cov          # the [LF] Pattern, no DPT
+
+    dev = res["devices"][0]
+    assert dev["order_number"] == "TST-DIM2", dev
+    assert dev["application_program"]["version"] == "v7", dev
+    assert dev["object_counts"]["master_catalog_total"] == 5
+
+    # DPT conversion DPST-x-y -> x.00y
+    o2 = [o for o in dev["comm_objects"] if o["number"] == 2][0]
+    assert o2["dpt"] == "3.007", o2
+    o1 = [o for o in dev["comm_objects"] if o["number"] == 1][0]
+    assert o1["dpt"] == "1.001" and o1["role"] == "cmd", o1
+    o5 = [o for o in dev["comm_objects"] if o["number"] == 5][0]
+    assert o5["dpt"] is None, "missing DatapointType must stay None, never guessed"
+
+    # per-channel block detection: [C1]/[C2] -> 2 instances of 2 objects, stride 2
+    blk = [b for b in dev["blocks"] if b["unit"] == "C"][0]
+    assert blk["instances"] == 2 and blk["objects_per_instance"] == 2, blk
+    assert blk["stride"] == 2, blk
+
+    # summary is compact (no full object lists)
+    sm = summary(res)
+    assert "comm_objects" not in sm["devices"][0]
+
+    # round-trip: emit YAML -> load as catalog -> decompose_device is catalog-exact
+    cat = os.path.join(d, "cat.yaml")
+    with open(cat, "w", encoding="utf-8") as fh:
+        fh.write(to_catalog_yaml(res))
+    os.environ["NICKOL_KNX_CATALOG"] = d
+    dl._CATALOG_INDEX = None
+    r = dl.decompose_device("TST-DIM2")
+    assert r["source"] == "catalog-exact", r
+    assert r["total_master_objects"] == 5, r
+    os.environ.pop("NICKOL_KNX_CATALOG", None)
+    dl._CATALOG_INDEX = None
+
+print("OK — appprog_parser: order->app map, DPST->DPT, honest missing-DPT, block/stride, YAML round-trip")
@@ -0,0 +1,78 @@
+"""decompose_device: generic-recipe fallback + opt-in local exact catalog.
+
+Self-contained: builds a tiny synthetic catalog in a temp dir, so the test does
+NOT depend on any locally-harvested (unshipped) catalog data.
+"""
+import sys, os, tempfile
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import nickol_knx_mcp.device_library as dl
+
+SAMPLE_CATALOG = """\
+schema_version: "0.1"
+manufacturer: "TestVendor"
+devices:
+  - order_number: "TV-DIM4"
+    name: "TestDimmer 4CH"
+    category: "actuator/dimmer"
+    application_program: { name: "TestDimmer", version: "v3" }
+    channels: { dimmer_channels: 4 }
+    object_counts: { master_catalog_total: 92 }
+    repeating_blocks:
+      - unit: "dimmer_channel"
+        instances: 4
+        objects_per_instance: 5
+        stride: 5
+        objects:
+          - { number: 1, name: "[C1] On/Off", dpt: "1.001", role: cmd, size_bits: 1 }
+          - { number: 2, name: "[C1] On/Off (Status)", dpt: "1.001", role: status, size_bits: 1 }
+          - { number: 3, name: "[C1] Relative Dimming", dpt: "3.007", role: cmd, size_bits: 4 }
+          - { number: 4, name: "[C1] Absolute Dimming", dpt: "5.001", role: cmd, size_bits: 8 }
+          - { number: 5, name: "[C1] Pattern", role: cmd, size_bits: 8 }   # no dpt -> stays None
+"""
+
+
+def _reset(catalog_path=None):
+    if catalog_path:
+        os.environ["NICKOL_KNX_CATALOG"] = catalog_path
+    else:
+        os.environ.pop("NICKOL_KNX_CATALOG", None)
+    dl._CATALOG_INDEX = None  # drop the module cache between modes
+
+
+# 1) No catalog configured -> generic recipe, behaviour unchanged
+_reset()
+r = dl.decompose_device("ZIO-MB24", 2)
+assert r["matched"] and r["source"] == "recipe-approximate", r
+assert r["total_ga"] == r["objects_per_unit"] * 2, r
+
+r = dl.decompose_device("no-such-device")
+assert r["matched"] is False and r["source"] == "recipe-approximate", r
+
+# 2) Local catalog configured -> exact model, matched by order number AND by name
+with tempfile.TemporaryDirectory() as d:
+    with open(os.path.join(d, "vendor.yaml"), "w", encoding="utf-8") as fh:
+        fh.write(SAMPLE_CATALOG)
+    _reset(d)
+
+    for q in ("TV-DIM4", "tv-dim4", "TestDimmer 4CH", "testdimmer 4ch"):
+        r = dl.decompose_device(q)
+        assert r["source"] == "catalog-exact", (q, r["source"])
+        assert r["order_number"] == "TV-DIM4", (q, r)
+        assert r["application_program"]["version"] == "v3", r
+        assert r["total_master_objects"] == 92, r
+        blk = r["blocks"][0]
+        assert blk["unit"] == "dimmer_channel" and blk["instances"] == 4, blk
+        assert blk["objects_per_instance"] == 5, blk
+        assert blk["first_instance_objects"][0]["dpt"] == "1.001", blk
+        assert blk["first_instance_objects"][4]["dpt"] is None, "no DPT must stay None, never guessed"
+
+    # a device NOT in the catalog still falls back to the generic recipe
+    r = dl.decompose_device("dimmer")
+    assert r["source"] == "recipe-approximate" and r["matched"], r
+
+# 3) env cleared -> back to recipe-only, no stale cache
+_reset()
+assert dl.decompose_device("TV-DIM4")["matched"] is False
+
+print("OK — device catalog wiring: recipe fallback + catalog-exact + name match + honest unverified DPT")