使用rasterio.open()时如何检查源与目标文件是否相同及相关问题
问题背景
我基于rasterio实现了GeoTIFF的读取、重投影与写入新文件的功能(转换逻辑比官方示例更复杂),目前已经处理了三类场景:
- 输出文件已存在时覆盖,当前实现代码:
try: os.remove(fname) except FileNotFoundError: pass
- 输出文件被其他程序占用时捕获异常,当前实现代码:
try: write_reprojected(...) except (PermissionError, OSError): print("output_file is opened elsewhere, close and try again")
- 误将源文件指定为目标文件时的处理,当前通过路径绝对地址对比检测:
if os.path.abspath(src.files[0]) == os.path.abspath(fname): #raise MyCustomException() print("Trying to write file to its old location")
我现在有两个疑问:
- 使用
rasterio.open()写入时,有没有更好的方法检查源文件(src)和目标文件(dst)是否为同一文件? - 是否存在移除/覆盖被其他程序占用文件的方法?
附完整实现代码:
import os import numpy as np import rasterio from rasterio.warp import calculate_default_transform, reproject, Resampling def write_reprojected(src, crs, transform, width, height, fname): # 检测是否试图写入源文件路径 if os.path.abspath(src.files[0]) == os.path.abspath(fname): #raise MyCustomException() print("Trying to write file to its old location") kwargs = src.meta.copy() kwargs.update({ 'crs': crs, 'transform': transform, 'width': width, 'height': height}) # 覆盖已存在的文件 try: os.remove(fname) except FileNotFoundError: pass with rasterio.open(fname, 'w', **kwargs) as dst: for i in range(1, src.count + 1): reproject( source=rasterio.band(src, i), destination=rasterio.band(dst, i), src_transform=src.transform, src_crs=src.crs, dst_transform=transform, dst_crs=crs, resampling=Resampling.bilinear) input_file = 'old_folder\\old_file.tif' output_folder = 'new_folder' output_file = 'new_file.tif' try: os.makedirs(output_folder) except FileExistsError: pass with rasterio.open(input_file, 'r') as src: transform, width, height = aligned_target( src.transform, src.width, src.height, [round(r) for r in src.res]) try: write_reprojected(src, src.crs, transform, width, height, os.path.join(output_folder, output_file)) except (PermissionError, OSError): print("output_file is opened elsewhere, close and try again")
问题解答
1. 更优的源文件与目标文件同一性检测方法
你当前用绝对路径字符串对比的方法是可行的,但可以通过文件系统的inode信息做更可靠的判断——因为不同路径可能指向同一个文件(比如硬链接),字符串对比会漏掉这种情况。
可以用os.stat()获取文件的设备ID和inode编号,只有当两者都相同时,才说明是同一个文件:
def is_same_file(src_path, dst_path): src_stat = os.stat(src_path) dst_stat = os.stat(dst_path) # 设备ID和inode编号都相同则为同一文件 return src_stat.st_dev == dst_stat.st_dev and src_stat.st_ino == dst_stat.st_ino # 在write_reprojected中替换原检测逻辑 src_path = os.path.abspath(src.name) # 用src.name替代src.files[0]更简洁 dst_path = os.path.abspath(fname) try: if is_same_file(src_path, dst_path): print("Trying to write file to its old location") # 这里可以抛出自定义异常或直接返回终止流程 return except FileNotFoundError: # 目标文件还不存在,无需处理 pass
2. 处理被其他程序占用的文件
在Windows系统中,当文件被其他程序锁定时,无法直接删除或覆盖——这是系统级的文件锁定机制,没有合法的绕过方法。可行的替代方案有:
- 保持当前的异常提示逻辑:这是最稳妥的方式,直接提示用户关闭占用程序后重试。
- 写入临时文件,待原文件释放后替换:先将重投影结果写入临时文件,若目标文件被占用,可提示用户临时文件位置,或轮询等待文件解锁后自动替换。示例逻辑:
import tempfile def write_reprojected(src, crs, transform, width, height, fname): # ... 省略其他逻辑 ... # 生成临时文件路径 tmp_path = tempfile.mktemp(suffix='.tif') try: # 先写入临时文件 with rasterio.open(tmp_path, 'w', **kwargs) as dst: for i in range(1, src.count + 1): reproject( source=rasterio.band(src, i), destination=rasterio.band(dst, i), src_transform=src.transform, src_crs=src.crs, dst_transform=transform, dst_crs=crs, resampling=Resampling.bilinear) # 尝试替换目标文件 try: os.replace(tmp_path, fname) except (PermissionError, OSError): print(f"目标文件被占用,临时文件已保存至: {tmp_path}") except Exception as e: # 出错时清理临时文件 if os.path.exists(tmp_path): os.remove(tmp_path) raise e
- 第三方工具强制解锁:比如Windows下的
Unlocker,但需要额外依赖,且可能带来系统风险,不推荐在代码中集成。
内容的提问来源于stack exchange,提问作者konstanze
相关产品推荐
相关产品推荐

