bot_dev1
ace09a6d16
新增 core/cockpit 模块，定义 iAOP-cockpit-layout-v1 布局资产 schema 的 权威实现，对齐 PRD 5.5「⑤ 配置化驾驶舱」验收口径（切换模板零改码）。 交付内容： - core/cockpit/layout.py：布局字段规范、合法性集合（主题/组件类型）、 validate_layout 校验器（结构+语义）、load_layout 解析器、 CockpitLayout/Widget/Grid 内存模型（dataclass）、LayoutValidationError。 - core/cockpit/__init__.py：对外导出。 - core/cockpit/tests/test_layout.py：30 个单测（PRD 示例/树脂模板兼容/ 各类非法/性能提示/解析/round-trip/不变入参）。 - core/cockpit/scripts/verify_layout_schema.py：端到端验证脚本（exit=0）。 兼容性：现有 templates/resin/dashboard/cockpit.resin.yaml 引用的 $schema: iAOP-cockpit-layout-v1 与本实现一致，树脂模板资产校验通过。
379 lines
14 KiB
Python
379 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""驾驶舱布局 JSON Schema 定义 + 校验器 + 解析器（issue #50 / PRD 5.5）。
|
||
|
||
本文件是 ``$schema: iAOP-cockpit-layout-v1`` 的权威实现：定义布局资产的
|
||
字段规范、合法性集合，以及一份与现有行业模板（如
|
||
``templates/resin/dashboard/cockpit.resin.yaml``）完全兼容的校验/解析管线。
|
||
|
||
布局资产结构（PRD 5.5 示例）::
|
||
|
||
{
|
||
"$schema": "iAOP-cockpit-layout-v1",
|
||
"title": "氯化车间驾驶舱",
|
||
"theme": "dark",
|
||
"widgets": [
|
||
{"type": "process_view", "src": "ti_four_state.svg", "x":0,"y":0,"w":6,"h":4},
|
||
{"type": "trend", "bind": "CLF-01.TEMP", "x":6,"y":0,"w":6,"h":2},
|
||
{"type": "kpi_card", "metric": "Ti_purity", "x":6,"y":2,"w":3,"h":2},
|
||
{"type": "alarm_panel", "x":0,"y":4,"w":12,"h":3}
|
||
]
|
||
}
|
||
|
||
字段规范（PRD 5.5「布局 JSON Schema 定义」+「配置点」）:
|
||
$schema string 必填 布局版本标识，固定 ``iAOP-cockpit-layout-v1``。
|
||
title string 必填 驾驶舱标题（行业模板级，如"氯化车间驾驶舱"）。
|
||
theme enum 必填 主题：``dark`` / ``light``。
|
||
widgets list 必填 组件清单，至少 1 个；每个 widget 见下表。
|
||
grid object 选填 栅格基线（默认 12 列）；见 ``Grid``。
|
||
|
||
widget 字段:
|
||
type enum 必填 组件类型（见 ``VALID_WIDGET_TYPES``）。
|
||
x, y int 必填 栅格左上角坐标（≥0）。
|
||
w, h int 必填 宽/高（>0）。
|
||
description string 选填 组件说明（实施/行业工程师可读）。
|
||
src string 条件必填 ``process_view`` 必填，流程图 SVG 资源名。
|
||
bind string 条件必填 ``trend`` 必填，绑定的点位/指标 ID（如 CLF-01.TEMP）。
|
||
metric string 条件必填 ``kpi_card`` 必填，指标键（如 Ti_purity）。
|
||
label string 选填 ``kpi_card`` 展示标签。
|
||
|
||
复杂仪表盘性能（PRD 5.5）由渲染层负责：组件 ≥ 30 或数据点 ≥ 500 时启用
|
||
虚拟滚动 + 采样降频 + WebWorker，保证首屏 ≤ 2s。本 schema 不强制该阈值，
|
||
仅在 ``LayoutValidationResult`` 中提示组件数，便于上层决策。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 布局版本标识（被 templates/resin/dashboard/cockpit.resin.yaml 的 $schema 引用）
|
||
# ---------------------------------------------------------------------------
|
||
LAYOUT_SCHEMA_ID: str = "iAOP-cockpit-layout-v1"
|
||
LAYOUT_SCHEMA_VERSION: int = 1
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 合法性集合
|
||
# ---------------------------------------------------------------------------
|
||
# 主题（PRD 5.5「配置点：主题」）
|
||
VALID_THEMES: Tuple[str, ...] = ("dark", "light")
|
||
|
||
# 组件类型（PRD 5.5「能力：四状态工艺流程视图、实时趋势、KPI卡片、告警面板、NL查询入口」）
|
||
VALID_WIDGET_TYPES: Tuple[str, ...] = (
|
||
"process_view", # 四状态工艺流程视图
|
||
"trend", # 实时趋势
|
||
"kpi_card", # KPI 卡片
|
||
"alarm_panel", # 告警面板
|
||
"nl_query", # 自然语言查询入口
|
||
)
|
||
|
||
# 默认栅格基线（PRD 5.5 示例使用 12 列；resin 模板亦按 12 列布局）
|
||
DEFAULT_GRID_COLUMNS: int = 12
|
||
|
||
# 性能提示阈值（PRD 5.5「复杂仪表盘性能」）
|
||
PERF_WIDGET_THRESHOLD: int = 30
|
||
|
||
# 各组件类型必填的特有字段（type → 字段名）
|
||
_WIDGET_REQUIRED_FIELDS: Dict[str, Tuple[str, ...]] = {
|
||
"process_view": ("src",),
|
||
"trend": ("bind",),
|
||
"kpi_card": ("metric",),
|
||
"alarm_panel": (),
|
||
"nl_query": (),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 异常 / 结果
|
||
# ---------------------------------------------------------------------------
|
||
class LayoutValidationError(ValueError):
|
||
"""布局资产校验失败。``load_layout`` 在校验不通过时抛出。
|
||
|
||
``errors`` 收集全部字段级错误，便于配置台「错误列表（行号+原因）」展示。
|
||
"""
|
||
|
||
def __init__(self, errors: List[str]):
|
||
super().__init__("; ".join(errors) if errors else "layout validation failed")
|
||
self.errors: List[str] = list(errors)
|
||
|
||
|
||
@dataclass
|
||
class LayoutValidationResult:
|
||
"""``validate_layout`` 的返回值，区分「是否合法」与「全部错误清单」。"""
|
||
|
||
ok: bool
|
||
errors: List[str] = field(default_factory=list)
|
||
widget_count: int = 0
|
||
perf_hint: Optional[str] = None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内存模型（dataclass）
|
||
# ---------------------------------------------------------------------------
|
||
@dataclass
|
||
class Widget:
|
||
"""单个驾驶舱组件的内存模型。"""
|
||
|
||
type: str
|
||
x: int
|
||
y: int
|
||
w: int
|
||
h: int
|
||
description: Optional[str] = None
|
||
# 以下为按 type 选填/必填的特有字段，统一存放，解析时已校验存在性
|
||
src: Optional[str] = None
|
||
bind: Optional[str] = None
|
||
metric: Optional[str] = None
|
||
label: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""序列化回布局资产 dict（仅保留有值/必填字段，便于发布）。"""
|
||
d: Dict[str, Any] = {
|
||
"type": self.type,
|
||
"x": self.x,
|
||
"y": self.y,
|
||
"w": self.w,
|
||
"h": self.h,
|
||
}
|
||
if self.description is not None:
|
||
d["description"] = self.description
|
||
if self.src is not None:
|
||
d["src"] = self.src
|
||
if self.bind is not None:
|
||
d["bind"] = self.bind
|
||
if self.metric is not None:
|
||
d["metric"] = self.metric
|
||
if self.label is not None:
|
||
d["label"] = self.label
|
||
return d
|
||
|
||
|
||
@dataclass
|
||
class Grid:
|
||
"""栅格基线（默认 12 列，可由行业模板覆盖）。"""
|
||
|
||
columns: int = DEFAULT_GRID_COLUMNS
|
||
|
||
|
||
@dataclass
|
||
class CockpitLayout:
|
||
"""一份完整驾驶舱布局的内存模型。
|
||
|
||
渲染层（Vue3 / 配置台）仅消费本对象：切换行业模板 = 加载另一份
|
||
``CockpitLayout``，**前端代码零改动**即满足 PRD 5.5 验收口径。
|
||
"""
|
||
|
||
title: str
|
||
theme: str
|
||
widgets: List[Widget]
|
||
grid: Grid = field(default_factory=Grid)
|
||
schema: str = LAYOUT_SCHEMA_ID
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""序列化为可发布的布局资产 dict（结构对齐 PRD 5.5 示例）。"""
|
||
return {
|
||
"$schema": self.schema,
|
||
"title": self.title,
|
||
"theme": self.theme,
|
||
"grid": {"columns": self.grid.columns},
|
||
"widgets": [w.to_dict() for w in self.widgets],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 校验器
|
||
# ---------------------------------------------------------------------------
|
||
def _require_type(value: Any, name: str, expected: type, errors: List[str], ctx: str) -> None:
|
||
"""检查 ``value`` 是 ``expected`` 类型，失败则追加一条错误。"""
|
||
# bool 是 int 的子类，栅格坐标不应接受 bool；此处显式排除
|
||
if expected is int and isinstance(value, bool):
|
||
errors.append(f"{ctx}: {name} 必须是整数，实际为 bool")
|
||
return
|
||
if not isinstance(value, expected):
|
||
errors.append(f"{ctx}: {name} 必须是 {expected.__name__}，实际为 {type(value).__name__}")
|
||
|
||
|
||
def _validate_widget(widget: Any, index: int, grid_columns: int, errors: List[str]) -> None:
|
||
"""校验单个 widget dict，错误追加到 ``errors``。"""
|
||
ctx = f"widgets[{index}]"
|
||
if not isinstance(widget, dict):
|
||
errors.append(f"{ctx}: 组件必须是对象（dict）")
|
||
return
|
||
|
||
# type
|
||
wtype = widget.get("type")
|
||
if not isinstance(wtype, str) or not wtype:
|
||
errors.append(f"{ctx}: type 缺失或非字符串")
|
||
wtype = ""
|
||
elif wtype not in VALID_WIDGET_TYPES:
|
||
errors.append(
|
||
f"{ctx}: type '{wtype}' 非法，合法值 {list(VALID_WIDGET_TYPES)}"
|
||
)
|
||
|
||
# 栅格坐标 x/y/w/h（必填整数）
|
||
for key in ("x", "y", "w", "h"):
|
||
if key not in widget:
|
||
errors.append(f"{ctx}: 缺少必填字段 {key}")
|
||
else:
|
||
_require_type(widget[key], key, int, errors, ctx)
|
||
val = widget[key]
|
||
if isinstance(val, int) and not isinstance(val, bool):
|
||
if key in ("x", "y") and val < 0:
|
||
errors.append(f"{ctx}: {key} 必须 ≥ 0，实际 {val}")
|
||
if key in ("w", "h") and val <= 0:
|
||
errors.append(f"{ctx}: {key} 必须 > 0，实际 {val}")
|
||
# 越界：x + w 不应超过栅格列数（提示级，不阻断合法性，但记一条 warning 风格错误便于配置台纠正）
|
||
try:
|
||
x = widget["x"]
|
||
w = widget["w"]
|
||
if (
|
||
isinstance(x, int)
|
||
and isinstance(w, int)
|
||
and not isinstance(x, bool)
|
||
and not isinstance(w, bool)
|
||
and x + w > grid_columns
|
||
):
|
||
errors.append(
|
||
f"{ctx}: x+w={x + w} 超过栅格列数 {grid_columns}，布局会被压缩"
|
||
)
|
||
except (KeyError, TypeError):
|
||
pass
|
||
|
||
# type 特有必填字段
|
||
if wtype in _WIDGET_REQUIRED_FIELDS:
|
||
for fname in _WIDGET_REQUIRED_FIELDS[wtype]:
|
||
val = widget.get(fname)
|
||
if not isinstance(val, str) or not val:
|
||
errors.append(f"{ctx}: type='{wtype}' 要求字段 {fname} 非空字符串")
|
||
|
||
# kpi_card 可选 label（若有则必须字符串）
|
||
label = widget.get("label")
|
||
if label is not None and not isinstance(label, str):
|
||
errors.append(f"{ctx}: label 必须是字符串")
|
||
|
||
|
||
def validate_layout(data: Any) -> LayoutValidationResult:
|
||
"""对一份布局资产（已解析的 dict）做完整校验。
|
||
|
||
返回 ``LayoutValidationResult``：``ok`` 表示是否通过，``errors`` 收集全部
|
||
字段级错误，``widget_count`` / ``perf_hint`` 供上层做性能决策。
|
||
"""
|
||
errors: List[str] = []
|
||
|
||
if not isinstance(data, dict):
|
||
return LayoutValidationResult(ok=False, errors=["布局根必须是对象（dict）"])
|
||
|
||
# $schema
|
||
schema = data.get("$schema")
|
||
if schema is None:
|
||
errors.append("缺少 $schema 字段")
|
||
elif schema != LAYOUT_SCHEMA_ID:
|
||
errors.append(
|
||
f"$schema='{schema}' 不被支持，当前版本 '{LAYOUT_SCHEMA_ID}'"
|
||
)
|
||
|
||
# title
|
||
title = data.get("title")
|
||
if not isinstance(title, str) or not title.strip():
|
||
errors.append("title 缺失或为空字符串")
|
||
|
||
# theme
|
||
theme = data.get("theme")
|
||
if theme is None:
|
||
errors.append("缺少 theme 字段")
|
||
elif theme not in VALID_THEMES:
|
||
errors.append(f"theme='{theme}' 非法，合法值 {list(VALID_THEMES)}")
|
||
|
||
# grid（可选）
|
||
grid_columns = DEFAULT_GRID_COLUMNS
|
||
grid = data.get("grid")
|
||
if grid is not None:
|
||
if not isinstance(grid, dict):
|
||
errors.append("grid 必须是对象（dict）")
|
||
else:
|
||
cols = grid.get("columns")
|
||
if cols is None:
|
||
errors.append("grid.columns 缺失")
|
||
else:
|
||
_require_type(cols, "columns", int, errors, "grid")
|
||
if isinstance(cols, int) and not isinstance(cols, bool) and cols <= 0:
|
||
errors.append(f"grid.columns 必须 > 0，实际 {cols}")
|
||
grid_columns = cols
|
||
elif isinstance(cols, int) and not isinstance(cols, bool):
|
||
grid_columns = cols
|
||
|
||
# widgets
|
||
widgets = data.get("widgets")
|
||
if widgets is None:
|
||
errors.append("缺少 widgets 字段")
|
||
widgets = []
|
||
if not isinstance(widgets, list):
|
||
errors.append("widgets 必须是数组（list）")
|
||
widgets = []
|
||
elif len(widgets) == 0:
|
||
errors.append("widgets 不能为空（至少 1 个组件）")
|
||
|
||
for idx, w in enumerate(widgets):
|
||
_validate_widget(w, idx, grid_columns, errors)
|
||
|
||
widget_count = len(widgets) if isinstance(widgets, list) else 0
|
||
perf_hint = None
|
||
if widget_count >= PERF_WIDGET_THRESHOLD:
|
||
perf_hint = (
|
||
f"组件数 {widget_count} ≥ {PERF_WIDGET_THRESHOLD}，渲染层应启用"
|
||
"虚拟滚动 + 采样降频 + WebWorker（PRD 5.5）"
|
||
)
|
||
|
||
return LayoutValidationResult(
|
||
ok=len(errors) == 0,
|
||
errors=errors,
|
||
widget_count=widget_count,
|
||
perf_hint=perf_hint,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 解析器
|
||
# ---------------------------------------------------------------------------
|
||
def _coerce_int(v: Any) -> int:
|
||
"""已校验为 int（非 bool）后取值。"""
|
||
return v # type: ignore[return-value]
|
||
|
||
|
||
def load_layout(data: Dict[str, Any]) -> CockpitLayout:
|
||
"""从已解析的 dict 构造 ``CockpitLayout``；校验失败抛 ``LayoutValidationError``。
|
||
|
||
调用前通常先用 ``validate_layout`` 判断，但本方法内部仍会再校验一次，
|
||
确保构造出的内存模型恒为合法资产。
|
||
"""
|
||
result = validate_layout(data)
|
||
if not result.ok:
|
||
raise LayoutValidationError(result.errors)
|
||
|
||
grid_raw = data.get("grid") or {}
|
||
grid = Grid(columns=grid_raw.get("columns", DEFAULT_GRID_COLUMNS))
|
||
|
||
widgets: List[Widget] = []
|
||
for w in data["widgets"]:
|
||
widgets.append(
|
||
Widget(
|
||
type=w["type"],
|
||
x=_coerce_int(w["x"]),
|
||
y=_coerce_int(w["y"]),
|
||
w=_coerce_int(w["w"]),
|
||
h=_coerce_int(w["h"]),
|
||
description=w.get("description"),
|
||
src=w.get("src"),
|
||
bind=w.get("bind"),
|
||
metric=w.get("metric"),
|
||
label=w.get("label"),
|
||
)
|
||
)
|
||
|
||
return CockpitLayout(
|
||
title=data["title"],
|
||
theme=data["theme"],
|
||
widgets=widgets,
|
||
grid=grid,
|
||
schema=data.get("$schema", LAYOUT_SCHEMA_ID),
|
||
)
|