785 lines
31 KiB
Python
785 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Ti-2 配方动态优化 · 优化问题建模（约束/目标定义）（Issue #78 / PRD 5.3 ②）。
|
||
|
||
承接 PRD 5.3「② 配方动态优化」与超参包示例（``objective`` / ``features`` /
|
||
``target``）：把"给定质量目标 + 工艺约束，求最优配方/参数"这条链路**模板化、
|
||
可配置、可测试**，且与 #79 求解器、#80 跨工序寻优、#81 可解释建议解耦。
|
||
|
||
PRD 设计口径
|
||
------------
|
||
- 架构表（PRD §5.3）：``工艺优化/配方推荐 | 优化/推荐 | 入：质量目标+约束；
|
||
出：参数/配方建议 | ② 配方动态优化 | 高（需闭环反馈）``。
|
||
- 模板化技术路径：超参包驱动——``objective``、输入特征清单、``target`` 等可变量
|
||
外置为 JSON 超参包，切换模板仅改此包；跨行业差异落资产，不落代码。
|
||
- 风险表：二期交付（一期数据门槛不足），故本期**先把问题建模沉淀为可校验的声明
|
||
式规格**，为 #79 求解器、#80 跨工序寻优、#81 可解释建议提供**统一的问题描述
|
||
契约**；先有"能跑通、可测试"的模型，数据就绪后接求解器（#79/#80）。
|
||
|
||
本模块交付
|
||
----------
|
||
1. **决策变量 ``DecisionVariable``**：配方/工艺可调参数的声明式规格——变量名、
|
||
单位、取值域（``Bounds`` 连续区间 / ``Choices`` 离散枚举）、初值、是否整型、
|
||
工艺含义（``meaning``，供 #81 可解释建议引用）。
|
||
2. **目标函数规格 ``ObjectiveSpec``**：``Sense``（minimize/maximize）+ 目标项
|
||
（``ObjectiveTerm``：系数 × 变量，线性目标）+ 目标 ``target``（PRD 超参包字段）。
|
||
3. **约束规格 ``ConstraintSpec``**：``ConstraintKind``（box / linear / ratio /
|
||
forbidden）统一描述工艺约束（温度上下限、配方配比、禁止组合等）。
|
||
4. **问题模型 ``OptimizationProblem``**：聚合变量 + 目标 + 约束，提供校验
|
||
（``validate``，聚合并列出全部错误，便于配置台一次性反馈）、声明式加载
|
||
（零第三方依赖 YAML 子集解析，与 data-bus/rag-kb/impurity-forecast 同款）。
|
||
|
||
设计要点
|
||
--------
|
||
- **零运行时依赖**（纯标准库）：与内核既有模块一致，便于离线/隔离网部署。
|
||
- **求解器无关**：本模块只描述"问题"，``solve`` 留给 #79 注入；便于换行业复用、
|
||
单测无需真实求解器。
|
||
- **可解释前置**：变量 ``meaning`` + 约束 ``reason`` 字段，为 #81 优化建议"可溯源"
|
||
预留引用依据（对齐 PRD"要求结果可解释、可溯源，要引用依据"）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
|
||
|
||
# 缺失值统一用 float('nan')，与 impurity-forecast 一致，便于上层判空屏蔽。
|
||
NAN = float("nan")
|
||
|
||
|
||
class ProblemError(ValueError):
|
||
"""配方优化问题建模错误（未知变量 / 越界 / 约束矛盾 / 重复定义等）。"""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 决策变量
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class DomainKind(str, Enum):
|
||
"""决策变量取值域类型。"""
|
||
|
||
BOUNDS = "bounds" # 连续区间 [low, high]（如温度 800~900℃）
|
||
CHOICES = "choices" # 离散枚举（如催化剂型号 A/B/C）
|
||
|
||
|
||
@dataclass
|
||
class DecisionVariable:
|
||
"""一个可调配方/工艺参数的声明式规格。
|
||
|
||
``bounds`` 与 ``choices`` 二选一（由 ``kind`` 决定）：
|
||
- ``bounds``：``[low, high]``，``integer=True`` 时取整；
|
||
- ``choices``：离散可选值列表（任意可比较的标量，多为 float/str）。
|
||
"""
|
||
|
||
name: str
|
||
kind: DomainKind
|
||
meaning: str = "" # 工艺含义，供 #81 可解释建议引用
|
||
unit: str = "" # 单位（℃、m³/h、kg、…）
|
||
bounds: Optional[Tuple[float, float]] = None
|
||
choices: Optional[List[Any]] = None
|
||
initial: Optional[float] = None # 当前工况/配方初值
|
||
integer: bool = False # 仅 bounds 连续域生效
|
||
|
||
def __post_init__(self) -> None:
|
||
if not self.name or not str(self.name).strip():
|
||
raise ProblemError("DecisionVariable.name 不能为空")
|
||
if self.kind == DomainKind.BOUNDS:
|
||
if self.bounds is None:
|
||
raise ProblemError(f"变量 {self.name!r} kind=bounds 但未提供 bounds")
|
||
low, high = self.bounds
|
||
if _is_num(low) and _is_num(high) and low > high:
|
||
raise ProblemError(
|
||
f"变量 {self.name!r} bounds 下界 {low} 大于上界 {high}")
|
||
if self.integer and self.bounds is not None:
|
||
low, high = self.bounds
|
||
if _is_num(low) and float(low).is_integer() is False:
|
||
raise ProblemError(
|
||
f"变量 {self.name!r} integer=True 但下界 {low} 非整")
|
||
if _is_num(high) and float(high).is_integer() is False:
|
||
raise ProblemError(
|
||
f"变量 {self.name!r} integer=True 但上界 {high} 非整")
|
||
elif self.kind == DomainKind.CHOICES:
|
||
if not self.choices:
|
||
raise ProblemError(f"变量 {self.name!r} kind=choices 但 choices 为空")
|
||
else: # pragma: no cover - 枚举穷尽
|
||
raise ProblemError(f"变量 {self.name!r} 未知 kind={self.kind!r}")
|
||
|
||
def contains(self, value: Any) -> bool:
|
||
"""取值是否落在该变量合法域内。"""
|
||
if self.kind == DomainKind.BOUNDS and self.bounds is not None:
|
||
if not _is_num(value):
|
||
return False
|
||
low, high = self.bounds
|
||
if self.integer and float(value).is_integer() is False:
|
||
return False
|
||
return low <= value <= high
|
||
# choices
|
||
return value in (self.choices or [])
|
||
|
||
def clamp(self, value: Any) -> Any:
|
||
"""把越界的连续域取值夹回合法区间（离散域不夹，原值返回）。"""
|
||
if self.kind == DomainKind.BOUNDS and self.bounds is not None and _is_num(value):
|
||
low, high = self.bounds
|
||
value = max(low, min(high, value))
|
||
if self.integer:
|
||
value = float(round(value))
|
||
return value
|
||
return value
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {
|
||
"name": self.name,
|
||
"kind": self.kind.value,
|
||
"meaning": self.meaning,
|
||
"unit": self.unit,
|
||
"integer": self.integer,
|
||
}
|
||
if self.kind == DomainKind.BOUNDS:
|
||
d["bounds"] = list(self.bounds) if self.bounds else None
|
||
else:
|
||
d["choices"] = list(self.choices) if self.choices else None
|
||
if self.initial is not None:
|
||
d["initial"] = self.initial
|
||
return d
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "DecisionVariable":
|
||
name = d.get("name")
|
||
if not isinstance(name, str):
|
||
raise ProblemError("DecisionVariable 缺少 name 字段")
|
||
kind_raw = d.get("kind", "bounds")
|
||
try:
|
||
kind = DomainKind(str(kind_raw))
|
||
except ValueError as e:
|
||
raise ProblemError(f"变量 {name!r} 未知 kind={kind_raw!r}") from e
|
||
bounds = d.get("bounds")
|
||
choices = d.get("choices")
|
||
if kind == DomainKind.BOUNDS and bounds is not None:
|
||
if (not isinstance(bounds, (list, tuple))) or len(bounds) != 2:
|
||
raise ProblemError(f"变量 {name!r} bounds 必须是 [low, high]")
|
||
bounds = (float(bounds[0]), float(bounds[1]))
|
||
if kind == DomainKind.CHOICES and choices is not None:
|
||
choices = list(choices)
|
||
return cls(
|
||
name=name,
|
||
kind=kind,
|
||
meaning=str(d.get("meaning", "")),
|
||
unit=str(d.get("unit", "")),
|
||
bounds=bounds,
|
||
choices=choices,
|
||
initial=d.get("initial"),
|
||
integer=bool(d.get("integer", False)),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 目标函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class Sense(str, Enum):
|
||
"""优化方向。"""
|
||
|
||
MINIMIZE = "minimize"
|
||
MAXIMIZE = "maximize"
|
||
|
||
@property
|
||
def label(self) -> str:
|
||
return {Sense.MINIMIZE: "最小化", Sense.MAXIMIZE: "最大化"}[self]
|
||
|
||
|
||
@dataclass
|
||
class ObjectiveTerm:
|
||
"""线性目标项：``coefficient * variable``（变量名引用 ``DecisionVariable.name``）。"""
|
||
|
||
variable: str
|
||
coefficient: float = 1.0
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {"variable": self.variable, "coefficient": self.coefficient}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "ObjectiveTerm":
|
||
if "variable" not in d:
|
||
raise ProblemError("ObjectiveTerm 缺少 variable 字段")
|
||
return cls(variable=str(d["variable"]), coefficient=float(d.get("coefficient", 1.0)))
|
||
|
||
|
||
@dataclass
|
||
class ObjectiveSpec:
|
||
"""目标函数声明式规格（线性加权，对齐 PRD 超参包 ``objective`` 字段）。
|
||
|
||
形如 ``sense(coef1*var1 + coef2*var2 + ...)``，目标质量 ``target`` 为达标量
|
||
（如 ``Ti_purity ≥ 99.5%`` 中的 99.5），仅记录、不参与求解，供 #81 可解释。
|
||
"""
|
||
|
||
sense: Sense = Sense.MAXIMIZE
|
||
terms: List[ObjectiveTerm] = field(default_factory=list)
|
||
target: Optional[str] = None # PRD 超参包 ``target``：如 "Ti_purity"
|
||
target_value: Optional[float] = None # 达标量（可选）
|
||
description: str = ""
|
||
|
||
def evaluate(self, assignment: Dict[str, float]) -> float:
|
||
"""给定一组变量取值，计算目标函数值（未知变量按 0 计）。"""
|
||
total = 0.0
|
||
for t in self.terms:
|
||
v = assignment.get(t.variable)
|
||
if _is_num(v):
|
||
total += t.coefficient * v
|
||
return total
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {
|
||
"sense": self.sense.value,
|
||
"terms": [t.to_dict() for t in self.terms],
|
||
}
|
||
if self.target is not None:
|
||
d["target"] = self.target
|
||
if self.target_value is not None:
|
||
d["target_value"] = self.target_value
|
||
if self.description:
|
||
d["description"] = self.description
|
||
return d
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "ObjectiveSpec":
|
||
sense_raw = d.get("sense", "maximize")
|
||
try:
|
||
sense = Sense(str(sense_raw))
|
||
except ValueError as e:
|
||
raise ProblemError(f"未知 sense={sense_raw!r}") from e
|
||
terms = [ObjectiveTerm.from_dict(t) for t in d.get("terms", [])]
|
||
tv = d.get("target_value")
|
||
return cls(
|
||
sense=sense,
|
||
terms=terms,
|
||
target=d.get("target"),
|
||
target_value=float(tv) if _is_num(tv) else None,
|
||
description=str(d.get("description", "")),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 约束
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class ConstraintKind(str, Enum):
|
||
"""约束类型（统一描述常见工艺约束）。"""
|
||
|
||
BOX = "box" # 变量上下界（冗余于 DecisionVariable.bounds，供"运行期收紧"）
|
||
LINEAR = "linear" # 线性不等式 Σ a_i*x_i (</<=/>/>=) b
|
||
RATIO = "ratio" # 配比约束：x_a / x_b (op) value
|
||
FORBIDDEN = "forbidden" # 禁止组合：若干变量取值组合不允许
|
||
|
||
|
||
_LINEAR_OPS = {"<", "<=", ">", ">=", "==", "!="}
|
||
|
||
|
||
@dataclass
|
||
class ConstraintSpec:
|
||
"""约束声明式规格。
|
||
|
||
每条约束带 ``reason``（工艺依据，供 #81 可解释建议"可溯源、引用依据"）。
|
||
"""
|
||
|
||
kind: ConstraintKind
|
||
reason: str = ""
|
||
# box
|
||
variable: Optional[str] = None
|
||
bounds: Optional[Tuple[float, float]] = None
|
||
# linear
|
||
coefficients: Optional[Dict[str, float]] = None
|
||
op: str = "<="
|
||
rhs: float = 0.0
|
||
# ratio
|
||
numerator: Optional[str] = None
|
||
denominator: Optional[str] = None
|
||
value: float = 0.0
|
||
# forbidden
|
||
combination: Optional[Dict[str, Any]] = None
|
||
|
||
def __post_init__(self) -> None:
|
||
if self.kind == ConstraintKind.LINEAR and self.op not in _LINEAR_OPS:
|
||
raise ProblemError(f"线性约束非法 op={self.op!r}")
|
||
if self.kind == ConstraintKind.LINEAR and not self.coefficients:
|
||
raise ProblemError("线性约束 coefficients 不能为空")
|
||
|
||
# ---- 校验（返回错误信息列表，不抛异常，便于聚合） ------------------
|
||
|
||
def errors_against(self, variables: Dict[str, DecisionVariable]) -> List[str]:
|
||
"""对该约束引用的变量是否存在做静态校验，返回错误信息列表。"""
|
||
errs: List[str] = []
|
||
if self.kind == ConstraintKind.BOX:
|
||
if not self.variable:
|
||
errs.append("box 约束缺少 variable")
|
||
elif self.variable not in variables:
|
||
errs.append(f"box 约束引用未知变量 {self.variable!r}")
|
||
elif self.kind == ConstraintKind.LINEAR:
|
||
for vname in (self.coefficients or {}):
|
||
if vname not in variables:
|
||
errs.append(f"线性约束引用未知变量 {vname!r}")
|
||
elif self.kind == ConstraintKind.RATIO:
|
||
for fld, vname in (("numerator", self.numerator),
|
||
("denominator", self.denominator)):
|
||
if not vname:
|
||
errs.append(f"ratio 约束缺少 {fld}")
|
||
elif vname not in variables:
|
||
errs.append(f"ratio 约束引用未知变量 {vname!r}")
|
||
elif self.kind == ConstraintKind.FORBIDDEN:
|
||
for vname in (self.combination or {}):
|
||
if vname not in variables:
|
||
errs.append(f"forbidden 约束引用未知变量 {vname!r}")
|
||
return errs
|
||
|
||
# ---- 可行性判定（给定取值，判断该约束是否满足） --------------------
|
||
|
||
def satisfied_by(self, assignment: Dict[str, Any]) -> bool:
|
||
"""给定一组变量取值，判断该约束是否被满足（未知变量视为未约束）。"""
|
||
if self.kind == ConstraintKind.BOX and self.bounds is not None and self.variable:
|
||
v = assignment.get(self.variable)
|
||
if not _is_num(v):
|
||
return True # 未知取值不判
|
||
low, high = self.bounds
|
||
return low <= v <= high
|
||
if self.kind == ConstraintKind.LINEAR and self.coefficients:
|
||
total = 0.0
|
||
unknown = False
|
||
for vname, coef in self.coefficients.items():
|
||
v = assignment.get(vname)
|
||
if not _is_num(v):
|
||
unknown = True
|
||
break
|
||
total += coef * v
|
||
if unknown:
|
||
return True
|
||
return _apply_op(total, self.op, self.rhs)
|
||
if self.kind == ConstraintKind.RATIO and self.numerator and self.denominator:
|
||
a = assignment.get(self.numerator)
|
||
b = assignment.get(self.denominator)
|
||
if not _is_num(a) or not _is_num(b) or b == 0:
|
||
return True
|
||
return _apply_op(a / b, self.op, self.value)
|
||
if self.kind == ConstraintKind.FORBIDDEN and self.combination:
|
||
# 组合中每个键值都命中才算"禁止组合"被触发
|
||
for vname, want in self.combination.items():
|
||
if assignment.get(vname) != want:
|
||
return True
|
||
return False
|
||
return True
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {"kind": self.kind.value}
|
||
if self.reason:
|
||
d["reason"] = self.reason
|
||
if self.kind == ConstraintKind.BOX:
|
||
d["variable"] = self.variable
|
||
d["bounds"] = list(self.bounds) if self.bounds else None
|
||
elif self.kind == ConstraintKind.LINEAR:
|
||
d["coefficients"] = dict(self.coefficients or {})
|
||
d["op"] = self.op
|
||
d["rhs"] = self.rhs
|
||
elif self.kind == ConstraintKind.RATIO:
|
||
d["numerator"] = self.numerator
|
||
d["denominator"] = self.denominator
|
||
d["op"] = self.op
|
||
d["value"] = self.value
|
||
elif self.kind == ConstraintKind.FORBIDDEN:
|
||
d["combination"] = dict(self.combination or {})
|
||
return d
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "ConstraintSpec":
|
||
kind_raw = d.get("kind")
|
||
try:
|
||
kind = ConstraintKind(str(kind_raw))
|
||
except ValueError as e:
|
||
raise ProblemError(f"未知约束 kind={kind_raw!r}") from e
|
||
bounds = d.get("bounds")
|
||
if kind == ConstraintKind.BOX and bounds is not None:
|
||
bounds = (float(bounds[0]), float(bounds[1]))
|
||
coefs = d.get("coefficients")
|
||
if coefs is not None:
|
||
coefs = {k: float(v) for k, v in coefs.items()}
|
||
return cls(
|
||
kind=kind,
|
||
reason=str(d.get("reason", "")),
|
||
variable=d.get("variable"),
|
||
bounds=bounds,
|
||
coefficients=coefs,
|
||
op=str(d.get("op", "<=")),
|
||
rhs=float(d.get("rhs", 0.0)),
|
||
numerator=d.get("numerator"),
|
||
denominator=d.get("denominator"),
|
||
value=float(d.get("value", 0.0)),
|
||
combination=d.get("combination"),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 优化问题
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def objective_factory() -> ObjectiveSpec:
|
||
"""dataclass 默认值工厂：空目标（最大化、无项）。"""
|
||
return ObjectiveSpec(sense=Sense.MAXIMIZE)
|
||
|
||
|
||
@dataclass
|
||
class OptimizationProblem:
|
||
"""配方优化问题模型（变量 + 目标 + 约束），求解器无关。
|
||
|
||
设计为「先建模、后求解」：``validate`` 做静态一致性校验（变量引用、域完整性），
|
||
``is_feasible`` 做取值可行性判定（运行期收紧约束 / 禁止组合），``solve`` 留给
|
||
#79 注入求解器，本模块不绑任何优化库。
|
||
"""
|
||
|
||
variables: List[DecisionVariable] = field(default_factory=list)
|
||
objective: ObjectiveSpec = field(default_factory=objective_factory)
|
||
constraints: List[ConstraintSpec] = field(default_factory=list)
|
||
problem_id: str = ""
|
||
template: str = "" # 如 "iAOP-Template-Ti"
|
||
description: str = ""
|
||
|
||
# ---- 变量索引 ----------------------------------------------------
|
||
|
||
@property
|
||
def variable_map(self) -> Dict[str, DecisionVariable]:
|
||
return {v.name: v for v in self.variables}
|
||
|
||
# ---- 校验 --------------------------------------------------------
|
||
|
||
def validate(self) -> List[str]:
|
||
"""聚合所有静态错误，返回错误信息列表（空列表表示通过）。"""
|
||
errs: List[str] = []
|
||
seen: set = set()
|
||
for v in self.variables:
|
||
if v.name in seen:
|
||
errs.append(f"重复定义变量 {v.name!r}")
|
||
seen.add(v.name)
|
||
vmap = self.variable_map
|
||
for t in self.objective.terms:
|
||
if t.variable not in vmap:
|
||
errs.append(f"目标项引用未知变量 {t.variable!r}")
|
||
for i, c in enumerate(self.constraints):
|
||
for e in c.errors_against(vmap):
|
||
errs.append(f"约束 #{i} ({c.kind.value}): {e}")
|
||
if self.objective.terms and not any(
|
||
t.variable in vmap for t in self.objective.terms
|
||
):
|
||
errs.append("目标函数所有项均引用未知变量")
|
||
return errs
|
||
|
||
# ---- 可行性判定 --------------------------------------------------
|
||
|
||
def is_feasible(self, assignment: Dict[str, Any]) -> bool:
|
||
"""给定一组变量取值，判断是否满足全部约束与变量域。"""
|
||
vmap = self.variable_map
|
||
for name, val in assignment.items():
|
||
v = vmap.get(name)
|
||
if v is not None and not v.contains(val):
|
||
return False
|
||
return all(c.satisfied_by(assignment) for c in self.constraints)
|
||
|
||
def violated_constraints(self, assignment: Dict[str, Any]) -> List[ConstraintSpec]:
|
||
"""返回被该取值违反的约束列表（供 #81 可解释建议引用依据）。"""
|
||
return [c for c in self.constraints if not c.satisfied_by(assignment)]
|
||
|
||
# ---- 序列化 ------------------------------------------------------
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {
|
||
"variables": [v.to_dict() for v in self.variables],
|
||
"objective": self.objective.to_dict(),
|
||
"constraints": [c.to_dict() for c in self.constraints],
|
||
}
|
||
if self.problem_id:
|
||
d["problem_id"] = self.problem_id
|
||
if self.template:
|
||
d["template"] = self.template
|
||
if self.description:
|
||
d["description"] = self.description
|
||
return d
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "OptimizationProblem":
|
||
return cls(
|
||
variables=[DecisionVariable.from_dict(v) for v in d.get("variables", [])],
|
||
objective=ObjectiveSpec.from_dict(d.get("objective", {})),
|
||
constraints=[ConstraintSpec.from_dict(c) for c in d.get("constraints", [])],
|
||
problem_id=str(d.get("problem_id", "")),
|
||
template=str(d.get("template", "")),
|
||
description=str(d.get("description", "")),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _is_num(x: object) -> bool:
|
||
return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
|
||
|
||
|
||
def _apply_op(left: float, op: str, right: float) -> bool:
|
||
"""应用比较算子（线性/配比约束共用）。"""
|
||
if op == "<":
|
||
return left < right
|
||
if op == "<=":
|
||
return left <= right
|
||
if op == ">":
|
||
return left > right
|
||
if op == ">=":
|
||
return left >= right
|
||
if op == "==":
|
||
return abs(left - right) < 1e-12
|
||
if op == "!=":
|
||
return abs(left - right) >= 1e-12
|
||
return False # pragma: no cover
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 零依赖 YAML 子集加载（与 data-bus / rag-kb / impurity-forecast 同款）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def load_problem(path: str) -> OptimizationProblem:
|
||
"""从声明式模板资产（YAML 子集）加载优化问题。
|
||
|
||
解析支持：缩进块、``key: value``、``- item``、行内 ``# 注释``、字符串/数字/
|
||
布尔、内联 ``[a, b]`` 列表与 ``{a: 1}`` 映射。足以覆盖本模板资产格式；
|
||
不引入第三方依赖，与内核既有模块一致。
|
||
"""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
text = fh.read()
|
||
data = _parse_yaml_subset(text)
|
||
if not isinstance(data, dict):
|
||
raise ProblemError(f"模板 {path} 顶层应为映射")
|
||
return OptimizationProblem.from_dict(data)
|
||
|
||
|
||
def _parse_yaml_subset(text: str) -> Any:
|
||
"""极简 YAML 子集解析器（仅供模板资产，非通用 YAML）。"""
|
||
# 去注释 + 去尾部空白，保留缩进
|
||
lines: List[str] = []
|
||
for raw in text.splitlines():
|
||
# 行内注释：仅在 "# " 前不是值的一部分时剥离；这里取保守策略——行首/值后
|
||
# 的 " #" 视为注释。冒号/方括号内的 # 不处理。
|
||
stripped = raw.rstrip()
|
||
if not stripped.strip():
|
||
continue
|
||
# 简单注释行
|
||
if stripped.lstrip().startswith("#"):
|
||
continue
|
||
# 去行尾注释（" #" 形式）
|
||
hash_idx = _find_inline_comment(stripped)
|
||
if hash_idx is not None:
|
||
stripped = stripped[:hash_idx].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]:
|
||
"""返回行内注释 ``#`` 的索引（无则 None），跳过 ``[...]``/``{...}`` 内的 #。"""
|
||
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 and i > 0 and line[i - 1] in (" ", "\t"):
|
||
return i
|
||
elif ch == "#" and depth == 0 and i == 0:
|
||
return i
|
||
return None
|
||
|
||
|
||
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]:
|
||
"""解析当前缩进层级的一个块，返回 (value, is_list_marker)。"""
|
||
if self.i >= len(self.lines):
|
||
return {}, False
|
||
line = self.lines[self.i]
|
||
cur_indent = self._indent(line)
|
||
if cur_indent < indent:
|
||
return {}, False
|
||
stripped = line.strip()
|
||
if stripped.startswith("- ") or stripped == "-":
|
||
return self._parse_list(cur_indent), True
|
||
return self._parse_mapping(cur_indent), False
|
||
|
||
def _parse_mapping(self, indent: int) -> Dict[str, Any]:
|
||
result: Dict[str, Any] = {}
|
||
# 实际子键缩进可能 > indent（如 "- key: v" 后 4 空格键、项缩进 2）。
|
||
# 用首行真实缩进对齐，避免误把合法子键当"孤立缩进"跳过。
|
||
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:
|
||
# 子块：用首行真实缩进解析（列表或映射），兼容 4 空格子键等
|
||
if self.i < len(self.lines) and self._indent(self.lines[self.i]) > effective:
|
||
child_indent = self._indent(self.lines[self.i])
|
||
val, _ = self.parse_block(child_indent)
|
||
result[key] = val
|
||
else:
|
||
result[key] = None
|
||
return result
|
||
|
||
def _parse_list(self, indent: int) -> List[Any]:
|
||
result: 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()
|
||
self.i += 1
|
||
# 后续更深缩进的行是否归属本项
|
||
if self.i < len(self.lines):
|
||
child_indent = self._indent(self.lines[self.i])
|
||
else:
|
||
child_indent = cur
|
||
has_deeper = child_indent > cur
|
||
if item_text:
|
||
# 可能是 "- key: value"（映射项）或 "- 标量"
|
||
if ":" in item_text and not item_text.startswith("["):
|
||
# 单行映射项的首键
|
||
key, sep, rest = item_text.partition(":")
|
||
kval = _parse_scalar(rest.strip()) if rest.strip() else None
|
||
if has_deeper:
|
||
# 把首键与后续子块合并：先解析子块，再把首键塞入
|
||
sub, _ = self.parse_block(child_indent)
|
||
item: Dict[str, Any] = sub if isinstance(sub, dict) else {}
|
||
item[key.strip()] = kval
|
||
else:
|
||
item = {key.strip(): kval}
|
||
result.append(item)
|
||
else:
|
||
if has_deeper:
|
||
# 标量头 + 子块（本模板未使用，保守取子块）
|
||
sub, _ = self.parse_block(child_indent)
|
||
result.append(sub)
|
||
else:
|
||
result.append(_parse_scalar(item_text))
|
||
else:
|
||
# "- " 后跟子块
|
||
if has_deeper:
|
||
sub, _ = self.parse_block(child_indent)
|
||
result.append(sub)
|
||
return result
|
||
|
||
|
||
def _parse_scalar(text: str) -> Any:
|
||
"""解析标量：数字/布尔/字符串/内联列表/内联映射。"""
|
||
text = text.strip()
|
||
if not text:
|
||
return ""
|
||
# 内联列表
|
||
if text.startswith("[") and text.endswith("]"):
|
||
inner = text[1:-1].strip()
|
||
if not inner:
|
||
return []
|
||
return [_parse_scalar(part.strip()) for part in _split_top(inner, ",")]
|
||
# 内联映射
|
||
if text.startswith("{") and text.endswith("}"):
|
||
inner = text[1:-1].strip()
|
||
if not inner:
|
||
return {}
|
||
out: Dict[str, Any] = {}
|
||
for part in _split_top(inner, ","):
|
||
k, sep, v = part.partition(":")
|
||
if sep:
|
||
out[k.strip()] = _parse_scalar(v.strip())
|
||
return out
|
||
low = text.lower()
|
||
if low == "true":
|
||
return True
|
||
if low == "false":
|
||
return False
|
||
if low in ("null", "none", "~"):
|
||
return None
|
||
# 数字
|
||
try:
|
||
if "." in text or "e" in low:
|
||
return float(text)
|
||
return int(text)
|
||
except ValueError:
|
||
# 去引号
|
||
if len(text) >= 2 and text[0] in "\"'" and text[-1] == text[0]:
|
||
return text[1:-1]
|
||
return text
|
||
|
||
|
||
def _split_top(text: str, sep: str) -> List[str]:
|
||
"""按分隔符切分顶层（跳过 []/{} 内的）。"""
|
||
parts: List[str] = []
|
||
depth = 0
|
||
cur: List[str] = []
|
||
in_str = False
|
||
for ch in text:
|
||
if ch == '"':
|
||
in_str = not in_str
|
||
cur.append(ch)
|
||
elif not in_str and ch in "[{":
|
||
depth += 1
|
||
cur.append(ch)
|
||
elif not in_str and ch in "]}":
|
||
depth = max(0, depth - 1)
|
||
cur.append(ch)
|
||
elif ch == sep and depth == 0:
|
||
parts.append("".join(cur))
|
||
cur = []
|
||
else:
|
||
cur.append(ch)
|
||
if cur:
|
||
parts.append("".join(cur))
|
||
return parts
|