diff --git a/src/core/handlers/step11_map_handler.py b/src/core/handlers/step11_map_handler.py index a8c7180..ea80346 100644 --- a/src/core/handlers/step11_map_handler.py +++ b/src/core/handlers/step11_map_handler.py @@ -243,79 +243,32 @@ class Step11MapHandler(BaseStepHandler): context.notify('step11_map', 'warning', f'共享上下文预计算失败: {e}，回退逐个处理') - # ── 并发生成 ── - # 克里金插值内部为 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)) - + # ── 串行生成 ── + # 注：克里金是内存密集型运算（32GB 下单个用 5-6GB）， + # 多线程并发竞争内存带宽，实际无提速，因此保持串行。 generated: List[str] = [] errors: Dict[str, str] = {} - if _max_workers > 1 and total > 1: - # ★ 主线程预先设置 matplotlib Agg 后端（避免多线程竞争） - import matplotlib + context.notify('step11_map', 'info', + f'串行生成 {total} 张专题图（克里金自适应分块）') + + 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: - 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}') + 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 diff --git a/src/gui/panels/step11_map_panel.py b/src/gui/panels/step11_map_panel.py index ba044ac..5d2cb71 100644 --- a/src/gui/panels/step11_map_panel.py +++ b/src/gui/panels/step11_map_panel.py @@ -7,7 +7,7 @@ Step10 面板 - 专题图生成 import os import traceback from pathlib import Path -from typing import List, Optional, Tuple +from typing import List, Optional from src.gui.panels._step_path_resolver import resolve_subdir, get_step_output_path, scan_work_dir_for_input @@ -100,104 +100,42 @@ class Step11MapBatchThread(QThread): f"[警告] 共享上下文失败: {e}，回退逐个处理", "warning" ) - # ── 批量处理（多线程并发克里金插值）── - # 克里金内部 numpy/scipy 运算释放 GIL，ThreadPoolExecutor 真并行 - import concurrent.futures - import threading + # ── 串行批量处理 ── + # 注：克里金插值是内存密集型运算（32GB 内存下单个即用 5-6GB）， + # 多线程并发会竞争内存带宽，实际耗时反而无改善，因此保持串行。 + for i, csv_p in enumerate(self.csv_paths): + if self._cancelled: + self.log_message.emit("专题图批量任务已被用户取消", "warning") + break + self.progress.emit(i + 1, n) + self.log_message.emit(f"专题图 [{i + 1}/{n}] {csv_p}", "info") - _all_csvs = list(self.csv_paths) # 快照副本 - # 过滤已存在的文件 - _pending: List[Tuple[str, str]] = [] # [(csv_path, output_file), ...] - _skipped = 0 - for csv_p in _all_csvs: stem = Path(csv_p).stem - out_f = ( + output_file = ( str(Path(self.output_dir_optional) / f'{stem}_distribution.png') if self.output_dir_optional else None ) - if out_f and ( - Path(out_f).exists() - or Path(out_f).with_suffix('.tif').exists() + # 已存在则跳过（兼容 tif 重定向后的文件名） + if output_file and ( + Path(output_file).exists() + or Path(output_file).with_suffix('.tif').exists() ): - _skipped += 1 - self.log_message.emit(f" → 跳过（已存在）: {stem}", "info") - else: - _pending.append((csv_p, out_f)) + self.log_message.emit(f" → 跳过（已存在）", "info") + continue - if _skipped > 0: - self.log_message.emit( - f"[跳过] {_skipped} 个已存在，待处理 {len(_pending)} 个", "info") - - _n_pending = len(_pending) - if _n_pending == 0: - self.log_message.emit("所有专题图均已存在，无需重新生成", "info") - self.finished_ok.emit(n) - return - - _max_workers = max(1, min(2, _n_pending, os.cpu_count() or 4)) - _completed = 0 - _lock = threading.Lock() - _done = threading.Event() # 取消信号 - - def _process_one(csv_path: str, output_file: str): - """单个 CSV → 分布图（在 ThreadPoolExecutor 线程中执行）""" - if _done.is_set(): - return None, csv_path - mapper.process_data( - csv_file=csv_path, - shp_file=boundary_shp, - output_file=output_file, - resolution=resolution, - output_format='tif', - shared_context=shared_ctx, - ) - return output_file, csv_path - - if _max_workers > 1 and _n_pending > 1: - self.log_message.emit( - f"[并发] {_n_pending} 个专题图，{_max_workers} 线程并行克里金", "info") - with concurrent.futures.ThreadPoolExecutor( - max_workers=_max_workers) as executor: - _futures = { - executor.submit(_process_one, csv_p, out_f): (csv_p, out_f) - for csv_p, out_f in _pending - } - for _future in concurrent.futures.as_completed(_futures): - if self._cancelled: - _done.set() - self.log_message.emit("专题图批量任务已被用户取消", "warning") - executor.shutdown(wait=False, cancel_futures=True) - break - csv_p, _ = _futures[_future] - try: - _future.result() - with _lock: - _completed += 1 - self.progress.emit(_completed + _skipped, n) - self.log_message.emit( - f"专题图 [{_completed}/{_n_pending}] ✓: " - f"{Path(csv_p).name}", "info") - except Exception as e: - with _lock: - _completed += 1 - self.log_message.emit( - f"专题图 [{_completed}/{_n_pending}] ✗: " - f"{Path(csv_p).name} — {e}", "error") - else: - for csv_p, out_f in _pending: - if self._cancelled: - self.log_message.emit("专题图批量任务已被用户取消", "warning") - break - _completed += 1 - self.progress.emit(_completed + _skipped, n) - self.log_message.emit( - f"专题图 [{_completed}/{_n_pending}] {csv_p}", "info") - try: - _process_one(csv_p, out_f) - except Exception as e: - self.log_message.emit(f" → 失败: {e}", "error") - continue + try: + mapper.process_data( + csv_file=csv_p, + shp_file=boundary_shp, + output_file=output_file, + resolution=resolution, + output_format='tif', + shared_context=shared_ctx, + ) + except Exception as e: + self.log_message.emit(f" → 失败: {e}", "error") + continue self.finished_ok.emit(n) except Exception as e: