如何程序化中断SciPy中耗时的curve_fit计算？

核心原因 #

SciPy的curve_fit依赖底层C扩展实现，Python线程的中断信号（比如抛出异常）无法穿透到C层代码，导致线程被中断后，计算仍在后台占用CPU资源。必须用独立进程隔离计算，通过终止进程的方式彻底停止运算。

具体实现步骤 #

1. 封装拟合逻辑为独立函数 #

把curve_fit调用封装成可在进程中执行的函数，返回结果或异常信息：

from scipy.optimize import curve_fit
import numpy as np

def fit_task(x_data, y_data, model_func, initial_guess):
    try:
        popt, pcov = curve_fit(
            model_func, x_data, y_data, 
            p0=initial_guess, maxfev=10**9
        )
        return (popt, pcov, None)
    except Exception as e:
        return (None, None, str(e))

2. 实现进程管理类（贴合PyQt生态） #

用multiprocessing.Process管理拟合进程，配合PyQt信号槽实现结果回调、超时控制和手动终止：

from PyQt6.QtCore import QObject, pyqtSignal, QTimer
import multiprocessing as mp
import queue

class FitController(QObject):
    fit_completed = pyqtSignal(object, object, str)  # popt, pcov, error_msg
    def __init__(self):
        super().__init__()
        self._process = None
        self._result_queue = mp.Queue()
        self._timeout_timer = QTimer(self)
        self._timeout_timer.setSingleShot(True)
        self._timeout_timer.timeout.connect(self.stop_fit)

    def start_fit(self, x_data, y_data, model_func, initial_guess, timeout=60):
        # 清理旧进程（如果存在）
        if self._process and self._process.is_alive():
            self._process.terminate()
            self._process.join()
        
        # 启动新进程
        self._process = mp.Process(
            target=self._process_wrapper,
            args=(x_data, y_data, model_func, initial_guess)
        )
        self._process.start()

        # 启动超时定时器
        if timeout > 0:
            self._timeout_timer.start(timeout * 1000)

        # 启动结果监听线程（避免阻塞主线程）
        from PyQt6.QtCore import QThread
        self._listener_thread = QThread()
        self._listener = ResultWatcher(self._result_queue)
        self._listener.moveToThread(self._listener_thread)
        self._listener.result_ready.connect(self._handle_result)
        self._listener_thread.started.connect(self._listener.watch)
        self._listener_thread.start()

    def stop_fit(self):
        if self._process and self._process.is_alive():
            self._process.terminate()
            self._process.join()
            self.fit_completed.emit(None, None, "计算已终止（超时/手动操作）")
        if self._listener_thread.isRunning():
            self._listener_thread.quit()
            self._listener_thread.wait()

    def _process_wrapper(self, x_data, y_data, model_func, initial_guess):
        result = fit_task(x_data, y_data, model_func, initial_guess)
        self._result_queue.put(result)

    def _handle_result(self, result):
        popt, pcov, error = result
        self.fit_completed.emit(popt, pcov, error)
        self._timeout_timer.stop()
        if self._listener_thread.isRunning():
            self._listener_thread.quit()
            self._listener_thread.wait()
        if self._process.is_alive():
            self._process.join()

# 监听进程结果队列的辅助类
class ResultWatcher(QObject):
    result_ready = pyqtSignal(object)
    def __init__(self, result_queue):
        super().__init__()
        self._queue = result_queue

    def watch(self):
        try:
            result = self._queue.get(block=True)
            self.result_ready.emit(result)
        except queue.Empty:
            pass

3. 在主窗口中集成使用 #

from PyQt6.QtWidgets import (
    QMainWindow, QPushButton, QVBoxLayout,
    QWidget, QLabel, QMessageBox
)
import numpy as np

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Curve Fit Controller")
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout(central_widget)

        self.start_btn = QPushButton("开始拟合")
        self.start_btn.clicked.connect(self.initiate_fit)
        layout.addWidget(self.start_btn)

        self.stop_btn = QPushButton("终止拟合")
        self.stop_btn.clicked.connect(self.stop_fit)
        layout.addWidget(self.stop_btn)

        self.status_label = QLabel("等待操作...")
        layout.addWidget(self.status_label)

        self.fit_controller = FitController()
        self.fit_controller.fit_completed.connect(self.on_fit_done)

    def initiate_fit(self):
        # 生成测试数据（替换为实际业务数据）
        x = np.linspace(0, 10, 100)
        y = 2.5 * x + 4 + np.random.normal(0, 0.8, size=x.shape)
        def linear_model(x, a, b):
            return a * x + b
        
        self.status_label.setText("拟合进行中...")
        self.fit_controller.start_fit(x, y, linear_model, [1, 3], timeout=30)

    def stop_fit(self):
        self.fit_controller.stop_fit()
        self.status_label.setText("拟合已终止")

    def on_fit_done(self, popt, pcov, error):
        if error:
            QMessageBox.warning(self, "拟合失败", error)
            self.status_label.setText("拟合失败")
        else:
            self.status_label.setText(f"拟合完成：a={popt[0]:.2f}, b={popt[1]:.2f}")

if __name__ == "__main__":
    from PyQt6.QtWidgets import QApplication
    import sys
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec())

关键注意事项 #

• 必须用进程而非线程：只有终止进程才能彻底停止底层C代码的运算，避免后台资源泄漏。
• 清理旧进程：每次启动新拟合前，务必终止并回收旧进程，防止僵尸进程占用资源。
• 超时与手动终止复用逻辑：stop_fit方法既可以被超时定时器触发，也可以被用户按钮调用。

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