diff --git a/src/core/handlers/step11_map_handler.py b/src/core/handlers/step11_map_handler.py index d3a43c6..a8c7180 100644 --- a/src/core/handlers/step11_map_handler.py +++ b/src/core/handlers/step11_map_handler.py @@ -243,32 +243,79 @@ class Step11MapHandler(BaseStepHandler): context.notify('step11_map', 'warning', f'共享上下文预计算失败: {e},回退逐个处理') - # ── 顺序生成(避免 Windows spawn 下 ProcessPoolExecutor 死锁)── - # 局部 Kriging 内部已做 16 块顺序分块,每块 ~20-30s, - # 每张图约 5-8 分钟。64 张 ≈ 5-8 小时,但进度完全透明可见。 + # ── 并发生成 ── + # 克里金插值内部为 numpy/scipy 运算(释放 GIL), + # 使用 ThreadPoolExecutor 并发处理多个 CSV,大幅缩短总耗时。 + # 注意:不使用 ProcessPoolExecutor(Windows spawn 会导致死锁)。 + _max_workers = int(config.get('kriging_workers', 2)) + _max_workers = max(1, min(_max_workers, total, os.cpu_count() or 4)) + generated: List[str] = [] errors: Dict[str, str] = {} - context.notify('step11_map', 'info', - f'顺序生成 {total} 张专题图(局部 Kriging 自适应分块)') - - for idx, csv_p in enumerate(csv_paths): - percent = int(idx / total * 100) - context.notify('step11_map', 'info', - f'专题图 [{idx+1}/{total}]: {Path(csv_p).name}') - - global_event_bus.publish('ProgressUpdate', { - 'percentage': percent, - 'message': f'Step11: {idx+1}/{total} {Path(csv_p).stem}', - }) - + if _max_workers > 1 and total > 1: + # ★ 主线程预先设置 matplotlib Agg 后端(避免多线程竞争) + import matplotlib try: - result_path, _ = _process_one_map(csv_p, base_kwargs, output_dir) - generated.append(result_path) - except Exception as e: - errors[csv_p] = str(e) - context.notify('step11_map', 'warning', - f'专题图 FAIL: {Path(csv_p).name} — {e}') + matplotlib.use('Agg', force=True) + except Exception: + pass + + import concurrent.futures + context.notify('step11_map', 'info', + f'并发生成 {total} 张专题图({_max_workers} 线程并行)') + + completed = 0 + with concurrent.futures.ThreadPoolExecutor( + max_workers=_max_workers) as executor: + future_map = { + executor.submit( + _process_one_map, csv_p, base_kwargs, output_dir + ): csv_p + for csv_p in csv_paths + } + for future in concurrent.futures.as_completed(future_map): + csv_p = future_map[future] + completed += 1 + try: + result_path, _ = future.result() + generated.append(result_path) + context.notify('step11_map', 'info', + f'专题图 [{completed}/{total}] ✓: ' + f'{Path(csv_p).name}') + except Exception as e: + errors[csv_p] = str(e) + context.notify('step11_map', 'warning', + f'专题图 [{completed}/{total}] ✗: ' + f'{Path(csv_p).name} — {e}') + + percent = int(completed / total * 100) + global_event_bus.publish('ProgressUpdate', { + 'percentage': percent, + 'message': f'Step11: {completed}/{total} ' + f'{Path(csv_p).stem}', + }) + else: + context.notify('step11_map', 'info', + f'顺序生成 {total} 张专题图(局部 Kriging 自适应分块)') + + for idx, csv_p in enumerate(csv_paths): + percent = int(idx / total * 100) + context.notify('step11_map', 'info', + f'专题图 [{idx+1}/{total}]: {Path(csv_p).name}') + + global_event_bus.publish('ProgressUpdate', { + 'percentage': percent, + 'message': f'Step11: {idx+1}/{total} {Path(csv_p).stem}', + }) + + try: + result_path, _ = _process_one_map(csv_p, base_kwargs, output_dir) + generated.append(result_path) + except Exception as e: + errors[csv_p] = str(e) + context.notify('step11_map', 'warning', + f'专题图 FAIL: {Path(csv_p).name} — {e}') step_end_time = time.time() elapsed = step_end_time - step_start_time