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', on='date').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(key='date', freq='M')).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, index=False, 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_ok=True) 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_ok=True) os.makedirs(wofost_dir, exist_ok=True) os.makedirs(apsim_dir, exist_ok=True) # 生成各模型文件 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读写
- 对于大型数据集，使用Pandas的
# 使用PyArrow加速CSV读写示例 def read_large_csv(filepath): return pd.read_csv(filepath, engine='pyarrow')- 扩展性设计：
- 使用配置文件管理站点信息和格式规范
- 支持插件式架构，方便添加新模型格式
# 插件式架构示例 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服务，方便团队其他成员使用。
