From aa5d9d939279a12ec1f6018b2d548e7b1c29fc83 Mon Sep 17 00:00:00 2001 From: duxin Date: Tue, 28 Jul 2026 16:45:46 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20GUI=20=E9=9D=A2=E6=9D=BF=E4=B8=93?= =?UTF-8?q?=E9=A2=98=E5=9B=BE=E7=94=9F=E6=88=90=E6=94=AF=E6=8C=81=E5=A4=9A?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E5=B9=B6=E5=8F=91=E5=85=8B=E9=87=8C=E9=87=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - step11_map_panel.run() 串行 for 循环改为 ThreadPoolExecutor 并发 - 默认 2 线程并行，通过过滤已存在文件跳过重复生成 - 支持取消信号：_done Event 通知 worker 线程提前退出 - 线程安全：_lock 保护共享计数器，Qt 信号从 QThread 安全 emit - 修复 Tuple 类型导入缺失 --- src/gui/panels/step11_map_panel.py | 118 ++++++++++++++++++++++------- 1 file changed, 91 insertions(+), 27 deletions(-) diff --git a/src/gui/panels/step11_map_panel.py b/src/gui/panels/step11_map_panel.py index 88aac1d..ba044ac 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 +from typing import List, Optional, Tuple from src.gui.panels._step_path_resolver import resolve_subdir, get_step_output_path, scan_work_dir_for_input @@ -100,40 +100,104 @@ class Step11MapBatchThread(QThread): f"[警告] 共享上下文失败: {e}，回退逐个处理", "warning" ) - # ── 批量处理 ── - 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") + # ── 批量处理（多线程并发克里金插值）── + # 克里金内部 numpy/scipy 运算释放 GIL，ThreadPoolExecutor 真并行 + import concurrent.futures + import threading + _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 - output_file = ( + out_f = ( str(Path(self.output_dir_optional) / f'{stem}_distribution.png') if self.output_dir_optional else None ) - # 已存在则跳过（兼容 tif 重定向后的文件名） - if output_file and ( - Path(output_file).exists() - or Path(output_file).with_suffix('.tif').exists() + if out_f and ( + Path(out_f).exists() + or Path(out_f).with_suffix('.tif').exists() ): - self.log_message.emit(f" → 跳过（已存在）", "info") - continue + _skipped += 1 + self.log_message.emit(f" → 跳过（已存在）: {stem}", "info") + else: + _pending.append((csv_p, out_f)) - 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 + 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 self.finished_ok.emit(n) except Exception as e: