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: