From 2aa70d078bcda956ea9faa682e30c48a81e0d3a5 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 03:58:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#78=20[Ti-?= =?UTF-8?q?2]=20=E9=85=8D=E6=96=B9=E4=BC=98=E5=8C=96=E9=97=AE=E9=A2=98?= =?UTF-8?q?=E5=BB=BA=E6=A8=A1=EF=BC=88=E5=86=B3=E7=AD=96=E5=8F=98=E9=87=8F?= =?UTF-8?q?/=E7=9B=AE=E6=A0=87/=E7=BA=A6=E6=9D=9F=E5=A3=B0=E6=98=8E?= =?UTF-8?q?=E5=BC=8F=E8=A7=84=E6=A0=BC+=E6=B1=82=E8=A7=A3=E5=99=A8?= =?UTF-8?q?=E6=97=A0=E5=85=B3+=E9=9B=B6=E4=BE=9D=E8=B5=96YAML=E5=8A=A0?= =?UTF-8?q?=E8=BD=BD=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- templates/ti-cl4/recipe-optim/README.md | 41 + templates/ti-cl4/recipe-optim/__init__.py | 5 + .../ti-cl4/recipe-optim/_sanity_check.py | 61 ++ .../config/recipe_optim.template.yaml | 68 ++ templates/ti-cl4/recipe-optim/problem.py | 784 ++++++++++++++++++ .../ti-cl4/recipe-optim/tests/_bootstrap.py | 26 + .../ti-cl4/recipe-optim/tests/test_problem.py | 291 +++++++ 7 files changed, 1276 insertions(+) create mode 100644 templates/ti-cl4/recipe-optim/README.md create mode 100644 templates/ti-cl4/recipe-optim/__init__.py create mode 100644 templates/ti-cl4/recipe-optim/_sanity_check.py create mode 100644 templates/ti-cl4/recipe-optim/config/recipe_optim.template.yaml create mode 100644 templates/ti-cl4/recipe-optim/problem.py create mode 100644 templates/ti-cl4/recipe-optim/tests/_bootstrap.py create mode 100644 templates/ti-cl4/recipe-optim/tests/test_problem.py diff --git a/templates/ti-cl4/recipe-optim/README.md b/templates/ti-cl4/recipe-optim/README.md new file mode 100644 index 0000000..d065ae5 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/README.md @@ -0,0 +1,41 @@ +# Ti-2 配方动态优化（recipe-optim） + +> 对应 issue #78（问题建模）/ #79（求解器集成）/ #80（跨工序寻优）/ #81（可解释建议）。 +> PRD §5.3 ②「配方动态优化」：`入：质量目标 + 约束；出：参数/配方建议`，二期交付。 + +本目录沉淀**配方优化**的声明式模板资产 + 求解器无关的问题建模引擎，跨行业差异落 +资产（YAML 模板），引擎零改动（PRD「超参包驱动」「模板化技术路径」）。 + +## 模块 + +- `problem.py` — 优化问题建模（**#78**）：决策变量 / 目标 / 约束的声明式规格 + + 校验 + 可行性判定 + 零依赖 YAML 子集加载。 +- `config/recipe_optim.template.yaml` — Template-Ti 配方优化模板资产。 +- `tests/` — 单元测试（`python -m unittest discover -s tests`）。 +- `_sanity_check.py` — 部署期一键自检（5 能力点）。 + +## 设计 + +1. **决策变量 `DecisionVariable`**：`bounds`（连续区间）/`choices`（离散枚举）， + 带 `meaning`/`unit`/`initial`，供 #81 可解释建议引用。 +2. **目标 `ObjectiveSpec`**：线性加权（min/max）+ PRD 超参包 `target` 字段。 +3. **约束 `ConstraintSpec`**：统一描述 box / linear / ratio / forbidden 工艺约束， + 每条带 `reason`（工艺依据，对齐 PRD"可解释、可溯源、引用依据"）。 +4. **问题 `OptimizationProblem`**：`validate` 聚合静态校验、`is_feasible` / + `violated_constraints` 做可行性判定，`solve` 留给 #79 注入求解器（求解器无关）。 + +## 与上下游的契约 + +- 下游 #79 求解器：消费 `OptimizationProblem`，产出满足约束的最优取值。 +- 下游 #80 跨工序寻优：复用同一变量/目标/约束模型描述跨工序关联。 +- 下游 #81 可解释建议：读 `violated_constraints` + 变量 `meaning` + 约束 `reason` + 产出"可溯源"建议。 + +## 运行 + +```bash +python -m unittest discover -s tests # 单元测试 +python _sanity_check.py # 部署期自检 +``` + +零第三方依赖（纯标准库），与内核既有模块一致，便于离线/隔离网部署。 diff --git a/templates/ti-cl4/recipe-optim/__init__.py b/templates/ti-cl4/recipe-optim/__init__.py new file mode 100644 index 0000000..bd0305a --- /dev/null +++ b/templates/ti-cl4/recipe-optim/__init__.py @@ -0,0 +1,5 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方动态优化模板包（Issue #78/#79/#80/#81）。 + +- ``problem``：优化问题建模（变量/目标/约束），求解器无关（#78）。 +""" diff --git a/templates/ti-cl4/recipe-optim/_sanity_check.py b/templates/ti-cl4/recipe-optim/_sanity_check.py new file mode 100644 index 0000000..eac776c --- /dev/null +++ b/templates/ti-cl4/recipe-optim/_sanity_check.py @@ -0,0 +1,61 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方优化问题建模 自检脚本（Issue #78）。 + +不依赖 unittest，直接加载模板资产并做能力点断言，便于 CI / 部署期一键核对。 +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from problem import ConstraintKind, OptimizationProblem, load_problem # noqa: E402 + +CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "config", "recipe_optim.template.yaml") + + +def main() -> int: + failures = [] + + # 1) 模板能加载 + p = load_problem(CONFIG) + if not isinstance(p, OptimizationProblem): + failures.append("load_problem 未返回 OptimizationProblem") + + # 2) 4 类约束齐全 + kinds = {c.kind for c in p.constraints} + expected = {ConstraintKind.BOX, ConstraintKind.LINEAR, + ConstraintKind.RATIO, ConstraintKind.FORBIDDEN} + if kinds != expected: + failures.append(f"约束种类不齐: {kinds} != {expected}") + + # 3) 静态校验通过 + errs = p.validate() + if errs: + failures.append(f"validate 未通过: {errs}") + + # 4) 可行性判定：合法取值可行、禁止组合不可行 + ok = p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0, + "feed_rate": 450, "catalyst": "A"}) + bad = p.is_feasible({"clf_temp": 950, "cl2_ratio": 1.0, + "feed_rate": 450, "catalyst": "A"}) + if not ok: + failures.append("合法取值被判为不可行") + if bad: + failures.append("越界取值(950℃)未被识别为不可行") + + # 5) 序列化往返无损 + rt = OptimizationProblem.from_dict(p.to_dict()) + if [v.name for v in rt.variables] != [v.name for v in p.variables]: + failures.append("序列化往返丢失变量") + + if failures: + print("❌ recipe-optim 自检失败:") + for f in failures: + print(" -", f) + return 1 + print("✅ recipe-optim 自检通过（5 能力点）") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/templates/ti-cl4/recipe-optim/config/recipe_optim.template.yaml b/templates/ti-cl4/recipe-optim/config/recipe_optim.template.yaml new file mode 100644 index 0000000..5c1a63d --- /dev/null +++ b/templates/ti-cl4/recipe-optim/config/recipe_optim.template.yaml @@ -0,0 +1,68 @@ +# Ti-2 配方动态优化 · 优化问题模板（Issue #78 / PRD 5.3 ②） +# +# 声明式描述「给定质量目标 + 工艺约束，求最优配方/参数」： +# 入：质量目标 + 约束；出：参数/配方建议（PRD §5.3 架构表）。 +# 切换行业/车间只改本模板资产，引擎零改动（PRD「超参包驱动」）。 +# +# 本示例对应「氯化车间/海绵钛（Template-Ti）」二期配方优化场景： +# 在 TiCl₄ 纯度达标（≥ target_value）前提下，寻优反应温度/氯气配比， +# 同时满足工艺约束（温度上限、氯气/钛配比、禁止组合）。 +problem_id: recipe_optim_ti +template: iAOP-Template-Ti +description: 氯化车间配方优化（二期，数据就绪后接 #79 求解器） + +variables: + - name: clf_temp + kind: bounds + meaning: 氯化炉反应温度 + unit: "℃" + bounds: [800, 920] + initial: 860 + - name: cl2_ratio + kind: bounds + meaning: 氯气与高钛渣配比 + unit: "ratio" + bounds: [0.8, 1.4] + initial: 1.0 + - name: feed_rate + kind: bounds + meaning: 进料速率 + unit: "kg/h" + bounds: [300, 600] + initial: 450 + integer: false + - name: catalyst + kind: choices + meaning: 催化剂型号 + choices: ["A", "B", "C"] + +objective: + sense: maximize + target: Ti_purity + target_value: 99.5 + description: 最大化 TiCl₄ 纯度（PRD 超参包 target 字段） + terms: + - variable: clf_temp + coefficient: 0.01 + - variable: cl2_ratio + coefficient: 2.0 + +constraints: + - kind: box + variable: clf_temp + bounds: [820, 900] + reason: 反应温度运行安全区间（运行期收紧，低于 820 反应不充分、高于 900 副产物激增） + - kind: linear + coefficients: {clf_temp: 1.0, feed_rate: -0.5} + op: "<=" + rhs: 700 + reason: 温度-进料耦合上限（防止局部过热） + - kind: ratio + numerator: cl2_ratio + denominator: cl2_ratio + op: ">=" + value: 0.0 + reason: 配比非负（占位示例，真实配比约束见工艺手册） + - kind: forbidden + combination: {catalyst: "C", clf_temp: 910} + reason: C 型催化剂禁止与 910℃ 高温组合（安全告警） diff --git a/templates/ti-cl4/recipe-optim/problem.py b/templates/ti-cl4/recipe-optim/problem.py new file mode 100644 index 0000000..cae6d0a --- /dev/null +++ b/templates/ti-cl4/recipe-optim/problem.py @@ -0,0 +1,784 @@ +# -*- 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 diff --git a/templates/ti-cl4/recipe-optim/tests/_bootstrap.py b/templates/ti-cl4/recipe-optim/tests/_bootstrap.py new file mode 100644 index 0000000..d52d096 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/_bootstrap.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""测试引导：把连字符目录挂载为可导入包（与 core 模块同款模式）。 + +- ``templates/ti-cl4/recipe-optim`` → 包名 ``recipe_optim``。 +本引擎零内核依赖（纯标准库），仅挂载自身包即可。 +""" +import importlib.util +import os +import sys + +PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +def _load_package(name: str, path: str) -> None: + """按文件路径完整加载一个包（执行其 __init__.py）。""" + if name in sys.modules: + return + init_py = os.path.join(path, "__init__.py") + spec = importlib.util.spec_from_file_location( + name, init_py, submodule_search_locations=[path]) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + + +_load_package("recipe_optim", PKG_DIR) diff --git a/templates/ti-cl4/recipe-optim/tests/test_problem.py b/templates/ti-cl4/recipe-optim/tests/test_problem.py new file mode 100644 index 0000000..71ddce6 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_problem.py @@ -0,0 +1,291 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方优化问题建模 单元测试（Issue #78）。 + +覆盖： +- 决策变量域（bounds/choices/integer/越界/夹紧）； +- 目标函数（线性求值、min/max、target 记录）； +- 约束（box/linear/ratio/forbidden 可行性判定 + 未知变量静态校验）； +- OptimizationProblem（聚合校验、可行性、违反约束枚举、序列化往返）； +- load_problem YAML 子集加载（模板资产）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: E402 挂载 recipe_optim 包 + +from recipe_optim.problem import ( # noqa: E402 + ConstraintKind, + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + ObjectiveTerm, + OptimizationProblem, + ProblemError, + Sense, + load_problem, +) + +CONFIG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "recipe_optim.template.yaml", +) + + +def _ti_problem() -> OptimizationProblem: + """构造一份与 config 同构的内存问题（用于无需 YAML 的断言）。""" + return OptimizationProblem( + variables=[ + DecisionVariable("clf_temp", DomainKind.BOUNDS, "温度", "℃", + bounds=(800, 920), initial=860), + DecisionVariable("cl2_ratio", DomainKind.BOUNDS, "配比", "ratio", + bounds=(0.8, 1.4), initial=1.0), + DecisionVariable("catalyst", DomainKind.CHOICES, "催化剂", "", + choices=["A", "B", "C"]), + ], + objective=ObjectiveSpec( + sense=Sense.MAXIMIZE, + target="Ti_purity", + target_value=99.5, + terms=[ObjectiveTerm("clf_temp", 0.01), ObjectiveTerm("cl2_ratio", 2.0)], + ), + constraints=[ + ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", bounds=(820, 900), + reason="温度安全区间"), + ConstraintSpec(ConstraintKind.LINEAR, coefficients={"clf_temp": 1.0}, + op="<=", rhs=900), + ConstraintSpec(ConstraintKind.FORBIDDEN, combination={"catalyst": "C"}), + ], + ) + + +class TestDecisionVariable(unittest.TestCase): + def test_bounds_valid_and_contains(self): + v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 10.0)) + self.assertTrue(v.contains(5)) + self.assertTrue(v.contains(0)) + self.assertTrue(v.contains(10)) + self.assertFalse(v.contains(-0.1)) + self.assertFalse(v.contains(10.1)) + self.assertFalse(v.contains("x")) + + def test_bounds_reversed_rejected(self): + with self.assertRaises(ProblemError): + DecisionVariable("t", DomainKind.BOUNDS, bounds=(10.0, 0.0)) + + def test_choices_domain(self): + v = DecisionVariable("cat", DomainKind.CHOICES, choices=["A", "B"]) + self.assertTrue(v.contains("A")) + self.assertFalse(v.contains("Z")) + + def test_integer_bounds_noninteger_rejected(self): + with self.assertRaises(ProblemError): + DecisionVariable("n", DomainKind.BOUNDS, bounds=(1.5, 5.0), integer=True) + + def test_integer_contains_and_clamp(self): + v = DecisionVariable("n", DomainKind.BOUNDS, bounds=(0.0, 10.0), integer=True) + self.assertFalse(v.contains(1.5)) + self.assertTrue(v.contains(3)) + # clamp 把越界夹回 + 取整 + self.assertEqual(v.clamp(12.4), 10.0) + self.assertEqual(v.clamp(-3), 0.0) + self.assertEqual(v.clamp(4.7), 5.0) + + def test_continuous_clamp(self): + v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 10.0)) + self.assertEqual(v.clamp(15), 10) + self.assertEqual(v.clamp(-2), 0) + + def test_empty_name_rejected(self): + with self.assertRaises(ProblemError): + DecisionVariable(" ", DomainKind.BOUNDS, bounds=(0, 1)) + + def test_missing_domain_payload_rejected(self): + with self.assertRaises(ProblemError): + DecisionVariable("t", DomainKind.BOUNDS) + with self.assertRaises(ProblemError): + DecisionVariable("t", DomainKind.CHOICES, choices=[]) + + def test_roundtrip(self): + v = DecisionVariable("t", DomainKind.BOUNDS, "温度", "℃", + bounds=(1.0, 2.0), initial=1.5) + v2 = DecisionVariable.from_dict(v.to_dict()) + self.assertEqual(v2.bounds, v.bounds) + self.assertEqual(v2.initial, v.initial) + + +class TestObjective(unittest.TestCase): + def test_evaluate_maximize(self): + obj = ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("a", 2.0), + ObjectiveTerm("b", -1.0)]) + self.assertEqual(obj.evaluate({"a": 3, "b": 1}), 5.0) + # 未知变量按 0 + self.assertEqual(obj.evaluate({"a": 3}), 6.0) + + def test_minimize_label(self): + self.assertEqual(Sense.MINIMIZE.label, "最小化") + self.assertEqual(Sense.MAXIMIZE.label, "最大化") + + def test_unknown_sense_rejected(self): + with self.assertRaises(ProblemError): + ObjectiveSpec.from_dict({"sense": "extreme"}) + + def test_roundtrip_with_target(self): + obj = ObjectiveSpec(Sense.MAXIMIZE, target="Ti_purity", target_value=99.5, + terms=[ObjectiveTerm("a", 0.5)]) + obj2 = ObjectiveSpec.from_dict(obj.to_dict()) + self.assertEqual(obj2.target, "Ti_purity") + self.assertEqual(obj2.target_value, 99.5) + self.assertEqual(obj2.terms[0].coefficient, 0.5) + + +class TestConstraint(unittest.TestCase): + def test_box_satisfied(self): + c = ConstraintSpec(ConstraintKind.BOX, variable="t", bounds=(0, 10)) + self.assertTrue(c.satisfied_by({"t": 5})) + self.assertFalse(c.satisfied_by({"t": 11})) + # 未知取值不判 + self.assertTrue(c.satisfied_by({})) + + def test_linear_ops(self): + c = ConstraintSpec(ConstraintKind.LINEAR, coefficients={"a": 1, "b": 1}, + op="<=", rhs=10) + self.assertTrue(c.satisfied_by({"a": 4, "b": 6})) + self.assertFalse(c.satisfied_by({"a": 6, "b": 6})) + # 部分未知视为未约束 + self.assertTrue(c.satisfied_by({"a": 4})) + + def test_ratio_constraint(self): + c = ConstraintSpec(ConstraintKind.RATIO, numerator="x", denominator="y", + op=">=", value=0.5) + self.assertTrue(c.satisfied_by({"x": 1, "y": 2})) + self.assertFalse(c.satisfied_by({"x": 1, "y": 4})) + # 分母为 0 视为未约束 + self.assertTrue(c.satisfied_by({"x": 1, "y": 0})) + + def test_forbidden_combination(self): + c = ConstraintSpec(ConstraintKind.FORBIDDEN, combination={"cat": "C"}) + self.assertFalse(c.satisfied_by({"cat": "C"})) + self.assertTrue(c.satisfied_by({"cat": "A"})) + # 多键需全部命中才算触发 + c2 = ConstraintSpec(ConstraintKind.FORBIDDEN, + combination={"cat": "C", "t": 900}) + self.assertTrue(c2.satisfied_by({"cat": "C", "t": 100})) + self.assertFalse(c2.satisfied_by({"cat": "C", "t": 900})) + + def test_linear_bad_op_rejected(self): + with self.assertRaises(ProblemError): + ConstraintSpec(ConstraintKind.LINEAR, coefficients={"a": 1}, op="~=") + with self.assertRaises(ProblemError): + ConstraintSpec(ConstraintKind.LINEAR, op="<=") + + def test_errors_against_unknown_var(self): + c = ConstraintSpec(ConstraintKind.BOX, variable="missing", bounds=(0, 1)) + self.assertTrue(c.errors_against({"t": object()})) + + def test_roundtrip(self): + c = ConstraintSpec(ConstraintKind.LINEAR, reason="x", + coefficients={"a": 1.0}, op=">=", rhs=5) + c2 = ConstraintSpec.from_dict(c.to_dict()) + self.assertEqual(c2.op, ">=") + self.assertEqual(c2.rhs, 5) + + +class TestOptimizationProblem(unittest.TestCase): + def test_validate_ok(self): + self.assertEqual(_ti_problem().validate(), []) + + def test_validate_duplicate_variable(self): + p = _ti_problem() + p.variables.append(DecisionVariable("clf_temp", DomainKind.BOUNDS, + bounds=(0, 1))) + errs = p.validate() + self.assertTrue(any("重复定义" in e for e in errs)) + + def test_validate_unknown_var_in_objective(self): + p = _ti_problem() + p.objective.terms.append(ObjectiveTerm("nope")) + errs = p.validate() + self.assertTrue(any("nope" in e for e in errs)) + + def test_validate_unknown_var_in_constraint(self): + p = _ti_problem() + p.constraints.append(ConstraintSpec(ConstraintKind.BOX, variable="ghost", + bounds=(0, 1))) + errs = p.validate() + self.assertTrue(any("ghost" in e for e in errs)) + + def test_is_feasible_and_violated(self): + p = _ti_problem() + # 合法取值 + self.assertTrue(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0, + "catalyst": "A"})) + # 越出 box 收紧（820~900） + self.assertFalse(p.is_feasible({"clf_temp": 910, "cl2_ratio": 1.0, + "catalyst": "A"})) + # forbidden 组合 + self.assertFalse(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0, + "catalyst": "C"})) + viol = p.violated_constraints({"clf_temp": 950, "cl2_ratio": 1.0, + "catalyst": "A"}) + self.assertGreater(len(viol), 0) + + def test_roundtrip(self): + p = _ti_problem() + p2 = OptimizationProblem.from_dict(p.to_dict()) + self.assertEqual([v.name for v in p2.variables], + [v.name for v in p.variables]) + self.assertEqual(p2.objective.sense, p.objective.sense) + self.assertEqual(len(p2.constraints), len(p.constraints)) + + +class TestLoadProblemYaml(unittest.TestCase): + def test_load_template(self): + self.assertTrue(os.path.exists(CONFIG_PATH), f"缺少模板 {CONFIG_PATH}") + p = load_problem(CONFIG_PATH) + names = [v.name for v in p.variables] + self.assertEqual(names, ["clf_temp", "cl2_ratio", "feed_rate", "catalyst"]) + # 校验通过 + self.assertEqual(p.validate(), []) + # 目标与约束就位 + self.assertEqual(p.objective.target, "Ti_purity") + self.assertEqual(p.objective.target_value, 99.5) + kinds = {c.kind for c in p.constraints} + self.assertEqual(kinds, {ConstraintKind.BOX, ConstraintKind.LINEAR, + ConstraintKind.RATIO, ConstraintKind.FORBIDDEN}) + # 合法初值可行 + self.assertTrue(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0, + "feed_rate": 450, "catalyst": "A"})) + + def test_load_choices_parsed(self): + p = load_problem(CONFIG_PATH) + cat = p.variable_map["catalyst"] + self.assertEqual(cat.kind, DomainKind.CHOICES) + self.assertEqual(cat.choices, ["A", "B", "C"]) + + def test_integer_flag_parsed(self): + # feed_rate integer: false + p = load_problem(CONFIG_PATH) + self.assertFalse(p.variable_map["feed_rate"].integer) + + +class TestScalarParser(unittest.TestCase): + """直接覆盖 _parse_scalar 的边界（内联列表/映射/数字/字符串/布尔）。""" + + def test_scalars(self): + from recipe_optim.problem import _parse_scalar + self.assertEqual(_parse_scalar("1"), 1) + self.assertEqual(_parse_scalar("1.5"), 1.5) + self.assertIs(_parse_scalar("true"), True) + self.assertIs(_parse_scalar("False"), False) + self.assertEqual(_parse_scalar("99.5"), 99.5) + self.assertEqual(_parse_scalar('"A"'), "A") + self.assertEqual(_parse_scalar("[1, 2, 3]"), [1, 2, 3]) + self.assertEqual(_parse_scalar("[A, B]"), ["A", "B"]) + self.assertEqual(_parse_scalar("{a: 1, b: 2}"), {"a": 1, "b": 2}) + + +if __name__ == "__main__": + unittest.main()