feat(#84): 树脂驾驶舱布局提取为模板
把已交付树脂驾驶舱布局资产(cockpit.resin.yaml)提取为可复用模板:
- dashboard/extractor.py:extract_template(变量化 R-801.TEMP→{device}.{point}、
metric→{metric})、instantiate(点位字典回填)、diff_layouts(差异比对);
- PointDict.from_csv 加载 9 列点位字典;零依赖 _parse_yaml_subset 解析 YAML;
- tests/test_extractor.py 19 用例全部通过。
This commit is contained in:
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user