aps-agent/server/engines/base.py

# ============================================================
# 引擎抽象接口（moduleId: engines-base, 可重生 ✅）
# 对应 plan.md §9.1 ISchedulingEngine：多引擎可插拔的统一契约
# ============================================================
from __future__ import annotations                    # 前向类型引用

from abc import ABC, abstractmethod                   # 抽象基类工具
from typing import Any                                 # 类型标注
from pydantic import BaseModel, Field                  # 入参契约校验

from server.contracts import ScheduleResult            # 引擎输出契约


class EngineParams(BaseModel):
    """引擎入参：一次排产请求的全部参数（与 legacy runScheduling params 对齐）。"""
    orderIds: list[int] = Field(default_factory=list)  # 目标订单 ID（空=全部待排）
    engineType: str = "RULE"                           # 请求的引擎类型（RULE/CP/GA/HYBRID/OPTIMIZE）
    strategyTemplate: str = "COMPREHENSIVE"            # 策略模板（排序规则）
    planningHorizonDays: int = 14                      # 计划展望期（天）
    startDate: str | None = None                       # 排产起始日 YYYY-MM-DD（None=明天）
    includeUnapproved: bool = False                    # OR-03：是否纳入未批准订单（试排 what-if）
    includeForecast: bool = False                      # OR-05：是否纳入 ACTIVE 预测订单（试排）
    constraints: dict[str, bool] = Field(              # 约束开关（SC-04 从约束映射；兼容旧键）
        default_factory=lambda: {
            "materialKit": True, "equipment": True, "personnel": True, "changeover": True,
            "capacity": True, "dueDate": True, "tooling": True, "freeze": True,
        })
    triggerType: str = "MANUAL"                        # 触发类型（手动/自动滚动）
    name: str | None = None                            # 版本名称（可选）
    # SC-06：可选覆盖（None=不启用，保持历史默认行为）
    deliveryBufferRatio: float | None = None           # 交期缓冲比；<1 则有效交期提前
    freezeWindowHours: float | None = None             # 冻结窗：排产起点后移小时数
    # SC-03：CP-SAT 时限（秒）；GA/HYBRID 后续复用
    timeLimitSeconds: float | None = None


class ISchedulingEngine(ABC):
    """排产引擎统一接口（§9.1）：输入订单/主数据/约束，输出版本+工单+冲突+KPI。"""

    # 引擎标识（子类覆写）
    name: str = "BASE"
    # 是否支持时限内产出可行解（anytime，M5 的 CP/LNS 用）
    supports_anytime: bool = False

    @abstractmethod
    def solve(self, world: dict[str, Any], params: EngineParams, next_id) -> ScheduleResult:
        """执行排产（权力等级 P1：只写传入的 world 字典，落盘由调用方决定）。

        Args:
            world:   世界状态字典（引擎向其中追加版本/PO/WO/冲突/日志）
            params:  排产参数（EngineParams）
            next_id: ID 发号函数 next_id(kind)->int（由 WorldStore 提供）
        Returns:
            ScheduleResult 排产结果摘要（含证据引用）
        """
        ...                                            # 由具体引擎实现