feat: GUI 面板专题图生成支持多线程并发克里金
- step11_map_panel.run() 串行 for 循环改为 ThreadPoolExecutor 并发 - 默认 2 线程并行,通过过滤已存在文件跳过重复生成 - 支持取消信号:_done Event 通知 worker 线程提前退出 - 线程安全:_lock 保护共享计数器,Qt 信号从 QThread 安全 emit - 修复 Tuple 类型导入缺失
This commit is contained in:
@ -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:
|
||||
|
||||
Reference in New Issue
Block a user