bot_dev1
42d206e503
新增 templates/ti-cl4/impurity-forecast：声明式特征工程引擎，特征以 FeatureSpec 描述（EMA/RollingStd/RateOfChange 等 7 算子），换行业只改模板配置 features.template.yaml，引擎零改动（PRD 5.3：特征工程层跨行业差异落在 FeatureSpec，不落代码）。 - features.py：FeatureSpec 声明 + 校验 + 7 算子 + 时序对齐 + 阈值 breach + 零依赖 YAML 解析 - config/features.template.yaml：炉温/氯气/炉压/炉层 9 条特征（对齐点位字典 point_id） - tests/test_features.py：24 项单测（校验/算子/对齐/breach/配置/端到端提前量）全通过 - _sanity_check.py：冒烟脚本（配置加载 + transform + breach 可观测） 验收：一期阈值+无监督上线，提前量信号可观测（PRD 提前≥30min、误报率≤8%口径）。
560 lines
22 KiB
Python
560 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""炉层杂质预警 · 声明式特征工程引擎（Issue #70 / PRD 5.3 异常·杂质预警）。
|
||
|
||
PRD 5.3 明确：**特征工程层的跨行业差异落在 FeatureSpec，不落代码**。
|
||
即特征以声明式规格描述（如 ``EMA(炉温, 5min)``、``RollingStd(氯气流量, 10)``、
|
||
``RateOfChange(炉压)``），由工艺模板定义、本引擎解释执行；换行业只改模板
|
||
配置（``config/features.template.yaml``），引擎零改动。
|
||
|
||
一期（Template-Ti）落地「炉层杂质预警」（PRD 5.3 ③，验收：提前 ≥ 30min、
|
||
误报率 ≤ 8%）。数据门槛低，先以**阈值 + 无监督**上线，3 个月后转监督
|
||
（PRD 4.1）；故本期特征工程面向无监督异常评分，同时产出监督可用的特征矩阵。
|
||
|
||
设计要点
|
||
--------
|
||
1. **声明式 FeatureSpec**：每条特征声明 ``kind``（算子）+ ``point``（来源测点）
|
||
+ ``params``（窗口/周期等），引擎按 kind 分派到内置算子。
|
||
2. **内置算子**（零第三方依赖，纯标准库）：
|
||
- ``raw`` 原始值透传；
|
||
- ``ema`` 指数滑动平均（平滑、去噪）；
|
||
- ``rolling_std`` 滚动标准差（波动度）；
|
||
- ``rate_of_change`` 变化率（速率预警）；
|
||
- ``rolling_mean`` 滚动均值；
|
||
- ``rolling_min`` / ``rolling_max`` 滚动极值（配合阈值）。
|
||
3. **时序对齐**：按时间戳对齐多测点为特征向量，缺失测点用 ``NaN`` 占位
|
||
并记录缺失率（误报率治理输入）。
|
||
4. **零依赖 YAML 子集解析**（与 data-bus / rag-kb 同款），解析模板资产。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Callable, Dict, List, Optional, Sequence, Tuple
|
||
|
||
# 缺失值统一用 float('nan')，便于上层用 math.isnan 判定与屏蔽。
|
||
NAN = float("nan")
|
||
|
||
|
||
class FeatureSpecError(ValueError):
|
||
"""FeatureSpec 声明或执行错误（未知算子 / 缺参 / 窗口非法等）。"""
|
||
|
||
|
||
class FeatureKind(str, Enum):
|
||
"""内置特征算子（声明式 FeatureSpec 的 ``kind`` 取值）。"""
|
||
|
||
RAW = "raw" # 原始值透传
|
||
EMA = "ema" # 指数滑动平均：params={"alpha": 0.2}
|
||
ROLLING_STD = "rolling_std" # 滚动标准差：params={"window": 10}
|
||
ROLLING_MEAN = "rolling_mean" # 滚动均值
|
||
ROLLING_MIN = "rolling_min" # 滚动最小值
|
||
ROLLING_MAX = "rolling_max" # 滚动最大值
|
||
RATE_OF_CHANGE = "rate_of_change" # 变化率：(x[t]-x[t-w])/x[t-w]
|
||
|
||
@property
|
||
def label(self) -> str:
|
||
return {
|
||
FeatureKind.RAW: "原始值",
|
||
FeatureKind.EMA: "指数滑动平均",
|
||
FeatureKind.ROLLING_STD: "滚动标准差",
|
||
FeatureKind.ROLLING_MEAN: "滚动均值",
|
||
FeatureKind.ROLLING_MIN: "滚动最小值",
|
||
FeatureKind.ROLLING_MAX: "滚动最大值",
|
||
FeatureKind.RATE_OF_CHANGE: "变化率",
|
||
}[self]
|
||
|
||
|
||
# 算子注册表：kind 名 → 算子实现。未知 kind 在注册阶段即拒绝（避免拼写漂移）。
|
||
KIND_REGISTRY: Dict[str, FeatureKind] = {k.value: k for k in FeatureKind}
|
||
|
||
|
||
@dataclass
|
||
class FeatureSpec:
|
||
"""单条声明式特征规格（模板配置中的一行特征声明）。
|
||
|
||
Attributes:
|
||
name: 特征输出名（特征向量列名，工艺可读，如 ``炉温_ema5``）。
|
||
kind: 算子（见 :class:`FeatureKind`）。
|
||
point: 来源测点 id（对齐点位字典 point_id，如 ``CLF-01.TEMP``）。
|
||
params: 算子参数（如 EMA 的 alpha、rolling_* 的 window）。
|
||
unit: 特征单位（可选，用于驾驶舱展示）。
|
||
threshold: 预警阈值（可选，无监督阈值上线的判定边界）。
|
||
"""
|
||
|
||
name: str
|
||
kind: FeatureKind
|
||
point: str
|
||
params: Dict[str, float] = field(default_factory=dict)
|
||
unit: str = ""
|
||
threshold: Optional[float] = None
|
||
|
||
def __post_init__(self) -> None:
|
||
if not self.name:
|
||
raise FeatureSpecError("FeatureSpec.name 不能为空")
|
||
if not self.point:
|
||
raise FeatureSpecError(f"特征 {self.name!r} 缺少 point（来源测点）")
|
||
# 参数合法性校验：滚动/变化率类必须有正整数 window
|
||
if self.kind in (FeatureKind.ROLLING_STD, FeatureKind.ROLLING_MEAN,
|
||
FeatureKind.ROLLING_MIN, FeatureKind.ROLLING_MAX,
|
||
FeatureKind.RATE_OF_CHANGE):
|
||
w = self.params.get("window")
|
||
if w is None:
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r}（{self.kind.value}）缺少 window 参数")
|
||
try:
|
||
wf = float(w)
|
||
except (TypeError, ValueError) as exc:
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r} window 必须是整数，实际 {w!r}") from exc
|
||
if wf != int(wf):
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r} window 必须是整数，实际 {w!r}")
|
||
wi = int(wf)
|
||
if wi <= 0:
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r} window 必须 > 0，实际 {wi}")
|
||
self.params["window"] = wi
|
||
if self.kind is FeatureKind.EMA:
|
||
alpha = self.params.get("alpha")
|
||
if alpha is None:
|
||
raise FeatureSpecError(f"特征 {self.name!r}（ema）缺少 alpha 参数")
|
||
try:
|
||
af = float(alpha)
|
||
except (TypeError, ValueError) as exc:
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r} alpha 必须是数值，实际 {alpha!r}") from exc
|
||
if not (0.0 < af <= 1.0):
|
||
raise FeatureSpecError(
|
||
f"特征 {self.name!r} alpha 须在 (0,1]，实际 {af}")
|
||
self.params["alpha"] = af
|
||
|
||
def describe(self) -> str:
|
||
"""工艺可读描述，如 ``炉温_ema5 = ema(CLF-01.TEMP, alpha=0.2)``。"""
|
||
pa = ", ".join(f"{k}={v}" for k, v in self.params.items())
|
||
return f"{self.name} = {self.kind.value}({self.point}{', ' + pa if pa else ''})"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算子实现：输入为按时间排序的标量序列（可能含 NAN），输出等长变换序列。
|
||
# 滚动窗口在序列前段（样本不足 window 个）输出 NAN，表示"尚不足以计算"。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _is_num(x: object) -> bool:
|
||
return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
|
||
|
||
|
||
def _rolling_window(values: Sequence[float], window: int,
|
||
reducer) -> List[float]:
|
||
"""通用滚动归约：前 window-1 个位置输出 NAN。"""
|
||
out: List[float] = []
|
||
buf: List[float] = []
|
||
for v in values:
|
||
if _is_num(v):
|
||
buf.append(float(v))
|
||
# 非数值视为缺失，不进缓冲区（窗口按"有效样本数"计数，更稳健）
|
||
if len(buf) >= window:
|
||
out.append(float(reducer(buf[-window:])))
|
||
else:
|
||
out.append(NAN)
|
||
return out
|
||
|
||
|
||
def _op_raw(values: Sequence[float], params: Dict[str, float]) -> List[float]:
|
||
return [float(v) if _is_num(v) else NAN for v in values]
|
||
|
||
|
||
def _op_ema(values: Sequence[float], params: Dict[str, float]) -> List[float]:
|
||
alpha = float(params["alpha"])
|
||
out: List[float] = []
|
||
prev: Optional[float] = None
|
||
for v in values:
|
||
if not _is_num(v):
|
||
out.append(NAN)
|
||
continue
|
||
x = float(v)
|
||
prev = x if prev is None else (alpha * x + (1.0 - alpha) * prev)
|
||
out.append(prev)
|
||
return out
|
||
|
||
|
||
def _op_rate_of_change(values: Sequence[float],
|
||
params: Dict[str, float]) -> List[float]:
|
||
window = int(params["window"])
|
||
out: List[float] = []
|
||
num: List[float] = []
|
||
for v in values:
|
||
if _is_num(v):
|
||
num.append(float(v))
|
||
if len(num) >= window + 1:
|
||
base = num[-(window + 1)]
|
||
cur = num[-1]
|
||
out.append((cur - base) / base if base else NAN)
|
||
else:
|
||
out.append(NAN)
|
||
return out
|
||
|
||
|
||
# kind → 算子函数 注册（FeatureEngine 分派用）
|
||
OPERATORS: Dict[FeatureKind, Callable[[Sequence[float], Dict[str, float]], List[float]]] = {
|
||
FeatureKind.RAW: _op_raw,
|
||
FeatureKind.EMA: _op_ema,
|
||
FeatureKind.ROLLING_STD: lambda v, p: _rolling_window(v, int(p["window"]),
|
||
lambda w: _std(w)),
|
||
FeatureKind.ROLLING_MEAN: lambda v, p: _rolling_window(v, int(p["window"]),
|
||
lambda w: sum(w) / len(w)),
|
||
FeatureKind.ROLLING_MIN: lambda v, p: _rolling_window(v, int(p["window"]), min),
|
||
FeatureKind.ROLLING_MAX: lambda v, p: _rolling_window(v, int(p["window"]), max),
|
||
FeatureKind.RATE_OF_CHANGE: _op_rate_of_change,
|
||
}
|
||
|
||
|
||
def _std(samples: Sequence[float]) -> float:
|
||
"""总体标准差（无监督波动度特征；零依赖实现）。"""
|
||
n = len(samples)
|
||
if n == 0:
|
||
return NAN
|
||
mean = sum(samples) / n
|
||
var = sum((x - mean) ** 2 for x in samples) / n
|
||
return math.sqrt(var)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 特征值 / 特征向量 / 引擎
|
||
# ---------------------------------------------------------------------------
|
||
|
||
FeatureValue = float # 单个特征值（可能为 NAN）
|
||
|
||
|
||
@dataclass
|
||
class FeatureVector:
|
||
"""某时刻对齐后的特征向量（多特征列 + 时间戳 + 缺失率）。"""
|
||
|
||
timestamp: float
|
||
values: Dict[str, FeatureValue] # name → 特征值
|
||
missing_rate: float = 0.0 # 本时刻缺失特征占比（误报率治理输入）
|
||
|
||
@property
|
||
def is_complete(self) -> bool:
|
||
"""所有特征均非缺失（监督训练样本需完整向量）。"""
|
||
return all(_is_num(v) for v in self.values.values())
|
||
|
||
def breach_features(self) -> List[str]:
|
||
"""返回超阈值 breach 的特征名（无监督阈值上线判定）。"""
|
||
# 阈值判定由 FeatureEngine 注入（见 engine.breach），这里仅占位。
|
||
return []
|
||
|
||
|
||
class FeatureEngine:
|
||
"""声明式特征工程引擎：解释 FeatureSpec 列表，对时序样本计算特征矩阵。
|
||
|
||
换行业只改模板配置（FeatureSpec 列表），引擎零改动（PRD 5.3）。
|
||
|
||
用法::
|
||
|
||
engine = FeatureEngine(specs)
|
||
vectors = engine.transform(samples)
|
||
for vec in vectors:
|
||
if vec.is_complete:
|
||
... # 喂给无监督评分器或监督训练
|
||
"""
|
||
|
||
def __init__(self, specs: Sequence[FeatureSpec]):
|
||
if not specs:
|
||
raise FeatureSpecError("FeatureEngine 至少需要一条 FeatureSpec")
|
||
# 同名特征直接拒绝（避免特征矩阵列冲突）
|
||
seen = set()
|
||
for s in specs:
|
||
if s.name in seen:
|
||
raise FeatureSpecError(f"特征名重复：{s.name!r}")
|
||
seen.add(s.name)
|
||
self.specs: List[FeatureSpec] = list(specs)
|
||
# 按来源测点聚合，减少重复取数
|
||
self._by_point: Dict[str, List[FeatureSpec]] = {}
|
||
for s in self.specs:
|
||
self._by_point.setdefault(s.point, []).append(s)
|
||
|
||
# -- 配置资产 ---------------------------------------------------------
|
||
|
||
@classmethod
|
||
def from_template_config(cls, path: str) -> "FeatureEngine":
|
||
"""从模板特征配置 YAML 资产构建引擎（零第三方依赖）。"""
|
||
return cls(load_feature_config(path).specs)
|
||
|
||
# -- 计算 -------------------------------------------------------------
|
||
|
||
def required_points(self) -> List[str]:
|
||
"""引擎依赖的全部来源测点 id（去重保序）。"""
|
||
seen, out = set(), []
|
||
for s in self.specs:
|
||
if s.point not in seen:
|
||
seen.add(s.point)
|
||
out.append(s.point)
|
||
return out
|
||
|
||
def transform(self, samples: Sequence[Dict[str, object]],
|
||
ts_key: str = "ts") -> List[FeatureVector]:
|
||
"""把时序样本流变换为按时间对齐的特征向量序列。
|
||
|
||
Args:
|
||
samples: 按 时间升序 排列的样本列表；每条样本是 ``{ts_key: epoch秒,
|
||
point_id: value, ...}`` 形态的 dict（对齐点位字典 point_id）。
|
||
ts_key: 时间戳键名（默认 ``ts``）。
|
||
|
||
Returns:
|
||
与 samples 等长的 FeatureVector 列表（按时间对齐）。
|
||
"""
|
||
if not samples:
|
||
return []
|
||
# 1) 按测点抽取时间序列（保持原顺序）
|
||
point_series: Dict[str, List[float]] = {p: [] for p in self._by_point}
|
||
timestamps: List[float] = []
|
||
for sample in samples:
|
||
ts = sample.get(ts_key)
|
||
try:
|
||
timestamps.append(float(ts) if ts is not None else NAN)
|
||
except (TypeError, ValueError):
|
||
timestamps.append(NAN)
|
||
for p in point_series:
|
||
v = sample.get(p)
|
||
point_series[p].append(float(v) if _is_num(v) else NAN)
|
||
|
||
# 2) 对每个测点的序列逐特征计算
|
||
# feature_columns[name] = 与时间等长的特征值序列
|
||
feature_columns: Dict[str, List[float]] = {}
|
||
for point, series in point_series.items():
|
||
for spec in self._by_point[point]:
|
||
op = OPERATORS.get(spec.kind)
|
||
if op is None: # 理论上 __post_init__ 已拦截，防御性
|
||
raise FeatureSpecError(f"未实现的算子 {spec.kind.value!r}")
|
||
feature_columns[spec.name] = op(series, spec.params)
|
||
|
||
# 3) 按时间戳对齐为特征向量
|
||
n = len(samples)
|
||
names = [s.name for s in self.specs]
|
||
vectors: List[FeatureVector] = []
|
||
for i in range(n):
|
||
row = {name: feature_columns[name][i] for name in names}
|
||
missing = sum(1 for v in row.values() if not _is_num(v))
|
||
vectors.append(FeatureVector(
|
||
timestamp=timestamps[i],
|
||
values=row,
|
||
missing_rate=missing / len(names) if names else 0.0,
|
||
))
|
||
return vectors
|
||
|
||
# -- 无监督阈值判定（一期上线口径） -----------------------------------
|
||
|
||
def breach(self, vector: FeatureVector) -> List[Tuple[str, float]]:
|
||
"""返回超阈值的 ``(特征名, 当前值)`` 列表（无监督阈值上线判定）。
|
||
|
||
一期 PRD 5.3 ③：阈值 + 无监督先上线；3 个月后转监督。本方法支持
|
||
FeatureSpec 声明的 ``threshold``（绝对值越界即 breach）。
|
||
"""
|
||
out: List[Tuple[str, float]] = []
|
||
spec_by_name = {s.name: s for s in self.specs}
|
||
for name, val in vector.values.items():
|
||
if not _is_num(val):
|
||
continue
|
||
spec = spec_by_name.get(name)
|
||
if spec is None or spec.threshold is None:
|
||
continue
|
||
if val > spec.threshold:
|
||
out.append((name, val))
|
||
return out
|
||
|
||
def describe(self) -> List[str]:
|
||
"""返回全部特征的工艺可读描述（文档/审计用）。"""
|
||
return [s.describe() for s in self.specs]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 模板配置资产（零依赖 YAML 子集解析，对齐 data-bus / rag-kb）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class FeatureTemplateConfig:
|
||
"""模板特征配置：模板元信息 + FeatureSpec 列表。"""
|
||
|
||
template: str
|
||
version: str
|
||
specs: List[FeatureSpec]
|
||
description: str = ""
|
||
|
||
|
||
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}``。
|
||
|
||
其余（标量 / 引号串）退化为 :func:`_parse_scalar`。flow map 用于
|
||
``params: {alpha: 0.2, window: 10}`` 这种紧凑声明。
|
||
"""
|
||
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 FeatureSpecError(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 = []
|
||
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 FeatureSpecError(f"features.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 FeatureSpecError(
|
||
f"features.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）")
|
||
if ":" not in t:
|
||
raise FeatureSpecError(f"features.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 FeatureSpecError(f"features.yaml 第 {no} 行 {key!r} 缺少值")
|
||
sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
|
||
if sub_indent <= indent:
|
||
raise FeatureSpecError(f"features.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]:
|
||
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 FeatureSpecError("features.yaml 顶层必须是 map")
|
||
if next_i < len(lines):
|
||
raise FeatureSpecError(
|
||
f"features.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点")
|
||
return value
|
||
|
||
|
||
def load_feature_config(path: str) -> FeatureTemplateConfig:
|
||
"""从模板特征 YAML 资产加载配置。
|
||
|
||
期望结构（详见 ``config/features.template.yaml``）::
|
||
|
||
template: ti-cl4
|
||
version: 1.0.0
|
||
description: 炉层杂质预警特征工程
|
||
specs:
|
||
- name: 炉温_raw
|
||
kind: raw
|
||
point: CLF-01.TEMP
|
||
- name: 炉温_ema5
|
||
kind: ema
|
||
point: CLF-01.TEMP
|
||
params: {alpha: 0.2}
|
||
threshold: 900.0
|
||
"""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = _load_yaml_text(fh.read())
|
||
|
||
template = str(data.get("template", "")).strip()
|
||
if not template:
|
||
raise FeatureSpecError("features.yaml 缺少 template 字段")
|
||
version = str(data.get("version", "1.0.0")).strip() or "1.0.0"
|
||
description = str(data.get("description", "")).strip()
|
||
|
||
raw_specs = data.get("specs") or []
|
||
if not isinstance(raw_specs, list):
|
||
raise FeatureSpecError("features.yaml specs 必须是 list")
|
||
specs: List[FeatureSpec] = []
|
||
for idx, item in enumerate(raw_specs):
|
||
if not isinstance(item, dict):
|
||
raise FeatureSpecError(f"features.yaml specs[{idx}] 必须是 map")
|
||
name = str(item.get("name", "")).strip()
|
||
kind_name = str(item.get("kind", "")).strip()
|
||
if kind_name not in KIND_REGISTRY:
|
||
raise FeatureSpecError(
|
||
f"features.yaml specs[{idx}] 未知算子 {kind_name!r}"
|
||
f"（应为 {sorted(KIND_REGISTRY)}）")
|
||
point = str(item.get("point", "")).strip()
|
||
unit = str(item.get("unit", "")).strip()
|
||
raw_params = item.get("params") or {}
|
||
if not isinstance(raw_params, dict):
|
||
raise FeatureSpecError(f"features.yaml specs[{idx}] params 必须是 map")
|
||
params: Dict[str, float] = {}
|
||
for pk, pv in raw_params.items():
|
||
try:
|
||
params[pk] = float(pv)
|
||
except (TypeError, ValueError) as exc:
|
||
raise FeatureSpecError(
|
||
f"features.yaml specs[{idx}] 参数 {pk}={pv!r} 不是数值") from exc
|
||
threshold = item.get("threshold")
|
||
if threshold not in (None, ""):
|
||
try:
|
||
threshold = float(threshold)
|
||
except (TypeError, ValueError) as exc:
|
||
raise FeatureSpecError(
|
||
f"features.yaml specs[{idx}] threshold 不是数值") from exc
|
||
else:
|
||
threshold = None
|
||
specs.append(FeatureSpec(
|
||
name=name, kind=KIND_REGISTRY[kind_name], point=point,
|
||
params=params, unit=unit, threshold=threshold,
|
||
))
|
||
return FeatureTemplateConfig(
|
||
template=template, version=version, specs=specs, description=description)
|