如何每隔10秒从OR-Tools的SCIP求解器输出中提取特定数据？

一、OR-Tools内置回调方案（推荐） #

OR-Tools提供MIP回调机制，无需解析控制台输出，直接从求解器获取结构化数据，是最可靠的方案。

实现步骤 #

1. 自定义回调类，继承pywraplp.Solver.MIPCallback，重写Callback方法实现定时捕获逻辑。
2. 将回调注册到求解器，求解过程中自动触发数据捕获。

代码示例 #

import time
from ortools.linear_solver import pywraplp

class GapMonitorCallback(pywraplp.Solver.MIPCallback):
    def __init__(self, solver, interval=10):
        super().__init__()
        self.solver = solver
        self.interval = interval
        self.last_check_time = time.time()

    def Callback(self, where):
        # 仅在MIP搜索阶段处理
        if where == pywraplp.Solver.CALLBACK_MIP:
            current_time = time.time()
            # 每隔指定时间捕获一次数据
            if current_time - self.last_check_time >= self.interval:
                gap = self.solver.MIPGap() * 100  # 转换为百分比格式
                primal_bound = self.solver.Objective().Value()
                dual_bound = self.solver.Objective().BestBound()
                completed = self.solver.SolveProgress() * 100  # 求解完成度

                print(f"[{time.strftime('%H:%M:%S')}] Gap: {gap:.2f}%, Primal Bound: {primal_bound:.2e}, Dual Bound: {dual_bound:.2e}, Completed: {completed:.2f}%")
                
                self.last_check_time = current_time

# 初始化求解器
solver = pywraplp.Solver.CreateSolver('SCIP')
solver.EnableOutput()

# 注册回调
callback = GapMonitorCallback(solver, interval=10)
solver.SetCallback(callback)

# 此处添加你的模型构建代码（变量、约束、目标函数等）

# 启动求解
status = solver.Solve()

关键说明 #

• MIPGap()：直接获取当前对偶间隙，返回小数（如0.0755对应7.55%）。
• Objective().Value()：获取当前最优原始边界值。
• Objective().BestBound()：获取当前最优对偶边界值。
• SolveProgress()：获取求解完成进度（范围0-1）。
• 回调会在MIP搜索的节点/迭代阶段触发，通过时间判断实现定时捕获。

二、外部捕获控制台输出方案 #

如果内置回调无法覆盖需求（比如需要SCIP输出的特定字段），可通过子进程捕获输出并解析。

为什么sys模块重定向无效？ #

SCIP是底层C++实现的求解器，其输出直接写入系统级stdout/stderr，绕过了Python的sys.stdout通道，因此Python层面的重定向无法捕获这些输出。

实现思路 #

1. 将求解逻辑封装为独立脚本。
2. 主脚本通过subprocess.Popen启动子进程，捕获其输出。
3. 实时读取输出行，用正则表达式解析目标数据，定时记录。

代码示例 #

子脚本solve_model.py

from ortools.linear_solver import pywraplp

def main():
    solver = pywraplp.Solver.CreateSolver('SCIP')
    solver.EnableOutput()

    # 此处添加你的模型构建代码

    status = solver.Solve()

if __name__ == "__main__":
    main()

主监控脚本

import time
import subprocess
import re

def parse_scip_output(line):
    # 匹配SCIP输出行的正则表达式，需根据实际输出格式调整
    pattern = r'\d+s\|\s+\d+\|\s+\d+\|\s+\d+\|\s+[\d.]+\|\s+\d+M\|\s+\d+\|\s+\d+\|\s+\d+\|\s+\d+\|\s+\d+\|\s+\d+\|\s+\d+k?\|\s+\d+\|\s+[\d.e+-]+\|\s+[\d.e+-]+\|\s+([\d.]+%)\|\s+([\d.]+%)'
    match = re.match(pattern, line)
    if match:
        return {
            'gap': match.group(1),
            'completed': match.group(2)
        }
    return None

def main():
    process = subprocess.Popen(
        ['python', 'solve_model.py'],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True
    )
    last_check_time = time.time()
    interval = 10

    while True:
        line = process.stdout.readline()
        if not line and process.poll() is not None:
            break
        
        parsed_data = parse_scip_output(line.strip())
        if parsed_data:
            current_time = time.time()
            if current_time - last_check_time >= interval:
                print(f"[{time.strftime('%H:%M:%S')}] Gap: {parsed_data['gap']}, Completed: {parsed_data['completed']}")
                last_check_time = current_time

if __name__ == "__main__":
    main()

三、SCIP日志文件监控方案 #

让SCIP将输出写入日志文件，通过监控文件变化实时解析：

1. 初始化求解器时设置日志参数：

# 指定SCIP日志文件路径
solver.SetSolverSpecificParametersAsString("logfile=scip_log.txt")

2. 使用watchdog等文件监控库，实时读取日志新增内容，解析提取目标数据，实现定时捕获。

内容的提问来源于stack exchange，提问作者gery001