From ebc03f711881b291659a06628d666eb7989b0ddb Mon Sep 17 00:00:00 2001 From: Nikolay1 Date: Wed, 15 Jul 2026 21:26:11 +0200 Subject: [PATCH] room-library R1: template format + resolver/IR + composer core Add the Room Template Library core: the public YAML template contract (schema_version, locale-neutral semantic slot_id, ru/en labels, per-slot basic/comfort presets, area_m2 as a provenance-tagged hint) with 6 built-in rooms (bedroom, children, living, kitchen, bathroom, corridor) + SCHEMA.md. room_library.py implements a function-first pipeline: templates+params -> resolved IR dataclasses (NOT GARecord) -> deterministic, permutation-invariant allocation (main=domain, middle=role, sub sequential; hard error on sub exhaustion) -> a real .knxproj ZIP that is re-read via load_project (same path as third-party projects, so generation never touches the classifier). Includes validate_room_template, compose(), manifest + device BOM proposal builders. Co-Authored-By: Claude Opus 4.8 --- nickol_knx_mcp/room_library.py | 799 ++++++++++++++++++++ nickol_knx_mcp/room_templates/SCHEMA.md | 106 +++ nickol_knx_mcp/room_templates/bathroom.yaml | 49 ++ nickol_knx_mcp/room_templates/bedroom.yaml | 63 ++ nickol_knx_mcp/room_templates/children.yaml | 55 ++ nickol_knx_mcp/room_templates/corridor.yaml | 43 ++ nickol_knx_mcp/room_templates/kitchen.yaml | 49 ++ nickol_knx_mcp/room_templates/living.yaml | 56 ++ 8 files changed, 1220 insertions(+) create mode 100644 nickol_knx_mcp/room_library.py create mode 100644 nickol_knx_mcp/room_templates/SCHEMA.md create mode 100644 nickol_knx_mcp/room_templates/bathroom.yaml create mode 100644 nickol_knx_mcp/room_templates/bedroom.yaml create mode 100644 nickol_knx_mcp/room_templates/children.yaml create mode 100644 nickol_knx_mcp/room_templates/corridor.yaml create mode 100644 nickol_knx_mcp/room_templates/kitchen.yaml create mode 100644 nickol_knx_mcp/room_templates/living.yaml diff --git a/nickol_knx_mcp/room_library.py b/nickol_knx_mcp/room_library.py new file mode 100644 index 0000000..9789955 --- /dev/null +++ b/nickol_knx_mcp/room_library.py @@ -0,0 +1,799 @@ +"""Room Template Library — R1 (vertical slice). + +Compose a NEW KNX project from a list of parametrised *room templates*. + +Design contract (see ``room_templates/SCHEMA.md``): + + * A room template is a locale-neutral YAML file. Its identity is a semantic + ``slot_id`` (an ASCII identifier), **never** its human name — ``labels.ru`` / + ``labels.en`` are presentation only, so a translation never changes identity. + * Each template declares ``parameters`` (window count, lighting-circuit count, + …). ``area_m2`` is a **hint** that seeds defaults, carrying provenance; it is + never treated as a normative fact and, in R1, never changes GA counts. + * Each template declares functional **slots**. A slot carries per-slot + ``basic`` / ``comfort`` presets (NOT one monolithic room level), so a house + can mix "comfort climate + basic lighting" per room. + * Slots reference **function types** (lighting_switch, lighting_dimmer, + shutter, climate_floor, presence). A function type expands into a fixed set + of KNX communication objects (command + status), each with a canonical DPT. + +The composition pipeline is deliberately layered and pure: + + templates + room specs + -> resolved functional model (IR dataclasses; NOT ``GARecord``) + -> address allocation (main = domain, middle = role, sub sequential) + -> a real ``.knxproj`` ZIP (ETS project/22 XML) + -> re-read through :func:`project.load_project` (the SAME reader used for + third-party projects) so the generator never touches the classifier. + +Everything downstream (ETS export, linters, manifest, BOM) runs off the re-read +project, exactly like a hand-made ETS file. The server never writes to a bus. + +R1 scope: NEW projects only, dry-run by default. Docking into an existing +project (allocation lockfile, drift detection) and the exact device-resolver are +R2 — see ``docs/roadmap/room-library/implementation-plan.md``. +""" + +from __future__ import annotations + +import io +import json +import os +import zipfile +from dataclasses import dataclass, field +from typing import Any, Optional + +# --------------------------------------------------------------------------- # +# Public format contract +# --------------------------------------------------------------------------- # +SCHEMA_VERSION = 1 +SUPPORTED_LANGUAGES = ("ru", "en") + +# Default 3-level taxonomy (CLAUDE.md methodology — main group = function domain). +DOMAIN_MAIN: dict[str, int] = { + "central": 0, + "lighting": 1, + "shutter": 2, + "hvac": 3, + "sensor": 4, + "energy": 5, + "diagnostics": 6, + "reserve": 7, +} + +DOMAIN_LABELS: dict[str, dict[str, str]] = { + "central": {"ru": "Центральные функции", "en": "Central"}, + "lighting": {"ru": "Освещение", "en": "Lighting"}, + "shutter": {"ru": "Шторы", "en": "Shutters"}, + "hvac": {"ru": "Климат", "en": "HVAC"}, + "sensor": {"ru": "Датчики", "en": "Sensors"}, + "energy": {"ru": "Энергия", "en": "Energy"}, + "diagnostics": {"ru": "Диагностика", "en": "Diagnostics"}, + "reserve": {"ru": "Резерв", "en": "Reserve"}, +} + +# Middle group = role/sub-function inside a domain. Command roles and their +# feedback roles live in DISTINCT, predictable middles (the pairing engine keys +# off name tokens, not adjacency — see CLAUDE.md). +MIDDLE_ROLE: dict[str, dict[str, int]] = { + "lighting": {"cmd_onoff": 0, "dimming": 1, "brightness_value": 2, + "status_onoff": 3, "brightness_status": 4}, + "shutter": {"move": 0, "stop": 1, "position": 2, "slat": 3, + "position_status": 4}, + "hvac": {"onoff": 0, "mode": 1, "setpoint": 2, "fan": 3, "status": 4, + "actual_temp": 5}, + "sensor": {"presence": 0, "illuminance": 1, "climate": 2}, + "central": {"scene": 0, "command": 1}, +} + +MIDDLE_LABELS: dict[str, dict[int, dict[str, str]]] = { + "lighting": {0: {"ru": "Вкл/Выкл", "en": "On/Off"}, + 1: {"ru": "Диммирование", "en": "Dimming"}, + 2: {"ru": "Значение яркости", "en": "Brightness value"}, + 3: {"ru": "Статус вкл/выкл", "en": "On/Off status"}, + 4: {"ru": "Статус яркости", "en": "Brightness status"}}, + "shutter": {0: {"ru": "Движение", "en": "Move"}, + 1: {"ru": "Стоп", "en": "Stop"}, + 2: {"ru": "Позиция", "en": "Position"}, + 3: {"ru": "Ламели", "en": "Slats"}, + 4: {"ru": "Статус позиции", "en": "Position status"}}, + "hvac": {0: {"ru": "Вкл/Выкл", "en": "On/Off"}, + 1: {"ru": "Режим", "en": "Mode"}, + 2: {"ru": "Уставка", "en": "Setpoint"}, + 3: {"ru": "Вентилятор", "en": "Fan"}, + 4: {"ru": "Статусы", "en": "Statuses"}, + 5: {"ru": "Температура факт", "en": "Actual temperature"}}, + "sensor": {0: {"ru": "Присутствие/Движение", "en": "Presence/Motion"}, + 1: {"ru": "Освещённость", "en": "Illuminance"}, + 2: {"ru": "Климат в комнатах", "en": "Room climate"}}, + "central": {0: {"ru": "Сцены", "en": "Scenes"}, + 1: {"ru": "Центральные команды", "en": "Central commands"}}, +} + +MAX_SUB = 255 # 3-level sub range 0..255; sub 0 reserved as range anchor +MAX_MIDDLE = 7 # 3-level middle range 0..7 + + +# --------------------------------------------------------------------------- # +# Function-type object model (function-first: objects, not devices). +# Each object: role_key, ru label, en label, dpt (main, sub), kind, domain, +# middle_role. Command objects are paired to their status object by name tokens. +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ObjSpec: + role: str + label_ru: str + label_en: str + dpt_main: int + dpt_sub: int + kind: str # command | status | sensor + domain: str + middle_role: str + + def label(self, language: str) -> str: + return self.label_ru if language == "ru" else self.label_en + + @property + def dpst(self) -> str: + return f"DPST-{self.dpt_main}-{self.dpt_sub}" + + +FUNCTION_OBJECTS: dict[str, list[ObjSpec]] = { + "lighting_switch": [ + ObjSpec("onoff", "вкл/выкл", "on/off", 1, 1, "command", "lighting", "cmd_onoff"), + ObjSpec("onoff_status", "статус", "status", 1, 11, "status", "lighting", "status_onoff"), + ], + "lighting_dimmer": [ + ObjSpec("onoff", "вкл/выкл", "on/off", 1, 1, "command", "lighting", "cmd_onoff"), + ObjSpec("dimming", "диммирование", "dimming", 3, 7, "command", "lighting", "dimming"), + ObjSpec("brightness", "яркость", "brightness", 5, 1, "command", "lighting", "brightness_value"), + ObjSpec("onoff_status", "статус", "status", 1, 11, "status", "lighting", "status_onoff"), + ObjSpec("brightness_status", "яркость статус", "brightness status", 5, 1, "status", "lighting", "brightness_status"), + ], + "shutter": [ + ObjSpec("updown", "вверх/вниз", "up/down", 1, 8, "command", "shutter", "move"), + ObjSpec("stop", "стоп", "stop", 1, 10, "command", "shutter", "stop"), + ObjSpec("position", "позиция", "position", 5, 1, "command", "shutter", "position"), + ObjSpec("position_status", "позиция статус", "position status", 5, 1, "status", "shutter", "position_status"), + ], + "climate_floor": [ + ObjSpec("onoff", "вкл/выкл", "on/off", 1, 1, "command", "hvac", "onoff"), + ObjSpec("setpoint", "уставка", "setpoint", 9, 1, "command", "hvac", "setpoint"), + ObjSpec("mode", "режим", "mode", 20, 102, "command", "hvac", "mode"), + ObjSpec("onoff_status", "статус", "status", 1, 11, "status", "hvac", "status"), + ObjSpec("setpoint_status", "уставка статус", "setpoint status", 9, 1, "status", "hvac", "status"), + ObjSpec("mode_status", "режим статус", "mode status", 20, 102, "status", "hvac", "status"), + ObjSpec("actual_temp", "температура факт", "actual temperature", 9, 1, "sensor", "hvac", "actual_temp"), + ], + "presence": [ + ObjSpec("occupancy", "присутствие", "occupancy", 1, 18, "sensor", "sensor", "presence"), + ObjSpec("illuminance", "освещённость", "illuminance", 9, 4, "sensor", "sensor", "illuminance"), + ], +} + +# Function type -> candidate device recipe (device_library) for the BOM proposal. +BOM_RECIPE: dict[str, str] = { + "lighting_switch": "switch_output", + "lighting_dimmer": "dimmer_channel", + "shutter": "shutter_channel", + "climate_floor": "floor_heating_zone", + "presence": "presence_detector", +} + + +class RoomLibraryError(ValueError): + """Raised on an unrecoverable composition error (address exhaustion, bad + template, colliding room labels). Never a silent overflow.""" + + +# --------------------------------------------------------------------------- # +# Resolved intermediate model (IR) — deliberately separate dataclasses, NOT the +# public GARecord. The IR is the composition's own contract; GARecord only +# appears AFTER the generated project is re-read by the standard loader. +# --------------------------------------------------------------------------- # +@dataclass +class ResolvedObject: + role: str + name: str + dpt_main: int + dpt_sub: int + kind: str + domain: str + main: int + middle: int + sub: int + + @property + def address(self) -> str: + return f"{self.main}/{self.middle}/{self.sub}" + + @property + def dpst(self) -> str: + return f"DPST-{self.dpt_main}-{self.dpt_sub}" + + +@dataclass +class ResolvedSlot: + slot_id: str + label: str + function_type: str + index: int # 1-based index within a multi-instance slot + objects: list[ResolvedObject] = field(default_factory=list) + + +@dataclass +class ResolvedRoom: + template_id: str + label: str + preset: str + params: dict[str, Any] + slots: list[ResolvedSlot] = field(default_factory=list) + + +@dataclass +class ComposedHouse: + project_name: str + language: str + rooms: list[ResolvedRoom] = field(default_factory=list) + + @property + def objects(self) -> list[ResolvedObject]: + out: list[ResolvedObject] = [] + for r in self.rooms: + for s in r.slots: + out.extend(s.objects) + return out + + +# --------------------------------------------------------------------------- # +# Template loading +# --------------------------------------------------------------------------- # +def _templates_dir() -> str: + return os.path.join(os.path.dirname(os.path.abspath(__file__)), "room_templates") + + +def load_builtin_templates() -> dict[str, dict[str, Any]]: + """Load every built-in room template keyed by its semantic ``slot_id``.""" + import yaml + out: dict[str, dict[str, Any]] = {} + d = _templates_dir() + for fn in sorted(os.listdir(d)): + if not fn.endswith((".yaml", ".yml")): + continue + with open(os.path.join(d, fn), encoding="utf-8") as fh: + data = yaml.safe_load(fh) + if isinstance(data, dict) and data.get("slot_id"): + data.setdefault("_source_file", fn) + out[data["slot_id"]] = data + return out + + +def load_template_file(path: str) -> dict[str, Any]: + import yaml + with open(path, encoding="utf-8") as fh: + data = yaml.safe_load(fh) + if not isinstance(data, dict): + raise RoomLibraryError(f"{path}: template root must be a mapping") + data.setdefault("_source_file", os.path.basename(path)) + return data + + +# --------------------------------------------------------------------------- # +# Template validation (schema-valid rung of the validity ladder) +# --------------------------------------------------------------------------- # +_IDENT_OK = set("abcdefghijklmnopqrstuvwxyz0123456789_") + + +def _is_ident(s: Any) -> bool: + return isinstance(s, str) and bool(s) and set(s) <= _IDENT_OK + + +def _finding(severity: str, code: str, message: str, **extra: Any) -> dict[str, Any]: + f = {"severity": severity, "code": code, "message": message} + f.update(extra) + return f + + +def validate_template(tmpl: dict[str, Any]) -> dict[str, Any]: + """Validate one room template against the R1 schema. Report-only. + + Returns ``{ok, schema_version, slot_id, findings}``. ``ok`` is False when any + finding is an error. + """ + findings: list[dict[str, Any]] = [] + + sv = tmpl.get("schema_version") + if sv is None: + findings.append(_finding("error", "schema_version_missing", + "Template has no schema_version (breaks migration/compat).")) + elif sv != SCHEMA_VERSION: + findings.append(_finding("error", "schema_version_unsupported", + f"schema_version {sv} is not supported (this build understands " + f"{SCHEMA_VERSION}).")) + + slot_id = tmpl.get("slot_id") + if not _is_ident(slot_id): + findings.append(_finding("error", "slot_id_invalid", + "Template slot_id must be a locale-neutral ASCII identifier " + "(identity must not come from a human name).")) + + labels = tmpl.get("labels") or {} + for lang in SUPPORTED_LANGUAGES: + if not (labels.get(lang) or "").strip(): + findings.append(_finding("error", "label_missing", + f"Template is missing labels.{lang}.", lang=lang)) + + # parameters: each needs a default; area_m2 must be declared a hint with provenance. + params = tmpl.get("parameters") or {} + if not isinstance(params, dict): + findings.append(_finding("error", "parameters_invalid", "parameters must be a mapping.")) + params = {} + for pname, pdef in params.items(): + if not isinstance(pdef, dict) or "default" not in pdef: + findings.append(_finding("error", "parameter_no_default", + f"Parameter '{pname}' has no default.", parameter=pname)) + continue + if pname == "area_m2": + if pdef.get("role") != "hint": + findings.append(_finding("error", "area_not_hint", + "area_m2 must be declared role: hint — area is a defaults " + "recommender, never a normative fact.", parameter=pname)) + if not pdef.get("provenance"): + findings.append(_finding("warning", "area_no_provenance", + "area_m2 should carry provenance (source/user_overridden).", + parameter=pname)) + + # slots + slots = tmpl.get("slots") or [] + if not isinstance(slots, list) or not slots: + findings.append(_finding("error", "slots_missing", "Template declares no slots.")) + slots = [] + seen_slots: set[str] = set() + for i, slot in enumerate(slots): + if not isinstance(slot, dict): + findings.append(_finding("error", "slot_invalid", f"Slot #{i} is not a mapping.")) + continue + sid = slot.get("slot_id") + if not _is_ident(sid): + findings.append(_finding("error", "slot_id_invalid", + f"Slot #{i} slot_id must be a locale-neutral identifier.", + slot=sid)) + elif sid in seen_slots: + findings.append(_finding("error", "slot_id_duplicate", + f"Slot id '{sid}' is used more than once.", slot=sid)) + else: + seen_slots.add(sid) + slabels = slot.get("labels") or {} + for lang in SUPPORTED_LANGUAGES: + if not (slabels.get(lang) or "").strip(): + findings.append(_finding("error", "slot_label_missing", + f"Slot '{sid}' is missing labels.{lang}.", + slot=sid, lang=lang)) + presets = slot.get("presets") or {} + if not isinstance(presets, dict) or not {"basic", "comfort"} <= set(presets): + findings.append(_finding("error", "slot_presets_missing", + f"Slot '{sid}' must define both basic and comfort presets.", + slot=sid)) + presets = presets if isinstance(presets, dict) else {} + for pname, pdef in presets.items(): + if not isinstance(pdef, dict): + findings.append(_finding("error", "preset_invalid", + f"Slot '{sid}' preset '{pname}' must be a mapping.", + slot=sid, preset=pname)) + continue + if not pdef.get("enabled", False): + continue # a disabled preset needs no function/multiplicity + ftype = pdef.get("function") + if ftype not in FUNCTION_OBJECTS: + findings.append(_finding("error", "unknown_function", + f"Slot '{sid}' preset '{pname}' uses unknown function " + f"'{ftype}'. Known: {sorted(FUNCTION_OBJECTS)}.", + slot=sid, preset=pname, function=ftype)) + mult = pdef.get("multiplicity") or {} + if not _mult_is_valid(mult, params): + findings.append(_finding("error", "multiplicity_invalid", + f"Slot '{sid}' preset '{pname}' multiplicity {mult} is " + "invalid (need {fixed: N} or {param: }).", + slot=sid, preset=pname)) + + errors = sum(1 for f in findings if f["severity"] == "error") + return { + "ok": errors == 0, + "schema_version": sv, + "slot_id": slot_id, + "source_file": tmpl.get("_source_file"), + "findings_summary": {"errors": errors, + "warnings": sum(1 for f in findings if f["severity"] == "warning"), + "total": len(findings)}, + "findings": findings, + } + + +# Public alias (the MCP tool name mirrors this). +validate_room_template = validate_template + + +def _mult_is_valid(mult: Any, params: dict[str, Any]) -> bool: + if not isinstance(mult, dict): + return False + if "fixed" in mult: + return isinstance(mult["fixed"], int) and mult["fixed"] >= 0 + if "param" in mult: + return mult["param"] in params + return False + + +# --------------------------------------------------------------------------- # +# Resolution: templates + room specs -> ResolvedRoom (function-first IR) +# --------------------------------------------------------------------------- # +def _canonical_key(spec: dict[str, Any]) -> str: + """Order-independent identity of a room request, so permuting the room list + yields byte-identical output (permutation invariance).""" + return json.dumps({ + "template": spec.get("template"), + "preset": spec.get("preset", "basic"), + "params": spec.get("params") or {}, + "slot_presets": spec.get("slot_presets") or {}, + "label": spec.get("label"), + }, sort_keys=True, ensure_ascii=False) + + +def _resolve_multiplicity(mult: dict[str, Any], params: dict[str, Any]) -> int: + if "fixed" in mult: + return max(0, int(mult["fixed"])) + if "param" in mult: + return max(0, int(params.get(mult["param"], 0))) + return 0 + + +def _room_params(tmpl: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + params: dict[str, Any] = {} + for pname, pdef in (tmpl.get("parameters") or {}).items(): + params[pname] = pdef.get("default") + for pname, val in (override or {}).items(): + params[pname] = val + return params + + +def _instance_labels(rooms: list[dict[str, Any]], templates: dict[str, dict[str, Any]], + language: str) -> list[str]: + """Assign a unique, human-readable zone label to each (canonically sorted) + room. Same template used N times -> base label, base 2, base 3 …""" + labels: list[str] = [] + counts: dict[str, int] = {} + for spec in rooms: + tmpl = templates[spec["template"]] + base = spec.get("label") or (tmpl.get("labels") or {}).get(language) or spec["template"] + counts[base] = counts.get(base, 0) + 1 + labels.append(base if counts[base] == 1 else f"{base} {counts[base]}") + return labels + + +def resolve_house(rooms: list[dict[str, Any]], templates: dict[str, dict[str, Any]], + language: str = "ru", project_name: str = "Room Library house") -> ComposedHouse: + """Resolve room specs into an allocated :class:`ComposedHouse`. + + Deterministic and permutation-invariant: rooms are sorted by a canonical key + before allocation, so the same *set* of rooms always yields identical + addresses regardless of input order. + """ + if language not in SUPPORTED_LANGUAGES: + raise RoomLibraryError(f"language must be one of {SUPPORTED_LANGUAGES}") + if not rooms: + raise RoomLibraryError("compose needs at least one room") + + # composition-valid rung: every referenced template must exist & validate. + for spec in rooms: + tid = spec.get("template") + if tid not in templates: + raise RoomLibraryError(f"unknown template '{tid}'. Known: {sorted(templates)}") + v = validate_template(templates[tid]) + if not v["ok"]: + raise RoomLibraryError(f"template '{tid}' is invalid: {v['findings']}") + + ordered = sorted(rooms, key=_canonical_key) + labels = _instance_labels(ordered, templates, language) + if len(set(labels)) != len(labels): + raise RoomLibraryError(f"room labels collide: {labels} — give distinct 'label' overrides") + + # sub counter per (main, middle); sub 0 is the range anchor, GAs start at 1. + sub_counter: dict[tuple[int, int], int] = {} + + def alloc(domain: str, middle_role: str) -> tuple[int, int, int]: + main = DOMAIN_MAIN[domain] + middle = MIDDLE_ROLE[domain][middle_role] + if middle > MAX_MIDDLE: + raise RoomLibraryError( + f"middle group {middle} out of range for domain '{domain}'") + key = (main, middle) + nxt = sub_counter.get(key, 0) + 1 + if nxt > MAX_SUB: + raise RoomLibraryError( + f"address space exhausted in {main}/{middle}/* (>{MAX_SUB} sub-addresses); " + "split the domain across more middle groups or projects — refusing to overflow") + sub_counter[key] = nxt + return main, middle, nxt + + house = ComposedHouse(project_name=project_name, language=language) + for spec, zone_label in zip(ordered, labels): + tmpl = templates[spec["template"]] + room_preset = spec.get("preset", "basic") + slot_presets = spec.get("slot_presets") or {} + params = _room_params(tmpl, spec.get("params") or {}) + room = ResolvedRoom(template_id=spec["template"], label=zone_label, + preset=room_preset, params=params) + + for slot in tmpl.get("slots") or []: + sid = slot["slot_id"] + eff_preset = slot_presets.get(sid, room_preset) + pdef = (slot.get("presets") or {}).get(eff_preset) or {} + if not pdef.get("enabled", False): + continue + ftype = pdef["function"] + n = _resolve_multiplicity(pdef.get("multiplicity") or {}, params) + slot_label = (slot.get("labels") or {}).get(language, sid) + for idx in range(1, n + 1): + inst_label = slot_label if n == 1 else f"{slot_label} {idx}" + rslot = ResolvedSlot(slot_id=sid, label=inst_label, + function_type=ftype, index=idx) + for obj in FUNCTION_OBJECTS[ftype]: + main, middle, sub = alloc(obj.domain, obj.middle_role) + name = f"{zone_label} {inst_label} {obj.label(language)}" + rslot.objects.append(ResolvedObject( + role=obj.role, name=name, + dpt_main=obj.dpt_main, dpt_sub=obj.dpt_sub, + kind=obj.kind, domain=obj.domain, + main=main, middle=middle, sub=sub)) + room.slots.append(rslot) + house.rooms.append(room) + return house + + +# --------------------------------------------------------------------------- # +# .knxproj ZIP writer (ETS project/22). We generate a real archive and re-read +# it with the standard loader; the generator never builds GARecord directly. +# --------------------------------------------------------------------------- # +_XML_ESCAPE = {"&": "&", "<": "<", ">": ">", '"': """} + + +def _esc(s: str) -> str: + return "".join(_XML_ESCAPE.get(c, c) for c in str(s)) + + +def _master_xml() -> str: + """knx_master.xml — copied from the shipped demo project when available, + otherwise a minimal stub (xknxproject only needs MediumType MT-0 resolvable).""" + for cand in ( + os.path.join(os.path.dirname(os.path.abspath(__file__)), + "..", "examples", "demo-home", "demo-home.knxproj"), + ): + cand = os.path.normpath(cand) + if os.path.isfile(cand): + try: + with zipfile.ZipFile(cand) as z: + return z.read("knx_master.xml").decode("utf-8") + except Exception: # noqa: BLE001 + pass + # minimal fallback master + return ('\n' + '\n' + ' \n' + ' \n' + ' \n\n') + + +def build_knxproj_bytes(house: ComposedHouse, proj_id: str = "P-ROOMLIB") -> bytes: + """Serialise a :class:`ComposedHouse` to a real ``.knxproj`` ZIP (bytes).""" + objs = house.objects + # group by main -> middle for GroupRanges + by_main: dict[int, dict[int, list[ResolvedObject]]] = {} + domain_by_main: dict[int, str] = {} + for o in objs: + by_main.setdefault(o.main, {}).setdefault(o.middle, []).append(o) + domain_by_main[o.main] = o.domain + + out: list[str] = [] + w = out.append + w('') + w('') + w(f' ') + w(' ') + w(f' ') + w(' ') + w(f' ') + w(f' ' + f'') + w(' ') + w(' ') + w(' ') + w(' ') + for main in sorted(by_main): + domain = domain_by_main[main] + mname = DOMAIN_LABELS.get(domain, {}).get(house.language, domain) + rs, re_ = main << 11, (main << 11) + 2047 + w(f' ') + for middle in sorted(by_main[main]): + midlabel = MIDDLE_LABELS.get(domain, {}).get(middle, {}).get( + house.language, f"Middle {middle}") + ms, me = (main << 11) | (middle << 8), ((main << 11) | (middle << 8)) + 255 + w(f' ') + for o in sorted(by_main[main][middle], key=lambda x: x.sub): + raw = (o.main << 11) | (o.middle << 8) | o.sub + w(f' ') + w(' ') + w(' ') + w(' ') + w(' ') + w(' ') + w(' ') + w(' ') + w('') + zero_xml = "\n".join(out) + + project_xml = ( + '\n' + '\n' + f' \n' + f' \n' + ' \n' + '\n') + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr(f"{proj_id}/project.xml", project_xml) + z.writestr(f"{proj_id}/0.xml", zero_xml) + z.writestr("knx_master.xml", _master_xml()) + z.writestr(f"{proj_id}.signature", "nickol-knx-mcp room-library synthetic (unsigned)") + return buf.getvalue() + + +# --------------------------------------------------------------------------- # +# Manifest + BOM +# --------------------------------------------------------------------------- # +def build_manifest(house: ComposedHouse) -> dict[str, Any]: + """Deterministic allocation manifest (the negative-oracle surface).""" + rooms_out: list[dict[str, Any]] = [] + total_ga = 0 + for room in house.rooms: + slots_out: list[dict[str, Any]] = [] + for slot in room.slots: + slots_out.append({ + "slot_id": slot.slot_id, + "label": slot.label, + "function": slot.function_type, + "index": slot.index, + "objects": [{"role": o.role, "address": o.address, "name": o.name, + "dpt": f"{o.dpt_main}.{o.dpt_sub:03d}", "kind": o.kind, + "domain": o.domain} for o in slot.objects], + }) + total_ga += len(slot.objects) + rooms_out.append({ + "template": room.template_id, + "label": room.label, + "preset": room.preset, + "params": room.params, + "slots": slots_out, + }) + # per-domain / per-main usage + usage: dict[str, dict[str, Any]] = {} + for o in house.objects: + key = f"{o.main} {o.domain}" + u = usage.setdefault(key, {"main": o.main, "domain": o.domain, "count": 0, + "max_sub_used": 0}) + u["count"] += 1 + u["max_sub_used"] = max(u["max_sub_used"], o.sub) + return { + "schema_version": SCHEMA_VERSION, + "project_name": house.project_name, + "language": house.language, + "taxonomy": {str(v): k for k, v in DOMAIN_MAIN.items()}, + "totals": {"rooms": len(house.rooms), "group_addresses": total_ga}, + "allocation": sorted(usage.values(), key=lambda x: x["main"]), + "rooms": rooms_out, + } + + +def build_bom_proposal(house: ComposedHouse) -> dict[str, Any]: + """Propose candidate devices from the device library for the resolved + functions. A PROPOSAL, not an exact BOM — exact set-cover selection is R2.""" + from .device_library import decompose_device + # count function-type instances (each ResolvedSlot is one channel/zone/unit) + counts: dict[str, int] = {} + for room in house.rooms: + for slot in room.slots: + counts[slot.function_type] = counts.get(slot.function_type, 0) + 1 + + items: list[dict[str, Any]] = [] + for ftype, n in sorted(counts.items()): + recipe = BOM_RECIPE.get(ftype) + dev = decompose_device(recipe, channels=n) if recipe else {"matched": False} + items.append({ + "function": ftype, + "channels_required": n, + "candidate_recipe": recipe, + "unit": dev.get("unit"), + "objects_per_unit": dev.get("objects_per_unit"), + "recipe_note": dev.get("note"), + }) + return { + "note": "Candidate devices only (function-first). Exact device selection with " + "channel/price optimisation is R2; verify channel counts against real " + "product datasheets before ordering.", + "items": items, + } + + +# --------------------------------------------------------------------------- # +# Top-level compose +# --------------------------------------------------------------------------- # +def compose(rooms: list[dict[str, Any]], language: str = "ru", + project_name: str = "Room Library house", + templates: Optional[dict[str, dict[str, Any]]] = None) -> dict[str, Any]: + """Compose a NEW house from room specs and return all artifacts + linter run. + + ``rooms`` items: ``{template, preset?, slot_presets?, params?, label?}``. + Returns a dict with: manifest, ets_xml, ets_csv, bom, lint (round-trip through + the standard loader), and the raw ``.knxproj`` bytes are NOT returned (the + server persists them on request). + """ + from .project import load_project + from .analyze import validate_naming, detect_missing_status, detect_dpt_issues + from .policy import check_policy, load_policy + + templates = templates if templates is not None else load_builtin_templates() + house = resolve_house(rooms, templates, language=language, project_name=project_name) + knxproj = build_knxproj_bytes(house) + + # Round-trip: write to a temp file and re-read with the SAME loader used for + # third-party projects, then lint the re-read model (not the IR). + import tempfile + tf = tempfile.NamedTemporaryFile(suffix=".knxproj", delete=False) + try: + tf.write(knxproj) + tf.close() + loaded = load_project(tf.name) + finally: + os.unlink(tf.name) + + from .generate_ets import generate_ets_xml, generate_ets_csv + naming = validate_naming(loaded) + missing = detect_missing_status(loaded) + dpt = detect_dpt_issues(loaded) + policy = check_policy(loaded, load_policy(None)) + all_findings = naming + missing + dpt + policy.get("findings", []) + errors = sum(1 for f in all_findings if f.get("severity") == "error") + warnings = sum(1 for f in all_findings if f.get("severity") == "warning") + + manifest = build_manifest(house) + return { + "project_name": project_name, + "language": language, + "totals": manifest["totals"], + "lint": { + "reread_group_addresses": len(loaded.gas), + "errors": errors, + "warnings": warnings, + "clean": errors == 0 and warnings == 0, + "naming": naming, + "missing_status": missing, + "dpt": dpt, + "policy": {"summary": policy.get("findings_summary"), + "findings": policy.get("findings")}, + }, + "manifest": manifest, + "bom": build_bom_proposal(house), + "ets_xml": generate_ets_xml(loaded), + "ets_csv": generate_ets_csv(loaded), + "_knxproj_bytes": knxproj, + } diff --git a/nickol_knx_mcp/room_templates/SCHEMA.md b/nickol_knx_mcp/room_templates/SCHEMA.md new file mode 100644 index 0000000..cff8613 --- /dev/null +++ b/nickol_knx_mcp/room_templates/SCHEMA.md @@ -0,0 +1,106 @@ +# Room Template format — public contract (schema_version 1) + +A room template is a **locale-neutral** YAML file describing the *functions* of a +room. `compose_rooms` turns a list of rooms into a new KNX project (group-address +structure + ETS import files + a device BOM proposal). + +> Stability: the schema is a public contract — a breaking change forces user +> migrations. Every template MUST carry `schema_version`. Identity of a template +> and of every slot is a semantic **ID**, never a human name; renaming a label +> never changes identity or the generated addresses. + +## Top-level keys + +| key | required | meaning | +|-----|----------|---------| +| `schema_version` | yes | `1`. Guards compatibility/migration. | +| `slot_id` | yes | Locale-neutral ASCII identifier (`[a-z0-9_]+`). The template's identity. | +| `labels` | yes | `{ru: …, en: …}` — presentation only (RU/EN). | +| `parameters` | yes | Named parameters with a `default` (and provenance). | +| `slots` | yes | List of functional slots (see below). | +| `automation_intents` | no | **Non-executable** metadata declarations only. | + +## Parameters + +Each parameter has a `default`. `area_m2` is special: it is a **hint** that seeds +defaults, not a normative fact — it MUST be declared `role: hint` and carry +`provenance`. In R1 area does not change GA counts; it documents the sizing +assumption only. + +```yaml +parameters: + lighting_circuits: + default: 2 + provenance: {source: preset_rule, note: "why this default"} + area_m2: + default: 28 + role: hint + provenance: {source: preset_default, user_overridden: false, note: "recommender only"} +``` + +## Slots and per-slot presets + +A slot is a functional block. Each slot declares **both** a `basic` and a +`comfort` preset — presets are *per-slot*, not one monolithic room level, so a +house can pick e.g. comfort climate with basic lighting. A preset is either +`{enabled: false}` or `{enabled: true, function: , multiplicity: {...}}`. + +```yaml +slots: + - slot_id: main_light + labels: {ru: основной свет, en: main light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} +``` + +`multiplicity` is `{fixed: N}` or `{param: }` — how many +instances (circuits, windows, zones) the slot expands to. + +### Function types (function-first) + +Each function type expands into a fixed set of KNX communication objects, each +with a canonical DPT and a command/status role: + +| function | objects (role · DPT) | +|----------|----------------------| +| `lighting_switch` | on/off `1.001` · status `1.011` | +| `lighting_dimmer` | on/off `1.001` · dimming `3.007` · brightness `5.001` · status `1.011` · brightness-status `5.001` | +| `shutter` | up/down `1.008` · stop `1.010` · position `5.001` · position-status `5.001` | +| `climate_floor` | on/off `1.001` · setpoint `9.001` · mode `20.102` · +3 statuses · actual-temp `9.001` | +| `presence` | occupancy `1.018` · illuminance `9.004` | + +Every controllable command gets its status object — the generated project passes +`check_missing_status`, `check_dpt`, `check_naming` and `check_policy` cleanly. + +## Address allocation (default taxonomy) + +`main` = function domain, `middle` = role/sub-function, `sub` = sequential: + +``` +0 Central · 1 Lighting · 2 Shutters · 3 HVAC · 4 Sensors · 5 Energy · 6 Diag · 7 Reserve +``` + +Allocation is deterministic and **permutation-invariant** (rooms are sorted by a +canonical key before allocation), so the same *set* of rooms always yields the +same addresses regardless of input order. Sub-address exhaustion (> 255 in a +middle group) raises a hard error — never a silent overflow. + +## automation_intents (declarations only) + +Logic (presence→light, wind→shutter) is **not** executed by the templates. It is +declared as metadata for the upper layer (Home Assistant / the KNX program): + +```yaml +automation_intents: + - intent: presence_lights_off + description: "Turn lights off when unoccupied." + criticality: convenience # or safety_related + implementation: external +``` + +## Not in R1 + +Docking into an existing project (allocation lockfile, drift detection), exact +device selection with channel/price optimisation, and premium presets are R2 — +see `docs/roadmap/room-library/implementation-plan.md`. diff --git a/nickol_knx_mcp/room_templates/bathroom.yaml b/nickol_knx_mcp/room_templates/bathroom.yaml new file mode 100644 index 0000000..4b8886e --- /dev/null +++ b/nickol_knx_mcp/room_templates/bathroom.yaml @@ -0,0 +1,49 @@ +# Room Template — Bathroom / WC (Санузел) +schema_version: 1 +slot_id: bathroom +labels: + ru: Санузел + en: Bathroom + +parameters: + lighting_circuits: + default: 1 + provenance: {source: preset_rule, note: "1 ceiling circuit"} + windows: + default: 0 + provenance: {source: preset_rule, note: "many bathrooms are windowless"} + area_m2: + default: 6 + role: hint + provenance: {source: preset_default, user_overridden: false} + +slots: + - slot_id: main_light + labels: {ru: свет, en: light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + + - slot_id: mirror_light + labels: {ru: свет зеркала, en: mirror light} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: lighting_switch, multiplicity: {fixed: 1}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: floor_heating + labels: {ru: тёплый пол, en: floor heating} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: climate_floor, multiplicity: {fixed: 1}} + +automation_intents: + - intent: humidity_ventilation + description: "Run ventilation above a humidity threshold." + criticality: convenience + implementation: external diff --git a/nickol_knx_mcp/room_templates/bedroom.yaml b/nickol_knx_mcp/room_templates/bedroom.yaml new file mode 100644 index 0000000..5942ca3 --- /dev/null +++ b/nickol_knx_mcp/room_templates/bedroom.yaml @@ -0,0 +1,63 @@ +# Room Template — Bedroom (Спальня) +# Public contract: see SCHEMA.md. Identity is the semantic slot_id, NOT the name. +schema_version: 1 +slot_id: bedroom +labels: + ru: Спальня + en: Bedroom + +# Parameters seed defaults. area_m2 is a HINT (recommender), never normative: +# in R1 it does not change GA counts, it only documents the sizing assumption. +parameters: + lighting_circuits: + default: 1 + provenance: {source: preset_rule, note: "1 dimmable ceiling circuit for a standard bedroom"} + windows: + default: 2 + provenance: {source: preset_rule, note: "typical master bedroom has 2 windows"} + area_m2: + default: 14 + role: hint + provenance: {source: preset_default, user_overridden: false, + note: "typical; seeds defaults only, not a normative fact"} + +slots: + # Main ceiling light: a plain switch in basic, a dimmer in comfort (per-slot + # preset, so a house can pick comfort climate but basic light). + - slot_id: main_light + labels: {ru: основной свет, en: main light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} + + - slot_id: bedside_light + labels: {ru: прикроватный свет, en: bedside light} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: lighting_switch, multiplicity: {fixed: 2}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: presence + labels: {ru: датчик, en: sensor} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: presence, multiplicity: {fixed: 1}} + + - slot_id: floor_heating + labels: {ru: тёплый пол, en: floor heating} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: climate_floor, multiplicity: {fixed: 1}} + +# Non-executable declarations only (metadata). The engine does NOT run these — +# safety-related logic stays external (Home Assistant / the KNX program). +automation_intents: + - intent: presence_lights_off + description: "Turn bedroom lights off when unoccupied." + criticality: convenience + implementation: external diff --git a/nickol_knx_mcp/room_templates/children.yaml b/nickol_knx_mcp/room_templates/children.yaml new file mode 100644 index 0000000..396778a --- /dev/null +++ b/nickol_knx_mcp/room_templates/children.yaml @@ -0,0 +1,55 @@ +# Room Template — Children's room (Детская) +schema_version: 1 +slot_id: children +labels: + ru: Детская + en: Children + +parameters: + lighting_circuits: + default: 1 + provenance: {source: preset_rule, note: "1 ceiling circuit"} + windows: + default: 1 + provenance: {source: preset_rule} + area_m2: + default: 12 + role: hint + provenance: {source: preset_default, user_overridden: false} + +slots: + - slot_id: main_light + labels: {ru: основной свет, en: main light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} + + - slot_id: desk_light + labels: {ru: свет стола, en: desk light} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: lighting_switch, multiplicity: {fixed: 1}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: presence + labels: {ru: датчик, en: sensor} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: presence, multiplicity: {fixed: 1}} + + - slot_id: floor_heating + labels: {ru: тёплый пол, en: floor heating} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: climate_floor, multiplicity: {fixed: 1}} + +automation_intents: + - intent: presence_lights_off + description: "Turn children's-room lights off when unoccupied." + criticality: convenience + implementation: external diff --git a/nickol_knx_mcp/room_templates/corridor.yaml b/nickol_knx_mcp/room_templates/corridor.yaml new file mode 100644 index 0000000..21e0567 --- /dev/null +++ b/nickol_knx_mcp/room_templates/corridor.yaml @@ -0,0 +1,43 @@ +# Room Template — Corridor / Hallway (Коридор) +schema_version: 1 +slot_id: corridor +labels: + ru: Коридор + en: Corridor + +parameters: + lighting_circuits: + default: 1 + provenance: {source: preset_rule, note: "1 ceiling circuit"} + windows: + default: 0 + provenance: {source: preset_rule} + area_m2: + default: 5 + role: hint + provenance: {source: preset_default, user_overridden: false} + +slots: + - slot_id: main_light + labels: {ru: свет, en: light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: presence + labels: {ru: датчик, en: sensor} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: presence, multiplicity: {fixed: 1}} + +automation_intents: + - intent: presence_lights + description: "Switch corridor lights on presence, off after a timeout." + criticality: convenience + implementation: external diff --git a/nickol_knx_mcp/room_templates/kitchen.yaml b/nickol_knx_mcp/room_templates/kitchen.yaml new file mode 100644 index 0000000..a2027e5 --- /dev/null +++ b/nickol_knx_mcp/room_templates/kitchen.yaml @@ -0,0 +1,49 @@ +# Room Template — Kitchen (Кухня) +schema_version: 1 +slot_id: kitchen +labels: + ru: Кухня + en: Kitchen + +parameters: + lighting_circuits: + default: 2 + provenance: {source: preset_rule, note: "ceiling + island/general — 2 circuits"} + windows: + default: 1 + provenance: {source: preset_rule} + area_m2: + default: 12 + role: hint + provenance: {source: preset_default, user_overridden: false} + +slots: + - slot_id: main_light + labels: {ru: свет, en: light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} + + - slot_id: worktop_light + labels: {ru: свет столешницы, en: worktop light} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: lighting_switch, multiplicity: {fixed: 1}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: presence + labels: {ru: датчик, en: sensor} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: presence, multiplicity: {fixed: 1}} + +automation_intents: + - intent: presence_lights_off + description: "Turn kitchen lights off when unoccupied." + criticality: convenience + implementation: external diff --git a/nickol_knx_mcp/room_templates/living.yaml b/nickol_knx_mcp/room_templates/living.yaml new file mode 100644 index 0000000..df9266a --- /dev/null +++ b/nickol_knx_mcp/room_templates/living.yaml @@ -0,0 +1,56 @@ +# Room Template — Living room (Гостиная) +schema_version: 1 +slot_id: living +labels: + ru: Гостиная + en: Living room + +parameters: + lighting_circuits: + default: 3 + provenance: {source: preset_rule, note: "chandelier + spots + accent — 3 circuits"} + windows: + default: 3 + provenance: {source: preset_rule} + area_m2: + default: 28 + role: hint + provenance: {source: preset_default, user_overridden: false, + note: ">18 m² typically justifies multiple circuits — recommender only"} + +slots: + - slot_id: main_light + labels: {ru: свет, en: light} + presets: + basic: {enabled: true, function: lighting_switch, multiplicity: {param: lighting_circuits}} + comfort: {enabled: true, function: lighting_dimmer, multiplicity: {param: lighting_circuits}} + + - slot_id: accent_light + labels: {ru: акцентный свет, en: accent light} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: lighting_switch, multiplicity: {fixed: 2}} + + - slot_id: window + labels: {ru: окно, en: window} + presets: + basic: {enabled: true, function: shutter, multiplicity: {param: windows}} + comfort: {enabled: true, function: shutter, multiplicity: {param: windows}} + + - slot_id: presence + labels: {ru: датчик, en: sensor} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: presence, multiplicity: {fixed: 1}} + + - slot_id: floor_heating + labels: {ru: тёплый пол, en: floor heating} + presets: + basic: {enabled: false} + comfort: {enabled: true, function: climate_floor, multiplicity: {fixed: 1}} + +automation_intents: + - intent: scene_lighting + description: "Recall living-room lighting scenes (movie/evening/bright)." + criticality: convenience + implementation: external