"""V2-native CP-SAT solver for the APS closed loop (round 89, slice 1). 输入是 APS 的 `SchedulingProblemV2`，输出是 `SchedulingSolutionV2`：资源分配和 时序由 OR-Tools CP-SAT 决定，准入、校验、版本物化和审计仍然全部归 APS。 本切片只做「问题 -> 解」的原生求解：不写 flex* 行、不物化版本，也不改 WorldStore。候选一旦物化，仍必须经过 `validate_solution` 才会成为 APS 版本。 """ from __future__ import annotations import math import time from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Any from server.aps_domain.scheduling_problem_v2 import ( PeggingAllocation, ResourceKind, ScheduledActivity, ScheduledResourceAllocation, SchedulingProblemV2, SchedulingSolutionV2, SolveStatus, SolutionProvenance, UnscheduledRequirement, scheduling_problem_hash, ) SOLVER_ID = "optimize-cpsat" SOLVER_VERSION = "0.1.0" DEFAULT_TIME_LIMIT_SECONDS = 20.0 DEFAULT_SEED = 42 @dataclass class CpsatOutcome: """一次 CP-SAT 求解的解与可审计元数据。""" solution: SchedulingSolutionV2 meta: dict[str, Any] = field(default_factory=dict) def _minutes_between(base: datetime, moment: datetime | None) -> int | None: """把时间点换算成相对基准的整数分钟（向下取整）。""" if moment is None: return None return int(math.floor((moment - base).total_seconds() / 60.0)) def _minutes_ceil(base: datetime, moment: datetime | None) -> int | None: """把时间点换算成相对基准的整数分钟（向上取整，用于日历右端点）。""" if moment is None: return None return int(math.ceil((moment - base).total_seconds() / 60.0)) def _merge_windows(windows: list[tuple[int, int]]) -> list[tuple[int, int]]: merged: list[tuple[int, int]] = [] for start, end in sorted(windows): if end <= start: continue if merged and start <= merged[-1][1]: merged[-1] = (merged[-1][0], max(merged[-1][1], end)) else: merged.append((start, end)) return merged def _blocked_windows(resource: Any, base: datetime, horizon: int) -> list[tuple[int, int]]: """资源在 [0, horizon] 内不可排的分钟区间 = 日历与维保之外的补集。""" open_windows: list[tuple[int, int]] = [] for interval in resource.calendarIntervals: start = _minutes_between(base, interval.start) end = _minutes_ceil(base, interval.end) if start is None or end is None: continue start = max(0, start) end = min(horizon, end) if end <= start: continue open_windows.append((start, end)) maintenance: list[tuple[int, int]] = [] for interval in resource.maintenanceIntervals: start = _minutes_between(base, interval.start) end = _minutes_ceil(base, interval.end) if start is None or end is None: continue start, end = max(0, start), min(horizon, end) if end > start: maintenance.append((start, end)) merged_open = _merge_windows(open_windows) blocked: list[tuple[int, int]] = [] cursor = 0 for start, end in merged_open: if start > cursor: blocked.append((cursor, start)) cursor = max(cursor, end) if cursor < horizon: blocked.append((cursor, horizon)) return _merge_windows([*blocked, *maintenance]) def _duration_minutes(value: float) -> int: return max(1, int(math.ceil(float(value)))) def _eligible_resources(activity: Any, resource_ids: set[str]) -> list[str]: candidates: list[str] = [] for requirement in activity.resourceRequirements: if requirement.kind != ResourceKind.EQUIPMENT: continue candidates.extend(requirement.eligibleResourceIds) if not candidates: candidates.extend(activity.eligibleResourceIds) unique = [rid for rid in dict.fromkeys(candidates) if rid in resource_ids] return unique def solve_problem_v2( problem: SchedulingProblemV2, *, time_limit_seconds: float = DEFAULT_TIME_LIMIT_SECONDS, seed: int = DEFAULT_SEED, horizon_minutes: int | None = None, ) -> CpsatOutcome: """用 CP-SAT 求一个 `SchedulingProblemV2` 候选解。 目标：最小化总拖期（`objectivePolicy` 里 tardiness 权重）。资源分配为每个工序 在合格设备中选一台，同设备工序不重叠，工序链按前驱顺序串行，并且不允许落在 设备日历与维保之外。 """ from ortools.sat.python import cp_model started = time.perf_counter() base = problem.planningStart span = int(math.floor((problem.planningEnd - base).total_seconds() / 60.0)) horizon = span if horizon_minutes is None else min(span, int(horizon_minutes)) if horizon <= 0: raise ValueError("planning window must be positive") resources_by_id = {resource.resourceId: resource for resource in problem.resources} equipment_resources = [r for r in problem.resources if r.kind == ResourceKind.EQUIPMENT] equipment_ids = {r.resourceId for r in equipment_resources} model = cp_model.CpModel() start_vars: dict[str, Any] = {} end_vars: dict[str, Any] = {} presence: dict[tuple[str, str], Any] = {} intervals_by_resource: dict[str, list[Any]] = {r.resourceId: [] for r in equipment_resources} unschedulable: list[str] = [] for activity in problem.activities: duration = _duration_minutes(activity.durationMin) eligible = _eligible_resources(activity, equipment_ids) if not eligible or duration > horizon: unschedulable.append(activity.activityId) continue release = int(max(0, _minutes_between(base, activity.materialReleaseAt) or 0)) if release + duration > horizon: unschedulable.append(activity.activityId) continue start = model.NewIntVar(release, horizon - duration, f"start:{activity.activityId}") end = model.NewIntVar(release + duration, horizon, f"end:{activity.activityId}") model.Add(end == start + duration) start_vars[activity.activityId] = start end_vars[activity.activityId] = end picks = [] for resource_id in eligible: chosen = model.NewBoolVar(f"pick:{activity.activityId}:{resource_id}") presence[(activity.activityId, resource_id)] = chosen picks.append(chosen) intervals_by_resource[resource_id].append( model.NewOptionalFixedSizeIntervalVar( start, duration, chosen, f"interval:{activity.activityId}:{resource_id}" ) ) model.AddExactlyOne(picks) for activity in problem.activities: target = start_vars.get(activity.activityId) if target is None: continue for predecessor_id in activity.predecessorActivityIds: predecessor_end = end_vars.get(predecessor_id) if predecessor_end is not None: model.Add(target >= predecessor_end) blocked_total = 0 for resource in equipment_resources: blocked = _blocked_windows(resource, base, horizon) blocked_total += len(blocked) for index, (start, end) in enumerate(blocked): intervals_by_resource[resource.resourceId].append( model.NewFixedSizeIntervalVar(start, end - start, f"closed:{resource.resourceId}:{index}") ) if intervals_by_resource[resource.resourceId]: model.AddNoOverlap(intervals_by_resource[resource.resourceId]) activities_by_requirement: dict[str, list[str]] = {} for activity in problem.activities: activities_by_requirement.setdefault(activity.requirementId, []).append(activity.activityId) tardiness_weight = float((problem.objectivePolicy.weights or {}).get("tardiness", 1.0) or 1.0) tardiness_terms = [] capacity = sum(_duration_minutes(a.durationMin) for a in problem.activities) or 1 due_offsets = [ offset for offset in (_minutes_between(base, requirement.requiredAt) for requirement in problem.requirements) if offset is not None ] # 交期可能早于计划起点（历史欠交），拖期上界必须把这段「已经迟到」的量算进去， # 否则 tardy 变量的域会把模型判成不可行。 already_late = max(0, -(min(due_offsets) if due_offsets else 0)) tardy_upper = horizon + already_late + capacity + 1 for requirement in problem.requirements: activity_ids = activities_by_requirement.get(requirement.requirementId) or [] if not activity_ids: continue ends = [end_vars[aid] for aid in activity_ids if aid in end_vars] if not ends: continue completion = model.NewIntVar(0, horizon, f"completion:{requirement.requirementId}") model.AddMaxEquality(completion, ends) due = _minutes_between(base, requirement.requiredAt) if due is None: continue tardy = model.NewIntVar(0, tardy_upper, f"tardy:{requirement.requirementId}") model.Add(tardy >= completion - due) tardiness_terms.append((requirement.requirementId, tardy)) if tardiness_terms: model.Minimize( sum(int(round(tardiness_weight * 1000)) * term for _, term in tardiness_terms) ) solver = cp_model.CpSolver() solver.parameters.max_time_in_seconds = float(time_limit_seconds) solver.parameters.random_seed = int(seed) solver.parameters.num_search_workers = 1 # 单线程保证可复现 status = solver.Solve(model) elapsed = time.perf_counter() - started status_name = solver.StatusName(status) solve_status = { cp_model.OPTIMAL: SolveStatus.OPTIMAL, cp_model.FEASIBLE: SolveStatus.FEASIBLE, }.get(status, SolveStatus.INFEASIBLE if status == cp_model.INFEASIBLE else SolveStatus.ERROR) scheduled: list[ScheduledActivity] = [] if status in (cp_model.OPTIMAL, cp_model.FEASIBLE): for activity in problem.activities: if activity.activityId not in start_vars: continue start_minute = int(solver.Value(start_vars[activity.activityId])) chosen_resource = None for resource_id in _eligible_resources(activity, equipment_ids): pick = presence.get((activity.activityId, resource_id)) if pick is not None and solver.Value(pick): chosen_resource = resource_id break if chosen_resource is None: continue end_minute = start_minute + _duration_minutes(activity.durationMin) units = float(activity.requiredResourceUnits or 1.0) scheduled.append( ScheduledActivity( activityId=activity.activityId, activityIdentity=activity.activityIdentity, requirementId=activity.requirementId, operationId=activity.operationId, sequence=activity.sequence, resourceId=chosen_resource, start=base + timedelta(minutes=start_minute), end=base + timedelta(minutes=end_minute), resourceUnits=units, resourceAllocations=( ScheduledResourceAllocation( resourceId=chosen_resource, kind=ResourceKind.EQUIPMENT, units=units, ), ), ) ) scheduled_ids = {row.activityId for row in scheduled} unscheduled_ids = set(unschedulable) | { activity.activityId for activity in problem.activities if activity.activityId not in scheduled_ids } unscheduled = tuple( UnscheduledRequirement( requirementId=requirement.requirementId, quantity=float(requirement.quantity), reasonCode="UNSCHEDULED_ACTIVITY", details="存在未落到候选解的制造活动", ) for requirement in problem.requirements if any( activity_id in unscheduled_ids for activity_id in activities_by_requirement.get(requirement.requirementId, []) ) ) total_tardiness = 0.0 if tardiness_terms and status in (cp_model.OPTIMAL, cp_model.FEASIBLE): total_tardiness = sum(float(solver.Value(term)) for _, term in tardiness_terms) if solve_status in (SolveStatus.OPTIMAL, SolveStatus.FEASIBLE) and unscheduled: solve_status = SolveStatus.PARTIAL objective_values = { "totalTardiness": total_tardiness, "scheduledActivities": float(len(scheduled)), "totalActivities": float(len(problem.activities)), } best_bound = None objective_value = total_tardiness gap = None if status in (cp_model.OPTIMAL, cp_model.FEASIBLE) and tardiness_terms: best_bound = float(solver.BestObjectiveBound()) / (tardiness_weight * 1000.0) objective_value = float(solver.ObjectiveValue()) / (tardiness_weight * 1000.0) if objective_value > 0: gap = max(0.0, (objective_value - best_bound) / objective_value) generated_at = datetime.now(tz=base.tzinfo or None) solution = SchedulingSolutionV2( problemId=problem.problemId, solveStatus=solve_status, objectiveValues=objective_values, bestBound=best_bound, gap=gap, activities=tuple(scheduled), pegging=(), unscheduledRequirements=unscheduled, provenance=SolutionProvenance( runId=f"cpsat:{problem.problemId}:{int(started * 1000)}", solverId=SOLVER_ID, solverVersion=SOLVER_VERSION, generatedAt=generated_at, businessDate=problem.businessDate, problemHash=scheduling_problem_hash(problem), sourceRevision=problem.sourceRevision, sourceFingerprints=problem.sourceFingerprints, ), ) meta = { "solverId": SOLVER_ID, "solverVersion": SOLVER_VERSION, "ortoolsStatus": status_name, "wallTimeSeconds": round(elapsed, 3), "timeLimitSeconds": float(time_limit_seconds), "seed": int(seed), "horizonMinutes": horizon, "activityCount": len(problem.activities), "resourceCount": len(equipment_resources), "blockedWindowCount": blocked_total, "precedenceEdges": sum(len(a.predecessorActivityIds) for a in problem.activities), "objective": "weightedTardiness", "objectiveValue": objective_value, } return CpsatOutcome(solution=solution, meta=meta)