bot_dev1
c6dc7d2344
新增 templates/ti-cl4/quality-forecast/features.py： - PointDict：解析 point_dict CSV（9 列，对齐 core/edge-gateway），检索/存在性校验 - FeatureSpec：声明式特征（source/transform/window/meaning），换行业只改清单 - FeatureExtractor：按清单从时序样本抽取特征矩阵，缺失值 NaN 占位 - 9 算子：raw/mean/std/min/max/range/diff/slope/ratio - 零依赖 YAML 子集加载（与 recipe-optim/data-bus 同款） - config/features.template.yaml：7 个默认特征（炉温/配比/CO波动/炉层/TiCl₄纯度） - 22 用例全通过；纯标准库零运行时依赖。
630 lines
22 KiB
Python
630 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Ti-1 氯化车间质量预测 · 特征工程（基于点位字典）（Issue #68 / PRD §5.3 ①）。
|
||
|
||
承接 PRD §5.3 ①「① 质量预测」与父 Issue #10「[Template-Ti 一期] ① 质量预测 +
|
||
③ 炉层杂质预警」：把"DCS 点表 → 可训练的特征矩阵"这条链路**模板化、可配置、
|
||
可测试**，且与 #69 模型训练、#73 模型部署解耦。
|
||
|
||
PRD 设计口径
|
||
------------
|
||
- 架构表（PRD §5.3）：``质量预测 | 预测 | 入：DCS实时数据+LIMS；
|
||
出：质量指标预测值(纯度/杂质) | ① 质量预测 | 中``。
|
||
- 模板化技术路径：特征清单（``FeatureSpec``）外置为 YAML/JSON 超参包，
|
||
切换模板/行业只改特征清单，特征工程代码零改动（PRD §5.3「换行业只改 Recipe」）。
|
||
- 数据门槛：一期客户 DCS 点表未到位时启用默认通用点位集完成框架验证
|
||
（PRD §13 缺省策略），故本模块**不依赖真实历史数据**——用合成/默认点位即可
|
||
完整跑通特征抽取，单测零外部数据依赖。
|
||
|
||
本模块交付
|
||
----------
|
||
1. **点位字典加载 ``PointDict``**：解析 ``point_dict.default.csv``
|
||
（device_id/point_id/name/unit/...，与 ``core/edge-gateway`` 同款 9 列），
|
||
提供 ``by_point_id`` / ``by_device`` 检索与点位存在性校验。
|
||
2. **特征规格 ``FeatureSpec``**：声明式特征——``name``、``source``（点位 point_id
|
||
或常量）、``transform``（聚合算子 raw/mean/std/min/max/diff/ratio/…）、
|
||
``window``（时间窗，秒）、``meaning``（工艺含义，供 #69/#73 可解释引用）。
|
||
3. **特征抽取器 ``FeatureExtractor``**：按特征清单从时序样本（``Sample`` 列表）
|
||
抽取特征向量；缺失值用 ``NaN`` 占位（与 impurity-forecast / recipe-optim 一致，
|
||
便于上层判空屏蔽）；输出有序 ``FeatureMatrix``（行=样本时刻，列=特征）。
|
||
4. **声明式加载**：从 YAML/JSON 特征清单加载（零第三方依赖 YAML 子集解析，
|
||
与 recipe-optim / data-bus / rag-kb 同款）。
|
||
|
||
设计要点
|
||
--------
|
||
- **零运行时依赖**（纯标准库）：CSV 用 ``csv``、YAML 子集自实现、统计用 ``math``
|
||
与手写聚合（不依赖 numpy/pandas），便于隔离网部署。
|
||
- **可解释前置**：特征 ``meaning`` 字段，为 #69 模型可解释性预留引用依据。
|
||
- **可校验**：``FeatureSpec.validate`` 聚合列出全部错误（未知点位/非法算子/负窗
|
||
口），便于配置台一次性反馈。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import math
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
|
||
|
||
# 缺失值统一用 float('nan')，与 impurity-forecast / recipe-optim 一致。
|
||
NAN = float("nan")
|
||
|
||
# 点位字典 CSV 表头（与 core/edge-gateway/config/point_dict.example.csv 对齐，9 列）
|
||
POINT_COLUMNS = [
|
||
"device_id", "point_id", "name", "unit", "dataType",
|
||
"sampleRate", "qualityCode", "opcNode", "protocol",
|
||
]
|
||
|
||
# 允许的特征变换算子（与 impurity-forecast 特征口径对齐）
|
||
ALLOWED_TRANSFORMS = {
|
||
"raw", "mean", "std", "min", "max", "range", "diff", "ratio", "slope",
|
||
}
|
||
|
||
|
||
class FeatureError(ValueError):
|
||
"""特征工程错误（未知点位 / 非法算子 / 窗口非法 / 重复特征名等）。"""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 点位字典
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class Point:
|
||
"""点位字典一行。"""
|
||
|
||
device_id: str
|
||
point_id: str
|
||
name: str
|
||
unit: str
|
||
data_type: str = "float"
|
||
sample_rate: int = 1000
|
||
quality_code: str = "true"
|
||
opc_node: str = ""
|
||
protocol: str = ""
|
||
|
||
@classmethod
|
||
def from_row(cls, row: Dict[str, str]) -> "Point":
|
||
return cls(
|
||
device_id=(row.get("device_id") or "").strip(),
|
||
point_id=(row.get("point_id") or "").strip(),
|
||
name=(row.get("name") or "").strip(),
|
||
unit=(row.get("unit") or "").strip(),
|
||
data_type=(row.get("dataType") or "float").strip(),
|
||
sample_rate=int(float(row.get("sampleRate") or 1000)),
|
||
quality_code=(row.get("qualityCode") or "true").strip(),
|
||
opc_node=(row.get("opcNode") or "").strip(),
|
||
protocol=(row.get("protocol") or "").strip(),
|
||
)
|
||
|
||
|
||
class PointDict:
|
||
"""点位字典：解析 CSV，提供检索与存在性校验。"""
|
||
|
||
def __init__(self, points: Sequence[Point]):
|
||
self._by_id: Dict[str, Point] = {p.point_id: p for p in points}
|
||
self._by_device: Dict[str, List[Point]] = {}
|
||
for p in points:
|
||
self._by_device.setdefault(p.device_id, []).append(p)
|
||
self.points: Tuple[Point, ...] = tuple(points)
|
||
|
||
@classmethod
|
||
def from_csv(cls, path: str) -> "PointDict":
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
rows = list(csv.DictReader(fh))
|
||
if not rows:
|
||
raise FeatureError(f"点位字典为空: {path}")
|
||
header = list(rows[0].keys())
|
||
missing = [c for c in POINT_COLUMNS if c not in header]
|
||
if missing:
|
||
raise FeatureError(f"点位字典缺列: {missing}")
|
||
return cls([Point.from_row(r) for r in rows])
|
||
|
||
def has(self, point_id: str) -> bool:
|
||
return point_id in self._by_id
|
||
|
||
def by_point_id(self, point_id: str) -> Point:
|
||
if point_id not in self._by_id:
|
||
raise FeatureError(f"未知点位: {point_id}")
|
||
return self._by_id[point_id]
|
||
|
||
def by_device(self, device_id: str) -> List[Point]:
|
||
return list(self._by_device.get(device_id, []))
|
||
|
||
def point_ids(self) -> List[str]:
|
||
return list(self._by_id.keys())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 特征规格
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class Transform(str, Enum):
|
||
RAW = "raw"
|
||
MEAN = "mean"
|
||
STD = "std"
|
||
MIN = "min"
|
||
MAX = "max"
|
||
RANGE = "range"
|
||
DIFF = "diff"
|
||
RATIO = "ratio"
|
||
SLOPE = "slope"
|
||
|
||
|
||
@dataclass
|
||
class FeatureSpec:
|
||
"""声明式特征规格。
|
||
|
||
- ``source`` 形如 ``CLF-01.TEMP``（点位 point_id）或常量数值；
|
||
- ``transform`` 聚合算子（raw/mean/std/min/max/range/diff/ratio/slope）；
|
||
- ``window`` 时间窗（秒，仅滚动窗算子有意义；raw/diff 用最近两点）；
|
||
- ``denominator`` 仅 ratio 算子使用（另一个 point_id 或常量）；
|
||
- ``meaning`` 工艺含义（#69/#73 可解释性引用）。
|
||
"""
|
||
|
||
name: str
|
||
source: str
|
||
transform: str = "raw"
|
||
window: float = 60.0
|
||
denominator: Optional[str] = None
|
||
meaning: str = ""
|
||
unit: str = ""
|
||
|
||
def validate(self, point_dict: Optional[PointDict] = None) -> List[str]:
|
||
errors: List[str] = []
|
||
if not self.name:
|
||
errors.append("特征 name 不能为空")
|
||
if self.transform not in ALLOWED_TRANSFORMS:
|
||
errors.append(f"特征 {self.name}: 非法 transform={self.transform}")
|
||
if self.window < 0:
|
||
errors.append(f"特征 {self.name}: window 不能为负 (={self.window})")
|
||
if self.transform == "ratio" and not self.denominator:
|
||
errors.append(f"特征 {self.name}: ratio 算子需指定 denominator")
|
||
# 点位存在性（source/denominator 形如 point_id 时校验）
|
||
if point_dict is not None:
|
||
for label, val in (("source", self.source),
|
||
("denominator", self.denominator)):
|
||
if val and not _is_constant(val) and not point_dict.has(val):
|
||
errors.append(f"特征 {self.name}: {label}={val} 不在点位字典")
|
||
return errors
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "FeatureSpec":
|
||
return cls(
|
||
name=str(d.get("name", "")).strip(),
|
||
source=str(d.get("source", "")).strip(),
|
||
transform=str(d.get("transform", "raw")).strip(),
|
||
window=float(d.get("window", 60.0)),
|
||
denominator=(str(d.get("denominator")).strip()
|
||
if d.get("denominator") else None),
|
||
meaning=str(d.get("meaning", "")).strip(),
|
||
unit=str(d.get("unit", "")).strip(),
|
||
)
|
||
|
||
|
||
def _is_constant(val: str) -> bool:
|
||
"""source/denominator 是否为常量数值（而非 point_id）。"""
|
||
try:
|
||
float(val)
|
||
return True
|
||
except (TypeError, ValueError):
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 时序样本
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class Sample:
|
||
"""一个采样时刻的多点位读数。
|
||
|
||
- ``ts`` 时间戳（秒，单调不减）；
|
||
- ``values`` point_id → 数值；缺失点位视为无读数。
|
||
"""
|
||
|
||
ts: float
|
||
values: Dict[str, float] = field(default_factory=dict)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 特征抽取
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class FeatureExtractor:
|
||
"""按特征清单从时序样本抽取特征向量。
|
||
|
||
用法::
|
||
|
||
ext = FeatureExtractor(specs, point_dict)
|
||
matrix = ext.extract(samples)
|
||
# matrix.rows[i] 是一个有序特征向量；matrix.names 是列名
|
||
"""
|
||
|
||
def __init__(self, specs: Sequence[FeatureSpec],
|
||
point_dict: Optional[PointDict] = None,
|
||
*, strict: bool = True):
|
||
self.specs: Tuple[FeatureSpec, ...] = tuple(specs)
|
||
self.point_dict = point_dict
|
||
if strict:
|
||
errors = self.validate()
|
||
if errors:
|
||
raise FeatureError("特征清单校验失败:\n " + "\n ".join(errors))
|
||
# 重复特征名检查
|
||
names = [s.name for s in self.specs]
|
||
dup = {n for n in names if names.count(n) > 1}
|
||
if dup and strict:
|
||
raise FeatureError(f"重复特征名: {sorted(dup)}")
|
||
|
||
def validate(self) -> List[str]:
|
||
errors: List[str] = []
|
||
for s in self.specs:
|
||
errors.extend(s.validate(self.point_dict))
|
||
return errors
|
||
|
||
@property
|
||
def names(self) -> List[str]:
|
||
return [s.name for s in self.specs]
|
||
|
||
def extract(self, samples: Sequence[Sample]) -> "FeatureMatrix":
|
||
rows: List[List[float]] = []
|
||
# 滚动窗：按 window 秒选取 <= ts 的历史样本
|
||
win = [s.window for s in self.specs]
|
||
max_window = max(win) if win else 0.0
|
||
ordered = sorted(samples, key=lambda s: s.ts)
|
||
for cur in ordered:
|
||
window_samples = [
|
||
s for s in ordered
|
||
if cur.ts - max_window <= s.ts <= cur.ts
|
||
]
|
||
row = [self._compute(spec, cur, window_samples)
|
||
for spec in self.specs]
|
||
rows.append(row)
|
||
return FeatureMatrix(names=self.names, rows=rows)
|
||
|
||
# 单特征计算 ------------------------------------------------------------
|
||
def _compute(self, spec: FeatureSpec, cur: Sample,
|
||
window_samples: Sequence[Sample]) -> float:
|
||
series = _series(spec.source, window_samples, self.point_dict)
|
||
denom_series = (
|
||
_series(spec.denominator, window_samples, self.point_dict)
|
||
if spec.denominator else []
|
||
)
|
||
tf = spec.transform
|
||
if tf == "raw":
|
||
return _last_or_nan(series)
|
||
if tf == "mean":
|
||
return _mean(series)
|
||
if tf == "std":
|
||
return _std(series)
|
||
if tf == "min":
|
||
return _min(series)
|
||
if tf == "max":
|
||
return _max(series)
|
||
if tf == "range":
|
||
return _range(series)
|
||
if tf == "diff":
|
||
return _diff(series)
|
||
if tf == "slope":
|
||
return _slope(series, spec.window)
|
||
if tf == "ratio":
|
||
return _ratio(_last_or_nan(series), _last_or_nan(denom_series))
|
||
# 不应到达（已 validate）
|
||
return NAN
|
||
|
||
|
||
@dataclass
|
||
class FeatureMatrix:
|
||
"""特征抽取结果：有序特征名 + 行向量集合。"""
|
||
|
||
names: List[str]
|
||
rows: List[List[float]]
|
||
|
||
def column(self, name: str) -> List[float]:
|
||
idx = self.names.index(name)
|
||
return [r[idx] for r in self.rows]
|
||
|
||
def to_records(self) -> List[Dict[str, float]]:
|
||
return [dict(zip(self.names, row)) for row in self.rows]
|
||
|
||
def drop_nan_rows(self) -> "FeatureMatrix":
|
||
"""丢弃任一特征为 NaN 的行（数据门槛不足时常用）。"""
|
||
clean = [r for r in self.rows if not any(math.isnan(v) for v in r)]
|
||
return FeatureMatrix(names=list(self.names), rows=clean)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 聚合算子（纯标准库）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _series(source: str, samples: Sequence[Sample],
|
||
point_dict: Optional[PointDict]) -> List[Tuple[float, float]]:
|
||
"""取一个 source 的 (ts, value) 序列。常量源展开为各样本时刻。"""
|
||
if _is_constant(source):
|
||
const = float(source)
|
||
return [(s.ts, const) for s in samples]
|
||
return [(s.ts, s.values[source]) for s in samples
|
||
if source in s.values and not math.isnan(s.values[source])]
|
||
|
||
|
||
def _last_or_nan(series: Sequence[Tuple[float, float]]) -> float:
|
||
return series[-1][1] if series else NAN
|
||
|
||
|
||
def _values(series: Sequence[Tuple[float, float]]) -> List[float]:
|
||
return [v for _, v in series]
|
||
|
||
|
||
def _mean(series: Sequence[Tuple[float, float]]) -> float:
|
||
vs = _values(series)
|
||
return sum(vs) / len(vs) if vs else NAN
|
||
|
||
|
||
def _std(series: Sequence[Tuple[float, float]]) -> float:
|
||
vs = _values(series)
|
||
n = len(vs)
|
||
if n < 2:
|
||
return NAN if n == 0 else 0.0
|
||
mu = sum(vs) / n
|
||
var = sum((v - mu) ** 2 for v in vs) / (n - 1)
|
||
return math.sqrt(var)
|
||
|
||
|
||
def _min(series: Sequence[Tuple[float, float]]) -> float:
|
||
vs = _values(series)
|
||
return min(vs) if vs else NAN
|
||
|
||
|
||
def _max(series: Sequence[Tuple[float, float]]) -> float:
|
||
vs = _values(series)
|
||
return max(vs) if vs else NAN
|
||
|
||
|
||
def _range(series: Sequence[Tuple[float, float]]) -> float:
|
||
vs = _values(series)
|
||
return (max(vs) - min(vs)) if vs else NAN
|
||
|
||
|
||
def _diff(series: Sequence[Tuple[float, float]]) -> float:
|
||
if len(series) < 2:
|
||
return NAN
|
||
return series[-1][1] - series[-2][1]
|
||
|
||
|
||
def _slope(series: Sequence[Tuple[float, float]], window: float) -> float:
|
||
"""最小二乘斜率（值/秒）；样本不足返回 NaN。"""
|
||
if len(series) < 2:
|
||
return NAN
|
||
xs = [t for t, _ in series]
|
||
# 时间窗外的样本不参与（已由 caller 截窗，这里再以 window 收敛）
|
||
if window and window > 0:
|
||
tmax = max(xs)
|
||
kept = [(t, v) for t, v in series if t >= tmax - window]
|
||
if len(kept) < 2:
|
||
return NAN
|
||
xs = [t for t, _ in kept]
|
||
ys = [v for _, v in kept]
|
||
else:
|
||
ys = [v for _, v in series]
|
||
n = len(xs)
|
||
xbar = sum(xs) / n
|
||
ybar = sum(ys) / n
|
||
num = sum((xs[i] - xbar) * (ys[i] - ybar) for i in range(n))
|
||
den = sum((xs[i] - xbar) ** 2 for i in range(n))
|
||
return num / den if den else NAN
|
||
|
||
|
||
def _ratio(a: float, b: float) -> float:
|
||
if math.isnan(a) or math.isnan(b) or b == 0:
|
||
return NAN
|
||
return a / b
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 声明式加载（零第三方依赖 YAML 子集解析，与 recipe-optim 同款）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def load_feature_specs(text: str,
|
||
point_dict: Optional[PointDict] = None,
|
||
*, strict: bool = True) -> FeatureExtractor:
|
||
"""从 YAML/JSON 文本加载特征清单并构造 FeatureExtractor。
|
||
|
||
支持的 YAML 子集：``features:`` 顶层键，下为 ``- name/source/transform/...``
|
||
列表项。也兼容 JSON（``{"features": [...]}``）。
|
||
"""
|
||
text = text.strip()
|
||
data: Any
|
||
if text.startswith("{") or text.startswith("["):
|
||
import json
|
||
data = json.loads(text)
|
||
else:
|
||
data = _parse_yaml_subset(text)
|
||
if not isinstance(data, dict):
|
||
raise FeatureError("特征清单顶层应为映射（含 features 键）")
|
||
raw_features = data.get("features")
|
||
if not isinstance(raw_features, list):
|
||
raise FeatureError("特征清单缺少 features 列表")
|
||
specs = [FeatureSpec.from_dict(f) for f in raw_features if isinstance(f, dict)]
|
||
if not specs:
|
||
raise FeatureError("特征清单 features 为空")
|
||
return FeatureExtractor(specs, point_dict, strict=strict)
|
||
|
||
|
||
def _parse_yaml_subset(text: str) -> Any:
|
||
"""极简 YAML 子集解析器（仅供模板资产，非通用 YAML）。
|
||
|
||
支持：注释（# ...）、映射（key: value）、列表（- item）、嵌套缩进、
|
||
基本标量（int/float/str/bool/null）。与 recipe-optim / data-bus 同款。
|
||
"""
|
||
lines: List[str] = []
|
||
for raw in text.splitlines():
|
||
stripped = raw.rstrip()
|
||
if not stripped.strip():
|
||
continue
|
||
if stripped.lstrip().startswith("#"):
|
||
continue
|
||
hi = _find_inline_comment(stripped)
|
||
if hi is not None:
|
||
stripped = stripped[:hi].rstrip()
|
||
if stripped:
|
||
lines.append(stripped)
|
||
parser = _YamlParser(lines)
|
||
return parser.parse_block(0)[0] if lines else {}
|
||
|
||
|
||
def _find_inline_comment(line: str) -> Optional[int]:
|
||
depth = 0
|
||
in_str = False
|
||
for i, ch in enumerate(line):
|
||
if ch == '"':
|
||
in_str = not in_str
|
||
elif not in_str:
|
||
if ch in "[{":
|
||
depth += 1
|
||
elif ch in "]}":
|
||
depth = max(0, depth - 1)
|
||
elif ch == "#" and depth == 0:
|
||
if i == 0 or line[i - 1] in (" ", "\t"):
|
||
return i
|
||
return None
|
||
|
||
|
||
def _parse_scalar(raw: str) -> Any:
|
||
raw = raw.strip()
|
||
if not raw:
|
||
return None
|
||
if raw.startswith('"') and raw.endswith('"'):
|
||
return raw[1:-1]
|
||
if raw.startswith("[") or raw.startswith("{"):
|
||
import json
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
return raw
|
||
low = raw.lower()
|
||
if low == "true":
|
||
return True
|
||
if low == "false":
|
||
return False
|
||
if low in ("null", "~", "none"):
|
||
return None
|
||
try:
|
||
return int(raw)
|
||
except ValueError:
|
||
pass
|
||
try:
|
||
return float(raw)
|
||
except ValueError:
|
||
pass
|
||
return raw
|
||
|
||
|
||
class _YamlParser:
|
||
"""递归下降的 YAML 子集解析器（按缩进分层）。"""
|
||
|
||
def __init__(self, lines: List[str]) -> None:
|
||
self.lines = lines
|
||
self.i = 0
|
||
|
||
def _indent(self, line: str) -> int:
|
||
return len(line) - len(line.lstrip(" "))
|
||
|
||
def parse_block(self, indent: int) -> Tuple[Any, bool]:
|
||
if self.i >= len(self.lines):
|
||
return {}, False
|
||
line = self.lines[self.i]
|
||
cur = self._indent(line)
|
||
if cur < indent:
|
||
return {}, False
|
||
stripped = line.strip()
|
||
if stripped.startswith("- ") or stripped == "-":
|
||
return self._parse_list(cur), True
|
||
return self._parse_mapping(cur), False
|
||
|
||
def _parse_mapping(self, indent: int) -> Dict[str, Any]:
|
||
result: Dict[str, Any] = {}
|
||
effective = indent
|
||
if self.i < len(self.lines):
|
||
first = self._indent(self.lines[self.i])
|
||
if first > indent:
|
||
effective = first
|
||
while self.i < len(self.lines):
|
||
line = self.lines[self.i]
|
||
cur = self._indent(line)
|
||
if cur < effective:
|
||
break
|
||
if cur > effective:
|
||
self.i += 1
|
||
continue
|
||
stripped = line.strip()
|
||
if stripped.startswith("- "):
|
||
break
|
||
key, sep, rest = stripped.partition(":")
|
||
if not sep:
|
||
self.i += 1
|
||
continue
|
||
key = key.strip()
|
||
rest = rest.strip()
|
||
self.i += 1
|
||
if rest:
|
||
result[key] = _parse_scalar(rest)
|
||
else:
|
||
# 子块
|
||
if self.i < len(self.lines):
|
||
nxt = self._indent(self.lines[self.i])
|
||
if nxt > effective:
|
||
val, _ = self.parse_block(nxt)
|
||
result[key] = val
|
||
return result
|
||
|
||
def _parse_list(self, indent: int) -> List[Any]:
|
||
items: List[Any] = []
|
||
while self.i < len(self.lines):
|
||
line = self.lines[self.i]
|
||
cur = self._indent(line)
|
||
if cur < indent:
|
||
break
|
||
if cur > indent:
|
||
self.i += 1
|
||
continue
|
||
stripped = line.strip()
|
||
if not stripped.startswith("-"):
|
||
break
|
||
item_text = stripped[1:].strip()
|
||
if not item_text:
|
||
# 子块（嵌套映射/列表）
|
||
if self.i + 1 < len(self.lines):
|
||
nxt = self._indent(self.lines[self.i + 1])
|
||
if nxt > cur:
|
||
self.i += 1
|
||
val, _ = self.parse_block(nxt)
|
||
items.append(val)
|
||
continue
|
||
self.i += 1
|
||
items.append(None)
|
||
continue
|
||
# "- key: value" 形式 → 该 item 是映射
|
||
if ":" in item_text and not item_text.startswith('"'):
|
||
k, sep, v = item_text.partition(":")
|
||
if sep:
|
||
item: Dict[str, Any] = {k.strip(): _parse_scalar(v.strip())}
|
||
self.i += 1
|
||
# 后续同缩进的 key 归入同一 item
|
||
if self.i < len(self.lines):
|
||
child_indent = self._indent(self.lines[self.i])
|
||
if child_indent > cur:
|
||
sub, _ = self.parse_block(child_indent)
|
||
if isinstance(sub, dict):
|
||
item.update(sub)
|
||
items.append(item)
|
||
continue
|
||
items.append(_parse_scalar(item_text))
|
||
self.i += 1
|
||
return items
|