feat(#38): 跨工序寻优模型模板化（固定主干+配方加载，PRD 5.3 ③跨工序寻优）
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,26 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""测试引导：把连字符目录 ``core/model-framework`` 加载为可导入包
|
||||
``model_framework``，使测试可 ``from model_framework import ...``。
|
||||
|
||||
与仓库内各 core 模块的测试引导同款模式（importlib 完整加载包，执行
|
||||
``__init__.py``，保持顶层导出可用）。
|
||||
"""
|
||||
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:
|
||||
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("model_framework", PKG_DIR)
|
||||
@@ -0,0 +1,340 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""跨工序寻优模型模板化单元测试（issue #38）。
|
||||
|
||||
覆盖：
|
||||
- 数据对象（DecisionVariable / Stage / Constraint / Objective / Recipe）的
|
||||
构造、校验、序列化往返；
|
||||
- 受限表达式求值 ``_safe_eval``（拒绝危险内建/属性访问）；
|
||||
- 四种求解器（grid / random / analytic / stub）的可行解搜索与目标最大化；
|
||||
- 主干 ``CrossProcessOptimizer.optimize`` + ``build_from_recipe``；
|
||||
- 采纳率口径（PRD 5.3 ≥ 60%）与可解释建议（StageSuggestion 方向）；
|
||||
- 样例配方（Ti / 树脂）均能加载并寻优跑通（验收口径）。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if HERE not in sys.path:
|
||||
sys.path.insert(0, HERE)
|
||||
|
||||
import _bootstrap # noqa: E402 加载 model_framework 包
|
||||
|
||||
from model_framework import ( # noqa: E402
|
||||
Constraint,
|
||||
CrossProcessOptError,
|
||||
CrossProcessOptimizer,
|
||||
DecisionVariable,
|
||||
Objective,
|
||||
OptimizationResult,
|
||||
Recipe,
|
||||
Stage,
|
||||
StageSuggestion,
|
||||
SOLVERS,
|
||||
build_from_recipe,
|
||||
list_sample_recipes,
|
||||
load_recipe,
|
||||
register_solver,
|
||||
sample_recipe_path,
|
||||
stub_solver,
|
||||
)
|
||||
|
||||
|
||||
def _two_stage_recipe(solver: str = "grid") -> Recipe:
|
||||
"""构造一个简单的两工序寻优配方用于测试。"""
|
||||
s1 = Stage(
|
||||
name="upstream",
|
||||
decision_vars=(
|
||||
DecisionVariable("u_temp", 100, 200, step=20, default=120),
|
||||
),
|
||||
transfer_vars=("u_yield",),
|
||||
proxy="(u_temp - 100) / 100",
|
||||
)
|
||||
s2 = Stage(
|
||||
name="downstream",
|
||||
decision_vars=(
|
||||
DecisionVariable("d_pressure", 1, 5, step=1, default=2),
|
||||
),
|
||||
transfer_vars=("quality",),
|
||||
proxy="u_yield * 0.5 + d_pressure * 0.1",
|
||||
)
|
||||
return Recipe(
|
||||
name="test-recipe",
|
||||
stages=(s1, s2),
|
||||
constraints=(
|
||||
Constraint("u_temp", "<=", 200, label="安全上限"),
|
||||
Constraint("d_pressure", ">=", 1, label="压力下限"),
|
||||
),
|
||||
objective=Objective("quality", "max", label="质量"),
|
||||
solver=solver,
|
||||
acceptance_floor=0.6,
|
||||
)
|
||||
|
||||
|
||||
class TestDataObjects(unittest.TestCase):
|
||||
"""数据对象构造、校验、序列化往返。"""
|
||||
|
||||
def test_decision_variable_grid_points(self):
|
||||
v = DecisionVariable("x", 0, 10, step=2)
|
||||
self.assertEqual(v.grid_points(), [0, 2, 4, 6, 8, 10])
|
||||
|
||||
def test_decision_variable_rejects_invalid_range(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
DecisionVariable("x", 10, 0)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
DecisionVariable("x", 0, 10, step=0)
|
||||
|
||||
def test_decision_variable_roundtrip(self):
|
||||
v = DecisionVariable("x", 1.5, 3.5, step=0.5, unit="MPa", default=2.0)
|
||||
v2 = DecisionVariable.from_dict(v.to_dict())
|
||||
self.assertEqual(v, v2)
|
||||
|
||||
def test_constraint_operators(self):
|
||||
ns = {"x": 5}
|
||||
self.assertTrue(Constraint("x", "<=", 5).satisfied(ns))
|
||||
self.assertTrue(Constraint("x", ">=", 5).satisfied(ns))
|
||||
self.assertTrue(Constraint("x", "==", 5).satisfied(ns))
|
||||
self.assertFalse(Constraint("x", "<=", 4).satisfied(ns))
|
||||
self.assertFalse(Constraint("x", ">=", 6).satisfied(ns))
|
||||
|
||||
def test_constraint_rejects_bad_op(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Constraint("x", "!=", 0)
|
||||
|
||||
def test_objective_score_min_inverts(self):
|
||||
obj = Objective("x", "min")
|
||||
# 最小化：x=5 的标准化分数应为 -5（越大越好 = 越小原值）
|
||||
self.assertAlmostEqual(obj.score({"x": 5}), -5.0)
|
||||
|
||||
def test_objective_rejects_bad_sense(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Objective("x", "avg")
|
||||
|
||||
def test_recipe_requires_stages(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=())
|
||||
|
||||
def test_recipe_rejects_bad_solver(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=(Stage(name="s"),), solver="magic")
|
||||
|
||||
def test_recipe_rejects_bad_acceptance(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=(Stage(name="s"),), acceptance_floor=1.5)
|
||||
|
||||
def test_recipe_roundtrip(self):
|
||||
r = _two_stage_recipe()
|
||||
r2 = Recipe.from_dict(r.to_dict())
|
||||
self.assertEqual(r, r2)
|
||||
self.assertEqual(r2.stages[0].decision_vars[0].name, "u_temp")
|
||||
|
||||
|
||||
class TestSafeEval(unittest.TestCase):
|
||||
"""受限表达式求值安全性。"""
|
||||
|
||||
def test_safe_eval_basic(self):
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
self.assertAlmostEqual(_safe_eval("1 + 2 * 3", {}), 7.0)
|
||||
self.assertAlmostEqual(_safe_eval("x + y", {"x": 1, "y": 2}), 3.0)
|
||||
self.assertAlmostEqual(_safe_eval("min(x, y)", {"x": 1, "y": 2}), 1.0)
|
||||
|
||||
def test_safe_eval_rejects_empty(self):
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
_safe_eval("", {})
|
||||
|
||||
def test_safe_eval_rejects_builtins(self):
|
||||
"""禁止访问 __import__ / open / 任意内建（沙箱保护）。"""
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
with self.assertRaises(Exception):
|
||||
_safe_eval("__import__('os')", {})
|
||||
with self.assertRaises(Exception):
|
||||
_safe_eval("open('x')", {})
|
||||
|
||||
|
||||
class TestSolvers(unittest.TestCase):
|
||||
"""四种求解器的可行解搜索与目标最大化。"""
|
||||
|
||||
def test_grid_solver_finds_feasible(self):
|
||||
r = _two_stage_recipe("grid")
|
||||
opt = CrossProcessOptimizer(r)
|
||||
res = opt.optimize()
|
||||
self.assertIsInstance(res, OptimizationResult)
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertGreaterEqual(res.objective_score, res.baseline_score)
|
||||
|
||||
def test_grid_solver_no_feasible_raises(self):
|
||||
# 矛盾约束：温度必须同时 <= 100 且 >= 200
|
||||
r = Recipe(
|
||||
name="infeasible",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 100, 300, step=50, default=150),)),),
|
||||
constraints=(Constraint("x", "<=", 100), Constraint("x", ">=", 200)),
|
||||
objective=Objective("x", "max"),
|
||||
solver="grid",
|
||||
)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
def test_random_solver_finds_feasible(self):
|
||||
r = _two_stage_recipe("random")
|
||||
res = CrossProcessOptimizer(r).optimize(seed=42)
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertEqual(res.solver, "random")
|
||||
|
||||
def test_random_solver_uses_solver_params(self):
|
||||
r = _two_stage_recipe("random")
|
||||
r = Recipe.from_dict({**r.to_dict(),
|
||||
"solver_params": {"n_samples": 50, "seed": 7}})
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
|
||||
def test_analytic_solver_single_var(self):
|
||||
# 单变量线性最大化目标：应在 high 边界取得最优
|
||||
r = Recipe(
|
||||
name="single",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 0, 10, step=1, default=2),)),),
|
||||
objective=Objective("x", "max", label="越大越好"),
|
||||
solver="analytic",
|
||||
)
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
self.assertEqual(res.objective_score, 10.0)
|
||||
# 建议把 x 从默认 2 上调到 10
|
||||
sug = res.suggestions[0]
|
||||
self.assertEqual(sug.new_value, 10.0)
|
||||
self.assertEqual(sug.direction, "上调")
|
||||
|
||||
def test_analytic_falls_back_to_grid_for_multi_var(self):
|
||||
r = _two_stage_recipe("analytic")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
# 多变量时 analytic 退化为 grid，仍能跑通
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
|
||||
def test_analytic_no_feasible_raises(self):
|
||||
r = Recipe(
|
||||
name="bad",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 0, 10, step=1, default=5),)),),
|
||||
constraints=(Constraint("x", ">=", 100),),
|
||||
objective=Objective("x", "max"),
|
||||
solver="analytic",
|
||||
)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
def test_stub_solver_returns_default(self):
|
||||
r = _two_stage_recipe("stub")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
# stub 直接取默认值，改善为 0
|
||||
self.assertEqual(res.improvement, 0.0)
|
||||
self.assertEqual(res.solver, "stub")
|
||||
|
||||
def test_unknown_solver_raises(self):
|
||||
r = Recipe.from_dict({**_two_stage_recipe().to_dict(), "solver": "grid"})
|
||||
# 临时篡改 recipe.solver 为非法值（绕过校验）测主干分支
|
||||
object.__setattr__(r, "solver", "voodoo")
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
|
||||
class TestAcceptanceAndSuggestions(unittest.TestCase):
|
||||
"""采纳率口径（PRD 5.3 ≥ 60%）与可解释建议。"""
|
||||
|
||||
def test_grid_improvement_marks_accepted(self):
|
||||
r = _two_stage_recipe("grid")
|
||||
# 默认值非最优，grid 应能找到更优解 → accepted
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
if res.improvement > 1e-9:
|
||||
self.assertTrue(res.accepted)
|
||||
self.assertGreaterEqual(res.acceptance, res.acceptance_floor)
|
||||
|
||||
def test_suggestion_direction(self):
|
||||
s_up = StageSuggestion("s", "x", 1.0, 3.0, 2.0)
|
||||
self.assertEqual(s_up.direction, "上调")
|
||||
s_down = StageSuggestion("s", "x", 3.0, 1.0, -2.0)
|
||||
self.assertEqual(s_down.direction, "下调")
|
||||
s_keep = StageSuggestion("s", "x", 2.0, 2.0, 0.0)
|
||||
self.assertEqual(s_keep.direction, "保持")
|
||||
|
||||
def test_result_to_dict_serializable(self):
|
||||
r = _two_stage_recipe("stub")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
d = res.to_dict()
|
||||
# 可 JSON 序列化
|
||||
json.dumps(d)
|
||||
self.assertIn("suggestions", d)
|
||||
self.assertIn("accepted", d)
|
||||
|
||||
|
||||
class TestSampleRecipes(unittest.TestCase):
|
||||
"""样例配方（Ti / 树脂）加载与寻优（验收口径）。"""
|
||||
|
||||
def test_sample_recipes_listed(self):
|
||||
names = list_sample_recipes()
|
||||
self.assertIn("recipe.ti.json", names)
|
||||
self.assertIn("recipe.resin.json", names)
|
||||
|
||||
def test_ti_recipe_loads_and_optimizes(self):
|
||||
opt = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
res = opt.optimize()
|
||||
self.assertEqual(res.solver, "grid")
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertGreaterEqual(res.objective_score, res.baseline_score)
|
||||
# 工序建议覆盖三道工序
|
||||
stages_covered = {s.stage for s in res.suggestions}
|
||||
self.assertEqual(stages_covered, {"氯化", "精制", "还原"})
|
||||
|
||||
def test_resin_recipe_loads_and_optimizes(self):
|
||||
opt = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
res = opt.optimize()
|
||||
self.assertEqual(res.solver, "random")
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
stages_covered = {s.stage for s in res.suggestions}
|
||||
self.assertEqual(stages_covered, {"反应", "水洗", "干燥"})
|
||||
|
||||
def test_two_recipes_same_engine_class(self):
|
||||
"""验收口径：同框架加载两套配方，寻优主干类零改动。"""
|
||||
opt_ti = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
opt_resin = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
self.assertIs(type(opt_ti), type(opt_resin))
|
||||
# 两套配方的工序拓扑确实不同
|
||||
self.assertNotEqual(opt_ti.recipe_meta["stages"],
|
||||
opt_resin.recipe_meta["stages"])
|
||||
|
||||
def test_load_recipe_from_temp_file(self):
|
||||
r = _two_stage_recipe()
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".json", delete=False, encoding="utf-8") as fh:
|
||||
json.dump(r.to_dict(), fh, ensure_ascii=False)
|
||||
path = fh.name
|
||||
try:
|
||||
r2 = load_recipe(path)
|
||||
self.assertEqual(r, r2)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
class TestRegisterSolver(unittest.TestCase):
|
||||
"""插件式求解器注册。"""
|
||||
|
||||
def test_register_custom_solver(self):
|
||||
called = {"n": 0}
|
||||
|
||||
def my_solver(recipe, **kw):
|
||||
called["n"] += 1
|
||||
return stub_solver(recipe, **kw)
|
||||
|
||||
register_solver("my", my_solver)
|
||||
self.assertIn("my", SOLVERS)
|
||||
# 直接构造主干并替换 recipe.solver 为已注册的自定义求解器
|
||||
r = _two_stage_recipe()
|
||||
object.__setattr__(r, "solver", "my")
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
self.assertEqual(called["n"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user