diff --git a/core/template-console/__init__.py b/core/template-console/__init__.py new file mode 100644 index 0000000..ca1afec --- /dev/null +++ b/core/template-console/__init__.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +"""⑤.7 模板配置台（Template Console）内核引擎 —— EPIC #9。 + +配置台是**跨模板通用的内核能力**：为实施工程师提供一个无代码的配置驱动 +界面，把"模型超参 / RAG / 布局"三类配置 + 点位字典 + 版本发布统一编排， +并将发布的配置**推送给内核**（edge-gateway / rag-kb / model-framework）。 + +本包拆为 6 个子模块，对应 6 个 issue（同一 feature 分支承载，单 PR 关联）： + +- ``rbac`` (#62) 三级 RBAC 权限（admin / engineer / readonly）； +- ``point_importer`` (#63) 点位字典 CSV 导入 + 自动校验页面（复用 + ``core/edge-gateway/point_dict`` 校验器，增加配置台 + 级结果聚合 + OPC 节点格式校验 + 模板选择）； +- ``config_store`` (#64) 配置项 CRUD（模型超参 / RAG / 布局三类，文件系统 + 版本化 JSON 存储）； +- ``preview`` (#65) 预览渲染引擎（布局/告警/查询 → 可预览结构化输出， + 对齐 iAOP-cockpit-layout-v1 widget 类型）； +- ``release`` (#66) 版本发布 + 回滚点（基于 config_store 快照，semver）； +- ``push_channel`` (#67) 配置台↔内核配置推送契约（JSON manifest + 校验和 + + 幂等性）。 + +设计原则（对齐 PRD「可解释可溯源」与既有内核范式）： +- 纯标准库零运行时依赖（无 pyyaml/numpy/pandas），YAML 子集用内置解析器； +- dataclass + Enum + 类型注解 + 中文 docstring； +- 关键决策均带 ``meaning`` / ``reason`` 字段，便于审计与可解释性。 +""" +from __future__ import annotations + +from .rbac import ( + Action, + Permission, + Role, + RoleKind, + User, + has_permission, +) + +__all__ = [ + "Action", + "Permission", + "Role", + "RoleKind", + "User", + "has_permission", +] + +#: 本包版本（对齐 EPIC #9 模板配置台交付节奏） +__version__ = "1.0.0" diff --git a/core/template-console/config_store.py b/core/template-console/config_store.py new file mode 100644 index 0000000..de513af --- /dev/null +++ b/core/template-console/config_store.py @@ -0,0 +1,291 @@ +# -*- coding: utf-8 -*- +"""⑤.7 配置项 CRUD 存储引擎 —— issue #64 / PRD ⑤.7。 + +配置台要管理三类业务配置：**模型超参 / RAG / 布局**。这些配置是模板交付物的 +"活"部分——实施工程师按现场调参，每次改动都要**可解释、可校验、可版本化** +（为 #66 发布/回滚提供快照源）。本模块提供基于文件系统的版本化 JSON 存储： + +- 三类配置各对应一个 JSON 文件（``model_params.json`` / ``rag_configs.json`` / + ``layout.json``），存放在一个 store 根目录下； +- 每条配置项是一个 ``ConfigItem``（key + value + 含义 + 校验规则）； +- 提供 ``list / get / upsert / delete`` CRUD，所有写操作都先**校验**再落盘， + 并记录 ``updated_by`` / ``reason``（对齐 PRD「可解释可溯源」）； +- 校验规则按类别内置（模型超参的范围/类型、RAG 的来源数、布局的 widget 类型）， + 非法值在 upsert 阶段即被拒绝，避免坏数据进入版本快照。 + +存储格式（每类一个 JSON，内容为 ``{items: [ConfigItem, ...], schema_version}``） +刻意简单、人可读，便于实施工程师直接查看/备份。 + +零运行时依赖：仅用 json / dataclass / Enum / 标准库。 +""" +from __future__ import annotations + +import json +import os +import re +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple + + +# --------------------------------------------------------------------------- +# 配置类别 +# --------------------------------------------------------------------------- + +class ConfigKind(str, Enum): + """三类业务配置（对齐 #64 需求）。""" + + MODEL_PARAM = "model_param" # 模型超参（学习率/迭代数/特征开关…） + RAG_CONFIG = "rag_config" # RAG 知识库配置（top_k/相似度阈值/来源…） + LAYOUT = "layout" # 驾驶舱布局（widget 列表） + + +#: 各类别对应的存储文件名 +KIND_FILENAME: Dict[ConfigKind, str] = { + ConfigKind.MODEL_PARAM: "model_params.json", + ConfigKind.RAG_CONFIG: "rag_configs.json", + ConfigKind.LAYOUT: "layout.json", +} + +#: 存储结构版本（schema 演进时升级，发布快照会带上） +STORE_SCHEMA_VERSION = 1 + +#: 驾驶舱布局允许的 widget 类型（对齐 iAOP-cockpit-layout-v1 / resin cockpit） +ALLOWED_WIDGET_TYPES = {"process_view", "trend", "kpi_card", "alarm_panel", "nl_query"} + + +# --------------------------------------------------------------------------- +# 配置项数据模型 +# --------------------------------------------------------------------------- + +@dataclass +class ConfigItem: + """一条配置项（可解释：带含义、更新人、原因）。""" + + key: str # 配置键（类别内唯一，如 learning_rate） + value: Any # 配置值（标量或结构化） + kind: ConfigKind # 所属类别 + meaning: str = "" # 业务含义（供配置台展示与审计） + updated_by: str = "system" # 最后修改人（对接 RBAC 用户名） + reason: str = "" # 本次修改原因（可解释可溯源） + updated_at: str = "" # ISO8601 时间戳 + + def to_dict(self) -> dict: + d = asdict(self) + d["kind"] = self.kind.value # 枚举序列化为字符串 + return d + + @classmethod + def from_dict(cls, raw: dict) -> "ConfigItem": + return cls( + key=raw["key"], + value=raw.get("value"), + kind=ConfigKind(raw.get("kind")), + meaning=raw.get("meaning", ""), + updated_by=raw.get("updated_by", "system"), + reason=raw.get("reason", ""), + updated_at=raw.get("updated_at", ""), + ) + + +def _now_iso() -> str: + """当前 UTC 时间 ISO8601（无时区歧义）。""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# 校验（按类别内置规则） +# --------------------------------------------------------------------------- + +@dataclass +class ValidationResult: + """配置项校验结果。""" + + ok: bool + errors: List[str] = field(default_factory=list) + + def __bool__(self) -> bool: + return self.ok + + +def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult: + """按类别校验配置项的 key/value 合法性。 + + 校验规则（配置台 upsert 前置门禁，防止坏数据进快照）： + - 通用：key 非空、匹配 ``[a-z0-9_.-]+``； + - model_param：value 为标量（int/float/bool/str）或标量列表； + - rag_config：top_k 为 1~50 的正整数、similarity_threshold 为 0~1 浮点、 + sources 为非空字符串列表； + - layout：value 为 widget 列表，每个 widget 有合法 type 与 x/y/w/h。 + """ + errors: List[str] = [] + if not key or not isinstance(key, str): + errors.append("key 不能为空") + elif not re.match(r"^[a-z0-9_.\-]+$", key): + errors.append(f"key '{key}' 仅允许小写字母/数字/._-") + + if kind == ConfigKind.MODEL_PARAM: + if not isinstance(value, (int, float, bool, str, list)): + errors.append("model_param 的 value 必须为标量或标量列表") + elif isinstance(value, list) and any( + not isinstance(v, (int, float, bool, str)) for v in value): + errors.append("model_param 列表 value 仅允许标量元素") + # 常见超参范围提示（软约束，仅对已知键） + if key == "learning_rate" and isinstance(value, (int, float)): + if not (0 < value < 1): + errors.append("learning_rate 应在 (0, 1) 区间") + if key == "iterations" and isinstance(value, int): + if value <= 0: + errors.append("iterations 必须为正整数") + + elif kind == ConfigKind.RAG_CONFIG: + if key == "top_k": + if not (isinstance(value, int) and 1 <= value <= 50): + errors.append("top_k 必须为 1~50 的整数") + elif key == "similarity_threshold": + if not (isinstance(value, (int, float)) and 0 <= value <= 1): + errors.append("similarity_threshold 必须为 0~1 的数") + elif key == "sources": + if not (isinstance(value, list) and value + and all(isinstance(s, str) and s for s in value)): + errors.append("sources 必须为非空字符串列表") + + elif kind == ConfigKind.LAYOUT: + if not isinstance(value, list): + errors.append("layout 的 value 必须为 widget 列表") + else: + for i, w in enumerate(value): + if not isinstance(w, dict): + errors.append(f"widget[{i}] 必须为对象") + continue + wt = w.get("type") + if wt not in ALLOWED_WIDGET_TYPES: + errors.append( + f"widget[{i}] 非法 type '{wt}'（合法：{sorted(ALLOWED_WIDGET_TYPES)})") + for coord in ("x", "y", "w", "h"): + if not isinstance(w.get(coord), int) or w.get(coord) < 0: + errors.append(f"widget[{i}] {coord} 必须为非负整数") + + return ValidationResult(ok=not errors, errors=errors) + + +# --------------------------------------------------------------------------- +# 存储引擎 +# --------------------------------------------------------------------------- + +class ConfigStore: + """基于文件系统的版本化配置存储（三类配置各一 JSON）。 + + 用法： + store = ConfigStore("/path/to/store") + store.upsert(ConfigKind.MODEL_PARAM, "learning_rate", 0.001, + meaning="学习率", updated_by="li", reason="首次标定") + items = store.list(ConfigKind.MODEL_PARAM) + """ + + def __init__(self, root: str) -> None: + self.root = root + os.makedirs(root, exist_ok=True) + + # -- 路径 -- + def _path(self, kind: ConfigKind) -> str: + return os.path.join(self.root, KIND_FILENAME[kind]) + + def _read(self, kind: ConfigKind) -> List[ConfigItem]: + path = self._path(kind) + if not os.path.isfile(path): + return [] + with open(path, "r", encoding="utf-8") as fh: + blob = json.load(fh) + return [ConfigItem.from_dict(r) for r in blob.get("items", [])] + + def _write(self, kind: ConfigKind, items: List[ConfigItem]) -> None: + blob = { + "schema_version": STORE_SCHEMA_VERSION, + "kind": kind.value, + "items": [it.to_dict() for it in items], + } + path = self._path(kind) + # 先写临时文件再替换，避免写一半被读到（原子写） + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(blob, fh, ensure_ascii=False, indent=2) + os.replace(tmp, path) + + # -- 查询 -- + def list(self, kind: ConfigKind) -> List[ConfigItem]: + """列出某类全部配置项。""" + return self._read(kind) + + def get(self, kind: ConfigKind, key: str) -> Optional[ConfigItem]: + """取单条配置项（不存在返回 None）。""" + for it in self._read(kind): + if it.key == key: + return it + return None + + # -- 写 -- + def upsert( + self, + kind: ConfigKind, + key: str, + value: Any, + meaning: str = "", + updated_by: str = "system", + reason: str = "", + ) -> ConfigItem: + """新增或更新一条配置项（先校验，再落盘）。 + + Raises: + ValueError: 校验失败（带全部错误明细）。 + """ + vr = validate_item(kind, key, value) + if not vr: + raise ValueError(f"配置项校验失败 [{kind.value}:{key}]：{'; '.join(vr.errors)}") + items = self._read(kind) + now = _now_iso() + existing_idx = next((i for i, it in enumerate(items) if it.key == key), None) + item = ConfigItem( + key=key, value=value, kind=kind, meaning=meaning, + updated_by=updated_by, reason=reason, updated_at=now, + ) + if existing_idx is None: + items.append(item) + else: + items[existing_idx] = item + self._write(kind, items) + return item + + def delete(self, kind: ConfigKind, key: str) -> bool: + """删除一条配置项。返回是否实际删除。""" + items = self._read(kind) + new_items = [it for it in items if it.key != key] + if len(new_items) == len(items): + return False + self._write(kind, new_items) + return True + + # -- 快照（供 #66 release 使用） -- + def snapshot(self) -> Dict[str, Any]: + """全量配置快照（三类聚合，供发布版本固化）。""" + return { + "schema_version": STORE_SCHEMA_VERSION, + "captured_at": _now_iso(), + "kinds": { + kind.value: [it.to_dict() for it in self._read(kind)] + for kind in ConfigKind + }, + } + + def restore(self, snapshot: Dict[str, Any]) -> None: + """从快照恢复全部配置（#66 回滚入口）。""" + kinds = snapshot.get("kinds", {}) + for kind in ConfigKind: + raw_items = kinds.get(kind.value, []) + items = [ConfigItem.from_dict(r) for r in raw_items] + self._write(kind, items) + + def item_counts(self) -> Dict[str, int]: + """各类配置项数量（配置台仪表盘用）。""" + return {kind.value: len(self._read(kind)) for kind in ConfigKind} diff --git a/core/template-console/point_importer.py b/core/template-console/point_importer.py new file mode 100644 index 0000000..667f645 --- /dev/null +++ b/core/template-console/point_importer.py @@ -0,0 +1,318 @@ +# -*- coding: utf-8 -*- +"""⑤.7 点位字典 CSV 导入 + 自动校验页面 —— issue #63 / PRD ⑤.7。 + +配置台的"导入页面"要解决：实施工程师拿着 DCS 点表（Excel 转 CSV）粘进配置台， +**一次性看到所有问题**（表头错/量纲错/重复点号/采样率非正/协议非法/OPC 节点 +格式错），而不是改一条报一条。本模块是导入页面的后端引擎。 + +**复用而非重造**：点位字典的 schema/加载/校验（量纲/数据类型/采样率/重复点号/ +协议）已由内核 ``core/edge-gateway/point_dict``（loader/validator/schema）实现 +并被边缘网关正式使用。本模块在其基础上增加**配置台专属**校验维度： + +1. OPC 节点格式校验（OPC UA 节点须形如 ``ns=<数字>;s=<名>`` 或 PLC 寄存器 + ``holding:<数字>`` / ``coil:<数字>``，与 simulator/opcua 驱动约定一致）； +2. 表头列顺序严格对齐（实施工程师照表填列，列序错位是高频错误）； +3. 行级结果聚合为 ``ImportRowIssue``（行号 + 严重级别 + 问题 + 修复建议）， + 供配置台前端逐行渲染、按严重级别过滤； +4. 模板选择（resin/ti）：不同行业模板的合法量纲集合不同（如树脂含 rpm/mmol·g⁻¹）， + 导入时按模板切换校验基线。 + +零运行时依赖：复用 ``point_dict`` 子包（纯标准库 csv/dataclass）。 +""" +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Tuple + +# 复用内核 edge-gateway 的点位字典加载/校验（schema/validator/loader） +# 在 tests/_bootstrap.py 中已把 core/edge-gateway 加入 sys.path， +# 故此处以顶层包 point_dict 引用（与 edge-gateway 自身测试一致）。 +from point_dict import ( # noqa: E402 + CSV_HEADERS, + VALID_PROTOCOLS, + VALID_UNITS, + Point, + PointDict, + load_point_dict_csv, +) +from point_dict import schema as _pd_schema # noqa: E402 + + +class Severity(str, Enum): + """问题严重级别（配置台前端据此着色/过滤）。""" + + ERROR = "error" # 阻断：不修复无法入库 + WARN = "warn" # 警告：可入库但建议复核（如 OPC 节点为空） + + +class TemplateKind(str, Enum): + """行业模板（决定合法量纲等校验基线）。""" + + TI = "ti" # 氯化/化工通用（templates/ti-cl4） + RESIN = "resin" # 吸附树脂（templates/resin，含 rpm / mmol/g） + + +# OPC 节点格式（与 simulator/opcua/s7 驱动约定一致）： +# ns=2;s=CLF.Temp —— OPC UA 节点（namespace + 字符串 id） +# holding:40010 / coil:1 —— Modbus 寄存器（保持/线圈 + 地址） +_OPC_UA_RE = re.compile(r"^ns=\d+;s=[^\s,]+$") +_MODBUS_RE = re.compile(r"^(holding|coil|input|discrete):(\d+)$") + + +def _unit_set(template: TemplateKind) -> set: + """按模板返回合法量纲集合（树脂含 rpm/mmol·g⁻¹ 等扩展）。""" + base = set(VALID_UNITS) + if template == TemplateKind.RESIN: + # VALID_UNITS 已含树脂扩展（rpm/mmol/g），直接复用 + return base + # ti 模板：移除树脂专属量纲，避免化工模板误用树脂量纲 + base.discard("rpm") + base.discard("mmol/g") + return base + + +@dataclass +class ImportRowIssue: + """导入页一行的问题（行号 + 严重级别 + 问题 + 修复建议，可解释）。""" + + row: int # CSV 行号（表头=1，数据从 2 起） + severity: Severity + code: str # 错误码（对齐 point_dict.validator 的 code + 本模块扩展） + message: str # 问题描述 + suggestion: str = "" # 修复建议（供配置台"一键修复"提示） + + def to_dict(self) -> dict: + return { + "row": self.row, "severity": self.severity.value, + "code": self.code, "message": self.message, + "suggestion": self.suggestion, + } + + +@dataclass +class ImportReport: + """导入校验报告（配置台导入页面数据模型）。""" + + template: TemplateKind + total_rows: int = 0 # 数据行数 + issues: List[ImportRowIssue] = field(default_factory=list) + loaded_points: int = 0 # 成功加载的点数 + file_path: str = "" + + @property + def ok(self) -> bool: + """无 ERROR 级问题即可入库（WARN 不阻断）。""" + return not any(i.severity == Severity.ERROR for i in self.issues) + + @property + def error_count(self) -> int: + return sum(1 for i in self.issues if i.severity == Severity.ERROR) + + @property + def warn_count(self) -> int: + return sum(1 for i in self.issues if i.severity == Severity.WARN) + + def summary(self) -> str: + """人类可读汇总（配置台导入结果横幅）。""" + status = "通过" if self.ok else "未通过" + return (f"导入校验{status}：{self.loaded_points} 点 / " + f"{self.total_rows} 行，错误 {self.error_count}，警告 {self.warn_count}") + + def to_dict(self) -> dict: + return { + "template": self.template.value, + "total_rows": self.total_rows, + "loaded_points": self.loaded_points, + "ok": self.ok, + "error_count": self.error_count, + "warn_count": self.warn_count, + "summary": self.summary(), + "issues": [i.to_dict() for i in self.issues], + } + + +# --------------------------------------------------------------------------- +# 校验扩展 +# --------------------------------------------------------------------------- + +def _validate_opc_node(point: Point) -> List[ImportRowIssue]: + """OPC 节点格式校验（配置台扩展维度）。 + + 约定（与驱动注册表对齐）： + - 协议 opcua：节点须匹配 ``ns=<数字>;s=<名>``； + - 协议 modbus：节点须匹配 ``holding/coil/input/discrete:<数字>``； + - 协议 simulator：节点可空，或任意上述格式（演示用，宽松）； + - 节点为空：WARN（可入库但运行时无法采集，建议补全）。 + """ + out: List[ImportRowIssue] = [] + node = (point.opc_node or "").strip() + if not node: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.WARN, code="empty_opc_node", + message=f"第{point.row_number}行 opcNode 为空", + suggestion="运行时无法采集，建议补全 OPC UA 节点或 PLC 寄存器地址", + )) + return out + + proto = (point.protocol or "").lower() + ok_ua = bool(_OPC_UA_RE.match(node)) + ok_mb = bool(_MODBUS_RE.match(node)) + if proto == "opcua" and not ok_ua: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.ERROR, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 不符合 OPC UA " + f"格式 ns=;s="), + suggestion="示例：ns=2;s=CLF.Temp", + )) + elif proto == "modbus" and not ok_mb: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.ERROR, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 不符合 Modbus " + f"格式 holding/coil/input/discrete:"), + suggestion="示例：holding:40010", + )) + elif proto not in ("opcua", "modbus") and not (ok_ua or ok_mb): + # simulator/s7/... 节点为空已 WARN；非空但格式都不符则 WARN（宽松） + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.WARN, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 既非 OPC UA 也非 " + f"Modbus 格式"), + suggestion="确认节点格式或清空（演示协议可空）", + )) + return out + + +def _validate_header_order(headers: List[str]) -> List[ImportRowIssue]: + """表头列顺序严格对齐（列序错位是实施工程师高频错误）。""" + out: List[ImportRowIssue] = [] + if not headers: + out.append(ImportRowIssue( + row=1, severity=Severity.ERROR, code="empty_header", + message="CSV 缺少表头行", + suggestion=f"表头应为：{','.join(CSV_HEADERS)}", + )) + return out + missing = [h for h in CSV_HEADERS if h not in headers] + for h in missing: + out.append(ImportRowIssue( + row=1, severity=Severity.ERROR, code="missing_column", + message=f"表头缺少必填列：{h}", + suggestion=f"补列 {h}（完整表头：{','.join(CSV_HEADERS)}）", + )) + if headers[: len(CSV_HEADERS)] != CSV_HEADERS and not missing: + out.append(ImportRowIssue( + row=1, severity=Severity.WARN, code="bad_column_order", + message=f"表头列顺序与标准不一致：{headers}", + suggestion=f"标准顺序：{','.join(CSV_HEADERS)}", + )) + return out + + +def _convert_validator_issues(report: "object", severity_for: Dict[str, Severity]) -> List[ImportRowIssue]: + """把内核 validator.ValidationReport.issues 转成 ImportRowIssue。""" + out: List[ImportRowIssue] = [] + for it in getattr(report, "issues", []): + sev = severity_for.get(it.code, Severity.ERROR) + out.append(ImportRowIssue( + row=it.row, severity=sev, code=it.code, message=it.message, + )) + return out + + +# 内核 validator 错误码 → 严重级别映射 +_SEVERITY_MAP: Dict[str, Severity] = { + "missing_column": Severity.ERROR, + "missing_field": Severity.ERROR, + "bad_unit": Severity.ERROR, + "bad_data_type": Severity.ERROR, + "bad_sample_rate": Severity.ERROR, + "bad_protocol": Severity.ERROR, + "dup_point": Severity.ERROR, +} + + +# --------------------------------------------------------------------------- +# 导入入口 +# --------------------------------------------------------------------------- + +def import_csv( + path: str, + template: TemplateKind = TemplateKind.TI, + extra_unit_check: bool = True, +) -> Tuple[PointDict, ImportReport]: + """导入并校验点位字典 CSV（配置台导入页面后端入口）。 + + Args: + path: CSV 文件路径（UTF-8，9 列表头）； + template: 行业模板（决定合法量纲集合，resin/ti）； + extra_unit_check: 是否按模板收窄量纲集合做额外校验。 + + Returns: + (PointDict, ImportReport)：加载的点位模型 + 校验报告。 + 报告 ``ok`` 为 True 即可入库；WARN 不阻断。 + """ + report = ImportReport(template=template, file_path=path) + + # 1) 表头校验（先读表头行） + import csv as _csv + with open(path, "r", encoding="utf-8-sig") as fh: + reader = _csv.reader(fh) + rows = list(reader) + headers = [c.strip() for c in rows[0]] if rows else [] + report.issues.extend(_validate_header_order(headers)) + + # 2) 加载 + 内核校验（量纲/数据类型/采样率/重复点号/协议） + point_dict = load_point_dict_csv(path) + report.loaded_points = len(point_dict) + report.total_rows = len(point_dict.points) + + from point_dict.validator import validate_point_dict + kernel_report = validate_point_dict(point_dict, headers) + report.issues.extend(_convert_validator_issues(kernel_report, _SEVERITY_MAP)) + + # 3) 模板级量纲收窄（resin 才允许 rpm/mmol·g⁻¹） + if extra_unit_check: + allowed_units = _unit_set(template) + for p in point_dict.points: + if p.unit and p.unit not in allowed_units: + # 内核 validator 已按全集校验过；这里只补充模板级差异提示 + if p.unit in ("rpm", "mmol/g") and template == TemplateKind.TI: + report.issues.append(ImportRowIssue( + row=p.row_number, severity=Severity.ERROR, + code="template_unit_mismatch", + message=(f"第{p.row_number}行 量纲 '{p.unit}' 为树脂模板专属，" + f"当前导入的是 {template.value} 模板"), + suggestion="切换模板为 resin，或修正量纲", + )) + + # 4) OPC 节点格式校验（配置台扩展） + for p in point_dict.points: + report.issues.extend(_validate_opc_node(p)) + + # 行号排序，便于配置台逐行展示 + report.issues.sort(key=lambda i: (i.row, i.code)) + return point_dict, report + + +def import_csv_string( + content: str, + template: TemplateKind = TemplateKind.TI, + encoding: str = "utf-8", +) -> Tuple[PointDict, ImportReport]: + """从 CSV 文本导入（配置台粘贴框场景，落临时文件后复用 import_csv）。""" + import tempfile + tmp = tempfile.NamedTemporaryFile( + mode="w", encoding=encoding, suffix=".csv", delete=False) + try: + tmp.write(content) + tmp.flush() + tmp.close() + return import_csv(tmp.name, template=template) + finally: + try: + os.unlink(tmp.name) + except OSError: + pass diff --git a/core/template-console/rbac.py b/core/template-console/rbac.py new file mode 100644 index 0000000..9ee3749 --- /dev/null +++ b/core/template-console/rbac.py @@ -0,0 +1,310 @@ +# -*- coding: utf-8 -*- +"""⑤.7 配置台三级 RBAC 权限模型 —— issue #62 / PRD ⑤.7。 + +配置台面向**多角色协作**：实施工程师配模板，行业工程师调参数，运维/管理者 +发布上线。直接对所有人开放写权限会带来误改与不可溯源风险。本模块用三级 +RBAC（基于角色的访问控制）锁定"谁能对哪类配置做什么"，并把每次权限判定 +的**理由**一并返回，对齐 PRD「可解释可溯源」。 + +三级角色（由低到高，后者继承前者全部权限）： + +- ``readonly`` （只读）：查看配置 / 预览 / 历史版本，不可写； +- ``engineer`` （行业工程师）：只读权限 + 编辑/校验/导入配置（模型超参 / + RAG / 布局 / 点位字典），但**不能发布与回滚**； +- ``admin`` （管理员）：工程师权限 + 发布 / 回滚 / 推送内核 / 用户管理。 + +权限判定核心为 ``has_permission(user, resource, action)``，返回 +``PermissionDecision``（allow + reason），便于配置台前端把"为什么拒绝" +直接展示给操作者，而不是一个干瘪的 403。 + +零运行时依赖：仅用 dataclass / Enum / 标准库。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Set + + +# --------------------------------------------------------------------------- +# 权限维度：资源 × 动作 +# --------------------------------------------------------------------------- + +class Resource(str, Enum): + """配置台可管控的资源（对齐 #63~#67 子模块）。""" + + POINT_DICT = "point_dict" # 点位字典（#63） + MODEL_PARAM = "model_param" # 模型超参配置（#64） + RAG_CONFIG = "rag_config" # RAG 知识库配置（#64） + LAYOUT = "layout" # 驾驶舱布局配置（#64/#65） + PREVIEW = "preview" # 预览（#65） + RELEASE = "release" # 版本发布/回滚（#66） + PUSH = "push" # 配置推送内核（#67） + USER = "user" # 用户/角色管理 + + +class Action(str, Enum): + """对资源可执行的动作。""" + + VIEW = "view" # 查看 / 预览 / 列表 + EDIT = "edit" # 新增 / 修改 / 删除 / 导入 / 校验 + PUBLISH = "publish" # 发布版本 / 回滚 / 推送内核 + MANAGE = "manage" # 用户与角色管理 + + +class RoleKind(str, Enum): + """三级角色枚举（值即配置资产中的角色标识）。""" + + READONLY = "readonly" + ENGINEER = "engineer" + ADMIN = "admin" + + +# 各资源的「写」动作等价集合：EDIT 含新增/修改/删除/导入/校验。 +# PUBLISH 含发布/回滚/推送。这样配置台前端只需关心粗粒度动作。 +_WRITE_ACTIONS: Set[Action] = {Action.EDIT, Action.PUBLISH, Action.MANAGE} + + +# --------------------------------------------------------------------------- +# 权限模型 +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Permission: + """一条权限授予（角色 → 资源 → 动作）。 + + ``meaning`` 解释该权限的业务含义，用于审计日志与配置台权限矩阵展示。 + 注意：权限**匹配**基于 ``resource:action``（资源×动作），与授予角色无关—— + 这正是角色继承能生效的关键（admin 继承 engineer 的 edit，匹配键相同）。 + ``role`` 仅作为审计元数据，记录"是谁授予的"。 + """ + + role: RoleKind + resource: Resource + action: Action + meaning: str = "" + + def key(self) -> str: + """权限匹配键（资源:动作）—— 角色继承据此累计。""" + return f"{self.resource.value}:{self.action.value}" + + def audit_key(self) -> str: + """审计唯一键（角色/资源/动作三元组，含授予者）。""" + return f"{self.role.value}:{self.resource.value}:{self.action.value}" + + +@dataclass +class Role: + """一个角色：权限集合 + 继承的父角色。""" + + kind: RoleKind + label: str # 中文展示名 + permissions: List[Permission] = field(default_factory=list) + inherits: Optional[RoleKind] = None # 继承的低一级角色 + description: str = "" # 角色职责说明（可解释性） + + def permission_keys(self) -> Set[str]: + """本角色直接授予的权限键集合。""" + return {p.key() for p in self.permissions} + + +@dataclass +class User: + """配置台用户。""" + + username: str + role: RoleKind + display_name: str = "" + # 可选资源级收窄：即便角色允许，列表中的资源也会被额外限制为只读。 + # 用于"只允许工程师改某几类配置"的细粒度场景。 + restricted_to_readonly: List[Resource] = field(default_factory=list) + + +@dataclass +class PermissionDecision: + """``has_permission`` 的判定结果（带理由，可解释）。""" + + allow: bool + reason: str # 人类可读的判定理由（允许/拒绝原因） + role: RoleKind + resource: Resource + action: Action + source: str = "explicit" # explicit（本角色直接授予）/ inherited（继承自父角色） + + +# --------------------------------------------------------------------------- +# 角色注册表：三级权限矩阵（对齐 PRD ⑤.7「三级 RBAC」） +# --------------------------------------------------------------------------- + +def _build_role_registry() -> Dict[RoleKind, Role]: + """构建三级角色及其权限矩阵。 + + 权限设计依据（PRD ⑤.7）： + - readonly：可查看所有配置/预览/历史，但不能改、不能发； + - engineer：在 readonly 基础上，可编辑/校验/导入四类业务配置， + 但**发布/回滚/推送/用户管理仍归 admin**（避免未经评审上线）； + - admin：在 engineer 基础上，可发布/回滚/推送 + 管理用户角色。 + """ + ro = Role( + kind=RoleKind.READONLY, + label="只读", + description="实施/运维只读角色：查看配置、预览、历史版本，不可写。", + permissions=[ + Permission(RoleKind.READONLY, Resource.POINT_DICT, Action.VIEW, + "查看点位字典与校验报告"), + Permission(RoleKind.READONLY, Resource.MODEL_PARAM, Action.VIEW, + "查看模型超参配置"), + Permission(RoleKind.READONLY, Resource.RAG_CONFIG, Action.VIEW, + "查看 RAG 知识库配置"), + Permission(RoleKind.READONLY, Resource.LAYOUT, Action.VIEW, + "查看驾驶舱布局配置"), + Permission(RoleKind.READONLY, Resource.PREVIEW, Action.VIEW, + "查看配置预览"), + Permission(RoleKind.READONLY, Resource.RELEASE, Action.VIEW, + "查看历史发布版本"), + ], + ) + + engineer = Role( + kind=RoleKind.ENGINEER, + label="行业工程师", + inherits=RoleKind.READONLY, + description="行业工程师：编辑/校验/导入业务配置，但不能发布与推送。", + permissions=[ + Permission(RoleKind.ENGINEER, Resource.POINT_DICT, Action.EDIT, + "导入/编辑/校验点位字典 CSV"), + Permission(RoleKind.ENGINEER, Resource.MODEL_PARAM, Action.EDIT, + "调整模型超参配置"), + Permission(RoleKind.ENGINEER, Resource.RAG_CONFIG, Action.EDIT, + "编辑 RAG 知识库配置"), + Permission(RoleKind.ENGINEER, Resource.LAYOUT, Action.EDIT, + "编辑驾驶舱布局配置"), + Permission(RoleKind.ENGINEER, Resource.PREVIEW, Action.VIEW, + "预览配置效果（编辑后必看）"), + ], + ) + + admin = Role( + kind=RoleKind.ADMIN, + label="管理员", + inherits=RoleKind.ENGINEER, + description="管理员：在工程师基础上负责发布/回滚/推送与用户管理。", + permissions=[ + Permission(RoleKind.ADMIN, Resource.RELEASE, Action.PUBLISH, + "发布新版本与回滚到历史版本"), + Permission(RoleKind.ADMIN, Resource.PUSH, Action.PUBLISH, + "把已发布配置推送给内核"), + Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE, + "管理用户与角色分配"), + Permission(RoleKind.ADMIN, Resource.POINT_DICT, Action.PUBLISH, + "确认点位字典上线（审批环节）"), + Permission(RoleKind.ADMIN, Resource.MODEL_PARAM, Action.PUBLISH, + "确认模型超参上线"), + Permission(RoleKind.ADMIN, Resource.LAYOUT, Action.PUBLISH, + "确认布局上线"), + ], + ) + + return {RoleKind.READONLY: ro, RoleKind.ENGINEER: engineer, RoleKind.ADMIN: admin} + + +_ROLES: Dict[RoleKind, Role] = _build_role_registry() + + +def get_role(kind: RoleKind) -> Role: + """获取角色定义。""" + return _ROLES[kind] + + +def all_roles() -> List[Role]: + """全部角色（按权限由低到高）。""" + return [_ROLES[RoleKind.READONLY], _ROLES[RoleKind.ENGINEER], _ROLES[RoleKind.ADMIN]] + + +def effective_permissions(kind: RoleKind) -> Set[str]: + """角色有效权限键（含继承链）。 + + 继承解析：admin 继承 engineer 继承 readonly，递归向上累计权限键。 + """ + role = _ROLES[kind] + keys: Set[str] = set(role.permission_keys()) + if role.inherits is not None: + keys |= effective_permissions(role.inherits) + return keys + + +# --------------------------------------------------------------------------- +# 判定 API +# --------------------------------------------------------------------------- + +def has_permission( + user: User, + resource: Resource, + action: Action, +) -> PermissionDecision: + """判定用户对某资源执行某动作是否被允许（带理由）。 + + 判定顺序： + 1. 计算角色有效权限（含继承），命中即允许并标注来源（本角色/继承）； + 2. 命中后若该资源在用户 ``restricted_to_readonly`` 列表且动作是写动作， + 则降级拒绝（细粒度收窄）； + 3. 未命中则拒绝，理由标注缺失的权限三元组。 + + Args: + user: 配置台用户； + resource: 目标资源； + action: 目标动作。 + + Returns: + PermissionDecision：allow + reason（可直接展示给操作者）。 + """ + target = f"{resource.value}:{action.value}" + eff = effective_permissions(user.role) + + # 细粒度收窄：即便角色允许，特定资源也被限制为只读 + if resource in user.restricted_to_readonly and action in _WRITE_ACTIONS: + return PermissionDecision( + allow=False, + reason=(f"用户 '{user.username}' 对资源 '{resource.value}' 被收窄为只读，" + f"禁止执行 '{action.value}' 动作"), + role=user.role, resource=resource, action=action, source="restricted", + ) + + if target in eff: + # 判定来源：本角色直接授予 or 继承自父角色 + own = get_role(user.role).permission_keys() + source = "explicit" if target in own else "inherited" + src_label = "本角色直接授予" if source == "explicit" else "继承自低级角色" + return PermissionDecision( + allow=True, + reason=(f"用户 '{user.username}'（{get_role(user.role).label}）" + f"允许对 '{resource.value}' 执行 '{action.value}'（{src_label}）"), + role=user.role, resource=resource, action=action, source=source, + ) + + return PermissionDecision( + allow=False, + reason=(f"用户 '{user.username}'（{get_role(user.role).label}）缺少权限 " + f"{user.role.value}:{resource.value}:{action.value}；" + f"该动作需更高角色或审批"), + role=user.role, resource=resource, action=action, source="denied", + ) + + +def can_publish(user: User) -> bool: + """便捷判定：用户是否具备发布（发布/回滚/推送）能力。""" + return has_permission(user, Resource.RELEASE, Action.PUBLISH).allow + + +def user_summary(user: User) -> Dict[str, object]: + """用户权限概览（供配置台用户卡片/审计日志展示）。""" + role = get_role(user.role) + return { + "username": user.username, + "display_name": user.display_name or user.username, + "role": user.role.value, + "role_label": role.label, + "description": role.description, + "effective_permission_count": len(effective_permissions(user.role)), + "restricted_to_readonly": [r.value for r in user.restricted_to_readonly], + } diff --git a/core/template-console/tests/_bootstrap.py b/core/template-console/tests/_bootstrap.py new file mode 100644 index 0000000..bcff38b --- /dev/null +++ b/core/template-console/tests/_bootstrap.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""测试引导：把 `core/template-console` 以包名 `template_console` 挂载到 sys.modules。 + +目录名 `template-console` 含连字符，无法直接以包名 import；挂载后模块内 +相对导入（`from .rbac import ...`）在 unittest 发现机制下可正常解析。 + +同时把兄弟内核目录 `core/edge-gateway` 加入 sys.path，使 point_importer +可复用其 `point_dict` 子包（loader/validator/schema），避免重复造轮子。 +""" +import os +import sys +import types + +# 1) 挂载 core/template-console 为 template_console 包 +CONSOLE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, CONSOLE_DIR) +if "template_console" not in sys.modules: + pkg = types.ModuleType("template_console") + pkg.__path__ = [CONSOLE_DIR] + sys.modules["template_console"] = pkg + +# 2) 暴露兄弟内核 edge-gateway/point_dict（#63 复用其校验器） +CORE_DIR = os.path.dirname(CONSOLE_DIR) +EDGE_GW_DIR = os.path.join(CORE_DIR, "edge-gateway") +if os.path.isdir(EDGE_GW_DIR) and EDGE_GW_DIR not in sys.path: + sys.path.insert(0, EDGE_GW_DIR) diff --git a/core/template-console/tests/test_config_store.py b/core/template-console/tests/test_config_store.py new file mode 100644 index 0000000..77790f7 --- /dev/null +++ b/core/template-console/tests/test_config_store.py @@ -0,0 +1,191 @@ +# -*- coding: utf-8 -*- +"""配置项 CRUD 存储引擎测试（issue #64）。 + +覆盖： +1. 三类配置 CRUD（list/get/upsert/delete）； +2. 原子写 + 持久化（重开 store 仍在）； +3. 校验规则（model_param/rag_config/layout，非法值拒绝）； +4. 快照 snapshot/restore（为 #66 提供基础）； +5. 可解释字段（meaning/reason/updated_by/updated_at 落盘）。 +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.config_store import ( # noqa: E402 + ALLOWED_WIDGET_TYPES, + ConfigItem, + ConfigKind, + ConfigStore, + ValidationResult, + validate_item, +) + + +class _TmpStore: + def __init__(self): + self._tmp = tempfile.mkdtemp() + self.store = ConfigStore(self._tmp) + + def cleanup(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + +class ValidationTest(unittest.TestCase): + """校验规则。""" + + def test_model_param_scalar_ok(self): + self.assertTrue(validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 0.001)) + + def test_model_param_learning_rate_range(self): + vr = validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 1.5) + self.assertFalse(vr) + self.assertTrue(any("learning_rate" in e for e in vr.errors)) + + def test_model_param_bad_key(self): + vr = validate_item(ConfigKind.MODEL_PARAM, "Bad Key!", 1) + self.assertFalse(vr) + + def test_rag_top_k_bounds(self): + self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 0)) + self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 51)) + self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "top_k", 10)) + + def test_rag_similarity_threshold(self): + self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.5)) + self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 1.5)) + + def test_rag_sources_must_be_nonempty_list(self): + self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", [])) + self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", ["", "x"])) + self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "sources", ["sop", "gb"])) + + def test_layout_widget_type(self): + bad = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}] + self.assertFalse(validate_item(ConfigKind.LAYOUT, "dashboard", bad)) + good = [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}] + self.assertTrue(validate_item(ConfigKind.LAYOUT, "dashboard", good)) + + def test_layout_widget_coords_nonneg_int(self): + bad = [{"type": "trend", "x": -1, "y": 0, "w": 1, "h": 1}] + vr = validate_item(ConfigKind.LAYOUT, "dashboard", bad) + self.assertFalse(vr) + + +class CrudTest(unittest.TestCase): + """CRUD + 持久化。""" + + def setUp(self): + self.ctx = _TmpStore() + self.store = self.ctx.store + + def tearDown(self): + self.ctx.cleanup() + + def test_upsert_and_get(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "iterations", 100, + meaning="迭代数", updated_by="li", reason="标定") + it = self.store.get(ConfigKind.MODEL_PARAM, "iterations") + self.assertIsNotNone(it) + self.assertEqual(it.value, 100) + self.assertEqual(it.updated_by, "li") + self.assertEqual(it.reason, "标定") + self.assertTrue(it.updated_at) # 时间戳已写 + + def test_upsert_rejects_invalid(self): + with self.assertRaises(ValueError): + self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 999) + + def test_upsert_overwrites(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.1) + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01, reason="调小") + it = self.store.get(ConfigKind.MODEL_PARAM, "lr") + self.assertEqual(it.value, 0.01) + self.assertEqual(it.reason, "调小") + + def test_list_and_delete(self): + self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 5) + self.store.upsert(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.6) + self.assertEqual(len(self.store.list(ConfigKind.RAG_CONFIG)), 2) + self.assertTrue(self.store.delete(ConfigKind.RAG_CONFIG, "top_k")) + self.assertIsNone(self.store.get(ConfigKind.RAG_CONFIG, "top_k")) + self.assertFalse(self.store.delete(ConfigKind.RAG_CONFIG, "nope")) + + def test_persistence_across_reopen(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + # 重开一个指向同一目录的 store + store2 = ConfigStore(self.ctx._tmp) + it = store2.get(ConfigKind.MODEL_PARAM, "lr") + self.assertIsNotNone(it) + self.assertEqual(it.value, 0.001) + + def test_json_file_is_human_readable(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001, meaning="学习率") + path = os.path.join(self.ctx._tmp, "model_params.json") + with open(path, encoding="utf-8") as fh: + blob = json.load(fh) + self.assertEqual(blob["schema_version"], 1) + self.assertEqual(blob["kind"], "model_param") + self.assertEqual(blob["items"][0]["meaning"], "学习率") + + +class SnapshotTest(unittest.TestCase): + """快照与恢复（#66 基础）。""" + + def setUp(self): + self.ctx = _TmpStore() + self.store = self.ctx.store + + def tearDown(self): + self.ctx.cleanup() + + def test_snapshot_captures_all_kinds(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8) + snap = self.store.snapshot() + self.assertIn("captured_at", snap) + self.assertEqual(set(snap["kinds"].keys()), + {"model_param", "rag_config", "layout"}) + self.assertEqual(len(snap["kinds"]["model_param"]), 1) + + def test_restore_replicates_state(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8) + snap = self.store.snapshot() + # 清空再恢复 + self.store.delete(ConfigKind.MODEL_PARAM, "lr") + self.store.delete(ConfigKind.RAG_CONFIG, "top_k") + self.store.restore(snap) + self.assertEqual(self.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001) + self.assertEqual(self.store.get(ConfigKind.RAG_CONFIG, "top_k").value, 8) + + def test_item_counts(self): + self.store.upsert(ConfigKind.LAYOUT, "dashboard", + [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}]) + counts = self.store.item_counts() + self.assertEqual(counts["layout"], 1) + self.assertEqual(counts["model_param"], 0) + + +class ConfigItemSerializationTest(unittest.TestCase): + """ConfigItem 序列化往返。""" + + def test_roundtrip(self): + it = ConfigItem(key="lr", value=0.1, kind=ConfigKind.MODEL_PARAM, + meaning="学习率", updated_by="li", reason="init", + updated_at="2026-01-01T00:00:00Z") + d = it.to_dict() + self.assertEqual(d["kind"], "model_param") + it2 = ConfigItem.from_dict(d) + self.assertEqual(it2.value, 0.1) + self.assertEqual(it2.kind, ConfigKind.MODEL_PARAM) + + +if __name__ == "__main__": + unittest.main() diff --git a/core/template-console/tests/test_point_importer.py b/core/template-console/tests/test_point_importer.py new file mode 100644 index 0000000..a2553f9 --- /dev/null +++ b/core/template-console/tests/test_point_importer.py @@ -0,0 +1,228 @@ +# -*- coding: utf-8 -*- +"""点位字典 CSV 导入 + 自动校验测试（issue #63）。 + +覆盖： +1. 合法 CSV 导入通过（ti / resin 两套模板）； +2. 表头校验（缺失列 / 列序错位）； +3. 内核校验复用（量纲/数据类型/采样率/重复点号/协议）； +4. OPC 节点格式校验（opcua/modbus/空）； +5. 模板级量纲收窄（rpm 仅 resin 允许）； +6. 报告 ok/汇总/字典化 + 粘贴框入口。 +""" +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.point_importer import ( # noqa: E402 + ImportReport, + ImportRowIssue, + Severity, + TemplateKind, + import_csv, + import_csv_string, +) + +GOOD_TI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol +CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua +CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,opcua +""" + +GOOD_RESIN = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol +R-801,R-801.TEMP,反应釜温度,℃,float,1000,true,ns=2;s=R801.Temp,opcua +R-801,R-801.AGIT,搅拌转速,rpm,float,1000,true,ns=2;s=R801.Agit,opcua +""" + +BAD_MULTI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol +CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,badnode,opcua +CLF-01,CLF-01.TEMP,炉压,kPa,badtype,0,true,ns=2;s=CLF.Pres,opcua +CLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus +""" + + +class _TmpCsv: + """临时 CSV 文件助手。""" + + def __init__(self, content): + self._tmp = tempfile.mkdtemp() + self.path = os.path.join(self._tmp, "points.csv") + with open(self.path, "w", encoding="utf-8") as fh: + fh.write(content) + + def cleanup(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + +class GoodImportTest(unittest.TestCase): + """合法 CSV 导入。""" + + def test_good_ti_imports_ok(self): + f = _TmpCsv(GOOD_TI) + try: + pd, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertTrue(rep.ok, rep.summary()) + self.assertEqual(rep.loaded_points, 2) + self.assertEqual(rep.error_count, 0) + finally: + f.cleanup() + + def test_good_resin_imports_ok_with_rpm(self): + f = _TmpCsv(GOOD_RESIN) + try: + pd, rep = import_csv(f.path, template=TemplateKind.RESIN) + self.assertTrue(rep.ok, rep.summary()) + # rpm 在 resin 模板合法 + self.assertEqual(rep.error_count, 0) + finally: + f.cleanup() + + def test_report_summary_and_dict(self): + f = _TmpCsv(GOOD_TI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("通过", rep.summary()) + d = rep.to_dict() + self.assertTrue(d["ok"]) + self.assertEqual(d["template"], "ti") + self.assertEqual(d["loaded_points"], 2) + finally: + f.cleanup() + + +class HeaderValidationTest(unittest.TestCase): + """表头校验。""" + + def test_missing_column_is_error(self): + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertFalse(rep.ok) + codes = [i.code for i in rep.issues if i.row == 1] + self.assertIn("missing_column", codes) + finally: + f.cleanup() + + def test_wrong_column_order_is_warn(self): + # 列齐全但顺序错（name 提前）→ WARN，不阻断 + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,protocol,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,opcua,ns=2;s=CLF.Temp\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("bad_column_order", [i.code for i in rep.issues]) + finally: + f.cleanup() + + +class KernelValidationTest(unittest.TestCase): + """复用内核校验（量纲/数据类型/采样率/重复点号）。""" + + def test_dup_point_detected(self): + bad = GOOD_TI + "CLF-01,CLF-01.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF.Temp2,opcua\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertFalse(rep.ok) + self.assertIn("dup_point", [i.code for i in rep.issues]) + finally: + f.cleanup() + + def test_bad_data_type_and_sample_rate(self): + f = _TmpCsv(BAD_MULTI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + codes = [i.code for i in rep.issues] + self.assertIn("bad_data_type", codes) + self.assertIn("bad_sample_rate", codes) + finally: + f.cleanup() + + def test_bad_protocol_detected(self): + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,unknownproto\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("bad_protocol", [i.code for i in rep.issues]) + finally: + f.cleanup() + + +class OpcNodeValidationTest(unittest.TestCase): + """OPC 节点格式校验（配置台扩展维度）。""" + + def test_bad_opcua_node_is_error(self): + # BAD_MULTI 第1行 opcNode=badnode 协议 opcua → ERROR + f = _TmpCsv(BAD_MULTI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + opc_issues = [i for i in rep.issues if i.code == "bad_opc_node"] + self.assertTrue(any(i.severity == Severity.ERROR for i in opc_issues)) + finally: + f.cleanup() + + def test_valid_modbus_node_ok(self): + # BAD_MULTI 第3行 holding:40010 modbus → 不报 bad_opc_node + f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus\n") + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertNotIn("bad_opc_node", [i.code for i in rep.issues + if i.severity == Severity.ERROR]) + finally: + f.cleanup() + + def test_empty_opc_node_is_warn(self): + f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,simulator\n") + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + empties = [i for i in rep.issues if i.code == "empty_opc_node"] + self.assertEqual(len(empties), 1) + self.assertEqual(empties[0].severity, Severity.WARN) + # 警告不阻断 + self.assertTrue(rep.ok) + finally: + f.cleanup() + + +class TemplateUnitTest(unittest.TestCase): + """模板级量纲收窄。""" + + def test_rpm_rejected_in_ti_template(self): + f = _TmpCsv(GOOD_RESIN) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + # rpm 是树脂专属，ti 模板应报 template_unit_mismatch + self.assertIn("template_unit_mismatch", [i.code for i in rep.issues]) + self.assertFalse(rep.ok) + finally: + f.cleanup() + + def test_rpm_allowed_in_resin_template(self): + f = _TmpCsv(GOOD_RESIN) + try: + _, rep = import_csv(f.path, template=TemplateKind.RESIN) + self.assertNotIn("template_unit_mismatch", [i.code for i in rep.issues]) + self.assertTrue(rep.ok, rep.summary()) + finally: + f.cleanup() + + +class ImportStringTest(unittest.TestCase): + """粘贴框入口（import_csv_string）。""" + + def test_import_from_string(self): + pd, rep = import_csv_string(GOOD_TI, template=TemplateKind.TI) + self.assertTrue(rep.ok) + self.assertEqual(len(pd), 2) + + def test_import_string_bad_csv(self): + bad = "device_id,point_id\nCLF-01,CLF-01.TEMP\n" # 缺列 + _, rep = import_csv_string(bad, template=TemplateKind.TI) + self.assertFalse(rep.ok) + + +if __name__ == "__main__": + unittest.main() diff --git a/core/template-console/tests/test_rbac.py b/core/template-console/tests/test_rbac.py new file mode 100644 index 0000000..4a26e4c --- /dev/null +++ b/core/template-console/tests/test_rbac.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +"""三级 RBAC 权限模型测试（issue #62）。 + +覆盖： +1. 三级角色权限矩阵正确（readonly/engineer/admin）； +2. 角色继承（admin 继承 engineer 继承 readonly）； +3. has_permission 允许/拒绝判定 + 理由可解释； +4. 细粒度收窄（restricted_to_readonly 把写动作降级拒绝）； +5. 便捷判定 can_publish / 用户概览。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.rbac import ( # noqa: E402 + Action, + Permission, + Resource, + Role, + RoleKind, + User, + all_roles, + can_publish, + effective_permissions, + get_role, + has_permission, + user_summary, +) + + +class RoleRegistryTest(unittest.TestCase): + """三级角色注册表。""" + + def test_three_roles_present(self): + roles = {r.kind for r in all_roles()} + self.assertEqual(roles, {RoleKind.READONLY, RoleKind.ENGINEER, RoleKind.ADMIN}) + + def test_role_labels_in_chinese(self): + self.assertEqual(get_role(RoleKind.READONLY).label, "只读") + self.assertEqual(get_role(RoleKind.ENGINEER).label, "行业工程师") + self.assertEqual(get_role(RoleKind.ADMIN).label, "管理员") + + def test_role_descriptions_explainable(self): + # 可解释性：每个角色都有职责说明 + for role in all_roles(): + self.assertTrue(role.description, f"{role.kind} 缺少 description") + + def test_inheritance_chain(self): + self.assertEqual(get_role(RoleKind.ADMIN).inherits, RoleKind.ENGINEER) + self.assertEqual(get_role(RoleKind.ENGINEER).inherits, RoleKind.READONLY) + self.assertIsNone(get_role(RoleKind.READONLY).inherits) + + def test_permission_key_format(self): + p = Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE) + # 匹配键为资源:动作（角色无关，便于继承）；审计键含授予角色 + self.assertEqual(p.key(), "user:manage") + self.assertEqual(p.audit_key(), "admin:user:manage") + + +class EffectivePermissionTest(unittest.TestCase): + """继承后的有效权限集合。""" + + def test_admin_inherits_engineer_and_readonly(self): + eff = effective_permissions(RoleKind.ADMIN) + # 匹配键为 resource:action：admin 拥有自身的 user:manage， + # 也继承 engineer 的 model_param:edit 与 readonly 的 layout:view + self.assertIn("user:manage", eff) + self.assertIn("model_param:edit", eff) + self.assertIn("layout:view", eff) + + def test_engineer_cannot_publish(self): + eff = effective_permissions(RoleKind.ENGINEER) + # 工程师不能发布/推送/管用户 + self.assertNotIn("release:publish", eff) + self.assertNotIn("push:publish", eff) + self.assertNotIn("user:manage", eff) + + def test_readonly_has_no_write(self): + eff = effective_permissions(RoleKind.READONLY) + for key in eff: + # 只读权限只能以 :view 结尾 + self.assertTrue(key.endswith(":view"), f"readonly 不应有写/发布权限: {key}") + + +class HasPermissionTest(unittest.TestCase): + """has_permission 判定 + 理由。""" + + def setUp(self): + self.ro = User("viewer", RoleKind.READONLY, "查看员") + self.eng = User("li_engineer", RoleKind.ENGINEER, "李工") + self.admin = User("root_admin", RoleKind.ADMIN, "管理员甲") + + def test_readonly_view_allowed(self): + d = has_permission(self.ro, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d.allow) + self.assertEqual(d.source, "explicit") + + def test_readonly_edit_denied(self): + d = has_permission(self.ro, Resource.LAYOUT, Action.EDIT) + self.assertFalse(d.allow) + self.assertIn("缺少权限", d.reason) + + def test_engineer_edit_allowed_inherited_view(self): + # 工程师编辑是本角色权限（explicit） + d_edit = has_permission(self.eng, Resource.LAYOUT, Action.EDIT) + self.assertTrue(d_edit.allow) + self.assertEqual(d_edit.source, "explicit") + # 工程师查看布局是继承自 readonly（inherited） + d_view = has_permission(self.eng, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d_view.allow) + self.assertEqual(d_view.source, "inherited") + + def test_engineer_publish_denied(self): + d = has_permission(self.eng, Resource.RELEASE, Action.PUBLISH) + self.assertFalse(d.allow) + + def test_admin_publish_allowed(self): + d = has_permission(self.admin, Resource.RELEASE, Action.PUBLISH) + self.assertTrue(d.allow) + self.assertEqual(d.source, "explicit") + + def test_admin_inherited_engineer_edit(self): + d = has_permission(self.admin, Resource.MODEL_PARAM, Action.EDIT) + self.assertTrue(d.allow) + self.assertEqual(d.source, "inherited") + + def test_decision_carries_reason(self): + # 可解释性：无论允许/拒绝，reason 非空且含用户名与资源 + for user in (self.ro, self.eng, self.admin): + d = has_permission(user, Resource.PUSH, Action.PUBLISH) + self.assertIn(user.username, d.reason) + self.assertIn(Resource.PUSH.value, d.reason) + + +class RestrictedUserTest(unittest.TestCase): + """细粒度收窄：restricted_to_readonly。""" + + def test_restricted_engineer_cannot_edit_that_resource(self): + # 工程师本可编辑布局，但被收窄为只读后应拒绝 + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + d = has_permission(u, Resource.LAYOUT, Action.EDIT) + self.assertFalse(d.allow) + self.assertEqual(d.source, "restricted") + + def test_restricted_engineer_can_still_view(self): + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + d = has_permission(u, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d.allow) + + def test_restricted_only_affects_named_resource(self): + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + # 模型超参未被收窄，仍可编辑 + d = has_permission(u, Resource.MODEL_PARAM, Action.EDIT) + self.assertTrue(d.allow) + + +class ConvenienceTest(unittest.TestCase): + """便捷判定与用户概览。""" + + def test_can_publish(self): + self.assertFalse(can_publish(User("v", RoleKind.READONLY))) + self.assertFalse(can_publish(User("e", RoleKind.ENGINEER))) + self.assertTrue(can_publish(User("a", RoleKind.ADMIN))) + + def test_user_summary(self): + s = user_summary(User("li", RoleKind.ENGINEER, "李工")) + self.assertEqual(s["username"], "li") + self.assertEqual(s["role"], "engineer") + self.assertEqual(s["role_label"], "行业工程师") + self.assertGreater(s["effective_permission_count"], 0) + self.assertEqual(s["restricted_to_readonly"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/templates/ti-cl4/dashboard/README.md b/templates/ti-cl4/dashboard/README.md new file mode 100644 index 0000000..20e9be0 --- /dev/null +++ b/templates/ti-cl4/dashboard/README.md @@ -0,0 +1,37 @@ +# 海绵钛驾驶舱布局资产 + 校验器（Issue #55 / PRD 5.5） + +> 父 Issue「⑤ Ti 行业布局模板（四状态流程视图）· 0.5d」 + +把海绵钛车间驾驶舱布局落为**对齐 iAOP-cockpit-layout-v1 的资产 + 可校验的纯标准库 +校验器**（无 node/前端构建环境，零运行时依赖）。 + +## 资产：`cockpit.ti.yaml` + +四状态工艺流程（PRD 4.2 海绵钛：**氯化 → 精制 → 还原 → 蒸馏**）： + +- `process_view` 主视图（首屏立即加载），`stages` 声明四状态覆盖； +- `trend` 实时趋势（氯化炉温度 `CLF-01.TEMP` / 氯气流量 `CLF-01.CL2`）； +- `kpi_card` KPI（TiCl₄纯度 `RF-01.PURITY` / 杂质 `RF-01.IMP` / 电耗 `E-01.KWH` / 蒸汽 `ST-01.STEAM`）； +- `alarm_panel` 告警面板 + `nl_query` NL 查询入口。 + +所有 `bind` 的 `point_id` 对齐 `templates/ti-cl4/point-dict/point_dict.default.csv`。 + +## 校验器：`layout_validator.py` + +`LayoutValidator(layout_yaml, point_dict_csv).validate()` → `LayoutReport`，校验： + +1. **widget 类型合法**：在 `iAOP-cockpit-layout-v1` 允许集合内（process_view/trend/kpi_card/alarm_panel/nl_query）； +2. **12 列网格不越界**：`0 ≤ x`、`x + w ≤ 12`、`y ≥ 0`、`w/h > 0`； +3. **bind point_id 在点位字典内**：防模板漂移（trend/kpi_card 的 bind 必须命中点字典）； +4. **四状态覆盖完整**：process_view 的 stages 必须覆盖氯化/精制/还原/蒸馏，order 单调递增、id 唯一。 + +零依赖 YAML 子集解析（复制 impurity-forecast 的 `_parse_yaml_subset`，无 pyyaml）。 + +## 测试 + +```bash +python -m unittest discover -s templates/ti-cl4/dashboard/tests -p "test_*.py" -v +``` + +覆盖正常 + 边界 + 错误（18 用例）：真实资产端到端通过、非法类型、网格越界/负坐标/零宽、 +bind 漂移、缺 process_view、缺必需状态、order 非单调、stage 重复、$schema 头、CSV 加载、空布局。 diff --git a/templates/ti-cl4/dashboard/__init__.py b/templates/ti-cl4/dashboard/__init__.py new file mode 100644 index 0000000..15ee51a --- /dev/null +++ b/templates/ti-cl4/dashboard/__init__.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +"""海绵钛驾驶舱布局资产 + 校验器包（Issue #55）。 + +对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）： +四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏）。 +""" +from .layout_validator import ( + LayoutError, + LayoutIssue, + LayoutReport, + LayoutValidator, + WidgetSpec, + ALLOWED_WIDGET_TYPES, + GRID_COLUMNS, + REQUIRED_STAGES, +) + +__all__ = [ + "LayoutError", + "LayoutIssue", + "LayoutReport", + "LayoutValidator", + "WidgetSpec", + "ALLOWED_WIDGET_TYPES", + "GRID_COLUMNS", + "REQUIRED_STAGES", +] diff --git a/templates/ti-cl4/dashboard/_sanity_check.py b/templates/ti-cl4/dashboard/_sanity_check.py new file mode 100644 index 0000000..a7bf846 --- /dev/null +++ b/templates/ti-cl4/dashboard/_sanity_check.py @@ -0,0 +1,33 @@ +# -*- coding: utf-8 -*- +"""海绵钛驾驶舱布局冒烟脚本（Issue #55）。 + +直接运行 ``python _sanity_check.py`` 验证：cockpit.ti.yaml + point_dict.default.csv +端到端校验通过（widget 类型/网格/bind/四状态全覆盖）。零第三方依赖。 +""" +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) + +from layout_validator import LayoutValidator # noqa: E402 + +LAYOUT_YAML = os.path.join(HERE, "cockpit.ti.yaml") +POINT_DICT_CSV = os.path.join( + HERE, os.pardir, "point-dict", "point_dict.default.csv") + + +def main() -> int: + report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate() + if not report.passed: + print("FAIL") + for issue in report.errors: + print(f" - [{issue.widget_id}] {issue.field}: {issue.reason}") + return 1 + print(f"OK: {report.widget_count} widgets，" + f"类型/网格/bind/四状态校验通过") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/templates/ti-cl4/dashboard/cockpit.ti.yaml b/templates/ti-cl4/dashboard/cockpit.ti.yaml new file mode 100644 index 0000000..adcf963 --- /dev/null +++ b/templates/ti-cl4/dashboard/cockpit.ti.yaml @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +# 海绵钛（Ti）车间驾驶舱布局资产（iAOP-Template-Ti，EPIC #12）。 +# +# 对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）： +# 切换行业模板后，驾驶舱按本布局自动重排，无需改前端代码。 +# widget 类型：process_view（工艺流程视图）/ trend（实时趋势）/ kpi_card（KPI卡片）/ +# alarm_panel（告警面板）/ nl_query（NL查询入口）。 +# +# 四状态工艺流程（PRD 4.2 海绵钛：氯化 → 精制 → 还原 → 蒸馏）： +# - 氯化 （CLF-01）：TiO₂ + Cl₂ + C → TiCl₄（沸腾氯化炉） +# - 精制 ：粗 TiCl₄ → 精 TiCl₄（除钒/除硅，常压精馏） +# - 还原 （RF-01）：TiCl₄ + Mg → 海绵钛（真空还原，Kroll 法） +# - 蒸馏 ：海绵钛 + 残余 Mg/MgCl₂ 分离（真空蒸馏） +# process_view 的 stages 字段声明四状态覆盖，layout_validator 校验完整性。 +# +# bind 的 point_id 对齐 templates/ti-cl4/point-dict/point_dict.default.csv。 +$schema: iAOP-cockpit-layout-v1 +title: 海绵钛车间驾驶舱 +theme: dark +widgets: + # ---- 四状态工艺流程主视图（首屏立即加载） --------------------------- + - type: process_view + src: ti_four_state.svg + x: 0 + y: 0 + w: 12 + h: 4 + description: 四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏） + stages: + - id: chlorination + name: 氯化 + device: CLF-01 + order: 1 + - id: purification + name: 精制 + order: 2 + - id: reduction + name: 还原 + device: RF-01 + order: 3 + - id: distillation + name: 蒸馏 + order: 4 + # ---- 实时趋势：氯化炉温度/氯气流量（工艺核心监控） ------------------- + - type: trend + bind: CLF-01.TEMP + x: 0 + y: 4 + w: 6 + h: 2 + description: 氯化炉温度实时趋势（沸腾氯化炉温 850±50℃） + - type: trend + bind: CLF-01.CL2 + x: 6 + y: 4 + w: 6 + h: 2 + description: 氯气流量实时趋势（流态化监控） + # ---- KPI 卡片：还原质量/能耗（海绵钛核心指标） ----------------------- + - type: kpi_card + metric: ticl4_purity + bind: RF-01.PURITY + label: TiCl₄纯度 + x: 0 + y: 6 + w: 3 + h: 2 + description: 还原 TiCl₄ 纯度（%，工艺 ≥ 99.9%） + - type: kpi_card + metric: ticl4_impurity + bind: RF-01.IMP + label: 杂质含量 + x: 3 + y: 6 + w: 3 + h: 2 + description: 还原杂质含量（%，越低越好） + - type: kpi_card + metric: energy_per_ton + bind: E-01.KWH + label: 累计电耗 + x: 6 + y: 6 + w: 3 + h: 2 + description: 车间累计电耗（kWh，单吨海绵钛综合能耗输入） + - type: kpi_card + metric: steam_flow + bind: ST-01.STEAM + label: 蒸汽流量 + x: 9 + y: 6 + w: 3 + h: 2 + description: 蒸汽流量（t/h，公用工程监控） + # ---- 告警面板 + NL 查询入口 ------------------------------------------ + - type: alarm_panel + x: 0 + y: 8 + w: 9 + h: 3 + description: 告警面板（氯化炉温/还原真空度/纯度/杂质异常） + - type: nl_query + x: 9 + y: 8 + w: 3 + h: 3 + description: 自然语言查询入口（工艺/质量/能耗问答） diff --git a/templates/ti-cl4/dashboard/layout_validator.py b/templates/ti-cl4/dashboard/layout_validator.py new file mode 100644 index 0000000..a48a86c --- /dev/null +++ b/templates/ti-cl4/dashboard/layout_validator.py @@ -0,0 +1,500 @@ +# -*- coding: utf-8 -*- +"""海绵钛驾驶舱布局校验器（Issue #55 / PRD 5.5「⑤ 配置化驾驶舱」）。 + +PRD 5.5：切换行业模板后，驾驶舱按布局资产自动重排，无需改前端代码。 +本模块把布局资产（``cockpit.ti.yaml``）落为**可校验的纯标准库资产 + 校验器**—— +给定布局 YAML + 点位字典 CSV，校验： + +1. **widget 类型合法**：在 PRD 5.5 ``iAOP-cockpit-layout-v1`` 允许集合内 + （process_view/trend/kpi_card/alarm_panel/nl_query）。 +2. **12 列网格不越界**：每个 widget ``0 ≤ x`` 且 ``x + w ≤ 12``，``y ≥ 0``、 + ``h > 0``；坐标为非负整数，w/h 正整数（网格对齐）。 +3. **bind 的 point_id 在点位字典内**：trend/kpi_card 的 ``bind`` 必须命中 + ``point_dict.default.csv`` 的 ``point_id`` 列（防模板漂移）。 +4. **四状态视图覆盖完整**：process_view 的 ``stages`` 必须覆盖工艺全流程 + （氯化/精制/还原/蒸馏），且 order 单调递增、id 唯一。 + +校验产出 :class:`LayoutReport`（PASS/FAIL + 逐条 :class:`LayoutIssue`， +每条 issue 带 ``reason`` 可解释）。 + +设计要点 +-------- +- **零依赖 YAML 子集解析**：复制 impurity-forecast features.py 的 + ``_parse_yaml_subset``（无 pyyaml），支持 map/list/标量/行内 flow map。 +- **纯标准库**：CSV 用标准库 csv，无 numpy/pyyaml 依赖。 +- **换行业只改资产**：校验器对任何对齐 ``iAOP-cockpit-layout-v1`` 的布局都适用。 + +用法:: + + report = LayoutValidator(layout_yaml, point_dict_csv).validate() + if not report.passed: + for issue in report.issues: + print(issue.severity, issue.widget_id, issue.reason) +""" +from __future__ import annotations + +import csv +import os +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Tuple + +#: iAOP-cockpit-layout-v1 允许的 widget 类型集合（对齐 resin _sanity_check）。 +ALLOWED_WIDGET_TYPES = frozenset({ + "process_view", "trend", "kpi_card", "alarm_panel", "nl_query", +}) + +#: 12 列网格（主流前端栅格标准，对齐 cockpit layout v1）。 +GRID_COLUMNS = 12 + +#: 海绵钛四状态工艺流程（PRD 4.2：氯化 → 精制 → 还原 → 蒸馏）。 +#: process_view 的 stages 必须覆盖这四个 id。 +REQUIRED_STAGES = ("chlorination", "purification", "reduction", "distillation") + + +class LayoutError(ValueError): + """布局资产解析/声明错误（YAML 格式错、表头缺字段等）。""" + + +class Severity(str, Enum): + """问题严重度。""" + + ERROR = "error" # 阻断：布局不可用（类型非法/越界/bind 缺失/状态缺失） + WARN = "warn" # 告警：可运行但不规范（重复/顺序乱） + + +@dataclass +class LayoutIssue: + """单条布局校验问题（含 reason 可解释）。""" + + severity: Severity + reason: str + widget_id: str = "" # 关联 widget（index 或 src/metric） + field: str = "" # 关联字段（type/x/bind/stages ...） + + @property + def is_error(self) -> bool: + return self.severity is Severity.ERROR + + +@dataclass +class LayoutReport: + """布局校验报告。""" + + issues: List[LayoutIssue] = field(default_factory=list) + widget_count: int = 0 + + @property + def errors(self) -> List[LayoutIssue]: + return [i for i in self.issues if i.is_error] + + @property + def passed(self) -> bool: + """通过 = 无 ERROR（WARN 不阻断）。""" + return not any(i.is_error for i in self.issues) + + def to_dict(self) -> dict: + return { + "passed": self.passed, + "widget_count": self.widget_count, + "error_count": len(self.errors), + "warn_count": len(self.issues) - len(self.errors), + "issues": [ + {"severity": i.severity.value, "widget_id": i.widget_id, + "field": i.field, "reason": i.reason} + for i in self.issues + ], + } + + +@dataclass +class WidgetSpec: + """单个 widget 的内存模型（从 YAML 解析）。""" + + index: int # 在 widgets 列表中的位置（0 起） + type: str + x: int = 0 + y: int = 0 + w: int = 1 + h: int = 1 + bind: str = "" # trend/kpi_card 绑定的 point_id + src: str = "" # process_view 的 SVG + metric: str = "" # kpi_card 的 metric + label: str = "" + description: str = "" + stages: List[Dict[str, object]] = field(default_factory=list) + + +# --------------------------------------------------------------------------- +# 零依赖 YAML 子集解析（复制自 impurity-forecast features.py，对齐 data-bus） +# --------------------------------------------------------------------------- + +def _parse_scalar(text: str) -> str: + """去掉标量两侧引号与行内注释。""" + t = text.split(" #", 1)[0].strip() + if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'): + return t[1:-1] + return t + + +def _parse_flow_value(text: str): + """解析 ``key: value`` 右侧值，支持行内 flow map ``{k: v, k: v}``。""" + t = text.split(" #", 1)[0].strip() + if t.startswith("{") and t.endswith("}"): + inner = t[1:-1].strip() + out: Dict[str, object] = {} + if not inner: + return out + for part in inner.split(","): + if ":" not in part: + raise LayoutError(f"flow map 项不是键值对：{part!r}") + k, _, v = part.partition(":") + out[k.strip()] = _parse_scalar(v) + return out + return _parse_scalar(text) + + +def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]: + out: List[Tuple[str, int]] = [] + for i, ln in enumerate(lines): + s = ln.strip() + if not s or s.startswith("#"): + continue + out.append((ln, i + 1)) + return out + + +def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int): + """递归解析 YAML 节点（map / list / scalar）。返回 (value, next_i)。""" + text, _ = lines[i] + # ---- list 节点 ---- + if text.lstrip(" ").startswith("- "): + items: List[object] = [] + while i < len(lines): + t, no = lines[i] + stripped = t.lstrip(" ") + if not stripped.startswith("- "): + break + lead_j = len(t) - len(t.lstrip(" ")) + if lead_j != indent: + break + item_text = stripped[2:].strip() + if not item_text: + raise LayoutError(f"cockpit.yaml 第 {no} 行：list 项为空") + if ":" in item_text: + map_indent = len(t) - len(t.lstrip(" ")) + 2 + lines[i] = (" " * map_indent + item_text, no) + v, i = _parse_node(lines, i, map_indent) + items.append(v) + else: + items.append(_parse_flow_value(item_text)) + i += 1 + return items, i + # ---- map 节点 ---- + result: Dict[str, object] = {} + while i < len(lines): + t, no = lines[i] + lead_j = len(t) - len(t.lstrip(" ")) + if lead_j < indent or t.lstrip(" ").startswith("- "): + break + if lead_j > indent: + raise LayoutError( + f"cockpit.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）") + if ":" not in t: + raise LayoutError(f"cockpit.yaml 第 {no} 行不是合法键值对：{t!r}") + key, _, rest = t.partition(":") + key = key.strip() + rest = rest.strip() + if rest: + result[key] = _parse_flow_value(rest) + i += 1 + continue + if i + 1 >= len(lines): + raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值") + sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" ")) + if sub_indent <= indent: + raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值（无嵌套）") + v, i = _parse_node(lines, i + 1, sub_indent) + result[key] = v + return result, i + + +def _load_yaml_text(text: str) -> Dict[str, object]: + """解析 YAML 文本为 dict（顶层必须是 map）。""" + lines = _strip_comments(text.splitlines()) + if not lines: + return {} + top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" ")) + value, next_i = _parse_node(lines, 0, top_indent) + if not isinstance(value, dict): + raise LayoutError("cockpit.yaml 顶层必须是 map") + if next_i < len(lines): + raise LayoutError( + f"cockpit.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点") + return value + + +# --------------------------------------------------------------------------- +# 点位字典加载（CSV → point_id 集合） +# --------------------------------------------------------------------------- + +def load_point_ids(csv_path: str) -> List[str]: + """从点位字典 CSV 加载全部 point_id（保序，对齐 CSV point_id 列）。 + + CSV 表头对齐 core/edge-gateway point_dict schema（第二列 point_id）。 + """ + if not os.path.isfile(csv_path): + raise LayoutError(f"点位字典 CSV 不存在：{csv_path}") + with open(csv_path, "r", encoding="utf-8") as fh: + rows = list(csv.reader(fh)) + if not rows: + raise LayoutError(f"点位字典 CSV 为空：{csv_path}") + header = [c.strip() for c in rows[0]] + if "point_id" not in header: + raise LayoutError( + f"点位字典 CSV 表头缺 point_id 列：{header}") + col = header.index("point_id") + ids: List[str] = [] + for i, row in enumerate(rows[1:], 2): + if len(row) <= col: + continue + pid = row[col].strip() + if pid: + ids.append(pid) + if not ids: + raise LayoutError(f"点位字典 CSV 无 point_id 数据行：{csv_path}") + return ids + + +# --------------------------------------------------------------------------- +# 校验器 +# --------------------------------------------------------------------------- + +class LayoutValidator: + """海绵钛驾驶舱布局校验器。 + + Args: + layout_yaml_path: 布局资产路径（cockpit.ti.yaml）。 + point_dict_csv_path: 点位字典 CSV 路径（point_dict.default.csv）。 + grid_columns: 网格列数（默认 12，对齐 cockpit layout v1）。 + required_stages: process_view 必须覆盖的 stage id（默认海绵钛四状态）。 + """ + + def __init__( + self, + layout_yaml_path: str, + point_dict_csv_path: Optional[str] = None, + grid_columns: int = GRID_COLUMNS, + required_stages: Tuple[str, ...] = REQUIRED_STAGES, + ) -> None: + if grid_columns <= 0: + raise LayoutError(f"grid_columns 必须 > 0，实际 {grid_columns}") + self.layout_path = layout_yaml_path + self.point_dict_path = point_dict_csv_path + self.grid_columns = int(grid_columns) + self.required_stages = tuple(required_stages) + + # ------------------------------------------------------------------ + def validate(self) -> LayoutReport: + """执行全部校验，返回报告。""" + report = LayoutReport() + # 1) 解析布局 YAML + try: + with open(self.layout_path, "r", encoding="utf-8") as fh: + data = _load_yaml_text(fh.read()) + except LayoutError: + raise + except OSError as exc: + raise LayoutError(f"布局 YAML 读取失败：{self.layout_path} ({exc})") from exc + + # schema 头校验 + schema = str(data.get("$schema", "")).strip() + if schema != "iAOP-cockpit-layout-v1": + report.issues.append(LayoutIssue( + severity=Severity.ERROR, + field="$schema", + reason=f"$schema 应为 'iAOP-cockpit-layout-v1'，实际 {schema!r}", + )) + + # 2) 解析 widgets + raw_widgets = data.get("widgets") or [] + if not isinstance(raw_widgets, list): + report.issues.append(LayoutIssue( + severity=Severity.ERROR, field="widgets", + reason=f"widgets 必须是 list，实际 {type(raw_widgets).__name__}")) + return report + widgets = self._parse_widgets(raw_widgets, report) + report.widget_count = len(widgets) + + if not widgets: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, field="widgets", + reason="布局无任何 widget")) + return report + + # 3) 加载点位字典（bind 校验需要） + point_ids: Optional[set] = None + if self.point_dict_path: + try: + point_ids = set(load_point_ids(self.point_dict_path)) + except LayoutError as exc: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, field="point_dict", + reason=str(exc))) + + # 4) 逐 widget 校验 + for w in widgets: + self._check_widget(w, point_ids, report) + + # 5) process_view 四状态覆盖 + self._check_process_views(widgets, report) + + return report + + # ------------------------------------------------------------------ + def _parse_widgets(self, raw_widgets: List[object], + report: LayoutReport) -> List[WidgetSpec]: + widgets: List[WidgetSpec] = [] + for idx, item in enumerate(raw_widgets): + if not isinstance(item, dict): + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=f"[{idx}]", + field="widgets", + reason=f"widgets[{idx}] 必须是 map，实际 {type(item).__name__}")) + continue + wtype = str(item.get("type", "")).strip() + widgets.append(WidgetSpec( + index=idx, + type=wtype, + x=_to_int(item.get("x"), 0), + y=_to_int(item.get("y"), 0), + w=_to_int(item.get("w"), 1), + h=_to_int(item.get("h"), 1), + bind=str(item.get("bind", "")).strip(), + src=str(item.get("src", "")).strip(), + metric=str(item.get("metric", "")).strip(), + label=str(item.get("label", "")).strip(), + description=str(item.get("description", "")).strip(), + stages=_as_list_of_dict(item.get("stages")), + )) + return widgets + + # ------------------------------------------------------------------ + def _check_widget(self, w: WidgetSpec, point_ids: Optional[set], + report: LayoutReport) -> None: + wid = f"[{w.index}]({w.type})" + # 4a) widget 类型合法 + if w.type not in ALLOWED_WIDGET_TYPES: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="type", + reason=f"非法 widget 类型 {w.type!r}（允许 {sorted(ALLOWED_WIDGET_TYPES)})")) + + # 4b) 12 列网格不越界（坐标非负整数、x+w ≤ columns、h>0） + if w.x < 0 or w.y < 0: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="grid", + reason=f"坐标不能为负：x={w.x} y={w.y}")) + if w.w <= 0 or w.h <= 0: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="grid", + reason=f"w/h 必须为正整数：w={w.w} h={w.h}")) + if w.x + w.w > self.grid_columns: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="grid", + reason=f"越出 {self.grid_columns} 列网格：x={w.x}+w={w.w}" + f"={w.x + w.w} > {self.grid_columns}")) + + # 4c) bind 的 point_id 必须在点位字典内（trend/kpi_card） + if w.bind: + if point_ids is not None and w.bind not in point_ids: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="bind", + reason=f"bind point_id {w.bind!r} 不在点位字典内" + f"（防模板漂移，对齐 point_dict.default.csv）")) + + # ------------------------------------------------------------------ + def _check_process_views(self, widgets: List[WidgetSpec], + report: LayoutReport) -> None: + """校验 process_view 的四状态覆盖完整。""" + pv = [w for w in widgets if w.type == "process_view"] + if not pv: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, field="process_view", + reason="布局缺少 process_view（四状态工艺流程主视图必需）")) + return + + covered: Dict[str, WidgetSpec] = {} # stage_id → widget + for w in pv: + wid = f"[{w.index}](process_view)" + if not w.stages: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="stages", + reason="process_view 缺少 stages 声明（四状态覆盖必需）")) + continue + + stage_ids: List[str] = [] + orders: List[int] = [] + seen: set = set() + for st in w.stages: + sid = str(st.get("id", "")).strip() + sname = str(st.get("name", "")).strip() + if not sid: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="stages", + reason=f"stage 缺少 id（name={sname!r}）")) + continue + if sid in seen: + report.issues.append(LayoutIssue( + severity=Severity.WARN, widget_id=wid, field="stages", + reason=f"stage id 重复：{sid!r}")) + continue + seen.add(sid) + stage_ids.append(sid) + covered.setdefault(sid, w) + order = st.get("order") + if order is not None: + try: + orders.append(int(order)) + except (TypeError, ValueError): + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="stages", + reason=f"stage {sid!r} order 非整数：{order!r}")) + + # order 单调递增校验 + if orders and len(orders) == len(stage_ids): + if orders != sorted(orders): + report.issues.append(LayoutIssue( + severity=Severity.ERROR, widget_id=wid, field="stages", + reason=f"stage order 非单调递增：{orders}")) + + # 必需四状态全覆盖 + missing = [s for s in self.required_stages if s not in covered] + if missing: + report.issues.append(LayoutIssue( + severity=Severity.ERROR, field="stages", + reason=f"process_view stages 未覆盖必需四状态：{missing}" + f"（氯化/精制/还原/蒸馏）")) + + +# --------------------------------------------------------------------------- +# 辅助 +# --------------------------------------------------------------------------- + +def _to_int(value: object, default: int) -> int: + """把 YAML 解析出的值（可能是 str/int）转为 int；失败返回 default。""" + if value is None or value == "": + return default + try: + return int(value) + except (TypeError, ValueError): + raise LayoutError(f"坐标值不是整数：{value!r}") + + +def _as_list_of_dict(value: object) -> List[Dict[str, object]]: + if not isinstance(value, list): + return [] + out: List[Dict[str, object]] = [] + for item in value: + if isinstance(item, dict): + out.append(item) + return out diff --git a/templates/ti-cl4/dashboard/tests/_bootstrap.py b/templates/ti-cl4/dashboard/tests/_bootstrap.py new file mode 100644 index 0000000..8de360c --- /dev/null +++ b/templates/ti-cl4/dashboard/tests/_bootstrap.py @@ -0,0 +1,13 @@ +# -*- coding: utf-8 -*- +"""测试引导：把 dashboard 测试根目录加入 sys.path，使 layout_validator 可导入。 + +dashboard 目录名是合法 Python 标识符，直接作为包导入；本引导把父目录 +（templates/ti-cl4/dashboard）挂到 sys.path，使 ``from layout_validator import ...`` +在 unittest 发现机制下可解析（与 core 模块测试引导同款）。 +""" +import os +import sys + +PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if PKG_DIR not in sys.path: + sys.path.insert(0, PKG_DIR) diff --git a/templates/ti-cl4/dashboard/tests/test_layout_validator.py b/templates/ti-cl4/dashboard/tests/test_layout_validator.py new file mode 100644 index 0000000..c9c56f7 --- /dev/null +++ b/templates/ti-cl4/dashboard/tests/test_layout_validator.py @@ -0,0 +1,309 @@ +# -*- coding: utf-8 -*- +"""海绵钛驾驶舱布局校验器测试（Issue #55）。 + +覆盖： +1. 真实 cockpit.ti.yaml + point_dict.default.csv 全部通过（端到端）； +2. widget 类型合法集合（非法类型 → ERROR）； +3. 12 列网格校验（越界/负坐标/非正 w/h）； +4. bind point_id 在点位字典内（漂移 → ERROR）； +5. process_view 四状态覆盖（缺 stage id / order 非单调 / 缺必需状态）； +6. $schema 头校验； +7. YAML 解析（flow map / 嵌套）； +8. 点位字典 CSV 加载（缺表头/空文件）； +9. 空布局 / 边界。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 (sys.path 挂载) + +from layout_validator import ( + GRID_COLUMNS, + LayoutError, + LayoutValidator, + REQUIRED_STAGES, + Severity, + load_point_ids, +) + +HERE = os.path.dirname(os.path.abspath(__file__)) +DASHBOARD_DIR = os.path.dirname(HERE) +LAYOUT_YAML = os.path.join(DASHBOARD_DIR, "cockpit.ti.yaml") +POINT_DICT_CSV = os.path.join( + DASHBOARD_DIR, os.pardir, "point-dict", "point_dict.default.csv") + + +def _write_layout(tmp_path: str, content: str) -> str: + """把布局内容写到临时文件，返回路径。""" + path = os.path.join(tmp_path, "cockpit.test.yaml") + with open(path, "w", encoding="utf-8") as fh: + fh.write(content) + return path + + +def _write_point_dict(tmp_path: str, ids: list) -> str: + """写一个最小点位字典 CSV（仅 point_id 列）。""" + path = os.path.join(tmp_path, "points.csv") + with open(path, "w", encoding="utf-8") as fh: + fh.write("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\n") + for i, pid in enumerate(ids): + fh.write(f"D{i},{pid},n,u,float,1000,true,n,simulator\n") + return path + + +# 最小合法布局模板（便于构造各类变形） +_VALID_LAYOUT = """\ +$schema: iAOP-cockpit-layout-v1 +title: 测试驾驶舱 +theme: dark +widgets: + - type: process_view + src: ti_four_state.svg + x: 0 + y: 0 + w: 12 + h: 4 + description: 四状态工艺流程 + stages: + - id: chlorination + name: 氯化 + order: 1 + - id: purification + name: 精制 + order: 2 + - id: reduction + name: 还原 + order: 3 + - id: distillation + name: 蒸馏 + order: 4 + - type: trend + bind: CLF-01.TEMP + x: 0 + y: 4 + w: 6 + h: 2 + description: 氯化炉温度 +""" + + +class TestEndToEndRealAssets(unittest.TestCase): + """真实 cockpit.ti.yaml + point_dict.default.csv 端到端校验。""" + + def test_real_layout_passes(self): + report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate() + if not report.passed: + for issue in report.errors: + print("ERROR:", issue.widget_id, issue.field, issue.reason) + self.assertTrue(report.passed, "真实布局应通过全部校验") + self.assertGreater(report.widget_count, 0) + + def test_real_layout_has_process_view_with_four_stages(self): + report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate() + # 无 stages 相关 ERROR + stage_errors = [i for i in report.errors if i.field == "stages"] + self.assertEqual(stage_errors, []) + + +class TestWidgetType(unittest.TestCase): + """widget 类型合法性。""" + + def test_invalid_widget_type_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + layout = _VALID_LAYOUT.replace("type: trend", "type: radar_chart") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + errors = [i for i in report.errors if i.field == "type"] + self.assertEqual(len(errors), 1) + self.assertIn("非法 widget 类型", errors[0].reason) + + +class TestGridBounds(unittest.TestCase): + """12 列网格校验。""" + + def test_x_plus_w_exceeds_columns(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + # trend x=10 w=6 → 16 > 12 + layout = _VALID_LAYOUT.replace( + " bind: CLF-01.TEMP\n x: 0\n y: 4\n w: 6\n h: 2", + " bind: CLF-01.TEMP\n x: 10\n y: 4\n w: 6\n h: 2") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + grid_errors = [i for i in report.errors if i.field == "grid" + and "越出" in i.reason] + self.assertEqual(len(grid_errors), 1) + + def test_negative_x_rejected(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + layout = _VALID_LAYOUT.replace(" x: 0\n y: 4\n w: 6\n h: 2", + " x: -1\n y: 4\n w: 6\n h: 2") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + neg = [i for i in report.errors if "坐标不能为负" in i.reason] + self.assertEqual(len(neg), 1) + + def test_zero_width_rejected(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + layout = _VALID_LAYOUT.replace(" x: 0\n y: 4\n w: 6\n h: 2", + " x: 0\n y: 4\n w: 0\n h: 2") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + wh = [i for i in report.errors if "w/h 必须为正整数" in i.reason] + self.assertEqual(len(wh), 1) + + +class TestBindPointId(unittest.TestCase): + """bind point_id 在点位字典内。""" + + def test_bind_not_in_dict_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = _write_layout(td, _VALID_LAYOUT) + # 点位字典不含 CLF-01.TEMP + csv_path = _write_point_dict(td, ["OTHER-01.X"]) + report = LayoutValidator(path, csv_path).validate() + bind_err = [i for i in report.errors if i.field == "bind"] + self.assertEqual(len(bind_err), 1) + self.assertIn("不在点位字典内", bind_err[0].reason) + + def test_bind_in_dict_passes(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = _write_layout(td, _VALID_LAYOUT) + csv_path = _write_point_dict(td, ["CLF-01.TEMP"]) + report = LayoutValidator(path, csv_path).validate() + bind_err = [i for i in report.errors if i.field == "bind"] + self.assertEqual(bind_err, []) + + +class TestProcessViewStages(unittest.TestCase): + """process_view 四状态覆盖。""" + + def test_missing_process_view_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + # 删除 process_view 块（保留 trend） + layout = """\ +$schema: iAOP-cockpit-layout-v1 +title: t +widgets: + - type: trend + bind: CLF-01.TEMP + x: 0 + y: 0 + w: 6 + h: 2 +""" + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + pv_err = [i for i in report.errors if i.field == "process_view"] + self.assertEqual(len(pv_err), 1) + + def test_missing_required_stage_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + # 删除 distillation stage + layout = _VALID_LAYOUT.replace( + " - id: distillation\n name: 蒸馏\n order: 4\n", "") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + missing = [i for i in report.errors if "未覆盖必需四状态" in i.reason] + self.assertEqual(len(missing), 1) + self.assertIn("distillation", missing[0].reason) + + def test_non_monotonic_order_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + # 把 reduction order 改为 5（> distillation 的 4）→ 非单调 + layout = _VALID_LAYOUT.replace(" - id: reduction\n name: 还原\n order: 3", + " - id: reduction\n name: 还原\n order: 5") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + order_err = [i for i in report.errors if "order 非单调递增" in i.reason] + self.assertEqual(len(order_err), 1) + + def test_duplicate_stage_id_warn(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + # 重复 chlorination（覆盖必需状态校验仍过，但 WARN 重复） + layout = _VALID_LAYOUT + """\ +""" + # 构造一个有重复 stage 的 process_view（替换 stages 块） + dup_layout = _VALID_LAYOUT.replace( + " - id: distillation\n name: 蒸馏\n order: 4", + " - id: distillation\n name: 蒸馏\n order: 4\n" + " - id: chlorination\n name: 氯化2\n order: 5") + path = _write_layout(td, dup_layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + dup = [i for i in report.issues if "stage id 重复" in i.reason] + self.assertEqual(len(dup), 1) + + +class TestSchemaHeader(unittest.TestCase): + """$schema 头校验。""" + + def test_wrong_schema_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + layout = _VALID_LAYOUT.replace("iAOP-cockpit-layout-v1", "some-other-schema") + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + schema_err = [i for i in report.errors if i.field == "$schema"] + self.assertEqual(len(schema_err), 1) + + +class TestPointDictLoader(unittest.TestCase): + """点位字典 CSV 加载。""" + + def test_load_point_ids(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + csv_path = _write_point_dict(td, ["A.X", "B.Y"]) + ids = load_point_ids(csv_path) + self.assertEqual(ids, ["A.X", "B.Y"]) + + def test_missing_csv_raises(self): + with self.assertRaises(LayoutError): + load_point_ids("/nonexistent/points.csv") + + def test_csv_missing_point_id_column_raises(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = os.path.join(td, "bad.csv") + with open(path, "w", encoding="utf-8") as fh: + fh.write("device_id,name\nD1,n\n") + with self.assertRaises(LayoutError): + load_point_ids(path) + + +class TestReportExport(unittest.TestCase): + """报告序列化 + 边界。""" + + def test_empty_layout_error(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + layout = """\ +$schema: iAOP-cockpit-layout-v1 +title: t +widgets: [] +""" + path = _write_layout(td, layout) + report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate() + self.assertFalse(report.passed) + + def test_report_to_dict(self): + report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate() + d = report.to_dict() + self.assertEqual(d["passed"], True) + self.assertIn("widget_count", d) + self.assertEqual(d["error_count"], 0) + + +if __name__ == "__main__": + unittest.main()