Python自动化处理气象数据为DSSAT/WOFOST/APSIM生成标准输入文件气象数据是作物生长模型的核心输入之一。无论是DSSAT、WOFOST还是APSIM都需要特定格式的气象文件才能正常运行。传统的手工处理方式不仅耗时耗力还容易出错。本文将展示如何用Python构建一个自动化流水线将原始气象数据高效转换为三种模型所需的格式。1. 准备工作与环境配置在开始之前我们需要确保环境配置正确。推荐使用Python 3.8版本并安装以下关键库pip install pandas numpy xarray netCDF4 openpyxl这些库将帮助我们处理各种格式的气象数据Pandas数据清洗与转换的核心工具NumPy数值计算支持xarray netCDF4处理NetCDF格式气象数据openpyxl读写Excel文件提示如果处理大量数据考虑使用Dask来提升性能特别是对于NetCDF文件。准备一个项目目录结构如下weather_pipeline/ ├── input/ # 存放原始气象数据 ├── output/ # 生成的标准格式文件 │ ├── dssat/ │ ├── wofost/ │ └── apsim/ └── scripts/ # Python脚本2. 原始数据读取与预处理气象数据可能来自多种来源和格式我们需要一个统一的读取接口。以下是处理CSV和NetCDF的示例代码def read_weather_data(filepath): 通用气象数据读取函数 if filepath.endswith(.csv): df pd.read_csv(filepath, parse_dates[date]) elif filepath.endswith(.nc): ds xr.open_dataset(filepath) df ds.to_dataframe().reset_index() else: raise ValueError(Unsupported file format) # 确保包含必要字段 required_cols [date, srad, tmax, tmin, rain] assert all(col in df.columns for col in required_cols) return df常见的数据质量问题及处理方法问题类型检测方法处理方案缺失值df.isnull().sum()线性插值或使用气候平均值单位不一致数值范围检查统一转换为标准单位时间不连续pd.date_range对比重采样或插值补全对于每小时数据需要降采样为日数据的情况def resample_to_daily(hourly_df): 将小时数据降采样为日数据 daily_df hourly_df.resample(D, ondate).agg({ srad: sum, tmax: max, tmin: min, rain: sum }) return daily_df.reset_index()3. DSSAT .WTH文件生成DSSAT的气象文件格式最为复杂需要特别注意以下几点文件名规范AAAA BB CC .WTH的8字符格式文件头信息包含站点元数据和气候参数数据列格式严格的列宽和精度要求3.1 关键参数计算DSSAT需要两个特殊气候参数TAV年平均温度和AMP温度年较差。计算代码如下def calculate_climate_params(df): 计算TAV和AMP参数 # 计算年平均温度 tav (df[tmax].mean() df[tmin].mean()) / 2 # 计算温度年较差AMP monthly_avg df.groupby(pd.Grouper(keydate, freqM)).mean() monthly_avg[tavg] (monthly_avg[tmax] monthly_avg[tmin]) / 2 amp monthly_avg[tavg].max() - monthly_avg[tavg].min() return tav, amp3.2 文件生成实现完整的.WTH文件生成函数def generate_dssat_file(df, site_info, output_dir): 生成DSSAT格式气象文件 # 计算气候参数 tav, amp calculate_climate_params(df) # 准备文件头 header ( f*WEATHER DATA : {site_info[code]}\n f INSI LAT LONG ELEV TAV AMP REFHT WNDHT\n f {site_info[code]} {site_info[lat]} {site_info[lon]} f{site_info[elev]} {tav:.1f} {amp:.1f} -99 -99\n DATE SRAD TMAX TMIN RAIN\n ) # 准备数据行 df[doy] df[date].dt.dayofyear df[year] df[date].dt.year % 100 # 取年份后两位 df[date_code] df[year] * 1000 df[doy] lines [] for _, row in df.iterrows(): line ( f{row[date_code]:5d} {row[srad]:5.1f} f{row[tmax]:5.1f} {row[tmin]:5.1f} {row[rain]:5.1f}\n ) lines.append(line) # 写入文件 filename f{site_info[code]}{site_info[year]}01.WTH with open(os.path.join(output_dir, filename), w) as f: f.write(header) f.writelines(lines) return filename4. WOFOST .XXX文件生成WOFOST的气象文件格式相对简单但需要注意单位转换辐射量从MJ/m²/d转换为KJ/m²/d乘以1000文件命名AAB .XXX格式XXX为3位年份代码关键生成代码def generate_wofost_file(df, site_info, output_dir): 生成WOFOST格式气象文件 # 单位转换 df[srad_kj] df[srad] * 1000 # MJ→KJ转换 # 准备文件内容 lines [ f{site_info[name]}\n, year day srad tmin tmax rain\n ] df[doy] df[date].dt.dayofyear for _, row in df.iterrows(): line ( f{row[date].year} {row[doy]} f{row[srad_kj]:.1f} {row[tmin]:.1f} f{row[tmax]:.1f} {row[rain]:.1f}\n ) lines.append(line) # 写入文件 year_code f{site_info[year] % 1000:03d} filename f{site_info[name]}{site_info[id]}.{year_code} with open(os.path.join(output_dir, filename), w) as f: f.writelines(lines) return filename5. APSIM .met文件生成APSIM提供了Excel模板来生成.met文件但我们可以用Python直接生成def generate_apsim_file(df, site_info, output_dir): 生成APSIM格式气象文件 # 准备数据框 apsim_df pd.DataFrame({ year: df[date].dt.year, day: df[date].dt.dayofyear, radn: df[srad], # MJ/m²/d maxt: df[tmax], # °C mint: df[tmin], # °C rain: df[rain], # mm vp: -99, # 缺省值 code: site_info[code] }) # 写入CSV临时文件 temp_file os.path.join(output_dir, temp.csv) apsim_df.to_csv(temp_file, indexFalse, float_format%.1f) # 重命名为.met final_file os.path.join(output_dir, f{site_info[code]}.met) os.rename(temp_file, final_file) return final_file6. 自动化流水线整合将所有功能整合为一个完整的处理流程def process_weather_pipeline(input_file, site_info, output_base_dir): 完整的自动化处理流水线 # 读取并预处理数据 df read_weather_data(input_file) # 创建输出目录 os.makedirs(output_base_dir, exist_okTrue) dssat_dir os.path.join(output_base_dir, dssat) wofost_dir os.path.join(output_base_dir, wofost) apsim_dir os.path.join(output_base_dir, apsim) os.makedirs(dssat_dir, exist_okTrue) os.makedirs(wofost_dir, exist_okTrue) os.makedirs(apsim_dir, exist_okTrue) # 生成各模型文件 dssat_file generate_dssat_file(df, site_info, dssat_dir) wofost_file generate_wofost_file(df, site_info, wofost_dir) apsim_file generate_apsim_file(df, site_info, apsim_dir) return { dssat: dssat_file, wofost: wofost_file, apsim: apsim_file }使用示例site_info { code: TEST, name: TestSite, id: 1, lat: 35.5, lon: 120.3, elev: 50, year: 21 # 2021年的后两位 } result process_weather_pipeline( input/weather_data.csv, site_info, output )7. 高级技巧与优化建议批量处理多个站点使用多进程加速处理from multiprocessing import Pool为每个站点创建独立的日志文件数据验证开发验证脚本检查生成文件的格式正确性与模型自带的示例文件进行对比异常处理添加对极端值的检测和警告实现自动重试机制处理临时文件错误性能优化对于大型数据集使用Pandas的chunksize参数考虑使用PyArrow加速CSV读写# 使用PyArrow加速CSV读写示例 def read_large_csv(filepath): return pd.read_csv(filepath, enginepyarrow)扩展性设计使用配置文件管理站点信息和格式规范支持插件式架构方便添加新模型格式# 插件式架构示例 class WeatherFormatConverter: def __init__(self): self.converters {} def register_format(self, name, converter_func): self.converters[name] converter_func def convert(self, format_name, df, site_info): return self.converters[format_name](df, site_info) # 使用示例 converter WeatherFormatConverter() converter.register_format(dssat, generate_dssat_file) converter.register_format(wofost, generate_wofost_file)在实际项目中这套自动化流程将气象数据处理时间从原来的数小时缩短到几分钟同时显著降低了人为错误的风险。一个常见的优化是将这个流水线封装为命令行工具或Web服务方便团队其他成员使用。