一、单个文件替换
cat replace_SingleFile.py import netCDF4 as nc import numpy as np file1 = '/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_202505/WRFOUT_T/wrfout_d01_2025-04-25_00:00:00' file2 = '/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_202404/WRFOUT/wrfout_d01_2024-03-26_00:00:00' ds1 = nc.Dataset(file1, 'r+') ds2 = nc.Dataset(file2, 'r') print(ds1.variables.keys()) print("file1 T2 shape:", ds1.variables['T2'].shape) print("file2 T2 shape:", ds2.variables['T2'].shape) print("file1 Times[0]:", b"".join(ds1.variables['Times'][0]).decode("ascii")) print("file2 Times[0]:", b"".join(ds2.variables['Times'][0]).decode("ascii")) if ds1.variables['T2'].shape != ds2.variables['T2'].shape: raise ValueError("T2 维度不一致，不能替换。") # 替换 src = ds2.variables['T2'][:].astype(np.float64) ds1.variables['T2'][:] = src ds1.sync() # 验证（读回 file1 的 T2 做差分） dst = ds1.variables['T2'][:].astype(np.float64) max_abs = np.max(np.abs(dst - src)) print("max_abs_diff(after write):", max_abs) ds1.close() ds2.close() if max_abs == 0.0: print("结论：替换成功（T2 完全一致）") else: print("结论：替换后仍存在差异（可能是精度/压缩/缺测导致）")
二、多个文件替换
cat replace_MultipleFiles.py import os import re import glob import netCDF4 as nc import numpy as np VAR = "T2" year_DST = 2025 year_SRC = 2024 month = 9 mm = f"{month:02d}" DST_DIR = f"/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_{year_DST}{month:02d}/WRFOUT_T" SRC_DIR = f"/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_{year_SRC}{month:02d}/WRFOUT" # 目标文件名：wrfout_d01_2025-04-25_00:00:00 PAT = re.compile(rf"^(wrfout_d01_){year_DST}-(\d{{2}}-\d{{2}}_\d{{2}}:\d{{2}}:\d{{2}})$") # 验证阈值：严格相等用 0.0；如担心浮点/压缩差异，可改 1e-6 VERIFY_TOL = 0.0 def replace_and_verify(dst_path, src_path): ds_dst = nc.Dataset(dst_path, "r+") ds_src = nc.Dataset(src_path, "r") # 基本检查 if VAR not in ds_dst.variables: ds_dst.close(); ds_src.close() return False, f"dst缺少{VAR}" if VAR not in ds_src.variables: ds_dst.close(); ds_src.close() return False, f"src缺少{VAR}" sh_dst = ds_dst.variables[VAR].shape sh_src = ds_src.variables[VAR].shape if sh_dst != sh_src: ds_dst.close(); ds_src.close() return False, f"{VAR}维度不一致 dst={sh_dst} src={sh_src}" # 读取源数据（float64 用于验证更稳） src_data = ds_src.variables[VAR][:].astype(np.float64) # 写入目标 ds_dst.variables[VAR][:] = src_data ds_dst.sync() # 读回目标用于验证 dst_data = ds_dst.variables[VAR][:].astype(np.float64) diff = dst_data - src_data absdiff = np.abs(diff) if np.all(np.isnan(absdiff)): max_abs = mean_abs = float("nan") else: max_abs = float(np.nanmax(absdiff)) mean_abs = float(np.nanmean(absdiff)) ds_dst.close() ds_src.close() ok = (max_abs <= VERIFY_TOL) msg = f" {VAR} verify max_abs={max_abs:.6g}, mean_abs={mean_abs:.6g}, tol={VERIFY_TOL}" return ok, msg def main(): dst_files = sorted(glob.glob(os.path.join(DST_DIR, f"wrfout_d01_{year_DST}-*"))) if not dst_files: print(f"未找到目标文件：{DST_DIR}/wrfout_d01_{year_DST}-*") return ok_cnt = skip_cnt = fail_cnt = 0 for idx, dst_path in enumerate(dst_files, start=1): dst_name = os.path.basename(dst_path) m = PAT.match(dst_name) if not m: print(f"[{idx}/{len(dst_files)}][SKIP] 文件名不符合预期格式：{dst_name}") skip_cnt += 1 continue prefix, rest = m.group(1), m.group(2) src_name = f"{prefix}{year_SRC}-{rest}" src_path = os.path.join(SRC_DIR, src_name) if not os.path.exists(src_path): print(f"[{idx}/{len(dst_files)}][SKIP] 源文件不存在：{src_name} （对应目标 {dst_name}）") skip_cnt += 1 continue # 关键提示：正在处理哪个文件，用哪个文件替换 print(f"\n[{idx}/{len(dst_files)}] 正在处理：{dst_name}") print(f" 将 {dst_name} 的 {VAR} 替换为 {src_name} 的 {VAR}") try: ok, info = replace_and_verify(dst_path, src_path) if ok: print(f" [OK] 完成替换并验证通过：{info}") ok_cnt += 1 else: print(f" [FAIL] 已执行替换，但验证未通过：{info}") fail_cnt += 1 except Exception as e: print(f" [ERROR] 处理失败：{repr(e)}") fail_cnt += 1 print("\n====== 汇总 ======") print(f"成功: {ok_cnt}") print(f"跳过: {skip_cnt}") print(f"失败: {fail_cnt}") if __name__ == "__main__": main()
三、检查替换结果
cat check_replace.py import netCDF4 as nc import numpy as np file1 = "/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_202505/WRFOUT_T/wrfout_d01_2025-04-25_00:00:00" file2 = "/public/home/acez11b6ht/apprepo/wrf_wps/4.5-intelmpi2017/app/WRF/test/em_real_202405/WRFOUT/wrfout_d01_2024-04-25_00:00:00" d1 = nc.Dataset(file1, "r") d2 = nc.Dataset(file2, "r") t1 = d1.variables["T2"][:].astype(np.float64) t2 = d2.variables["T2"][:].astype(np.float64) diff = t1 - t2 max_abs = np.nanmax(np.abs(diff)) mean_abs = np.nanmean(np.abs(diff)) print("T2 shape file1:", t1.shape) print("T2 shape file2:", t2.shape) print("max_abs_diff:", max_abs) print("mean_abs_diff:", mean_abs) # 再给一个严格判定 if max_abs == 0.0: print("结论：替换成功（T2 完全一致）") else: print("结论：不完全一致（请检查是否替换的是同一时间序列/是否有缺测或精度差异）") d1.close() d2.close()
