diff --git a/templates/resin/dashboard/extractor.py b/templates/resin/dashboard/extractor.py new file mode 100644 index 0000000..9ecdbe9 --- /dev/null +++ b/templates/resin/dashboard/extractor.py @@ -0,0 +1,484 @@ +# -*- coding: utf-8 -*- +"""树脂驾驶舱布局提取器 —— issue #84（父 EPIC #13「树脂模板封装」子任务）。 + +把已交付的树脂驾驶舱布局资产（`dashboard/cockpit.resin.yaml`）**抽取为可复用 +模板**，并提供「点位字典回填」能力，验证内核复制不同行业时的零改动假设 +（对齐 PRD 5.5「⑤ 配置化驾驶舱」：换行业只替换驾驶舱布局资产，前端代码不动）。 + +核心抽象：**布局资产 ↔ 布局模板 ↔ 点位字典** 三者解耦—— + +- 具体布局资产（`cockpit.resin.yaml`）含**硬编码**的点位绑定 + （如 `bind: R-801.TEMP`）与 KPI 指标名（如 `metric: resin_exchange_capacity`）； +- 布局模板把这些硬编码值**变量化**为占位符（`{device}.{point}` / `{metric}`）， + 仅保留布局骨架（$schema / title / theme / widgets 几何位置 / widget 类型）； +- 实例化时用**目标行业的点位字典**回填占位符，得到该行业的具体驾驶舱布局。 + +对外 API： + +- `extract_template(layout_yaml) -> dict` + 从具体布局资产抽取通用模板（变量化硬编码点位/指标）。 +- `instantiate(template, point_dict) -> dict` + 用点位字典回填占位符，生成具体布局（`bind` 用 `device.point` 填回）。 +- `diff_layouts(layout_a, layout_b) -> LayoutDiff` + 布局差异比对（schema/title/theme/widgets 几何与绑定差异）。 +- `PointDict.from_csv(path)` / `PointDict.lookup(device, point)` + 点位字典加载与查询（CSV 9 列，对齐内核 schema）。 + +纯标准库实现（无 yaml 依赖）：YAML 子集用 `_parse_yaml_subset` 解析； +与 `templates/ti-cl4/llm-scenarios/` 同款「dataclass + Enum + 类型注解 + +中文 docstring 引 PRD/EPIC」范式。 +""" +from __future__ import annotations + +import csv +import os +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Tuple, Union + +# ---------------------------------------------------------------------- +# 资产路径 +# ---------------------------------------------------------------------- +#: 模板根目录（templates/resin/） +TEMPLATE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +#: 驾驶舱布局资产（issue #84 的提取对象） +DEFAULT_LAYOUT_PATH = os.path.join( + TEMPLATE_ROOT, "dashboard", "cockpit.resin.yaml") +#: 点位字典资产（issue #82，回填占位符用） +DEFAULT_POINT_DICT_PATH = os.path.join( + TEMPLATE_ROOT, "point-dict", "point_dict.resin.csv") + +#: PRD 5.5 iAOP-cockpit-layout-v1 允许的 widget 类型 +ALLOWED_WIDGETS = { + "process_view", "trend", "kpi_card", "alarm_panel", "nl_query", +} + +#: 驾驶舱布局的 $schema（绑定布局模板版本） +COCKPIT_SCHEMA = "iAOP-cockpit-layout-v1" + +#: 占位符：设备位号 / 测点 / KPI 指标（变量化布局） +DEVICE_PLACEHOLDER = "{device}" +POINT_PLACEHOLDER = "{point}" +METRIC_PLACEHOLDER = "{metric}" +#: trend widget 的 bind 变量化形态（设备.测点） +BIND_TEMPLATE = f"{DEVICE_PLACEHOLDER}.{POINT_PLACEHOLDER}" + +#: 识别「设备.测点」绑定（如 `R-801.TEMP`），用于变量化与回填 +_BIND_RE = re.compile(r"^([A-Za-z0-9_\-]+)\.([A-Za-z0-9_\-]+)$") + + +# ---------------------------------------------------------------------- +# Widget 类型（枚举） +# ---------------------------------------------------------------------- +class WidgetType(str, Enum): + """PRD 5.5 允许的 widget 类型（str 继承便于直接作 YAML 值）。""" + + PROCESS_VIEW = "process_view" + TREND = "trend" + KPI_CARD = "kpi_card" + ALARM_PANEL = "alarm_panel" + NL_QUERY = "nl_query" + + @classmethod + def is_allowed(cls, value: str) -> bool: + """widget 类型是否在 PRD 5.5 schema 允许集合内。""" + return any(member.value == value for member in cls) + + +# ---------------------------------------------------------------------- +# 数据类 +# ---------------------------------------------------------------------- +@dataclass +class PointEntry: + """点位字典中的单条点位（CSV 9 列，对齐内核 schema）。""" + + device_id: str + point_id: str + name: str + unit: str + data_type: str + sample_rate: str + quality_code: str + opc_node: str + protocol: str + + @property + def point(self) -> str: + """短测点名：`R-801.TEMP` → `TEMP`（剥离设备前缀）。""" + prefix = f"{self.device_id}." + if self.point_id.startswith(prefix): + return self.point_id[len(prefix):] + return self.point_id + + +@dataclass +class PointDict: + """点位字典（按 device_id 索引点位集合）。 + + 用于 `instantiate` 把布局模板里的 `{device}.{point}` 占位符回填为 + 目标行业的具体点位绑定。 + """ + + entries: List[PointEntry] = field(default_factory=list) + + @classmethod + def from_csv(cls, path: str = DEFAULT_POINT_DICT_PATH) -> "PointDict": + """从 CSV 加载点位字典（9 列，对齐 point_dict.resin.csv schema）。""" + entries: List[PointEntry] = [] + with open(path, "r", encoding="utf-8") as fh: + reader = csv.reader(fh) + header = next(reader, None) + if header is None: + raise ValueError(f"点位字典 CSV 为空：{path}") + for row in reader: + if len(row) != 9: + raise ValueError( + f"点位字典 CSV 行列数异常（需 9 列）：{row}") + entries.append(PointEntry(*row)) + return cls(entries=entries) + + def devices(self) -> List[str]: + """去重保序返回所有设备位号。""" + seen: List[str] = [] + for e in self.entries: + if e.device_id not in seen: + seen.append(e.device_id) + return seen + + def lookup(self, device: str, point: Optional[str] = None) -> List[PointEntry]: + """查询某设备（可选过滤测点名）的点位。""" + out: List[PointEntry] = [] + for e in self.entries: + if e.device_id == device and (point is None or e.point == point): + out.append(e) + return out + + def first_device(self) -> str: + """首个设备位号（单设备趋势绑定回填的默认设备）。""" + return self.entries[0].device_id + + +@dataclass +class LayoutDiff: + """两份布局的差异（`diff_layouts` 返回）。""" + + schema_changed: bool = False + title_changed: bool = False + theme_changed: bool = False + widget_count_a: int = 0 + widget_count_b: int = 0 + geometry_changes: List[str] = field(default_factory=list) + binding_changes: List[str] = field(default_factory=list) + + @property + def is_identical(self) -> bool: + """两份布局是否完全一致。""" + return (not self.schema_changed and not self.title_changed + and not self.theme_changed + and not self.geometry_changes and not self.binding_changes + and self.widget_count_a == self.widget_count_b) + + def summary(self) -> List[str]: + """人读的差异摘要（可解释性：列每条差异的含义/原因）。""" + items: List[str] = [] + if self.schema_changed: + items.append("$schema 不一致（布局版本不同，可能不兼容）") + if self.title_changed: + items.append("title 不一致（驾驶舱标题文案不同）") + if self.theme_changed: + items.append("theme 不一致（明暗主题不同）") + if self.widget_count_a != self.widget_count_b: + items.append( + f"widget 数量不同：{self.widget_count_a} → {self.widget_count_b}") + for g in self.geometry_changes: + items.append(f"几何变更：{g}") + for b in self.binding_changes: + items.append(f"绑定变更：{b}") + if not items: + items.append("布局完全一致") + return items + + +# ---------------------------------------------------------------------- +# YAML 子集解析器（零依赖，复制自 ti-cl4 范式） +# ---------------------------------------------------------------------- +def _parse_value(token: str) -> Union[str, int, float, bool, None]: + """把标量 token 解析为 Python 类型（int/float/bool/None/str）。""" + token = token.strip() + if token == "": + return "" + if token.lower() == "true": + return True + if token.lower() == "false": + return False + if token.lower() in ("null", "none", "~"): + return None + # int + if re.fullmatch(r"-?\d+", token): + return int(token) + # float + if re.fullmatch(r"-?\d+\.\d+", token): + return float(token) + # 去引号 + if len(token) >= 2 and token[0] in "\"'" and token[-1] == token[0]: + return token[1:-1] + return token + + +def _is_bare_scalar(item_text: str) -> bool: + """列表项是否为裸标量（非 `key: value` 形式）。 + + 注意：`R-801.TEMP` 这类含冒号但被引号包裹或无分隔冒号的情况需判别。 + 本驾驶舱 YAML 中列表项的 mapping 形式形如 `type: trend`，分隔符 `: ` + 后有值；裸标量（无冒号）按标量处理。 + """ + # 含 ": " 或以 ":" 结尾 → 视作 mapping 起点 + return ": " not in item_text and not item_text.endswith(":") + + +def _parse_yaml_subset(text: str) -> dict: + """零依赖 YAML 子集解析器。 + + 仅支持本模块所读的驾驶舱布局资产用到的 YAML 语法： + + - 顶层 mapping（`key: value`）； + - 列表（`- item`），列表元素可为 mapping（多行 `key: value`）； + - 缩进表示嵌套（2 空格缩进）； + - 行内值支持 int/float/bool/None/带引号字符串/裸字符串； + - `#` 注释行、空行跳过； + - `key:`（空值）后跟 `- ` 列表项 → 该 key 的值解析为 list。 + + 复杂特性（多行字符串 / 流式语法 / 锚点）按需再加，本模板用不到。 + """ + root: dict = {} + # 栈元素：(indent, container, container_owner, owner_key) + # container_owner/owner_key 用于：当 container 是「按需创建的 list」时 + # 能回写到父 dict（解决 `widgets:` 后跟列表项的场景）。 + stack: List[Tuple[int, Union[dict, list], Optional[dict], Optional[str]]] = [ + (0, root, None, None)] + + def _indent(line: str) -> int: + return len(line) - len(line.lstrip(" ")) + + for raw in text.splitlines(): + stripped = raw.rstrip() + if not stripped.strip(): + continue + if stripped.lstrip().startswith("#"): + continue + indent = _indent(stripped) + content = stripped[indent:] + + # 弹栈到当前缩进的父级（>= 而非 >，保留同缩进兄弟） + while len(stack) > 1 and stack[-1][0] > indent: + stack.pop() + + if content.startswith("- "): + # 列表项 + item_text = content[2:].strip() + top_indent, container, owner, key = stack[-1] + # 若当前容器是「待创建 list」（dict 里某 key 值还指向 dict 自身的占位） + if isinstance(container, dict): + raise ValueError(f"列表项出现在非列表上下文：{raw!r}") + if ":" in item_text and not _is_bare_scalar(item_text): + # 列表项是 mapping 起点（如 `- type: trend`） + k, _, v = item_text.partition(":") + child: dict = {k.strip(): _parse_value(v) if v.strip() else None} + container.append(child) + stack.append((indent + 2, child, None, None)) + else: + container.append(_parse_value(item_text)) + else: + # mapping 项 + key, sep, val = content.partition(":") + if not sep: + continue + key = key.strip() + val = val.strip() + top_indent, container, owner, owner_key = stack[-1] + if not isinstance(container, dict): + raise ValueError(f"键值项出现在非 mapping 上下文：{raw!r}") + if val == "": + # 子结构占位：先建空 list（驾驶舱 widgets 后跟列表项）， + # 入栈供后续 `- ` 项填充；若实际跟的是 mapping，append 时会 + # 因类型不符报错（本资产不会发生）。 + child_list: list = [] + container[key] = child_list + stack.append((indent + 1, child_list, container, key)) + else: + container[key] = _parse_value(val) + + return root + + +# ---------------------------------------------------------------------- +# 布局加载 +# ---------------------------------------------------------------------- +def load_layout(path: str = DEFAULT_LAYOUT_PATH) -> dict: + """加载驾驶舱布局资产（零依赖 YAML 子集解析）。""" + with open(path, "r", encoding="utf-8") as fh: + text = fh.read() + return _parse_yaml_subset(text) + + +# ---------------------------------------------------------------------- +# 提取 / 实例化 / 比对 +# ---------------------------------------------------------------------- +def _is_device_point_binding(value: str) -> Optional[Tuple[str, str]]: + """识别 `设备.测点` 形态的绑定，返回 (device, point) 或 None。""" + m = _BIND_RE.match(value or "") + if m: + return m.group(1), m.group(2) + return None + + +def extract_template(layout_yaml: Union[str, dict]) -> dict: + """从具体布局资产抽取通用模板。 + + 把硬编码的点位绑定（`R-801.TEMP`）与 KPI 指标名 + （`resin_exchange_capacity`）**变量化**为占位符，保留布局骨架 + （$schema / title / theme / widgets 几何与类型 / description）。 + + Args: + layout_yaml: 具体布局资产路径、或已解析 dict、或 YAML 文本。 + + Returns: + 模板 dict，widget 的 `bind`/`metric` 被替换为占位符；并附带 + `placeholder_binding` 字段记录每个占位符的语义（可解释性）。 + """ + if isinstance(layout_yaml, str): + if os.path.isfile(layout_yaml): + layout = load_layout(layout_yaml) + else: + layout = _parse_yaml_subset(layout_yaml) + elif isinstance(layout_yaml, dict): + layout = layout_yaml + else: + raise TypeError(f"layout_yaml 类型不支持：{type(layout_yaml)}") + + template: dict = { + "$schema": layout.get("$schema", COCKPIT_SCHEMA), + "title": layout.get("title", ""), + "theme": layout.get("theme", "dark"), + "widgets": [], + # 占位符语义说明（提取记录，便于实例化时回填与可解释性） + "placeholder_binding": { + DEVICE_PLACEHOLDER: "设备位号（点位字典 device_id）", + POINT_PLACEHOLDER: "测点名（点位字典 point_id 去设备前缀）", + METRIC_PLACEHOLDER: "KPI 指标名（业务语义键）", + }, + } + for w in layout.get("widgets", []): + wt = dict(w) # 浅拷贝 + # 1) trend widget 的 bind：R-801.TEMP → {device}.{point} + if wt.get("type") == WidgetType.TREND.value and "bind" in wt: + dp = _is_device_point_binding(str(wt["bind"])) + if dp: + wt["bind"] = BIND_TEMPLATE + wt.setdefault("extracted_from", {})["bind"] = list(dp) + # 2) kpi_card 的 metric：变量化为 {metric}（保留 label） + if wt.get("type") == WidgetType.KPI_CARD.value and "metric" in wt: + wt.setdefault("extracted_from", {})["metric"] = wt["metric"] + wt["metric"] = METRIC_PLACEHOLDER + template["widgets"].append(wt) + return template + + +def instantiate(template: dict, point_dict: PointDict) -> dict: + """用点位字典把模板占位符回填为具体布局。 + + 回填策略（对齐 PRD 5.5 布局语义）： + + - trend widget：`{device}.{point}` 占位符 → 按模板中记录的 + `extracted_from.bind` 设备/测点回填；若模板已抽象到只剩占位符 + 且未记录来源，则用点位字典首个设备 + 同名测点回填。 + - kpi_card：`{metric}` → 回填为模板 `extracted_from.metric` 原值 + （指标名属业务语义键，不随设备变，保持原值最安全）。 + + Args: + template: `extract_template` 的输出。 + point_dict: 目标行业点位字典（`PointDict.from_csv`）。 + + Returns: + 具体布局 dict（去除 `extracted_from` / `placeholder_binding` + 内部字段，符合 iAOP-cockpit-layout-v1 schema）。 + """ + if not isinstance(template, dict): + raise TypeError(f"template 必须是 dict：{type(template)}") + if not point_dict.entries: + raise ValueError("点位字典为空，无法回填") + + devices = point_dict.devices() + out: dict = { + "$schema": template.get("$schema", COCKPIT_SCHEMA), + "title": template.get("title", ""), + "theme": template.get("theme", "dark"), + "widgets": [], + } + for w in template.get("widgets", []): + cw = dict(w) + src = cw.pop("extracted_from", None) or {} + if cw.get("type") == WidgetType.TREND.value and "bind" in cw: + bind = str(cw["bind"]) + if DEVICE_PLACEHOLDER in bind or POINT_PLACEHOLDER in bind: + # 优先用提取时记录的来源设备/测点 + device = src.get("bind", [None, None])[0] if src.get("bind") else None + point = src.get("bind", [None, None])[1] if len(src.get("bind", [])) > 1 else None + # 用点位字典校验/覆盖：若设备在字典中则用之，否则回退首个设备 + if not device or device not in devices: + device = devices[0] + if not point: + cands = point_dict.lookup(device) + point = cands[0].point if cands else "TEMP" + cw["bind"] = f"{device}.{point}" + if cw.get("type") == WidgetType.KPI_CARD.value and "metric" in cw: + if str(cw["metric"]) == METRIC_PLACEHOLDER: + # metric 是业务语义键，保留模板提取时的原值 + cw["metric"] = src.get("metric", "kpi_metric") + out["widgets"].append(cw) + return out + + +def diff_layouts(layout_a: dict, layout_b: dict) -> LayoutDiff: + """比对两份具体布局（或模板）的差异。 + + 返回 `LayoutDiff`：含 schema/title/theme/widgets 几何位置与 + bind/metric 绑定的逐条差异（用于换行业时回归比对，定位漂移）。 + """ + diff = LayoutDiff( + schema_changed=layout_a.get("$schema") != layout_b.get("$schema"), + title_changed=layout_a.get("title") != layout_b.get("title"), + theme_changed=layout_a.get("theme") != layout_b.get("theme"), + ) + wa = layout_a.get("widgets", []) or [] + wb = layout_b.get("widgets", []) or [] + diff.widget_count_a = len(wa) + diff.widget_count_b = len(wb) + + # 按位置（索引）对齐 widgets，比对几何与绑定 + n = max(len(wa), len(wb)) + for i in range(n): + a = wa[i] if i < len(wa) else None + b = wb[i] if i < len(wb) else None + if a is None or b is None: + diff.geometry_changes.append( + f"widget[{i}] {'缺失' if a is None else '新增'}") + continue + # 几何（x/y/w/h） + for k in ("x", "y", "w", "h"): + av, bv = a.get(k), b.get(k) + if av != bv: + diff.geometry_changes.append( + f"widget[{i}].{k}: {av} → {bv}") + # 类型 + if a.get("type") != b.get("type"): + diff.geometry_changes.append( + f"widget[{i}].type: {a.get('type')} → {b.get('type')}") + # 绑定（bind / metric） + for k in ("bind", "metric"): + av, bv = a.get(k), b.get(k) + if av != bv: + diff.binding_changes.append( + f"widget[{i}].{k}: {av!r} → {bv!r}") + return diff diff --git a/templates/resin/tests/test_extractor.py b/templates/resin/tests/test_extractor.py new file mode 100644 index 0000000..ab6c3ec --- /dev/null +++ b/templates/resin/tests/test_extractor.py @@ -0,0 +1,278 @@ +# -*- coding: utf-8 -*- +"""树脂驾驶舱布局提取器测试（issue #84）。 + +覆盖（>=8 用例）： +1. load_layout：驾驶舱布局资产解析（schema/title/theme/widgets 9 个）； +2. extract_template：trend bind 变量化（R-801.TEMP → {device}.{point}）； +3. extract_template：kpi metric 变量化（→ {metric}）； +4. extract_template：占位符语义（placeholder_binding）+ 提取来源记录； +5. PointDict.from_csv：21 条点位、device 去重； +6. PointDict.lookup：按设备/测点查询； +7. instantiate：用点位字典回填 bind（具体点位）； +8. instantiate：metric 回填为模板原指标名（业务语义键不变）； +9. instantiate：去除内部字段（extracted_from/placeholder_binding）； +10. diff_layouts：一致布局 → is_identical； +11. diff_layouts：bind 变更 → binding_changes； +12. diff_layouts：几何变更 → geometry_changes； +13. WidgetType.is_allowed / COCKPIT_SCHEMA 常量。 +14. 端到端：extract → instantiate → 与原布局一致（回填幂等性）。 +""" +import copy +import os +import sys +import unittest + +# 把 templates/resin/dashboard 挂到 sys.path 以便 `from extractor import ...` +_RESIN_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_DASHBOARD_DIR = os.path.join(_RESIN_DIR, "dashboard") +sys.path.insert(0, _DASHBOARD_DIR) + +from extractor import ( # noqa: E402 + BIND_TEMPLATE, + COCKPIT_SCHEMA, + DEVICE_PLACEHOLDER, + METRIC_PLACEHOLDER, + POINT_PLACEHOLDER, + DEFAULT_LAYOUT_PATH, + DEFAULT_POINT_DICT_PATH, + ALLOWED_WIDGETS, + LayoutDiff, + PointDict, + PointEntry, + WidgetType, + diff_layouts, + extract_template, + instantiate, + load_layout, +) + + +class TestYAMLSubsetParse(unittest.TestCase): + """零依赖 YAML 子集解析（驾驶舱布局资产）。""" + + def test_load_layout_basic(self): + """#1 驾驶舱资产解析：schema/title/theme/widgets 9 个。""" + layout = load_layout() + self.assertEqual(layout["$schema"], COCKPIT_SCHEMA) + self.assertEqual(layout["title"], "吸附树脂车间驾驶舱") + self.assertEqual(layout["theme"], "dark") + self.assertEqual(len(layout["widgets"]), 9) + + def test_widget_types_all_allowed(self): + """#13 所有 widget 类型在 PRD 5.5 schema 允许集合内。""" + layout = load_layout() + for w in layout["widgets"]: + self.assertIn(w["type"], ALLOWED_WIDGETS) + self.assertTrue(WidgetType.is_allowed(w["type"])) + + def test_parse_inline_scalar_types(self): + """YAML 行内标量解析：int/float/bool/None/带引号字符串。""" + from extractor import _parse_yaml_subset, _parse_value + self.assertEqual(_parse_value("12"), 12) + self.assertEqual(_parse_value("4.5"), 4.5) + self.assertIs(_parse_value("true"), True) + self.assertIs(_parse_value("false"), False) + self.assertIsNone(_parse_value("null")) + self.assertEqual(_parse_value("'x'"), "x") + # 缩进嵌套 + 列表项 mapping + doc = _parse_yaml_subset( + "title: T\nwidgets:\n - type: trend\n bind: R1.X\n x: 1\n") + self.assertEqual(doc["title"], "T") + self.assertEqual(doc["widgets"][0]["type"], "trend") + self.assertEqual(doc["widgets"][0]["bind"], "R1.X") + self.assertEqual(doc["widgets"][0]["x"], 1) + + +class TestExtractTemplate(unittest.TestCase): + """布局资产 → 通用模板（变量化）。""" + + def setUp(self): + self.layout = load_layout() + self.tmpl = extract_template(self.layout) + + def test_trend_bind_placeholderized(self): + """#2 trend bind：R-801.TEMP → {device}.{point}。""" + trend_binds = [w.get("bind") for w in self.tmpl["widgets"] + if w.get("type") == "trend"] + self.assertTrue(trend_binds) + for b in trend_binds: + self.assertEqual(b, BIND_TEMPLATE) + + def test_kpi_metric_placeholderized(self): + """#3 kpi metric：resin_exchange_capacity → {metric}。""" + kpi_metrics = [w.get("metric") for w in self.tmpl["widgets"] + if w.get("type") == "kpi_card"] + self.assertEqual(len(kpi_metrics), 4) + for m in kpi_metrics: + self.assertEqual(m, METRIC_PLACEHOLDER) + + def test_placeholder_semantics_and_source(self): + """#4 占位符语义（placeholder_binding）+ 提取来源记录。""" + # 占位符语义说明 + pb = self.tmpl["placeholder_binding"] + self.assertIn(DEVICE_PLACEHOLDER, pb) + self.assertIn(POINT_PLACEHOLDER, pb) + self.assertIn(METRIC_PLACEHOLDER, pb) + # trend widget 记录提取来源（device/point 二元组） + trend = next(w for w in self.tmpl["widgets"] + if w.get("type") == "trend") + self.assertEqual(trend["extracted_from"]["bind"], ["R-801", "TEMP"]) + # kpi widget 记录原指标名 + kpi = next(w for w in self.tmpl["widgets"] + if w.get("type") == "kpi_card") + self.assertEqual(kpi["extracted_from"]["metric"], + "resin_exchange_capacity") + + def test_extract_from_yaml_text_and_path(self): + """extract_template 接受路径 / dict / YAML 文本三种入参。""" + tmpl_from_path = extract_template(DEFAULT_LAYOUT_PATH) + self.assertEqual(tmpl_from_path["$schema"], COCKPIT_SCHEMA) + tmpl_from_dict = extract_template(self.layout) + self.assertEqual(tmpl_from_dict["title"], self.tmpl["title"]) + + +class TestPointDict(unittest.TestCase): + """点位字典加载与查询。""" + + def setUp(self): + self.pd = PointDict.from_csv() + + def test_csv_load_count_and_devices(self): + """#5 21 条点位 + device 去重保序。""" + self.assertEqual(len(self.pd.entries), 21) + self.assertEqual(self.pd.entries[0].device_id, "R-801") + self.assertEqual(self.pd.devices()[0], "R-801") + # device 去重（R-801 出现 6 次，去重后 1） + self.assertEqual(self.pd.devices().count("R-801"), 1) + + def test_lookup_and_point_short_name(self): + """#6 查询某设备测点 + point 短名（去设备前缀）。""" + r801 = self.pd.lookup("R-801") + self.assertTrue(r801) + # 短测点名剥离设备前缀 + temps = [e.point for e in r801 if e.point == "TEMP"] + self.assertEqual(temps, ["TEMP"]) + # 精确测点过滤 + temp_only = self.pd.lookup("R-801", "TEMP") + self.assertEqual(len(temp_only), 1) + self.assertEqual(temp_only[0].unit, "℃") + # 未知设备 → 空列表 + self.assertEqual(self.pd.lookup("NOPE"), []) + + +class TestInstantiate(unittest.TestCase): + """模板 + 点位字典 → 具体布局（回填占位符）。""" + + def setUp(self): + self.pd = PointDict.from_csv() + self.tmpl = extract_template(load_layout()) + self.concrete = instantiate(self.tmpl, self.pd) + + def test_backfill_trend_bind(self): + """#7 trend bind 回填为具体点位（R-801.TEMP）。""" + binds = [w.get("bind") for w in self.concrete["widgets"] + if w.get("type") == "trend"] + self.assertEqual(binds, ["R-801.TEMP", "R-801.AGIT"]) + # 不残留占位符 + for b in binds: + self.assertNotIn(DEVICE_PLACEHOLDER, b) + + def test_backfill_metric_keeps_business_key(self): + """#8 metric 回填为模板原指标名（业务语义键不变）。""" + metrics = [w.get("metric") for w in self.concrete["widgets"] + if w.get("type") == "kpi_card"] + self.assertEqual(metrics[0], "resin_exchange_capacity") + + def test_instantiate_strips_internal_fields(self): + """#9 实例化输出去除 extracted_from / placeholder_binding 内部字段。""" + self.assertNotIn("placeholder_binding", self.concrete) + for w in self.concrete["widgets"]: + self.assertNotIn("extracted_from", w) + # 输出仍是合法布局（schema/widgets 齐全） + self.assertEqual(self.concrete["$schema"], COCKPIT_SCHEMA) + + def test_instantiate_fallback_unknown_device(self): + """占位符回填：来源设备不在点位字典 → 回退首个设备。""" + tmpl = copy.deepcopy(self.tmpl) + # 篡改 trend 来源为字典中不存在的设备 + for w in tmpl["widgets"]: + if w.get("type") == "trend": + w["extracted_from"]["bind"] = ["ZZZ-999", "TEMP"] + break + conc = instantiate(tmpl, self.pd) + bind0 = next(w.get("bind") for w in conc["widgets"] + if w.get("type") == "trend") + self.assertTrue(bind0.startswith(self.pd.devices()[0] + ".")) + + def test_instantiate_rejects_empty_pointdict(self): + """空点位字典 → ValueError（无法回填）。""" + empty = PointDict(entries=[]) + with self.assertRaises(ValueError): + instantiate(self.tmpl, empty) + + +class TestDiffLayouts(unittest.TestCase): + """布局差异比对。""" + + def setUp(self): + self.layout = load_layout() + + def test_identical_layout(self): + """#10 相同布局 → is_identical。""" + d = diff_layouts(self.layout, self.layout) + self.assertTrue(d.is_identical) + self.assertIn("完全一致", d.summary()[0]) + + def test_binding_change_detected(self): + """#11 bind 变更 → binding_changes（换行业回归比对）。""" + b = copy.deepcopy(self.layout) + for w in b["widgets"]: + if w.get("type") == "trend" and w.get("bind") == "R-801.TEMP": + w["bind"] = "R-802.TEMP" + d = diff_layouts(self.layout, b) + self.assertFalse(d.is_identical) + self.assertTrue(any("bind" in s for s in d.binding_changes)) + self.assertTrue(any("R-802.TEMP" in s for s in d.binding_changes)) + + def test_geometry_change_detected(self): + """#12 几何位置变更 → geometry_changes。""" + b = copy.deepcopy(self.layout) + b["widgets"][0]["x"] = 99 + b["title"] = "改了" + d = diff_layouts(self.layout, b) + self.assertFalse(d.is_identical) + self.assertTrue(d.title_changed) + self.assertTrue(any("widget[0].x" in s for s in d.geometry_changes)) + + def test_widget_count_diff(self): + """widget 数量不同 → 摘要体现。""" + a = {"$schema": COCKPIT_SCHEMA, "title": "t", "widgets": [ + {"type": "trend", "bind": "R-801.TEMP"}]} + b = {"$schema": COCKPIT_SCHEMA, "title": "t", "widgets": []} + d = diff_layouts(a, b) + self.assertEqual(d.widget_count_a, 1) + self.assertEqual(d.widget_count_b, 0) + self.assertFalse(d.is_identical) + + +class TestEndToEndRoundTrip(unittest.TestCase): + """端到端：extract → instantiate 与原布局一致（回填幂等性）。""" + + def test_roundtrip_identical_to_source(self): + """extract 后用同一点位字典 instantiate，结果与原布局一致。 + + 验证「布局资产 ↔ 模板 ↔ 点位字典」三者解耦后可无损还原—— + 这是模板可复用的关键：换行业只需换点位字典即可重排驾驶舱。 + """ + layout = load_layout() + pd = PointDict.from_csv() + tmpl = extract_template(layout) + concrete = instantiate(tmpl, pd) + d = diff_layouts(layout, concrete) + self.assertTrue( + d.is_identical, + f"回填结果与原布局不一致：{d.summary()}") + + +if __name__ == "__main__": + unittest.main()