ArcGIS批量TIF影像归一化实战：从GDAL到Raster函数的性能优化

1次阅读

共计 2293 个字符，预计需要花费 6 分钟才能阅读完成。

在气象分析、土地分类等遥感应用中，影像归一化（Normalization）是常见的预处理步骤。它能够消除不同影像间的亮度差异，使得数据更具可比性。然而，传统的手动归一化操作存在几个明显缺陷：

耗时：面对成百上千的影像文件，手动操作几乎不可能完成
易错：人工操作容易遗漏文件或设置错误参数
不可复用 ：每次处理都需要重复相同的步骤，无法形成标准化流程

在 ArcGIS 生态中，我们主要考虑三种技术方案：

方案	I/ O 效率	内存占用	API 易用性	适用场景
GDAL	高	低	中等	需要精细控制底层操作时
ArcPy	中等	高	高	简单脚本任务
Raster Function	最高	最低	最高	批量并行处理

归一化公式一般为：(value – min) / (max – min)。Raster Functions 提供了直接的代数运算支持：

from arcgis.raster.functions import remap, raster_calculator

def normalize_raster(input_raster):
    min_val = input_raster.min()
    max_val = input_raster.max()
    return (input_raster - min_val) / (max_val - min_val)

from multiprocessing import Pool
import os

def process_file(input_path, output_dir):
    # 处理逻辑...
    pass

def batch_process(input_files, output_dir, workers=4):
    with Pool(workers) as p:
        p.starmap(process_file, [(f, output_dir) for f in input_files])

import tempfile
from contextlib import contextmanager

@contextmanager
def tempfile_pool(size=10):
    temp_files = []
    try:
        yield temp_files
    finally:
        for f in temp_files:
            try:
                os.remove(f)
            except:
                pass

import arcpy
import logging
from tqdm import tqdm
from arcgis.raster import Raster

# 配置日志
logging.basicConfig(filename='normalize.log', level=logging.INFO)

def normalize_tif(input_path, output_path):
    try:
        raster = Raster(input_path)
        normalized = (raster - raster.min()) / (raster.max() - raster.min())
        normalized.save(output_path)
        logging.info(f'Success: {input_path}')
    except Exception as e:
        logging.error(f'Failed {input_path}: {str(e)}')

def batch_normalize(input_dir, output_dir):
    tif_files = [f for f in os.listdir(input_dir) if f.endswith('.tif')]

    for file in tqdm(tif_files):
        in_path = os.path.join(input_dir, file)
        out_path = os.path.join(output_dir, f'norm_{file}')
        normalize_tif(in_path, out_path)

通过调整 arcpy.env.compression 选项，测试不同分块大小对处理速度的影响：

for chunk_size in [256, 512, 1024, 2048]:
    arcpy.env.compression = f'LZW {chunk_size}' 
    # 运行测试并记录时间

import tracemalloc

tracemalloc.start()
# 运行可疑代码
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
for stat in top_stats[:10]:
    print(stat)