optimizer.py TTC-DDT optimizer. 核心思路 1. DDT 分值权重更高且候选弧段更少所以先安排 DDT。 2. 对未安排的 DDT 做“增广链”修复尝试把挡路的 DDT 挪到其它弧段。 3. DDT 稳定后再安排 TTC并优先复用已经被 DDT 占用的 TTCDDT 弧段。 4. 全程使用可回滚的本地调度器最后返回 assignment 列表给平台复核。 from__future__importannotationsimporttimefromtypingimportDict,List,Optional,Sequence,Set,Tuplefrompipeline.builderimportAssignmentBuilderfrompipeline.scoringimportcal_balanced_score DEFAULT_OPTIMIZER_SECONDS520.0defoptimize(state,builder:AssignmentBuilder,config:Dict):平台调用入口返回 assignment[task_id] arc_id 或 -1。start_timetime.time()deadlinestart_timeDEFAULT_OPTIMIZER_SECONDS# AssignmentBuilder 没有撤销接口而增广链需要“先挪走任务、失败再回滚”。# 因此这里使用一个约束逻辑等价、但支持 unassign 的本地调度器。schedulerMutableScheduler(state)ddt_tasks[taskfortaskinstate.task_listiftask.task_typeDDT]ttc_tasks[taskfortaskinstate.task_listiftask.task_typeTTC]# 任务越稀缺越早处理候选弧段少的任务如果留到后面通常更难补救。ddt_tasks.sort(keytask_order_key)ttc_tasks.sort(keytask_order_key)# 第一阶段快速生成 DDT 优先的可行解。fortaskinddt_tasks:try_place_with_augment(state,scheduler,task.id,depth0,deadlinedeadline)# 第二阶段对未完成 DDT 做增广链修复。深度 3 性价比最高# 深度 4 只在剩余时间内限时尝试给 TTC 补排和最终校验保留余量。ddt_deadlinemin(deadline-25.0,start_time495.0)fordepthin(1,2,3):iftime.time()deadline:breakfortaskinddt_tasks:iftime.time()ddt_deadline:breakifscheduler.task_to_arc[task.id]0:try_place_with_augment(state,scheduler,task.id,depth,ddt_deadline)depth4_deadlinemin(deadline-25.0,start_time495.0)fortaskinddt_tasks:iftime.time()depth4_deadline:breakifscheduler.task_to_arc[task.id]0:try_place_with_augment(state,scheduler,task.id,4,depth4_deadline)# 第三阶段在 DDT 方案基础上补 TTC优先复用 TTCDDT 共享弧段。fortaskinttc_tasks:iftime.time()deadline:breakforarc_idinordered_arcs(state,scheduler,task):ifscheduler.assign(task.id,arc_id):breakreturnscheduler.build()classMutableScheduler:支持 assign/unassign 的轻量调度器约束逻辑与 AssignmentBuilder 保持一致。def__init__(self,state):self.statestate self.task_to_arc[-1]*len(state.task_list)self.antenna_usage:Dict[int,List[Tuple[int,int,int]]]{}self.orbit_type_usage:Dict[Tuple[int,int,int],int]{}self.arc_task_counts:Dict[int,int]{}self.arc_tasks:Dict[int,List[int]]{}defbuild(self)-List[int]:returnself.task_to_arc[:]defcan_assign(self,task_id:int,arc_id:int)-bool:taskself.state.task_list[task_id]arcself.state.arc_list[arc_id]ifself.task_to_arc[task_id]0:returnFalseifarc_idnotintask.optional_arc_ids:returnFalse# 普通弧段只能安排一个任务TTCDDT 弧段最多允许一对同卫星 TTC/DDT 共用。current_arc_countself.arc_task_counts.get(arc_id,0)ifarc.antenna_functionTTCDDT:ifcurrent_arc_count1:returnFalseelifcurrent_arc_count0:returnFalse# 同一卫星、同一轨道、同一任务类型不能重复安排。key(arc.satellite_index,arc.orbit_index,task.task_type_flag)ifself.orbit_type_usage.get(key)isnotNone:returnFalse# 检查同一天线上的占用时间窗口是否冲突。任务实际占用窗口要包含准备和释放时间。begin_timearc.start_time-task.prepare_time end_timearc.end_timetask.release_timeforother_begin,other_end,other_task_idinself.antenna_usage.get(arc.antenna_index,[]):ifend_timeother_beginorother_endbegin_time:continueother_arc_idself.task_to_arc[other_task_id]other_taskself.state.task_list[other_task_id]other_arcself.state.arc_list[other_arc_id]# 唯一允许重叠的情况同一 TTCDDT 弧段、同一卫星、一 TTC 一 DDT。if(arc.antenna_functionTTCDDTandarc_idother_arc_idandarc.satellite_indexother_arc.satellite_indexandtask.task_type!other_task.task_type):continuereturnFalsereturnTruedefassign(self,task_id:int,arc_id:int)-bool:ifnotself.can_assign(task_id,arc_id):returnFalsetaskself.state.task_list[task_id]arcself.state.arc_list[arc_id]# 同步维护四类索引后续 can_assign/blockers_for 才能快速判断冲突来源。self.task_to_arc[task_id]arc_id self.arc_task_counts[arc_id]self.arc_task_counts.get(arc_id,0)1self.arc_tasks.setdefault(arc_id,[]).append(task_id)self.orbit_type_usage[(arc.satellite_index,arc.orbit_index,task.task_type_flag)]task_id self.antenna_usage.setdefault(arc.antenna_index,[]).append((arc.start_time-task.prepare_time,arc.end_timetask.release_time,task_id))returnTruedefunassign(self,task_id:int)-bool:arc_idself.task_to_arc[task_id]ifarc_id0:returnFalsetaskself.state.task_list[task_id]arcself.state.arc_list[arc_id]# 增广链失败时需要精确回滚所以这里要把 assign 写入的索引全部撤掉。self.task_to_arc[task_id]-1next_countself.arc_task_counts.get(arc_id,0)-1ifnext_count0:self.arc_task_counts[arc_id]next_countelse:self.arc_task_counts.pop(arc_id,None)task_idsself.arc_tasks.get(arc_id,[])iftask_idintask_ids:task_ids.remove(task_id)ifnottask_ids:self.arc_tasks.pop(arc_id,None)self.orbit_type_usage.pop((arc.satellite_index,arc.orbit_index,task.task_type_flag),None)usageself.antenna_usage.get(arc.antenna_index,[])item(arc.start_time-task.prepare_time,arc.end_timetask.release_time,task_id)ifiteminusage:usage.remove(item)ifnotusage:self.antenna_usage.pop(arc.antenna_index,None)returnTruedefblockers_for(self,task_id:int,arc_id:int)-Optional[Set[int]]:返回 task_id 放到 arc_id 时会挡住它的已安排任务集合。taskself.state.task_list[task_id]arcself.state.arc_list[arc_id]ifself.task_to_arc[task_id]0orarc_idnotintask.optional_arc_ids:returnNoneblockers:Set[int]set()# 弧段容量冲突普通弧段已有任务或 TTCDDT 弧段已经满员。current_arc_countself.arc_task_counts.get(arc_id,0)ifarc.antenna_functionTTCDDT:ifcurrent_arc_count1:blockers.update(self.arc_tasks.get(arc_id,[]))elifcurrent_arc_count0:blockers.update(self.arc_tasks.get(arc_id,[]))key(arc.satellite_index,arc.orbit_index,task.task_type_flag)orbit_task_idself.orbit_type_usage.get(key)iforbit_task_idisnotNone:blockers.add(orbit_task_id)# 天线时间窗口冲突找出所有和目标占用窗口重叠的已安排任务。begin_timearc.start_time-task.prepare_time end_timearc.end_timetask.release_timeforother_begin,other_end,other_task_idinself.antenna_usage.get(arc.antenna_index,[]):ifend_timeother_beginorother_endbegin_time:continueother_arc_idself.task_to_arc[other_task_id]other_taskself.state.task_list[other_task_id]other_arcself.state.arc_list[other_arc_id]if(arc.antenna_functionTTCDDTandarc_idother_arc_idandarc.satellite_indexother_arc.satellite_indexandtask.task_type!other_task.task_type):continueblockers.add(other_task_id)returnblockersdeftry_place_with_augment(state,scheduler:MutableScheduler,task_id:int,depth:int,deadline:float,seen:Optional[Set[int]]None,)-bool:尝试安排 task_id失败时递归挪走一个同类型阻塞任务。iftime.time()deadline:returnFalseifscheduler.task_to_arc[task_id]0:returnTrueseenseenorset()taskstate.task_list[task_id]forarc_idinordered_arcs(state,scheduler,task):# 先尝试直接插入直接成功是最便宜、也最稳定的操作。ifscheduler.assign(task_id,arc_id):returnTrueifdepth0:continueblockersscheduler.blockers_for(task_id,arc_id)# 只处理“单个同类型任务挡路”的情况。多个阻塞任务会导致回滚复杂、耗时也高。ifnotblockersorlen(blockers)!1:continueblocker_idnext(iter(blockers))ifblocker_idinseen:continueifstate.task_list[blocker_id].task_type!task.task_type:continueold_arc_idscheduler.task_to_arc[blocker_id]scheduler.unassign(blocker_id)# 把当前任务放进目标弧段再递归尝试给被挪走的任务寻找新位置。ifscheduler.assign(task_id,arc_id):iftry_place_with_augment(state,scheduler,blocker_id,depth-1,deadline,seen|{task_id},):returnTrue# 递归失败则撤销当前尝试恢复到尝试前状态。scheduler.unassign(task_id)scheduler.assign(blocker_id,old_arc_id)returnFalsedeftask_order_key(task)-Tuple[int,int,int,int,int]:候选弧段少的任务优先其次按日期、卫星和原始编号稳定排序。return(len(task.optional_arc_ids),task.day_index,task.satellite_index,task.sub_index,task.id,)defordered_arcs(state,scheduler:MutableScheduler,task)-List[int]:按启发式质量排序候选弧段避免随机占用关键资源。arcstask.optional_arc_ids[:]arcs.sort(keylambdaarc_id:arc_order_key(state,scheduler,task,arc_id))returnarcsdefarc_order_key(state,scheduler:MutableScheduler,task,arc_id:int)-Tuple[int,int,int,float,int,int]:arcstate.arc_list[arc_id]function_score0# DDT 优先拿 DDT/TTCDDT 资源TTC 则优先复用已有 DDT 的 TTCDDT 弧段。iftask.task_typeDDT:ifarc.antenna_functionTTCDDT:function_score-2000elifarc.antenna_functionDDT:function_score-1500else:ifscheduler.arc_task_counts.get(arc_id,0)0andarc.antenna_functionTTCDDT:function_score-100000elifarc.antenna_functionTTC:function_score-1000# 排序目标功能更合适、竞争更少、占用更短、仰角更高、时间更早。occupied_timearc.end_time-arc.start_timetask.prepare_timetask.release_timereturn(function_score,len(arc.optional_task_ids),occupied_time,-arc.elevation,arc.start_time,arc_id,)deffitness(state,assignment:Sequence[int])-float:计算课程总分100*TTC完成率 150*DDT完成率。ttc_finished,ttc_total,ddt_finished,ddt_totalcompletion_counts(state,assignment)ttc_ratettc_finished/ttc_totalifttc_totalelse0.0ddt_rateddt_finished/ddt_totalifddt_totalelse0.0_,_,total_scorecal_balanced_score(ttc_rate,ddt_rate)returntotal_scoredefcompletion_counts(state,assignment:Sequence[int])-Tuple[int,int,int,int]:ttc_total0ddt_total0ttc_finished0ddt_finished0fortaskinstate.task_list:assignedtask.idlen(assignment)andassignment[task.id]0iftask.task_typeTTC:ttc_total1ifassigned:ttc_finished1eliftask.task_typeDDT:ddt_total1ifassigned:ddt_finished1returnttc_finished,ttc_total,ddt_finished,ddt_total