# -*- 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