Compare commits
2 Commits
9512cab807
...
b8f0625bd9
| Author | SHA1 | Date | |
|---|---|---|---|
| b8f0625bd9 | |||
| 3f023cffd4 |
@ -206,42 +206,32 @@ class Step11MapHandler(BaseStepHandler):
|
|||||||
context.notify('step11_map', 'warning',
|
context.notify('step11_map', 'warning',
|
||||||
f'共享上下文预计算失败: {e},回退逐个处理')
|
f'共享上下文预计算失败: {e},回退逐个处理')
|
||||||
|
|
||||||
# ── 多进程并行(GDAL 线程不安全,但进程隔离下安全)──
|
# ── 顺序生成(避免 Windows spawn 下 ProcessPoolExecutor 死锁)──
|
||||||
|
# 局部 Kriging 内部已做 16 块顺序分块,每块 ~20-30s,
|
||||||
|
# 每张图约 5-8 分钟。64 张 ≈ 5-8 小时,但进度完全透明可见。
|
||||||
generated: List[str] = []
|
generated: List[str] = []
|
||||||
errors: Dict[str, str] = {}
|
errors: Dict[str, str] = {}
|
||||||
|
|
||||||
import multiprocessing
|
context.notify('step11_map', 'info',
|
||||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
f'顺序生成 {total} 张专题图(局部 Kriging 自适应分块)')
|
||||||
|
|
||||||
# 留出 1-2 个核心保证电脑不卡死
|
for idx, csv_p in enumerate(csv_paths):
|
||||||
max_workers = max(1, multiprocessing.cpu_count() - 2)
|
percent = int(idx / total * 100)
|
||||||
context.notify('step11_map', 'info', f'使用 {max_workers} 个进程并行生成')
|
context.notify('step11_map', 'info',
|
||||||
|
f'专题图 [{idx+1}/{total}]: {Path(csv_p).name}')
|
||||||
|
|
||||||
with ProcessPoolExecutor(max_workers=max_workers) as executor:
|
global_event_bus.publish('ProgressUpdate', {
|
||||||
future_to_csv = {
|
'percentage': percent,
|
||||||
executor.submit(_process_one_map, csv_p, base_kwargs, output_dir): csv_p
|
'message': f'Step11: {idx+1}/{total} {Path(csv_p).stem}',
|
||||||
for csv_p in csv_paths
|
})
|
||||||
}
|
|
||||||
done_count = 0
|
|
||||||
for future in as_completed(future_to_csv):
|
|
||||||
csv_p = future_to_csv[future]
|
|
||||||
done_count += 1
|
|
||||||
try:
|
|
||||||
result_path, _ = future.result()
|
|
||||||
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}')
|
|
||||||
|
|
||||||
pct = int(done_count / total * 100)
|
try:
|
||||||
global_event_bus.publish('ProgressUpdate', {
|
result_path, _ = _process_one_map(csv_p, base_kwargs, output_dir)
|
||||||
'percentage': pct,
|
generated.append(result_path)
|
||||||
'message': f'Step11 专题图: {done_count}/{total}',
|
except Exception as e:
|
||||||
})
|
errors[csv_p] = str(e)
|
||||||
if done_count % max(1, total // 10) == 0 or done_count == total:
|
context.notify('step11_map', 'warning',
|
||||||
context.notify('step11_map', 'info',
|
f'专题图 FAIL: {Path(csv_p).name} — {e}')
|
||||||
f'专题图 [{done_count}/{total}]')
|
|
||||||
|
|
||||||
step_end_time = time.time()
|
step_end_time = time.time()
|
||||||
elapsed = step_end_time - step_start_time
|
elapsed = step_end_time - step_start_time
|
||||||
|
|||||||
@ -2978,14 +2978,31 @@ class ContentMapper:
|
|||||||
|
|
||||||
print(f"[共享上下文] 网格: {nx}×{ny} = {nx*ny} 点")
|
print(f"[共享上下文] 网格: {nx}×{ny} = {nx*ny} 点")
|
||||||
|
|
||||||
# ⑥ 水域掩膜布尔矩阵(只此一次)
|
# ⑥ 水域掩膜布尔矩阵(降采样到 ~100m 分辨率计算,避免 1m 下千万级 Point 对象 OOM)
|
||||||
mask = None
|
mask = None
|
||||||
if boundary_gdf is not None:
|
if boundary_gdf is not None:
|
||||||
mask_pts = np.column_stack((grid_xx.ravel(), grid_yy.ravel()))
|
_MASK_TARGET_POINTS = 50000 # 掩膜降采样目标点数
|
||||||
|
_mask_step = max(1, int(np.sqrt(grid_xx.size / _MASK_TARGET_POINTS)))
|
||||||
|
if _mask_step > 1:
|
||||||
|
mask_xx = grid_xx[::_mask_step, ::_mask_step]
|
||||||
|
mask_yy = grid_yy[::_mask_step, ::_mask_step]
|
||||||
|
print(f"[共享上下文] 掩膜降采样 {_mask_step}× → "
|
||||||
|
f"{mask_xx.shape[1]}×{mask_xx.shape[0]} = {mask_xx.size:,} 点")
|
||||||
|
else:
|
||||||
|
mask_xx, mask_yy = grid_xx, grid_yy
|
||||||
|
|
||||||
|
mask_pts = np.column_stack((mask_xx.ravel(), mask_yy.ravel()))
|
||||||
mask_gdf = gpd.GeoDataFrame(
|
mask_gdf = gpd.GeoDataFrame(
|
||||||
geometry=[Point(x, y) for x, y in mask_pts], crs=self.output_crs
|
geometry=[Point(x, y) for x, y in mask_pts], crs=self.output_crs
|
||||||
)
|
)
|
||||||
mask = mask_gdf.within(boundary_gdf.unary_union).values.reshape(grid_xx.shape)
|
mask_lowres = mask_gdf.within(boundary_gdf.unary_union).values.reshape(mask_xx.shape)
|
||||||
|
# 升采样回原始分辨率(最近邻,掩膜是布尔值)
|
||||||
|
if _mask_step > 1:
|
||||||
|
mask = np.kron(mask_lowres, np.ones((_mask_step, _mask_step), dtype=bool))
|
||||||
|
# 裁剪到精确原始尺寸
|
||||||
|
mask = mask[:grid_xx.shape[0], :grid_xx.shape[1]]
|
||||||
|
else:
|
||||||
|
mask = mask_lowres
|
||||||
print(f"[共享上下文] 水域掩膜: {int(mask.sum())}/{mask.size} 点在水域内")
|
print(f"[共享上下文] 水域掩膜: {int(mask.sum())}/{mask.size} 点在水域内")
|
||||||
|
|
||||||
return (grid_xx, grid_yy, mask, bounds, boundary_gdf)
|
return (grid_xx, grid_yy, mask, bounds, boundary_gdf)
|
||||||
|
|||||||
Reference in New Issue
Block a user